From 03225e31a90a19c3a8d359b8ebc7f70a2878d4c2 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 29 Jun 2026 11:13:46 +0800 Subject: [PATCH] =?UTF-8?q?refactor(dispatcher):=20compose=20=E7=BF=BB?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=20=E2=80=94=E2=80=94=20=E8=A1=A5=E5=B9=B3?= =?UTF-8?q?=E4=B8=89=E7=BC=BA=E5=8F=A3=E5=90=8E=E9=80=80=E5=BD=B9=20graph.?= =?UTF-8?q?go=20=E5=9C=A8=E6=9C=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EINO_COMPOSE 默认改为开（留 =0 逃生舱回退权威 graph.go）。翻默认前补平 compose 路径三个会静默吃掉治理能力的缺口： - HITL 审批：execDSLNode 无 approval 分支 → 审批节点被当未识别跳过、下游照跑； 补 case 调 approvalNode。 - 忠实度评测：executeGraph 把 refs 硬写 nil → 评测静默失效；runComposeGraph 改签名回传 refsOf(b)。 - 终态传播：rejected/fatalErr 不传播也不阻断下游 → 任务误判 done-空；加节点 lambda 入口守卫 + branch cond 守卫 + 终态上抛 errRejected/fatalErr，并让 runComposeConversation 模型失败置 fatalErr（对齐 graph.go 的 break 语义）。 新增等价测试：审批拒绝停下游、refs 回流。go test ./... 全绿。 soak 无回归后即可物理退役 graph.go。 Co-Authored-By: Claude Opus 4.8 --- .../internal/eino/compose_compiler.go | 40 +++++++++--- .../internal/eino/compose_compiler_test.go | 63 ++++++++++++++++++- .../internal/eino/compose_graph.go | 13 +++- 3 files changed, 103 insertions(+), 13 deletions(-) diff --git a/sundynix-dispatcher/internal/eino/compose_compiler.go b/sundynix-dispatcher/internal/eino/compose_compiler.go index 095a8a8..525ae44 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler.go @@ -24,11 +24,10 @@ func registerFlowMerge() { } // executeGraph 按灰度开关选编排实现：compose.Graph（Phase C）或自研 graph.go（默认/权威）。 -// 返回 (成稿, 检索来源, error)；来源供忠实度评测（compose 路径暂不提供来源 → 返回 nil）。 +// 返回 (成稿, 检索来源, error)；来源供忠实度评测，两条路径都回传（compose 已对齐 graph.go）。 func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) { if composeEnabled() { - ans, err := o.runComposeGraph(ctx, t, tr) - return ans, nil, err + return o.runComposeGraph(ctx, t, tr) } return o.runGraph(ctx, t, tr) } @@ -36,7 +35,9 @@ func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *e // runComposeGraph 把 DSL 图编译为 Eino compose.Graph 并执行（Phase C 编排归一）： // 节点体复用现有 execDSLNode；黑板进 compose 本地状态；branch 走 AddBranch； // DAG 触发模式让无依赖节点并行调度（效率）。编译失败即降级回自研 graph.go（安全网）。 -func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, error) { +// 返回 (成稿, 检索来源, error)：审批拒绝→errRejected、预算触顶/模型失败→fatalErr， +// 与 graph.go 终态严格对齐，否则任务会被误判 done-空。 +func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) { registerFlowMerge() flow, ferr := dsl.Parse(t.Graph) plan := dsl.Compile(t.Graph) @@ -52,7 +53,7 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr b.profile = o.fetchMemory(ctx, b.uid, b.query) b.history = o.fetchHistory(ctx, b.sid) o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") - return b.answer, nil + return b.answer, refsOf(b), b.fatalErr // 模型失败 → 上抛判 failed（对齐 graph.go） } // 邻接 + 入度（只认两端都存在的边）。 @@ -104,6 +105,10 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda( func(c context.Context, _ flowSignal) (flowSignal, error) { perr := compose.ProcessState(c, func(sc context.Context, bd *board) error { + // 上游审批拒绝 / 预算触顶 / 模型失败 → 跳过下游（对齐 graph.go 的 break 中止语义）。 + if bd.rejected || bd.fatalErr != nil { + return nil + } o.execDSLNode(sc, t, node, bd, plan, tr) return nil }) @@ -126,6 +131,9 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr cond := func(c context.Context, _ flowSignal) (map[string]bool, error) { chosen := map[string]bool{} _ = compose.ProcessState(c, func(sc context.Context, bd *board) error { + if bd.rejected || bd.fatalErr != nil { + return nil // 已中止 → 不选任何下游，下方收口到 END + } for _, tgt := range o.branchNode(brn, bd, outE[brn.ID], nodeByID, tr) { chosen[key(tgt)] = true } @@ -161,18 +169,28 @@ func (o *Orchestrator) runComposeGraph(ctx context.Context, t *contract.Task, tr r, cerr := g.Compile(ctx, compose.WithNodeTriggerMode(compose.AllPredecessor)) if cerr != nil { tr.info("task", "system", "compose 编译失败", "退回自研 graph.go："+cerr.Error()) - ans, _, gerr := o.runGraph(ctx, t, tr) // 降级路径丢弃 refs（compose 路径暂不评忠实度） - return ans, gerr + return o.runGraph(ctx, t, tr) // 降级回权威实现（带 refs / 终态） } if _, ierr := r.Invoke(ctx, flowSignal{}); ierr != nil { - tr.info("task", "system", "compose 执行告警", ierr.Error()) // 副作用已落 board；下方按需补一段答复 + tr.info("task", "system", "compose 执行告警", ierr.Error()) // 副作用已落 board；下方按终态收尾 + } + + // 终态对齐 graph.go：审批拒绝 / 预算触顶 / 模型失败要显式上抛，否则任务误判 done-空。 + 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.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") + if b.fatalErr != nil { // 兜底对话也可能触预算顶 / 模型失败 + return b.answer, nil, b.fatalErr + } } - return b.answer, nil + return b.answer, refsOf(b), nil // 成功：带回检索来源供忠实度评测 } // execDSLNode 执行一个非 branch 的 DSL 节点（compose 编译器用；节点体与 graph.go 一致， @@ -208,6 +226,10 @@ func (o *Orchestrator) execDSLNode(ctx context.Context, t *contract.Task, n dsl. 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 "approval": + // HITL 人工审批中断：阻塞等批准/拒绝，拒绝/超时置 b.rejected（节点入口守卫据此中止下游）。 + // 返回的「放行下游」清单在 compose 由静态边 + rejected 守卫接管，这里丢弃。 + o.approvalNode(ctx, t.ID, n, b, tr, nil) case "render": o.renderNode(ctx, t.ID, n, b, tr) case "map": diff --git a/sundynix-dispatcher/internal/eino/compose_compiler_test.go b/sundynix-dispatcher/internal/eino/compose_compiler_test.go index 081c7b1..0f93ef1 100644 --- a/sundynix-dispatcher/internal/eino/compose_compiler_test.go +++ b/sundynix-dispatcher/internal/eino/compose_compiler_test.go @@ -2,7 +2,11 @@ package eino import ( "context" + "encoding/json" + "errors" + "strings" "testing" + "time" "github.com/sundynix/sundynix-dispatcher/internal/harness" "github.com/sundynix/sundynix-dispatcher/internal/llm" @@ -34,13 +38,70 @@ func runBoth(t *testing.T, graph string) (interp, comp string) { t.Fatalf("runGraph: %v", err) } o2 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}} - a2, err := o2.runComposeGraph(context.Background(), task, &execTracer{}) + a2, _, err := o2.runComposeGraph(context.Background(), task, &execTracer{}) if err != nil { t.Fatalf("runComposeGraph: %v", err) } return a1, a2 } +// rejectWaiter 是恒拒绝的 HITL 审批替身（fail-safe 路径回归用）。 +type rejectWaiter struct{ note string } + +func (r *rejectWaiter) WaitApproval(context.Context, string, time.Duration) (*contract.ApprovalDecision, error) { + return &contract.ApprovalDecision{Approved: false, Note: r.note}, nil +} + +// TestComposeApprovalRejectStopsDownstream 钉死 compose 路径的 HITL 审批： +// 审批拒绝须置 errRejected、产出拒绝语，且下游 agent 绝不执行（否则等于审批形同虚设）。 +// 这是翻默认前最高危的回归点——execDSLNode 一度没有 approval 分支会把审批节点当未识别跳过。 +func TestComposeApprovalRejectStopsDownstream(t *testing.T) { + graph := `{"version":"1","nodes":[ + {"id":"in","kind":"input","config":{"text":"hi"}}, + {"id":"ap","kind":"approval","config":{"title":"上线审批"}}, + {"id":"a","kind":"agent","config":{"system":"机密操作"}} + ],"edges":[ + {"source":"in","target":"ap"},{"source":"ap","target":"a"} + ]}` + o := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), + sink: &fakeSink{}, approval: &rejectWaiter{note: "不批"}} + ans, refs, err := o.runComposeGraph(context.Background(), + &contract.Task{ID: "t_ap", Graph: []byte(graph)}, &execTracer{}) + if !errors.Is(err, errRejected) { + t.Fatalf("审批拒绝应返回 errRejected，got %v", err) + } + if !strings.Contains(ans, "已被拒绝") { + t.Fatalf("应产出拒绝语，got %q", ans) + } + if strings.Contains(ans, "ANS:") { + t.Fatalf("审批拒绝后下游 agent 不应执行，但成稿含 agent 产出: %q", ans) + } + if refs != nil { + t.Fatalf("拒绝终态 refs 应为 nil，got %v", refs) + } +} + +// TestComposeReturnsRefs 钉死 compose 路径回传检索来源——曾被硬写成 nil，导致忠实度评测静默失效。 +// 与 runGraph 同图对照：两路径都应回流含检索片段的 refs（喂 grounded judge）。 +func TestComposeReturnsRefs(t *testing.T) { + g := `{"nodes":[ + {"id":"i","kind":"input","config":{"text":"介绍杭州西湖"}}, + {"id":"r","kind":"retriever","config":{"kb":"travel"}}, + {"id":"a","kind":"agent","config":{"system":"你是导游"}} + ],"edges":[{"source":"i","target":"r"},{"source":"r","target":"a"}]}` + ll := &fakeLLM{ready: true, stream: func(m []llm.ChatMessage) string { return m[0].Content }} + o := newOrch(ll, kbTool(nil), &fakeSink{}, &fakeExec{}) + tk := &contract.Task{ID: "t_refs", Graph: json.RawMessage(g), Meta: map[string]any{contract.MetaUserID: "u42"}} + + _, refs, err := o.runComposeGraph(context.Background(), tk, &execTracer{}) + if err != nil { + t.Fatalf("runComposeGraph: %v", err) + } + if len(refs) == 0 || !strings.Contains(strings.Join(refs, ""), ragSnippet) { + t.Fatalf("compose 路径应回流含检索片段的 refs，got %v", refs) + } +} + // TestComposeEquivalentLinear 多节点线性图：input→memory→agent→output，两路径成稿应逐字一致。 func TestComposeEquivalentLinear(t *testing.T) { graph := `{"version":"1","nodes":[ diff --git a/sundynix-dispatcher/internal/eino/compose_graph.go b/sundynix-dispatcher/internal/eino/compose_graph.go index 9cc1a78..4190145 100644 --- a/sundynix-dispatcher/internal/eino/compose_graph.go +++ b/sundynix-dispatcher/internal/eino/compose_graph.go @@ -14,9 +14,10 @@ import ( "github.com/sundynix/sundynix-dispatcher/internal/harness" ) -// composeEnabled 报告是否启用 compose.Graph 编排路径（Phase C 灰度开关，默认关 → 走自研 graph.go）。 -// 并存策略：EINO_COMPOSE=1 时对话主流程改走 Eino compose 运行时，行为对齐后再逐步退役 graph.go。 -func composeEnabled() bool { return os.Getenv("EINO_COMPOSE") == "1" } +// 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 → 累计成稿），便于等价回归。 @@ -82,6 +83,9 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string sr, err := r.Stream(ctx, msgs, compose.WithCallbacks(composeTracer(tr, node))) if err != nil { tr.emit(node, "model", "error", "compose 图执行", err.Error(), time.Since(t0).Milliseconds()) + if b.fatalErr == nil { // 模型失败 → 标记致命错，让任务判 failed（对齐 runAgent，杜绝 done-空） + b.fatalErr = fmt.Errorf("agent 模型推理失败: %w", err) + } return } defer sr.Close() @@ -104,6 +108,9 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string } if rerr != nil { tr.emit(node, "model", "error", "compose 图执行", rerr.Error(), time.Since(t0).Milliseconds()) + if b.fatalErr == nil { // 流式中断也算失败（结果不完整）→ 判 failed，不静默 done-空 + b.fatalErr = fmt.Errorf("agent 模型推理失败: %w", rerr) + } return } if chunk.Content == "" {