refactor(dispatcher): compose 翻默认 —— 补平三缺口后退役 graph.go 在望

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 <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-06-29 11:13:46 +08:00
parent d2662a1f37
commit 03225e31a9
3 changed files with 103 additions and 13 deletions
@@ -24,11 +24,10 @@ func registerFlowMerge() {
} }
// executeGraph 按灰度开关选编排实现:compose.GraphPhase C)或自研 graph.go(默认/权威)。 // executeGraph 按灰度开关选编排实现:compose.GraphPhase 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) { func (o *Orchestrator) executeGraph(ctx context.Context, t *contract.Task, tr *execTracer) (string, []string, error) {
if composeEnabled() { if composeEnabled() {
ans, err := o.runComposeGraph(ctx, t, tr) return o.runComposeGraph(ctx, t, tr)
return ans, nil, err
} }
return o.runGraph(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 编排归一): // runComposeGraph 把 DSL 图编译为 Eino compose.Graph 并执行(Phase C 编排归一):
// 节点体复用现有 execDSLNode;黑板进 compose 本地状态;branch 走 AddBranch // 节点体复用现有 execDSLNode;黑板进 compose 本地状态;branch 走 AddBranch
// DAG 触发模式让无依赖节点并行调度(效率)。编译失败即降级回自研 graph.go(安全网)。 // 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() registerFlowMerge()
flow, ferr := dsl.Parse(t.Graph) flow, ferr := dsl.Parse(t.Graph)
plan := dsl.Compile(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.profile = o.fetchMemory(ctx, b.uid, b.query)
b.history = o.fetchHistory(ctx, b.sid) b.history = o.fetchHistory(ctx, b.sid)
o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") 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( _ = g.AddLambdaNode(key(node.ID), compose.InvokableLambda(
func(c context.Context, _ flowSignal) (flowSignal, error) { func(c context.Context, _ flowSignal) (flowSignal, error) {
perr := compose.ProcessState(c, func(sc context.Context, bd *board) 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) o.execDSLNode(sc, t, node, bd, plan, tr)
return nil 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) { cond := func(c context.Context, _ flowSignal) (map[string]bool, error) {
chosen := map[string]bool{} chosen := map[string]bool{}
_ = compose.ProcessState(c, func(sc context.Context, bd *board) error { _ = 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) { for _, tgt := range o.branchNode(brn, bd, outE[brn.ID], nodeByID, tr) {
chosen[key(tgt)] = true 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)) r, cerr := g.Compile(ctx, compose.WithNodeTriggerMode(compose.AllPredecessor))
if cerr != nil { if cerr != nil {
tr.info("task", "system", "compose 编译失败", "退回自研 graph.go"+cerr.Error()) tr.info("task", "system", "compose 编译失败", "退回自研 graph.go"+cerr.Error())
ans, _, gerr := o.runGraph(ctx, t, tr) // 降级路径丢弃 refs(compose 路径暂不评忠实度 return o.runGraph(ctx, t, tr) // 降级回权威实现(带 refs / 终态
return ans, gerr
} }
if _, ierr := r.Invoke(ctx, flowSignal{}); ierr != nil { 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 节点(纯工具/检索图)也要出一段答复。 // 图里无 agent 节点(纯工具/检索图)也要出一段答复。
if b.answer == "" { if b.answer == "" {
o.runComposeConversation(ctx, t.ID, b, plan.System, tr, "agent") 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 一致, // 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...)) merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...))
b.refs, b.toolOut = merged, nil b.refs, b.toolOut = merged, nil
tr.info("aggregate:"+n.ID, "system", labelOf(n, "汇聚"), "策略:"+firstNonEmpty(cstr(n.Config, "strategy"), "拼接")) 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": case "render":
o.renderNode(ctx, t.ID, n, b, tr) o.renderNode(ctx, t.ID, n, b, tr)
case "map": case "map":
@@ -2,7 +2,11 @@ package eino
import ( import (
"context" "context"
"encoding/json"
"errors"
"strings"
"testing" "testing"
"time"
"github.com/sundynix/sundynix-dispatcher/internal/harness" "github.com/sundynix/sundynix-dispatcher/internal/harness"
"github.com/sundynix/sundynix-dispatcher/internal/llm" "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) t.Fatalf("runGraph: %v", err)
} }
o2 := &Orchestrator{pool: echoLLM(), breaker: harness.NewCircuitBreaker(), sink: &fakeSink{}} 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 { if err != nil {
t.Fatalf("runComposeGraph: %v", err) t.Fatalf("runComposeGraph: %v", err)
} }
return a1, a2 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("审批拒绝应返回 errRejectedgot %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 应为 nilgot %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 路径应回流含检索片段的 refsgot %v", refs)
}
}
// TestComposeEquivalentLinear 多节点线性图:input→memory→agent→output,两路径成稿应逐字一致。 // TestComposeEquivalentLinear 多节点线性图:input→memory→agent→output,两路径成稿应逐字一致。
func TestComposeEquivalentLinear(t *testing.T) { func TestComposeEquivalentLinear(t *testing.T) {
graph := `{"version":"1","nodes":[ graph := `{"version":"1","nodes":[
@@ -14,9 +14,10 @@ import (
"github.com/sundynix/sundynix-dispatcher/internal/harness" "github.com/sundynix/sundynix-dispatcher/internal/harness"
) )
// composeEnabled 报告是否启用 compose.Graph 编排路径(Phase C 灰度开关,默认关 → 走自研 graph.go) // composeEnabled 报告是否启用 compose.Graph 编排路径。
// 并存策略:EINO_COMPOSE=1 时对话主流程改走 Eino compose 运行时,行为对齐后再逐步退役 graph.go。 // Phase C 灰度收尾:compose 已与 graph.go 终态对齐(refs / 审批拒绝 / 预算·模型失败 全部传播),
func composeEnabled() bool { return os.Getenv("EINO_COMPOSE") == "1" } // 默认翻为 compose 运行时;保留逃生舱 EINO_COMPOSE=0 → 回退权威 graph.go。soak 无回归后即可退役 graph.go。
func composeEnabled() bool { return os.Getenv("EINO_COMPOSE") != "0" }
// runConversation 是对话/模型节点的统一入口:按灰度开关选 compose.Graph 或自研 runAgent。 // runConversation 是对话/模型节点的统一入口:按灰度开关选 compose.Graph 或自研 runAgent。
// 二者对外行为一致(据黑板拼消息 → 流式回流 token → 累计成稿),便于等价回归。 // 二者对外行为一致(据黑板拼消息 → 流式回流 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))) sr, err := r.Stream(ctx, msgs, compose.WithCallbacks(composeTracer(tr, node)))
if err != nil { if err != nil {
tr.emit(node, "model", "error", "compose 图执行", err.Error(), time.Since(t0).Milliseconds()) 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 return
} }
defer sr.Close() defer sr.Close()
@@ -104,6 +108,9 @@ func (o *Orchestrator) runComposeConversation(ctx context.Context, taskID string
} }
if rerr != nil { if rerr != nil {
tr.emit(node, "model", "error", "compose 图执行", rerr.Error(), time.Since(t0).Milliseconds()) 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 return
} }
if chunk.Content == "" { if chunk.Content == "" {