From f238ae14553396e91f0e9f6bace9a69977ece4c1 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Tue, 23 Jun 2026 15:42:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(dispatcher):=20=E7=BC=96=E6=8E=92=E5=BC=8F?= =?UTF-8?q?=E5=A4=9A=E6=99=BA=E8=83=BD=E4=BD=93=E6=8E=A5=E5=8A=9B=20?= =?UTF-8?q?=E2=80=94=E2=80=94=20=E4=B8=8A=E6=B8=B8=20agent=20=E4=BA=A7?= =?UTF-8?q?=E5=87=BA=E6=B2=BF=E5=9B=BE=E4=BC=A0=E7=BB=99=E4=B8=8B=E6=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 让画布上串联的多个 agent 真正协作完成一件事（如 检索→撰写→审查 出报告）， 而非各自对原 query 重答： - board 增 agentOut（各上游 agent 产出按序）；RunCtx 增 Upstream，buildMessages 注入"前序协作 agent 的产出，请在此基础上继续"。 - runAgent / runReactAgent / runComposeConversation 三条路径统一：注入 b.agentOut 作上下文，产出经 recordAgentOutput 入黑板（append agentOut + 设为当前成稿 answer， 多 agent 时最后一个 agent 产出即最终成品）。 不需要 eino multiagent 组件——编排式协作由现有图引擎 + 上下文传递实现。 测试 TestAgentCollaborationPassesOutput：下游 agent 确见上游产出、成稿=下游产出； make test-go 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../internal/eino/agent_collab_test.go | 46 +++++++++++++++++++ sundynix-dispatcher/internal/eino/compile.go | 5 ++ .../internal/eino/compose_graph.go | 18 +++++--- sundynix-dispatcher/internal/eino/graph.go | 30 ++++++++---- .../internal/eino/react_agent.go | 14 ++++-- 5 files changed, 93 insertions(+), 20 deletions(-) create mode 100644 sundynix-dispatcher/internal/eino/agent_collab_test.go diff --git a/sundynix-dispatcher/internal/eino/agent_collab_test.go b/sundynix-dispatcher/internal/eino/agent_collab_test.go new file mode 100644 index 0000000..b479bd5 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/agent_collab_test.go @@ -0,0 +1,46 @@ +package eino + +import ( + "context" + "strings" + "testing" + + "github.com/sundynix/sundynix-dispatcher/internal/harness" + "github.com/sundynix/sundynix-dispatcher/internal/llm" + "github.com/sundynix/sundynix-shared/contract" +) + +// TestAgentCollaborationPassesOutput 验证编排式多智能体接力： +// agent1 的产出注入 agent2 的上下文，agent2 基于它继续；最终成稿 = 下游 agent 的产出。 +func TestAgentCollaborationPassesOutput(t *testing.T) { + graph := `{"version":"1","nodes":[ + {"id":"a1","kind":"agent","config":{"system":"研究"}}, + {"id":"a2","kind":"agent","config":{"system":"撰写"}} + ],"edges":[{"source":"a1","target":"a2"}]}` + + saw := false + ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { + var sys string + for _, x := range m { + if x.Role == "system" { + sys = x.Content + } + } + if strings.Contains(sys, "研究产出XYZ") { // agent1 的产出出现在 agent2 的 system → 接力成功 + saw = true + return "最终报告：基于上游研究撰写" + } + return "研究产出XYZ" + }} + o := &Orchestrator{pool: ll, breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}} + ans, err := o.runGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{}) + if err != nil { + t.Fatal(err) + } + if !saw { + t.Fatal("下游 agent 未看到上游 agent 的产出——协作没接力") + } + if !strings.Contains(ans, "最终报告") { + t.Fatalf("最终成稿应为下游 agent 产出，得 %q", ans) + } +} diff --git a/sundynix-dispatcher/internal/eino/compile.go b/sundynix-dispatcher/internal/eino/compile.go index 787d624..82527d2 100644 --- a/sundynix-dispatcher/internal/eino/compile.go +++ b/sundynix-dispatcher/internal/eino/compile.go @@ -19,6 +19,7 @@ type RunCtx struct { Profile string // 召回的画像 History []*schema.Message // 短期历史 ToolOut []string // 工具/检索节点产出（含参考资料） + Upstream []string // 前序协作 agent 的产出（多 agent 接力时注入，让下游基于上游继续） } // chatTemplate 是会话消息模板：系统提示词 + 历史占位 + 用户输入。 @@ -43,6 +44,10 @@ func buildMessages(ctx context.Context, rc *RunCtx) ([]*schema.Message, error) { sys.WriteString("\n\n以下是工具/检索得到的参考资料：\n") sys.WriteString(strings.Join(rc.ToolOut, "\n---\n")) } + if len(rc.Upstream) > 0 { + sys.WriteString("\n\n以下是前序协作 agent 的产出，请在此基础上继续完成你的部分（不要重头再来）：\n") + sys.WriteString(strings.Join(rc.Upstream, "\n---\n")) + } return chatTemplate.Format(ctx, map[string]any{ "system": sys.String(), "history": rc.History, diff --git a/sundynix-dispatcher/internal/eino/compose_graph.go b/sundynix-dispatcher/internal/eino/compose_graph.go index 7b14d40..e49d318 100644 --- a/sundynix-dispatcher/internal/eino/compose_graph.go +++ b/sundynix-dispatcher/internal/eino/compose_graph.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "os" + "strings" "time" "github.com/cloudwego/eino/compose" @@ -54,11 +55,12 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string rc := &RunCtx{ UserID: b.uid, SessionID: b.sid, - System: firstNonEmpty(system, defaultAgentSystem), - Query: b.query, - Profile: b.profile, - History: b.history, - ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + System: firstNonEmpty(system, defaultAgentSystem), + Query: b.query, + Profile: b.profile, + History: b.history, + ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力 } msgs, _ := buildMessages(ctx, rc) @@ -72,6 +74,7 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string defer sr.Close() chunks := 0 + var produced strings.Builder // 本节点产出（供下游 agent 接力） for { chunk, rerr := sr.Recv() if rerr == io.EOF { @@ -86,8 +89,9 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string } safe, _ := harness.RedactSecrets(chunk.Content) // 输出护栏：逐片脱敏 _ = o.sink.PublishToken(taskID, []byte(safe)) - b.answer += safe + produced.WriteString(safe) chunks++ } - tr.info(node, "system", "compose 图", fmt.Sprintf("%d 段输出 / %d 字（Eino compose 运行时）", chunks, len([]rune(b.answer)))) + o.recordAgentOutput(b, produced.String()) + tr.info(node, "system", "compose 图", fmt.Sprintf("%d 段输出 / %d 字（Eino compose 运行时）", chunks, len([]rune(produced.String())))) } diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index aa52d51..5152bf0 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -27,7 +27,8 @@ type board struct { refs []string // 检索 / 聚合得到的参考资料 toolOut []string // 工具节点产出 sections []reportSection // map 并行 fan-out 产出的分项成稿（供 render 多章渲染） - answer string // 终端 agent / map 的成稿（流式累计） + answer string // 当前成稿（多 agent 协作时 = 最近一个 agent 的产出 = 成品） + agentOut []string // 各上游 agent 的产出（按序），注入下游 agent 上下文以实现接力协作 } // runGraph 按 DSL 图的真实拓扑与连线执行（替代旧的线性拍平 compileFlow）。 @@ -230,16 +231,18 @@ func (o *Orchestrator) execToolNode(ctx context.Context, taskID string, n dsl.No // runAgent 执行 agent/模型节点：据黑板拼消息 → 流式回流 token → 累计成稿。 func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) { rc := &RunCtx{ - System: firstNonEmpty(system, defaultAgentSystem), - Query: b.query, - Profile: b.profile, - History: b.history, - ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + System: firstNonEmpty(system, defaultAgentSystem), + Query: b.query, + Profile: b.profile, + History: b.history, + ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力 } msgs, _ := buildMessages(ctx, rc) tr.emit(node, "model", "start", "模型流式推理", "", 0) t0 := time.Now() n, redacted := 0, 0 + var produced strings.Builder // 本节点自身产出（用于沿图向下游传递） send := func(s string) { if s == "" { return @@ -248,7 +251,7 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy safe, hit := harness.RedactSecrets(s) redacted += hit _ = o.sink.PublishToken(taskID, []byte(safe)) - b.answer += safe + produced.WriteString(safe) n++ } var err error @@ -264,8 +267,19 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy if redacted > 0 { tr.info(node, "system", "输出护栏", fmt.Sprintf("已脱敏 %d 处疑似密钥/令牌", redacted)) } + o.recordAgentOutput(b, produced.String()) // 产出入黑板：成当前成稿 + 供下游接力 tr.emit(node, "model", "end", "模型流式推理", - fmt.Sprintf("%d tokens / %d 字", n, len([]rune(b.answer))), time.Since(t0).Milliseconds()) + fmt.Sprintf("%d tokens / %d 字", n, len([]rune(produced.String()))), time.Since(t0).Milliseconds()) +} + +// recordAgentOutput 把一个 agent 节点的产出记入黑板：append 到 agentOut（供下游 agent 注入接力）， +// 并设为当前成稿 answer（多 agent 协作时，最后一个 agent 的产出即最终成品）。空产出忽略。 +func (o *Orchestrator) recordAgentOutput(b *board, out string) { + if strings.TrimSpace(out) == "" { + return + } + b.agentOut = append(b.agentOut, out) + b.answer = out } // renderNode 执行渲染节点：把当前成稿渲染成 Word（经 mcp-go report_render）。 diff --git a/sundynix-dispatcher/internal/eino/react_agent.go b/sundynix-dispatcher/internal/eino/react_agent.go index 8b2d733..52a87eb 100644 --- a/sundynix-dispatcher/internal/eino/react_agent.go +++ b/sundynix-dispatcher/internal/eino/react_agent.go @@ -6,6 +6,7 @@ import ( "fmt" "io" "log" + "strings" "time" "github.com/cloudwego/eino/components/tool" @@ -195,9 +196,10 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar System: firstNonEmpty(system, defaultAgentSystem), Query: b.query, // 自主 agent 不预注入画像：让它经 recall_user_memory 工具按需自取（否则模型直接答、不调工具）。 - Profile: "", - History: b.history, - ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + Profile: "", + History: b.history, + ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + Upstream: append([]string{}, b.agentOut...), // 前序协作 agent 产出 → 接力 } msgs, _ := buildMessages(ctx, rc) @@ -211,6 +213,7 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar defer sr.Close() chunks := 0 + var produced strings.Builder // 本节点产出（供下游 agent 接力） for { chunk, rerr := sr.Recv() if rerr == io.EOF { @@ -225,9 +228,10 @@ func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *boar } safe, _ := harness.RedactSecrets(chunk.Content) _ = o.sink.PublishToken(taskID, []byte(safe)) - b.answer += safe + produced.WriteString(safe) chunks++ } + o.recordAgentOutput(b, produced.String()) tr.emit(node, "model", "end", "ReAct 智能体", - fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(b.answer))), time.Since(t0).Milliseconds()) + fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(produced.String()))), time.Since(t0).Milliseconds()) }