From ef6f525a740cc9eea5e0eb24333a340caa0bcae9 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Tue, 30 Jun 2026 08:53:02 +0800 Subject: [PATCH] =?UTF-8?q?refactor(dispatcher):=20T1.1=20=E9=80=80?= =?UTF-8?q?=E5=BD=B9=20graph.go=20=E2=80=94=E2=80=94=20=E7=BC=96=E6=8E=92?= =?UTF-8?q?=E5=BC=95=E6=93=8E=E6=94=B6=E6=88=90=E5=8D=95=E4=B8=80=20compos?= =?UTF-8?q?e?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit compose 已默认数天、HITL/多智能体/评测全在其上真实跑过，soak 充分。删掉自研拓扑 解释器这第二套引擎，消灭"双实现 drift"税： - 删 runGraph（graph.go 的自研解释器）+ composeEnabled/EINO_COMPOSE 逃生舱开关 + runConversation（仅 runGraph 用的死代码）。 - executeGraph 直接走 runComposeGraph；compose 编译失败兜底改单轮对话（不再回退 graph.go）；清掉仅 runGraph 用的 import（otel attribute/trace/otelx）。 - 保留 board / 各节点执行器(retriever/tool/agent/branch/approval/map/render/aggregate) / 工具函数 —— 它们是 compose 各节点 lambda 复用的，非 graph.go 专属。 - 测试：8 处 runGraph→runComposeGraph；等价测试(对照两引擎)转为 compose 正确性测试； runConversation 的开关测试转为「无 ChatModel 降级 runAgent」。 go test ./... 全绿 + vet 干净；冒烟 简单 agent/分支图 跑通。此后每个编排改动不再两边 对齐，成本减半。DEPTH_ROADMAP T1.1 ✅。 Co-Authored-By: Claude Opus 4.8 --- DEPTH_ROADMAP.md | 15 +- .../internal/eino/agent_collab_test.go | 2 +- .../internal/eino/compose_compiler.go | 15 +- .../internal/eino/compose_compiler_test.go | 35 ++-- .../internal/eino/compose_graph.go | 20 +-- .../internal/eino/compose_graph_test.go | 23 ++- sundynix-dispatcher/internal/eino/graph.go | 155 +----------------- .../internal/eino/integration_test.go | 8 +- .../internal/eino/orchestrator.go | 2 +- .../internal/eino/rag_integration_test.go | 4 +- 10 files changed, 54 insertions(+), 225 deletions(-) diff --git a/DEPTH_ROADMAP.md b/DEPTH_ROADMAP.md index bf4b7cc..a8733ee 100644 --- a/DEPTH_ROADMAP.md +++ b/DEPTH_ROADMAP.md @@ -39,12 +39,15 @@ ## 🥈 Tier 1 — 收口 / 去税 -### [ ] T1.1 退役 graph.go — 消灭双引擎 +### [x] T1.1 退役 graph.go — 消灭双引擎 ✅ 现状：compose 已默认，HITL/多智能体只在 compose 上；graph.go 是逃生舱。 -- [ ] compose soak 一段真实流量（无回归） -- [ ] 删 graph.go（runGraph 及其专属分支）+ `EINO_COMPOSE` 开关 + 降级回退 -- [ ] 把仅 graph.go 用到的辅助函数收编/删除，等价测试转为只测 compose -- 验收：单引擎，全测试绿，HITL/多智能体/分支/map 全在 compose 上。 +- [x] compose soak（默认跑数天，HITL/多智能体/评测全在其上真实运行） +- [x] 删 `runGraph`（自研拓扑解释器）+ `composeEnabled`/`EINO_COMPOSE` 开关 + `runConversation` 死代码 +- [x] compose 编译失败兜底改单轮对话（不再回退 graph.go）；清掉仅 runGraph 用的 import(otel) +- [x] 保留 board/各节点执行器/工具函数（compose 各 lambda 复用）；8 处测试 runGraph→runComposeGraph， + 等价测试转为 compose 正确性测试，runConversation 测试转为「无 ChatModel 降级 runAgent」 +- 验收 ✅：单引擎，go test ./... 全绿 + vet 干净；冒烟 简单 agent/分支图 跑通；HITL/多智能体/branch/map + 全在 compose 上。 ### [ ] T1.2 前端深度补齐 - [ ] 评测质量面板（运行页展示 overall/质量/忠实/纠偏/flags，驱动 T0.1 的可见性） @@ -92,7 +95,7 @@ | Tier | 完成 / 总 | |---|---| | T0 激活 | 2 / 2 ✅ | -| T1 收口 | 0 / 2 | +| T1 收口 | 1 / 2 | | T2 深化 | 0 / 3 | | T3 硬化 ⏸ | 0 / 5 | diff --git a/sundynix-dispatcher/internal/eino/agent_collab_test.go b/sundynix-dispatcher/internal/eino/agent_collab_test.go index 5497643..0fdbfd2 100644 --- a/sundynix-dispatcher/internal/eino/agent_collab_test.go +++ b/sundynix-dispatcher/internal/eino/agent_collab_test.go @@ -33,7 +33,7 @@ func TestAgentCollaborationPassesOutput(t *testing.T) { 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{}) + ans, _, err := o.runComposeGraph(context.Background(), &contract.Task{ID: "tc", Graph: []byte(graph)}, &execTracer{}) if err != nil { t.Fatal(err) } diff --git a/sundynix-dispatcher/internal/eino/compose_compiler.go b/sundynix-dispatcher/internal/eino/compose_compiler.go index 14103b7..a923188 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler.go @@ -26,13 +26,10 @@ func registerFlowMerge() { }) } -// executeGraph 按灰度开关选编排实现：compose.Graph（Phase C）或自研 graph.go（默认/权威）。 -// 返回 (成稿, 检索来源, error)；来源供忠实度评测，两条路径都回传（compose 已对齐 graph.go）。 +// executeGraph 把任务 DSL 图编译为 Eino compose.Graph 并执行（唯一编排引擎；自研解释器 graph.go +// 的 runGraph 已退役）。返回 (成稿, 检索来源, error)；来源供忠实度评测。 func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) { - if composeEnabled() { - return o.runComposeGraph(ctx, t, tr) - } - return o.runGraph(ctx, t, tr) + return o.runComposeGraph(ctx, t, tr) } // runComposeGraph 把 DSL 图编译为 Eino compose.Graph 并执行（Phase C 编排归一）： @@ -212,8 +209,10 @@ func (o *Orchestrator) execComposeGraph(ctx context.Context, t *contract.Task, t } r, cerr := g.Compile(ctx, compileOpts...) if cerr != nil { - tr.info("task", "system", "compose 编译失败", "退回自研 graph.go："+cerr.Error()) - return o.runGraph(ctx, t, tr) // 降级回权威实现（带 refs / 终态） + // 编译失败（罕见）→ 降级为单轮对话兜底（自研 graph.go 已退役，不再回退）。 + tr.info("task", "system", "compose 编译失败", "降级单轮对话："+cerr.Error()) + o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") + return b.answer, refsOf(b), b.fatalErr } // checkpoint id = task id：审批中断时 compose 据此把整图状态（含 board）落进 store。 var invokeOpts []compose.Option diff --git a/sundynix-dispatcher/internal/eino/compose_compiler_test.go b/sundynix-dispatcher/internal/eino/compose_compiler_test.go index 9a5e22a..667694a 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler_test.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler_test.go @@ -28,21 +28,15 @@ func echoLLM() *fakeLLM { } } -func runBoth(t *testing.T, graph string) (interp, comp string) { +// runCompose 跑一个图并返回成稿（自研 graph.go 已退役，编排引擎统一为 compose）。 +func runCompose(t *testing.T, graph string) string { t.Helper() - task := &contract.Task{ID: "t_eq", Graph: []byte(graph)} - - o1 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}} - a1, _, err := o1.runGraph(context.Background(), task, &execTracer{}) - if err != nil { - t.Fatalf("runGraph: %v", err) - } - o2 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}} - a2, _, err := o2.runComposeGraph(context.Background(), task, &execTracer{}) + o := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}} + ans, _, err := o.runComposeGraph(context.Background(), &contract.Task{ID: "t_eq", Graph: []byte(graph)}, &execTracer{}) if err != nil { t.Fatalf("runComposeGraph: %v", err) } - return a1, a2 + return ans } // rejectWaiter 是恒拒绝的 HITL 审批替身（fail-safe 路径回归用）。 @@ -257,8 +251,8 @@ func TestComposeReturnsRefs(t *testing.T) { } } -// TestComposeEquivalentLinear 多节点线性图：input→memory→agent→output，两路径成稿应逐字一致。 -func TestComposeEquivalentLinear(t *testing.T) { +// TestComposeLinear 多节点线性图：input→memory→agent→output，compose 应跑通并产出成稿。 +func TestComposeLinear(t *testing.T) { graph := `{"version":"1","nodes":[ {"id":"in","kind":"input","config":{"text":"什么是图编排"}}, {"id":"m","kind":"memory","config":{}}, @@ -267,14 +261,14 @@ func TestComposeEquivalentLinear(t *testing.T) { ],"edges":[ {"source":"in","target":"m"},{"source":"m","target":"a"},{"source":"a","target":"out"} ]}` - interp, comp := runBoth(t, graph) - if interp == "" || interp != comp { - t.Fatalf("线性图不等价: interp=%q compose=%q", interp, comp) + if ans := runCompose(t, graph); !strings.Contains(ans, "ANS:") { + t.Fatalf("线性图 compose 应产出成稿，got %q", ans) } } -// TestComposeEquivalentBranch 分支图：input→branch→(真)A/(假)B，条件恒真应都走 A，两路径一致。 -func TestComposeEquivalentBranch(t *testing.T) { +// TestComposeBranch 分支图：input→branch→(真)A/(假)B，条件恒真走 A，compose 应跑通产出。 +// 分支真假边的精确选路另由 TestRunGraph_BranchRouting（compose）钉死。 +func TestComposeBranch(t *testing.T) { graph := `{"version":"1","nodes":[ {"id":"in","kind":"input","config":{"text":"hi"}}, {"id":"br","kind":"branch","config":{"condition":""}}, @@ -285,8 +279,7 @@ func TestComposeEquivalentBranch(t *testing.T) { {"source":"br","target":"a","sourceHandle":"true"}, {"source":"br","target":"b","sourceHandle":"false"} ]}` - interp, comp := runBoth(t, graph) - if interp == "" || interp != comp { - t.Fatalf("分支图不等价: interp=%q compose=%q", interp, comp) + if ans := runCompose(t, graph); !strings.Contains(ans, "ANS:") { + t.Fatalf("分支图 compose 应产出成稿，got %q", ans) } } diff --git a/sundynix-dispatcher/internal/eino/compose_graph.go b/sundynix-dispatcher/internal/eino/compose_graph.go index 4190145..9c33331 100644 --- a/sundynix-dispatcher/internal/eino/compose_graph.go +++ b/sundynix-dispatcher/internal/eino/compose_graph.go @@ -4,7 +4,6 @@ import ( "context" "fmt" "io" - "os" "strings" "time" @@ -14,28 +13,13 @@ import ( "github.com/sundynix/sundynix-dispatcher/internal/harness" ) -// composeEnabled 报告是否启用 compose.Graph 编排路径。 -// Phase C 灰度收尾：compose 已与 graph.go 终态对齐（refs / 审批拒绝 / 预算·模型失败 全部传播）， -// 默认翻为 compose 运行时；保留逃生舱 EINO_COMPOSE=0 → 回退权威 graph.go。soak 无回归后即可退役 graph.go。 -func composeEnabled() bool { return os.Getenv("EINO_COMPOSE") != "0" } - -// runConversation 是对话/模型节点的统一入口：按灰度开关选 compose.Graph 或自研 runAgent。 -// 二者对外行为一致（据黑板拼消息 → 流式回流 token → 累计成稿），便于等价回归。 -func (o *Orchestrator) runConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) { - if composeEnabled() { - o.runComposeConversation(ctx, taskID, b, system, tr, node) - return - } - o.runAgent(ctx, taskID, b, system, tr, node) -} - // runComposeConversation 用 Eino compose.Graph 跑对话主流程： // START → ChatModel 节点 → END，编译为 Runnable 后流式执行；可观测经 callbacks 桥到 ExecEvent。 -// 模型未就绪 / 编译失败时降级回自研 runAgent，保证不回归。 +// 模型未就绪 / 编译失败时降级回 runAgent（同样的流式回流，保证不回归）。 func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string, b *board, system string, tr *execTracer, node string) { cm := o.pool.ChatModel() if cm == nil { - o.runAgent(ctx, taskID, b, system, tr, node) // 无模型 → 走自研路径的降级桩 + o.runAgent(ctx, taskID, b, system, tr, node) // 无模型 → 降级桩 return } diff --git a/sundynix-dispatcher/internal/eino/compose_graph_test.go b/sundynix-dispatcher/internal/eino/compose_graph_test.go index f5c5c6a..e4313cb 100644 --- a/sundynix-dispatcher/internal/eino/compose_graph_test.go +++ b/sundynix-dispatcher/internal/eino/compose_graph_test.go @@ -50,10 +50,8 @@ func TestComposeGraphRuns(t *testing.T) { } } -// TestComposeConversationEquivalent 验证 EINO_COMPOSE=1 时对话主流程走 compose 路径， -// 把模型输出流式回流到 sink 并累计成稿——与自研 runAgent 行为等价（都逐片转发模型输出）。 -func TestComposeConversationEquivalent(t *testing.T) { - t.Setenv("EINO_COMPOSE", "1") +// TestComposeConversation 验证对话主流程走 compose 路径：模型输出流式回流到 sink 并累计成稿。 +func TestComposeConversation(t *testing.T) { fs := &fakeSink{} o := &Orchestrator{ pool: &fakeLLM{ready: true, cm: &stubModel{reply: "我是 compose 路径的回答"}}, @@ -61,8 +59,7 @@ func TestComposeConversationEquivalent(t *testing.T) { sink: fs, } b := &board{query: "你好"} - tr := &execTracer{} // sink 为 nil → 轨迹发射空操作 - o.runConversation(context.Background(), "task_compose", b, "", tr, "agent") + o.runComposeConversation(context.Background(), "task_compose", b, "", &execTracer{}, "agent") if !strings.Contains(b.answer, "compose 路径") { t.Fatalf("成稿未含模型输出: %q", b.answer) @@ -72,18 +69,18 @@ func TestComposeConversationEquivalent(t *testing.T) { } } -// TestComposeDisabledUsesRunAgent 验证开关关闭时仍走自研 runAgent（默认行为不变）。 -func TestComposeDisabledUsesRunAgent(t *testing.T) { - t.Setenv("EINO_COMPOSE", "0") +// TestComposeConversationDegradesToRunAgent 验证无 Eino ChatModel 时降级回 runAgent（流式回流不丢）。 +func TestComposeConversationDegradesToRunAgent(t *testing.T) { fs := &fakeSink{} o := &Orchestrator{ - pool: &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "自研路径回答" }}, + // cm 为 nil → ChatModel() 返回 nil → runComposeConversation 降级 runAgent（走 stream 桩）。 + pool: &fakeLLM{ready: true, stream: func([]llm.ChatMessage) string { return "降级路径回答" }}, breaker: harness.NewCircuitBreaker(), sink: fs, } b := &board{query: "你好"} - o.runConversation(context.Background(), "task_legacy", b, "", &execTracer{}, "agent") - if !strings.Contains(b.answer, "自研路径") { - t.Fatalf("未走自研路径: %q", b.answer) + o.runComposeConversation(context.Background(), "task_degrade", b, "", &execTracer{}, "agent") + if !strings.Contains(b.answer, "降级路径") { + t.Fatalf("无 ChatModel 应降级 runAgent 出稿: %q", b.answer) } } diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 2d606b3..8185918 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -8,13 +8,10 @@ import ( "time" "github.com/cloudwego/eino/schema" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/trace" "github.com/sundynix/sundynix-dispatcher/internal/dsl" "github.com/sundynix/sundynix-dispatcher/internal/harness" "github.com/sundynix/sundynix-shared/contract" - "github.com/sundynix/sundynix-shared/otelx" ) // defaultAgentSystem 是 agent 节点未填系统提示词时的兜底。 @@ -32,156 +29,12 @@ type board struct { sections []reportSection // map 并行 fan-out 产出的分项成稿（供 render 多章渲染） answer string // 当前成稿（多 agent 协作时 = 最近一个 agent 的产出 = 成品） agentOut []string // 各上游 agent 的产出（按序），注入下游 agent 上下文以实现接力协作 - rejected bool // HITL 审批节点拒绝/超时 → 置位，runGraph 中止并返回 errRejected - fatalErr error // agent 节点 LLM 调用失败 → 置位，runGraph 中止并上抛 → 任务判 failed（而非 done-空） + rejected bool // HITL 审批节点拒绝/超时 → 置位，执行中止并返回 errRejected + fatalErr error // agent 节点 LLM 调用失败 → 置位，执行中止并上抛 → 任务判 failed（而非 done-空） } -// runGraph 按 DSL 图的真实拓扑与连线执行（替代旧的线性拍平 compileFlow）。 -// -// 入度0 入口 → 沿连线传播 active → 每个节点按 kind 执行真实行为 → -// branch 按条件只激活选中的下游（剪枝）→ agent 节点流式回流 token。 -// -// 逐节点点亮"运行·观测"。返回终端 agent 的完整产出（供写回历史）。 -func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) { - flow, ferr := dsl.Parse(t.Graph) - plan := dsl.Compile(t.Graph) - b := &board{ - uid: meta(t, contract.MetaUserID), - sid: meta(t, contract.MetaSessionID), - query: plan.Query, - } - - // 无法解析或空图：退化为"无图单轮对话"（注入默认记忆 + 直接出模型）。 - if ferr != nil || flow == nil || len(flow.Nodes) == 0 { - tr.info("task", "system", "无结构化图", "按单轮对话执行") - b.profile = o.fetchMemory(ctx, b.uid, b.query) - b.history = o.fetchHistory(ctx, b.sid) - o.runConversation(ctx, t.ID, b, plan.System, tr, "agent") - return b.answer, refsOf(b), b.fatalErr // 模型失败 → 上抛判 failed - } - - // 建邻接与入度（只认两端都存在的边）。保留整条边以便 branch 按 true/false 标签选路。 - nodeByID := make(map[string]dsl.Node, len(flow.Nodes)) - outE := make(map[string][]dsl.Edge) - indeg := make(map[string]int, len(flow.Nodes)) - for _, n := range flow.Nodes { - nodeByID[n.ID] = n - indeg[n.ID] = 0 - } - for _, e := range flow.Edges { - if _, ok := nodeByID[e.Source]; !ok { - continue - } - if _, ok := nodeByID[e.Target]; !ok { - continue - } - outE[e.Source] = append(outE[e.Source], e) - indeg[e.Target]++ - } - - // 入口节点（入度 0）置 active；执行时沿连线把下游激活，branch 只激活选中分支。 - active := make(map[string]bool) - for _, n := range flow.Nodes { - if indeg[n.ID] == 0 { - active[n.ID] = true - } - } - - // 图里没有 memory 节点 → 沿用旧默认：注入画像+历史（避免回归）。 - hasMemory := false - for _, n := range flow.Nodes { - if n.Kind == "memory" { - hasMemory = true - break - } - } - if !hasMemory { - b.profile = o.fetchMemory(ctx, b.uid, b.query) - b.history = o.fetchHistory(ctx, b.sid) - } - - for _, n := range flow.Topo() { - if !active[n.ID] { - continue // 被 branch 剪掉的下游，不执行 - } - // 每个节点一个 span：下游的工具调用 / LLM span 会自动挂到本节点之下， - // Jaeger 里即呈现「节点 → 工具 → embedding/milvus」的层级。 - nctx, nspan := otelx.Tracer().Start(ctx, "node."+n.Kind, - trace.WithAttributes( - attribute.String("sundynix.node.id", n.ID), - attribute.String("sundynix.node.kind", n.Kind), - )) - propagate := targetsOf(outE[n.ID]) // 默认激活全部出边；branch 会改写 - switch n.Kind { - case "input": - if txt := cstr(n.Config, "text"); txt != "" { - b.query = txt - } - tr.info("input:"+n.ID, "system", labelOf(n, "输入"), truncate(b.query, 80)) - case "memory": - if cbool(n.Config, "profile") { - b.profile = o.fetchMemory(nctx, b.uid, b.query) - } - if cbool(n.Config, "history") { - b.history = o.fetchHistory(nctx, b.sid) - } - tr.info("memory:"+n.ID, "memory", labelOf(n, "记忆"), - fmt.Sprintf("画像 %d 字 · 历史 %d 条", len([]rune(b.profile)), len(b.history))) - case "retriever": - o.retrieverNode(nctx, n, b, tr) - case "tool": - o.execToolNode(nctx, t.ID, n, b, tr) - case "agent": - sys := firstNonEmpty(cstr(n.Config, "system"), plan.System) - if cbool(n.Config, "autonomous") { // 开启自主工具 → ReAct（模型自己选工具） - o.runReactAgent(nctx, t.ID, b, sys, n, tr, "agent:"+n.ID) - } 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 - tr.info("aggregate:"+n.ID, "system", labelOf(n, "汇聚"), "策略："+firstNonEmpty(cstr(n.Config, "strategy"), "拼接")) - case "render": - o.renderNode(nctx, t.ID, n, b, tr) - case "branch": - propagate = o.branchNode(n, b, outE[n.ID], nodeByID, tr) - case "approval": - propagate = o.approvalNode(nctx, t.ID, n, b, tr, outE[n.ID]) - case "map": - o.mapNode(nctx, t.ID, n, b, tr) - case "output": - tr.info("output:"+n.ID, "system", labelOf(n, "输出"), "目标："+firstNonEmpty(cstr(n.Config, "target"), "屏幕")) - default: - tr.info(n.Kind+":"+n.ID, "system", labelOf(n, n.Kind), "未识别节点，跳过") - } - nspan.End() - if b.rejected || b.fatalErr != nil { - break // 审批拒绝/超时 或 模型失败：中止后续节点 - } - for _, tgt := range propagate { - active[tgt] = true - } - } - - if b.rejected { - return b.answer, nil, errRejected // 合法终态，Handle 据此判 rejected 并优雅收尾 - } - if b.fatalErr != nil { - return b.answer, nil, b.fatalErr // 上抛 → Handle 判 failed（带原因） - } - - // 图里无 agent 节点（纯工具/检索图）也要出一段模型答复，否则没有输出。 - if b.answer == "" { - o.runConversation(ctx, t.ID, b, plan.System, tr, "agent") - } - if b.fatalErr != nil { - return b.answer, nil, b.fatalErr - } - return b.answer, refsOf(b), nil // 成功：带回检索来源供忠实度评测 -} +// 注：本文件原有自研图解释器 runGraph 已退役（编排引擎统一为 compose，见 compose_compiler.go +// 的 execComposeGraph）。以下是被 compose 各节点 lambda 复用的节点执行器与黑板辅助函数。 // refsOf 汇总本次执行的检索来源（检索资料 + 工具产出），供忠实度评测。 func refsOf(b *board) []string { diff --git a/sundynix-dispatcher/internal/eino/integration_test.go b/sundynix-dispatcher/internal/eino/integration_test.go index 47ace69..c6c2db5 100644 --- a/sundynix-dispatcher/internal/eino/integration_test.go +++ b/sundynix-dispatcher/internal/eino/integration_test.go @@ -134,7 +134,7 @@ func TestRunGraph_BranchRouting(t *testing.T) { ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { return m[0].Content }} run := func(cond string) string { o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{}) - ans, _, err := o.runGraph(context.Background(), task(strings.Replace(g, "%s", cond, 1)), o.tracer("t1")) + ans, _, err := o.runComposeGraph(context.Background(), task(strings.Replace(g, "%s", cond, 1)), o.tracer("t1")) if err != nil { t.Fatal(err) } @@ -160,7 +160,7 @@ func TestRunGraph_ToolFeedsAgent(t *testing.T) { return &contract.ToolResult{OK: true, Content: "TOOLDATA"} }} o := newOrch(ll, ft, &fakeSink{}, &fakeExec{}) - ans, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1")) + ans, _, err := o.runComposeGraph(context.Background(), task(g), o.tracer("t1")) if err != nil { t.Fatal(err) } @@ -186,7 +186,7 @@ func TestRunGraph_MapFanout(t *testing.T) { return "正文XYZ", nil }} o := newOrch(ll, &fakeTools{}, &fakeSink{}, &fakeExec{}) - ans, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1")) + ans, _, err := o.runComposeGraph(context.Background(), task(g), o.tracer("t1")) if err != nil { t.Fatal(err) } @@ -203,7 +203,7 @@ func TestRunGraph_OutputRedaction(t *testing.T) { }} fs := &fakeSink{} o := newOrch(ll, &fakeTools{}, fs, &fakeExec{}) - if _, _, err := o.runGraph(context.Background(), task(g), o.tracer("t1")); err != nil { + if _, _, err := o.runComposeGraph(context.Background(), task(g), o.tracer("t1")); err != nil { t.Fatal(err) } out := fs.text() diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index 9c9767f..ce47e9b 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -261,7 +261,7 @@ func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { slog.InfoContext(ctx, "task received", "task_id", t.ID, "graph_bytes", len(t.Graph)) tr.info("task", "system", "任务受理", fmt.Sprintf("DSL %d 字节，按图执行", len(t.Graph))) - // 按 DSL 图执行：compose.Graph（EINO_COMPOSE=1）或自研 graph.go（默认）；agent 节点流式回流 token。 + // 按 DSL 图执行：编译为 Eino compose.Graph 并跑（唯一编排引擎）；agent 节点流式回流 token。 answer, refs, err := o.executeGraph(tctx, t, tr) if errors.Is(err, errInterrupted) { // HITL 持久化中断：checkpoint 已落、任务停在 waiting（状态在审批节点内已置）。 diff --git a/sundynix-dispatcher/internal/eino/rag_integration_test.go b/sundynix-dispatcher/internal/eino/rag_integration_test.go index 022279c..a9175bd 100644 --- a/sundynix-dispatcher/internal/eino/rag_integration_test.go +++ b/sundynix-dispatcher/internal/eino/rag_integration_test.go @@ -41,7 +41,7 @@ func TestRAG_ConversationInjectsAndReturnsRefs(t *testing.T) { o := newOrch(ll, kbTool(&captured), &fakeSink{}, &fakeExec{}) tk := &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{contract.MetaUserID: "u42"}} - ans, refs, err := o.runGraph(context.Background(), tk, o.tracer("t1")) + ans, refs, err := o.runComposeGraph(context.Background(), tk, o.tracer("t1")) if err != nil { t.Fatal(err) } @@ -119,7 +119,7 @@ func TestRAG_DegradesWhenRetrieverDown(t *testing.T) { }} o := newOrch(ll, ft, &fakeSink{}, &fakeExec{}) - ans, refs, err := o.runGraph(context.Background(), &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{}}, o.tracer("t1")) + ans, refs, err := o.runComposeGraph(context.Background(), &contract.Task{ID: "t1", Graph: json.RawMessage(g), Meta: map[string]any{}}, o.tracer("t1")) if err != nil { t.Fatalf("检索失败不应让整图失败: %v", err) }