From 66caeef35c55e2951fc4d9f5179cc960acd78195 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 29 Jun 2026 15:37:23 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E5=A4=9A=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E4=BD=93=E5=8D=8F=E5=90=8C=20v1=20=E2=80=94=E2=80=94=20coordin?= =?UTF-8?q?ator=20=E8=8A=82=E7=82=B9(Eino=C3=97Anthropic=20=E8=9E=8D?= =?UTF-8?q?=E5=90=88)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LLM 自主在 agent 间路由/委派:orchestrator=ReAct(Eino 出机器),编排认知按 Anthropic orchestrator-worker 配方(出脑子)。专家=包成工具的子 agent(agent-as-tool),lead 给 每个专家写定制简报(brief)后并行派发、综合。方案见 MULTI_AGENT.md。 为什么 agent-as-tool 而非 Eino host:host 的 specialist 拿原始输入(preHandler return state.msgs),传不了 lead 写的定制简报,而定制简报正是 Anthropic 多智能体的 精髓。agent-as-tool 让 orchestrator 自己 emit 工具调用、参数 brief 即简报。 = OpenAI agent.as_tool() / Anthropic 研究系统的 orchestrator-worker。 - coordinator.go: specialistTool(react.Agent/ChatModel 包成 InvokableTool,入参 brief, 精炼返回) + parseSpecialists/buildSpecialists(带工具→react,不带→ChatModel,MCP 工具 按 spec.tools 过滤) + runCoordinator(lead 提示词=Anthropic 配方) + leadOrchestratorPrompt。 - 双路接入 execDSLNode(compose)+ runGraph(graph.go)的 case coordinator。 - 护栏:禁套娃(专家是内联叶子)/ MaxStep / 专家 I/O 计入共享 Budget / 降级(无 ToolCallingModel 或 0 专家 → runAgent)。 - streamAgentReply:抽出 runReactAgent 与 runCoordinator 共用的流式回流尾段。 - 复用即得:evaluator-optimizer=harness 低分纠偏;成本天花板=预算护栏;上下文隔离= 专家独立 react.Agent;观测=每次派发落 agent 轨迹。 测试:parseSpecialists / agent-as-tool 包装(brief 透传+精炼返回+失败作观察) / 降级。 live 验证(真 deepseek):两专家**并行派发**、lead 给各自写**不同定制简报**、最终 **综合**(非拼接)成稿,评测 1.00。 Co-Authored-By: Claude Opus 4.8 --- MULTI_AGENT.md | 77 ++++++ .../internal/eino/compose_compiler.go | 2 + .../internal/eino/coordinator.go | 247 ++++++++++++++++++ .../internal/eino/coordinator_test.go | 91 +++++++ sundynix-dispatcher/internal/eino/graph.go | 2 + .../internal/eino/react_agent.go | 9 +- 6 files changed, 426 insertions(+), 2 deletions(-) create mode 100644 MULTI_AGENT.md create mode 100644 sundynix-dispatcher/internal/eino/coordinator.go create mode 100644 sundynix-dispatcher/internal/eino/coordinator_test.go diff --git a/MULTI_AGENT.md b/MULTI_AGENT.md new file mode 100644 index 0000000..c00d682 --- /dev/null +++ b/MULTI_AGENT.md @@ -0,0 +1,77 @@ +# 多智能体方案（Eino × Anthropic 融合） + +> 现状：单 agent 自主调工具（ReAct）已落地；多 agent 静态接力（DSL 图）已落地。 +> 缺的是 **LLM 自主在 agent 之间路由/委派** —— 即"多智能体协同"。本方案是它的落地设计。 + +## 设计原则：Eino 出机器，Anthropic 出脑子 + +| 层 | 谁的强项 | 怎么用 | +|---|---|---| +| 运行时机器 | **Eino** | `react.Agent` + ToolsNode（并行工具调用）+ 流式 + 预算/熔断 + callbacks 观测 + checkpoint | +| 编排认知 | **Anthropic** | lead 的"分解→定额→写定制简报→综合"全在**提示词**；worker 返回**精炼结论**而非原始堆料 | +| 闭环纪律 | 两者皆有现成 | evaluator-optimizer = 复用 harness **低分自动纠偏**；成本天花板 = 复用**预算护栏** | + +**关键判断：用 agent-as-tool，不用 `host.NewMultiAgent`。** +Eino 的 host 模式里，specialist 拿到的是**原始输入**（preHandler `return state.msgs`），host 的工具参数只是路由理由、不传给专家当任务。而 Anthropic 多智能体的精髓正是 **lead 给每个 worker 写定制简报**。host 表达不了，故用 **agent-as-tool**：orchestrator 自己 emit 工具调用，参数 `brief` 就是它写的简报。host 留作"无简报简化版"备选。 + +> 业界对照：本方案 = OpenAI Agents SDK 的 `agent.as_tool()` + handoff 思路，= Anthropic 研究系统的 orchestrator-worker。心智模型与头部项目通用、可迁移。 + +## v1 架构 + +``` +coordinator 节点 + └─ orchestrator = react.Agent（lead 提示词 = Anthropic 配方） + ├─ tool: specialist_A (= 包成工具的子 react.Agent / ChatModel) + ├─ tool: specialist_B + └─ (可选) MCP 工具 + 流程：lead 分解任务 → 给每个专家写 brief → 并行派发(并行 tool-call) + → 收齐精炼结论 → lead 在最后一轮综合成稿 +``` + +- **并行**：orchestrator 一轮 emit 多个 tool-call → Eino ToolsNode 并发跑专家（Anthropic 的并行 breadth）。 +- **综合**：ReAct 天然在最后一轮综合，不需要单独 summarizer。 +- **上下文隔离**：每个专家是独立 react.Agent、独立消息上下文（原生）。 +- **定制简报**：专家工具入参 `brief` = lead 写给它的子任务/期望输出/边界。 +- **精炼返回**：专家系统提示词追加"只回结论要点，不堆原文"，省 orchestrator 综合时的上下文。 + +## 与现有 harness 的融合点（零/少新代码） + +- **evaluator-optimizer**：协调者综合稿照常进 harness 评测 + 低分自动纠偏 —— 已有。 +- **成本天花板**：协调者 + 专家 I/O 共享 `harness.Budget`，触顶中止 —— 已有，专家 I/O 在 agent-as-tool 边界计入。 +- **观测**：每次派发落一条 `kind=agent` exec 事件（brief + 精炼结论）+ OTel 嵌套 span。 + +## 落地增量 + +### 增量 1 —— 后端核心（dispatcher/internal/eino/coordinator.go）✅ 本次 +- `specialistTool`：react.Agent / ChatModel 包成 InvokableTool，入参 `brief`，精炼返回。 +- `parseSpecialists` / `buildSpecialists`：从节点 config 的 `agents[]` 建专家工具集（带工具→react；不带→ChatModel；MCP 工具按 `spec.tools` 过滤）。 +- `runCoordinator`：orchestrator = react.Agent（lead 提示词），双路接入 `execDSLNode`（compose）+ `runGraph`（graph.go）的 `case "coordinator"`。 +- 护栏：禁套娃（专家是内联叶子，不能再是 coordinator）/ MaxStep / 共享 Budget / 降级（无 ToolCallingModel 或 0 专家 → 退 `runAgent`）。 +- `streamAgentReply`：抽出 runReactAgent 与 runCoordinator 共用的流式回流尾段。 + +### 增量 2 —— 前端（desktop Studio） +- nodeCatalog 加 `coordinator` 节点 + `agentList` 字段（子智能体卡片：名称/用途/系统提示词/工具多选）。 +- 运行轨迹：专家调用渲染成可展开子节点（显示 brief + 精炼结论）。 + +### 增量 3（可选，按场景）—— handoff / 可中断多智能体 +- 真需要"专家间多轮协商/控制权移交"才上：迁 `adk`（`SetSubAgents` + `Transfer`），复用 HITL 的 checkpoint 做可中断恢复。不推翻 v1。 + +## 开放决策 + +| 决策 | v1 取舍 | +|---|---| +| 专家定义 | 内联（写在协调者节点）；注册表（跨节点复用）留 v1.5 | +| Summarizer | 不单设，ReAct orchestrator 末轮自综合 | +| 专家带工具 | 支持但可选（无 tools = 纯对话专家，有 tools = react 子 agent） | + +## DSL 形态 + +```json +{ "id":"coord", "kind":"coordinator", "config":{ + "system":"任务背景（可选）", + "agents":[ + {"name":"legal", "use":"法律条款/合规问题", "system":"你是法律专家…", "tools":["wiki_search"]}, + {"name":"finance", "use":"财务/数字测算", "system":"你是财务专家…"} + ] +}} +``` diff --git a/sundynix-dispatcher/internal/eino/compose_compiler.go b/sundynix-dispatcher/internal/eino/compose_compiler.go index 86f2e50..14103b7 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler.go @@ -292,6 +292,8 @@ func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl. } else { o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID) } + case "coordinator": // 多智能体协调：orchestrator 自主把子任务派给专家(agent-as-tool)再综合 + o.runCoordinator(ctx, t.ID, b, firstNonEmpty(cstr(n.Config, "system"), plan.System), n, tr, "coordinator:"+n.ID) case "aggregate": merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...)) b.refs, b.toolOut = merged, nil diff --git a/sundynix-dispatcher/internal/eino/coordinator.go b/sundynix-dispatcher/internal/eino/coordinator.go new file mode 100644 index 0000000..92710ea --- /dev/null +++ b/sundynix-dispatcher/internal/eino/coordinator.go @@ -0,0 +1,247 @@ +package eino + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/cloudwego/eino/components/tool" + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/flow/agent/react" + "github.com/cloudwego/eino/schema" + + "github.com/sundynix/sundynix-dispatcher/internal/dsl" + "github.com/sundynix/sundynix-dispatcher/internal/harness" +) + +// 多智能体协同（Eino × Anthropic 融合，详见仓库 MULTI_AGENT.md）： +// orchestrator = react.Agent（机器=Eino），编排认知按 Anthropic orchestrator-worker 配方； +// 专家 = 包成工具的子 agent（agent-as-tool），lead 给每个专家写定制简报（brief）后并行派发、综合。 + +// leadOrchestratorPrompt 是协调者(lead)的编排提示词 —— Anthropic 配方：先分解、按复杂度定额（别过度派发）、 +// 给每个专家写明确简报、并行派发、收齐后综合消解冲突，不直接拼接专家原文。 +const leadOrchestratorPrompt = `你是多智能体协调者(lead)。你手下有若干专家，每个专家是你可以调用的一个工具。 +工作方式： +1. 先想清楚任务能否拆分、需要哪几个专家、各花多少功夫——简单任务用 1 个专家甚至自己直接答，复杂任务才并行派多个，不要过度派发。 +2. 调用专家时，在 brief 参数里给它写一份明确的简报：要它完成的子任务、期望的输出形式、边界与约束；不要把原问题原样丢给它。 +3. 可以在一轮里并行调用多个专家。收齐各专家的结论后，由你综合成一段连贯、完整的最终答复，消解专家之间的冲突，必要时点明哪个结论来自哪个专家。 +不要把专家的原始返回直接拼接给用户——要消化、综合。` + +// specialistCondensedSuffix 追加到每个专家的系统提示词：要求精炼返回，省 orchestrator 综合时的上下文。 +const specialistCondensedSuffix = "\n\n返回要求：只给结论与关键依据要点，简洁，不要堆砌原始材料或冗长背景。" + +// specialistSpec 是协调者节点配置里的一个子智能体(专家)定义。 +type specialistSpec struct { + Name string + Use string // 用途/擅长，作为工具描述供 lead 判断何时调 + System string // 专家人设 + Tools []string // 该专家可用的 MCP 工具名子集（空=纯对话专家） +} + +// parseSpecialists 从协调者节点 config 的 agents 列表解析出专家定义。 +func parseSpecialists(cfg map[string]any) []specialistSpec { + raw, _ := cfg["agents"].([]any) + var out []specialistSpec + for _, item := range raw { + m, ok := item.(map[string]any) + if !ok { + continue + } + s := specialistSpec{Name: cstr(m, "name"), Use: cstr(m, "use"), System: cstr(m, "system")} + if s.Name == "" { + continue + } + if tools, ok := m["tools"].([]any); ok { + for _, t := range tools { + if ts, ok := t.(string); ok && strings.TrimSpace(ts) != "" { + s.Tools = append(s.Tools, strings.TrimSpace(ts)) + } + } + } + out = append(out, s) + } + return out +} + +// specialistTool 把一个专家(react.Agent 或 ChatModel)包成 Eino InvokableTool（agent-as-tool）： +// 入参 brief = lead 写给它的定制简报；返回专家的精炼结论。每次派发落一条 agent 轨迹。 +type specialistTool struct { + info *schema.ToolInfo + name string + run func(ctx context.Context, brief string) (string, error) + tr *execTracer +} + +func (s *specialistTool) Info(context.Context) (*schema.ToolInfo, error) { return s.info, nil } + +// InvokableRun 跑一次专家派发：取 lead 写的 brief → 执行子 agent → 返回精炼结论给 orchestrator。 +// 专家失败作为"观察"返回（不中断协调）；专家 I/O 计入共享预算（Anthropic 的成本克制由 Budget 兜底）。 +func (s *specialistTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.Option) (string, error) { + var args struct { + Brief string `json:"brief"` + } + if argsJSON != "" { + _ = json.Unmarshal([]byte(argsJSON), &args) + } + brief := strings.TrimSpace(args.Brief) + if brief == "" { + brief = "（协调者未给简报，请按你的职责处理当前任务）" + } + if bud := harness.BudgetFrom(ctx); bud != nil { + bud.AddPrompt(brief) + } + end := s.tr.span("agent:"+s.name, "agent", "派发专家 "+s.name) + out, err := s.run(ctx, brief) + if err != nil { + end("专家执行失败", err) + return "专家 " + s.name + " 执行失败：" + err.Error(), nil + } + if bud := harness.BudgetFrom(ctx); bud != nil { + bud.AddComplete(out) + } + end("简报 "+truncate(brief, 100)+" → "+truncate(out, 160), nil) + return out, nil +} + +// buildSpecialists 据配置建专家工具集。带工具→react.Agent(过滤后的 MCP 工具子集)；不带→纯 ChatModel。 +// 无可用模型的专家跳过。专家是内联叶子（非图节点），天然不能再是 coordinator → 杜绝套娃递归。 +func (o *Orchestrator) buildSpecialists(ctx context.Context, specs []specialistSpec, b *board, taskID string, tr *execTracer) []tool.BaseTool { + allTools := o.agentTools(b, taskID, tr) // 全量 MCP 工具，下面按 spec.Tools 过滤 + byName := map[string]tool.BaseTool{} + for _, t := range allTools { + if info, err := t.Info(ctx); err == nil { + byName[info.Name] = t + } + } + var out []tool.BaseTool + for _, spec := range specs { + sys := firstNonEmpty(spec.System, defaultAgentSystem) + specialistCondensedSuffix + run := o.specialistRunner(ctx, spec, sys, byName) + if run == nil { + tr.info("coordinator", "system", "专家跳过", "无可用模型："+spec.Name) + continue + } + out = append(out, &specialistTool{ + name: spec.Name, tr: tr, run: run, + info: &schema.ToolInfo{ + Name: spec.Name, + Desc: firstNonEmpty(spec.Use, "专家 "+spec.Name), + ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{ + "brief": {Type: schema.String, Desc: "给该专家的明确简报：子任务、期望输出、边界约束", Required: true}, + }), + }, + }) + } + return out +} + +// specialistRunner 构造专家执行闭包：带工具且模型支持函数调用→react.Agent.Generate；否则→ChatModel.Generate。 +// 无可用模型返回 nil（该专家被跳过）。 +func (o *Orchestrator) specialistRunner(ctx context.Context, spec specialistSpec, sys string, byName map[string]tool.BaseTool) func(context.Context, string) (string, error) { + var tools []tool.BaseTool + for _, tn := range spec.Tools { + if t, ok := byName[tn]; ok { + tools = append(tools, t) + } + } + if len(tools) > 0 { + if tcm := o.pool.ToolCallingModel(); tcm != nil { + ag, err := react.NewAgent(ctx, &react.AgentConfig{ + ToolCallingModel: tcm, + ToolsConfig: compose.ToolsNodeConfig{Tools: tools}, + MaxStep: reactMaxStep(), + StreamToolCallChecker: streamHasToolCall, + }) + if err == nil { + return func(c context.Context, brief string) (string, error) { + msg, gerr := ag.Generate(c, []*schema.Message{schema.SystemMessage(sys), schema.UserMessage(brief)}) + if gerr != nil { + return "", gerr + } + return msg.Content, nil + } + } + } + // 工具型专家但模型不支持函数调用 → 退纯对话（下方） + } + cm := o.pool.ChatModel() + if cm == nil { + return nil + } + return func(c context.Context, brief string) (string, error) { + msg, gerr := cm.Generate(c, []*schema.Message{schema.SystemMessage(sys), schema.UserMessage(brief)}) + if gerr != nil { + return "", gerr + } + return msg.Content, nil + } +} + +// runCoordinator 执行多智能体协调节点：orchestrator = ReAct（专家=agent-as-tool），lead 提示词按 +// Anthropic 配方（分解→定制简报→并行派发→综合）。无 ToolCallingModel / 0 可用专家 → 降级单 agent。 +func (o *Orchestrator) runCoordinator(ctx context.Context, taskID string, b *board, system string, n dsl.Node, tr *execTracer, node string) { + specs := parseSpecialists(n.Config) + tcm := o.pool.ToolCallingModel() + if tcm == nil || len(specs) == 0 { + tr.info(node, "system", "协调者降级", "模型不支持函数调用或未配置专家，退回普通对话") + o.runAgent(ctx, taskID, b, system, tr, node) + return + } + specialists := o.buildSpecialists(ctx, specs, b, taskID, tr) + if len(specialists) == 0 { + tr.info(node, "system", "协调者降级", "无可用专家，退回普通对话") + o.runAgent(ctx, taskID, b, system, tr, node) + return + } + + ag, err := react.NewAgent(ctx, &react.AgentConfig{ + ToolCallingModel: tcm, + ToolsConfig: compose.ToolsNodeConfig{Tools: specialists}, + MaxStep: reactMaxStep(), + StreamToolCallChecker: streamHasToolCall, + }) + if err != nil { + tr.emit(node, "model", "error", "构建协调者", err.Error(), 0) + o.runAgent(ctx, taskID, b, system, tr, node) + return + } + + // lead 提示词 = Anthropic 编排配方 + 节点自定义任务背景。 + leadSys := leadOrchestratorPrompt + if s := strings.TrimSpace(system); s != "" && s != defaultAgentSystem { + leadSys += "\n\n任务背景：" + s + } + rc := &RunCtx{ + UserID: b.uid, SessionID: b.sid, + System: leadSys, + Query: b.query, + History: b.history, + ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力 + } + msgs, _ := buildMessages(ctx, rc) + if bud := harness.BudgetFrom(ctx); bud != nil { + for _, m := range msgs { + bud.AddPrompt(m.Content) + } + if bud.Exceeded() { + if b.fatalErr == nil { + b.fatalErr = errBudget + } + tr.emit(node, "system", "error", "token 预算", "已达单任务预算上限，中止", 0) + return + } + } + + tr.emit(node, "model", "start", "多智能体协调", fmt.Sprintf("%d 个专家可派发", len(specialists)), 0) + t0 := time.Now() + sr, err := ag.Stream(ctx, msgs) + if err != nil { + tr.emit(node, "model", "error", "多智能体协调", err.Error(), time.Since(t0).Milliseconds()) + return + } + defer sr.Close() + o.streamAgentReply(ctx, taskID, b, sr, tr, node, "多智能体协调", t0) +} diff --git a/sundynix-dispatcher/internal/eino/coordinator_test.go b/sundynix-dispatcher/internal/eino/coordinator_test.go new file mode 100644 index 0000000..e3aa664 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/coordinator_test.go @@ -0,0 +1,91 @@ +package eino + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/cloudwego/eino/schema" + + "github.com/sundynix/sundynix-dispatcher/internal/dsl" + "github.com/sundynix/sundynix-dispatcher/internal/harness" + "github.com/sundynix/sundynix-dispatcher/internal/llm" +) + +// TestParseSpecialists 钉死节点 config 的 agents 列表解析（含 tools 子集、无名跳过）。 +func TestParseSpecialists(t *testing.T) { + cfg := map[string]any{"agents": []any{ + map[string]any{"name": "legal", "use": "法律条款", "system": "你是律师", "tools": []any{"wiki_search", "kb_search"}}, + map[string]any{"name": "finance", "use": "财务测算"}, + map[string]any{"use": "无名字应被跳过"}, + }} + specs := parseSpecialists(cfg) + if len(specs) != 2 { + t.Fatalf("应解析出 2 个专家（无名跳过），got %d", len(specs)) + } + if specs[0].Name != "legal" || specs[0].Use != "法律条款" || len(specs[0].Tools) != 2 { + t.Fatalf("legal 解析有误: %+v", specs[0]) + } + if specs[1].Name != "finance" || len(specs[1].Tools) != 0 { + t.Fatalf("finance 解析有误: %+v", specs[1]) + } +} + +// TestSpecialistToolInvoke 钉死 agent-as-tool 包装：lead 写的 brief 透传给专家、结论回传、 +// 空 brief 兜底、专家失败作为"观察"返回而非中断协调。 +func TestSpecialistToolInvoke(t *testing.T) { + var gotBrief string + st := &specialistTool{ + name: "legal", tr: &execTracer{}, + info: &schema.ToolInfo{Name: "legal"}, + run: func(_ context.Context, brief string) (string, error) { gotBrief = brief; return "结论：条款合规", nil }, + } + + out, err := st.InvokableRun(context.Background(), `{"brief":"审查合同第3条违约金"}`) + if err != nil { + t.Fatal(err) + } + if gotBrief != "审查合同第3条违约金" { + t.Fatalf("lead 简报应透传给专家，got %q", gotBrief) + } + if out != "结论：条款合规" { + t.Fatalf("专家结论应回传，got %q", out) + } + + // 空 brief → 兜底文案（不把空串丢给专家）。 + st.run = func(_ context.Context, brief string) (string, error) { gotBrief = brief; return "ok", nil } + _, _ = st.InvokableRun(context.Background(), `{}`) + if !strings.Contains(gotBrief, "未给简报") { + t.Fatalf("空 brief 应兜底，got %q", gotBrief) + } + + // 专家失败 → 返回失败观察，err 为 nil（不中断 orchestrator 的 ReAct 循环）。 + st.run = func(_ context.Context, _ string) (string, error) { return "", fmt.Errorf("boom") } + out, err = st.InvokableRun(context.Background(), `{"brief":"x"}`) + if err != nil { + t.Fatalf("专家失败不应上抛 error（应作观察返回）: %v", err) + } + if !strings.Contains(out, "执行失败") { + t.Fatalf("应返回失败观察，got %q", out) + } +} + +// TestRunCoordinatorDegrade 钉死降级：模型不支持函数调用（ToolCallingModel=nil）→ 退回普通对话出稿， +// 不因"多智能体"而卡死或空答。 +func TestRunCoordinatorDegrade(t *testing.T) { + fs := &fakeSink{} + o := &Orchestrator{ + pool: &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "降级直答" }}, + breaker: harness.NewCircuitBreaker(), + sink: fs, + } + b := &board{query: "你好"} + n := dsl.Node{ID: "c", Kind: "coordinator", Config: map[string]any{ + "agents": []any{map[string]any{"name": "x", "use": "y"}}, + }} + o.runCoordinator(context.Background(), "t_co", b, "", n, &execTracer{}, "coordinator:c") + if !strings.Contains(b.answer, "降级直答") { + t.Fatalf("ToolCallingModel 缺失应降级 runAgent 出答案，got %q", b.answer) + } +} diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index bc553c7..2d606b3 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -138,6 +138,8 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT } else { o.runConversation(nctx, t.ID, b, sys, tr, "agent:"+n.ID) } + case "coordinator": // 多智能体协调：orchestrator 自主把子任务派给专家(agent-as-tool)再综合 + o.runCoordinator(nctx, t.ID, b, firstNonEmpty(cstr(n.Config, "system"), plan.System), n, tr, "coordinator:"+n.ID) case "aggregate": merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...)) b.refs, b.toolOut = merged, nil diff --git a/sundynix-dispatcher/internal/eino/react_agent.go b/sundynix-dispatcher/internal/eino/react_agent.go index 6a950c3..d92bffe 100644 --- a/sundynix-dispatcher/internal/eino/react_agent.go +++ b/sundynix-dispatcher/internal/eino/react_agent.go @@ -225,7 +225,12 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar return } defer sr.Close() + o.streamAgentReply(ctx, taskID, b, sr, tr, node, "ReAct 智能体", t0) +} +// streamAgentReply 把一个 agent / 协调者的流式输出回流到 sink：跨分片脱敏 + 计输出 token + +// 落产出(供下游接力) + 轨迹收尾。runReactAgent 与 runCoordinator 共用，杜绝两路尾段漂移。 +func (o *Orchestrator) streamAgentReply(ctx context.Context, taskID string, b *board, sr *schema.StreamReader[*schema.Message], tr *execTracer, node, label string, t0 time.Time) { chunks := 0 var produced strings.Builder // 本节点产出（供下游 agent 接力） red := harness.NewStreamRedactor() // 输出护栏：跨分片脱敏，杜绝密钥被切断而漏检 @@ -243,7 +248,7 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar break } if rerr != nil { - tr.emit(node, "model", "error", "ReAct 智能体", rerr.Error(), time.Since(t0).Milliseconds()) + tr.emit(node, "model", "error", label, rerr.Error(), time.Since(t0).Milliseconds()) return } if chunk.Content == "" { @@ -256,6 +261,6 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar bud.AddComplete(produced.String()) // 成本护栏：计入输出 token } o.recordAgentOutput(b, produced.String()) - tr.emit(node, "model", "end", "ReAct 智能体", + tr.emit(node, "model", "end", label, fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(produced.String()))), time.Since(t0).Milliseconds()) }