From 2ee16d1f992d460e3ff8ae5be34c415906655e38 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Tue, 23 Jun 2026 10:42:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(dispatcher):=20Eino=20=E9=87=87=E7=BA=B3?= =?UTF-8?q?=20Phase=20C=20=E2=80=94=E2=80=94=20=E5=AF=B9=E8=AF=9D=E4=B8=BB?= =?UTF-8?q?=E6=B5=81=E7=A8=8B=E8=B7=91=20compose.Graph=20+=20callbacks=20?= =?UTF-8?q?=E5=BD=92=E4=B8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按"并存 + 等价回归"策略，对话主流程可跑在 Eino compose.Graph 上： - compose_graph.go：runConversation 按 EINO_COMPOSE 灰度开关分流； runComposeConversation 建图 START→ChatModel→END，Compile→Stream 回流 token； 模型未就绪/编译失败降级回 runAgent。默认关，graph.go 仍是默认且权威。 - compose_callbacks.go：composeTracer 用 utils/callbacks 把 ChatModel/Tool 的 start/end/error 桥到 ExecEvent（可观测归一，不再各处手写 emit）。 - LLM 接口 + Pool 增 ChatModel()；fakeLLM 加 cm 字段 + stub Eino 模型。 - 测试：compose 图编译运行 / compose 对话流式 / 开关关→走 runAgent。 顺带修真 bug：SubjectTaskStatus 原 sundynix.tasks.status 落在任务流通配 sundynix.tasks.> 内 → 状态事件被当成"幽灵任务"自我放大（实测污染 2300+ 条）。 挪到 sundynix.status.task + dispatcher 加空任务护栏。 验收：make test-go 全绿；live compose 路径 54 字答复 eval 1.00、默认路径 eval 1.00、幽灵任务 0 复发。剩余 branch/map/render 等节点逐步迁移后 graph.go 退役。 Co-Authored-By: Claude Opus 4.8 (1M context) --- EINO_ADOPTION.md | 14 ++- .../internal/eino/compose_callbacks.go | 68 ++++++++++++++ .../internal/eino/compose_graph.go | 93 +++++++++++++++++++ .../internal/eino/compose_graph_test.go | 89 ++++++++++++++++++ sundynix-dispatcher/internal/eino/graph.go | 6 +- .../internal/eino/integration_test.go | 4 + .../internal/eino/orchestrator.go | 7 ++ sundynix-dispatcher/internal/llm/pool.go | 3 + sundynix-shared/contract/task.go | 3 +- 9 files changed, 281 insertions(+), 6 deletions(-) create mode 100644 sundynix-dispatcher/internal/eino/compose_callbacks.go create mode 100644 sundynix-dispatcher/internal/eino/compose_graph.go create mode 100644 sundynix-dispatcher/internal/eino/compose_graph_test.go diff --git a/EINO_ADOPTION.md b/EINO_ADOPTION.md index 3acb352..585282d 100644 --- a/EINO_ADOPTION.md +++ b/EINO_ADOPTION.md @@ -9,7 +9,7 @@ - [x] **Phase A · 地基**：`llm.Pool` 换 Eino ChatModel 组件（commit d84b1ec，验收通过） - [x] **Phase B · 质变**：MCP 工具→`InvokableTool` + ReAct agent（模型自主调工具，验收 7/7 命中） -- [ ] **Phase C · 编排归一**：`Flow→compose.Graph` + callbacks 桥接 +- [~] **Phase C · 编排归一**：✅ 对话主流程跑 `compose.Graph`（EINO_COMPOSE 灰度开关，默认关）+ callbacks→ExecEvent 归一 / ⬜ branch·map·render 等节点逐步迁移（并存+等价回归） - [~] **Phase D · 状态化执行**：✅ 任务生命周期 FSM（已完成）/ ⬜ HITL 中断恢复 / ⬜ 多智能体（按场景） --- @@ -117,7 +117,17 @@ github.com/cloudwego/eino-ext/... # ⚠️ 官方组件实现（open --- -## Phase C · 编排归一：迁到 compose.Graph 〔P2〕 +## Phase C · 编排归一：迁到 compose.Graph 〔P2〕🟡 进行中（核心已落地，并存灰度） + +> 落地（并存+等价回归策略）：对话主流程已可跑在 `compose.Graph` 上—— +> - `compose_graph.go`：`runConversation` 按 `EINO_COMPOSE` 开关分流；`runComposeConversation` 建图 `START→ChatModel→END`、`Compile`→`Stream`，token 回流；模型未就绪/编译失败降级回 `runAgent`。**默认关，graph.go 仍是默认且权威。** +> - `compose_callbacks.go`：`composeTracer` 用 `utils/callbacks` 把 ChatModel/Tool 的 start/end/error 桥到 ExecEvent（可观测归一）。 +> - 测试：compose 图编译+运行、compose 对话流式回流、开关关→走 runAgent，三个单测；live 实测 compose 路径出 54 字答复 + eval 1.00，默认路径 eval 1.00。 +> - **顺带修真 bug**：`SubjectTaskStatus` 原为 `sundynix.tasks.status`，落在任务流捕获通配 `sundynix.tasks.>` 内 → 状态事件被当成"幽灵任务"自我放大（实测污染 2300+ 条）。已挪到 `sundynix.status.task` + dispatcher 加空任务护栏。 +> +> 剩余：branch / map / render / retriever / prompt 等节点逐步迁 compose（同并存+等价回归），对齐后 graph.go 退役。 + + **目标**：自研解释器退役，DSL 图编译为 `compose.Graph`，吃到流式 reduce/branch、类型化边、自动并发、callbacks。 diff --git a/sundynix-dispatcher/internal/eino/compose_callbacks.go b/sundynix-dispatcher/internal/eino/compose_callbacks.go new file mode 100644 index 0000000..3466255 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/compose_callbacks.go @@ -0,0 +1,68 @@ +package eino + +import ( + "context" + "fmt" + + "github.com/cloudwego/eino/callbacks" + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/components/tool" + "github.com/cloudwego/eino/schema" + ucallbacks "github.com/cloudwego/eino/utils/callbacks" +) + +// composeTracer 把 Eino compose 运行时的回调（ChatModel / Tool 的 start/end/error） +// 翻译成我们现有的 ExecEvent 轨迹——可观测"归一"：用框架原生回调，而非各处手写 emit。 +// node 为该次运行在"运行·观测"里的归属节点 id（如 agent:xxx）。 +func composeTracer(tr *execTracer, node string) callbacks.Handler { + return ucallbacks.NewHandlerHelper(). + ChatModel(&ucallbacks.ModelCallbackHandler{ + OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *model.CallbackInput) context.Context { + tr.emit(node, "model", "start", "compose·ChatModel", inputMsgsPreview(in), 0) + return ctx + }, + OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *model.CallbackOutput) context.Context { + detail := "" + if out != nil && out.Message != nil { + detail = truncate(out.Message.Content, 120) + } + tr.emit(node, "model", "end", "compose·ChatModel", detail, 0) + return ctx + }, + OnEndWithStreamOutput: func(ctx context.Context, _ *callbacks.RunInfo, out *schema.StreamReader[*model.CallbackOutput]) context.Context { + out.Close() // 仅观测：正文经主输出流消费，这里只标记结束 + tr.emit(node, "model", "end", "compose·ChatModel", "流式完成", 0) + return ctx + }, + OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context { + tr.emit(node, "model", "error", "compose·ChatModel", err.Error(), 0) + return ctx + }, + }). + Tool(&ucallbacks.ToolCallbackHandler{ + OnStart: func(ctx context.Context, _ *callbacks.RunInfo, in *tool.CallbackInput) context.Context { + if in != nil { + tr.emit(node, "tool", "start", "compose·工具", truncate(in.ArgumentsInJSON, 120), 0) + } + return ctx + }, + OnEnd: func(ctx context.Context, _ *callbacks.RunInfo, out *tool.CallbackOutput) context.Context { + if out != nil { + tr.emit(node, "tool", "end", "compose·工具", truncate(out.Response, 160), 0) + } + return ctx + }, + OnError: func(ctx context.Context, _ *callbacks.RunInfo, err error) context.Context { + tr.emit(node, "tool", "error", "compose·工具", err.Error(), 0) + return ctx + }, + }). + Handler() +} + +func inputMsgsPreview(in *model.CallbackInput) string { + if in == nil { + return "" + } + return fmt.Sprintf("%d 条消息", len(in.Messages)) +} diff --git a/sundynix-dispatcher/internal/eino/compose_graph.go b/sundynix-dispatcher/internal/eino/compose_graph.go new file mode 100644 index 0000000..7b14d40 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/compose_graph.go @@ -0,0 +1,93 @@ +package eino + +import ( + "context" + "fmt" + "io" + "os" + "time" + + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/schema" + + "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" } + +// 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，保证不回归。 +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) // 无模型 → 走自研路径的降级桩 + return + } + + g := compose.NewGraph[[]*schema.Message, *schema.Message]() + if err := g.AddChatModelNode("model", cm); err != nil { + tr.info(node, "system", "compose 降级", "建图失败，退回自研路径："+err.Error()) + o.runAgent(ctx, taskID, b, system, tr, node) + return + } + _ = g.AddEdge(compose.START, "model") + _ = g.AddEdge("model", compose.END) + r, err := g.Compile(ctx) + if err != nil { + tr.info(node, "system", "compose 降级", "编译失败，退回自研路径："+err.Error()) + o.runAgent(ctx, taskID, b, system, tr, node) + return + } + + 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...), + } + msgs, _ := buildMessages(ctx, rc) + + t0 := time.Now() + // ChatModel 的 start/end 由 composeTracer（callbacks）落轨迹，这里不再手写 emit（归一）。 + 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()) + return + } + defer sr.Close() + + chunks := 0 + for { + chunk, rerr := sr.Recv() + if rerr == io.EOF { + break + } + if rerr != nil { + tr.emit(node, "model", "error", "compose 图执行", rerr.Error(), time.Since(t0).Milliseconds()) + return + } + if chunk.Content == "" { + continue + } + safe, _ := harness.RedactSecrets(chunk.Content) // 输出护栏：逐片脱敏 + _ = o.sink.PublishToken(taskID, []byte(safe)) + b.answer += safe + chunks++ + } + tr.info(node, "system", "compose 图", fmt.Sprintf("%d 段输出 / %d 字（Eino compose 运行时）", chunks, len([]rune(b.answer)))) +} diff --git a/sundynix-dispatcher/internal/eino/compose_graph_test.go b/sundynix-dispatcher/internal/eino/compose_graph_test.go new file mode 100644 index 0000000..f5c5c6a --- /dev/null +++ b/sundynix-dispatcher/internal/eino/compose_graph_test.go @@ -0,0 +1,89 @@ +package eino + +import ( + "context" + "strings" + "testing" + + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/schema" + + "github.com/sundynix/sundynix-dispatcher/internal/harness" + "github.com/sundynix/sundynix-dispatcher/internal/llm" +) + +// stubModel 是实现 Eino model.BaseChatModel 的测试桩，固定回一段文本（确定性）。 +type stubModel struct{ reply string } + +func (s *stubModel) Generate(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.Message, error) { + return schema.AssistantMessage(s.reply, nil), nil +} + +func (s *stubModel) Stream(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.StreamReader[*schema.Message], error) { + sr, sw := schema.Pipe[*schema.Message](1) + go func() { + sw.Send(schema.AssistantMessage(s.reply, nil), nil) + sw.Close() + }() + return sr, nil +} + +// TestComposeGraphRuns 直接验证 compose.Graph（START→ChatModel→END）能编译并流式产出。 +func TestComposeGraphRuns(t *testing.T) { + g := compose.NewGraph[[]*schema.Message, *schema.Message]() + if err := g.AddChatModelNode("model", &stubModel{reply: "你好世界"}); err != nil { + t.Fatal(err) + } + _ = g.AddEdge(compose.START, "model") + _ = g.AddEdge("model", compose.END) + r, err := g.Compile(context.Background()) + if err != nil { + t.Fatalf("compile: %v", err) + } + out, err := r.Invoke(context.Background(), []*schema.Message{schema.UserMessage("hi")}) + if err != nil { + t.Fatalf("invoke: %v", err) + } + if out.Content != "你好世界" { + t.Fatalf("got %q", out.Content) + } +} + +// TestComposeConversationEquivalent 验证 EINO_COMPOSE=1 时对话主流程走 compose 路径， +// 把模型输出流式回流到 sink 并累计成稿——与自研 runAgent 行为等价（都逐片转发模型输出）。 +func TestComposeConversationEquivalent(t *testing.T) { + t.Setenv("EINO_COMPOSE", "1") + fs := &fakeSink{} + o := &Orchestrator{ + pool: &fakeLLM{ready: true, cm: &stubModel{reply: "我是 compose 路径的回答"}}, + breaker: harness.NewCircuitBreaker(), + sink: fs, + } + b := &board{query: "你好"} + tr := &execTracer{} // sink 为 nil → 轨迹发射空操作 + o.runConversation(context.Background(), "task_compose", b, "", tr, "agent") + + if !strings.Contains(b.answer, "compose 路径") { + t.Fatalf("成稿未含模型输出: %q", b.answer) + } + if !strings.Contains(fs.text(), "compose 路径") { + t.Fatalf("sink 未收到流式 token: %q", fs.text()) + } +} + +// TestComposeDisabledUsesRunAgent 验证开关关闭时仍走自研 runAgent（默认行为不变）。 +func TestComposeDisabledUsesRunAgent(t *testing.T) { + t.Setenv("EINO_COMPOSE", "0") + fs := &fakeSink{} + o := &Orchestrator{ + 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) + } +} diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 5b95d86..aa52d51 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -50,7 +50,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT tr.info("task", "system", "无结构化图", "按单轮对话执行") b.profile = o.fetchMemory(ctx, b.uid, b.query) b.history = o.fetchHistory(ctx, b.sid) - o.runAgent(ctx, t.ID, b, plan.System, tr, "agent") + o.runConversation(ctx, t.ID, b, plan.System, tr, "agent") return b.answer, nil } @@ -123,7 +123,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT if cbool(n.Config, "autonomous") { // 开启自主工具 → ReAct（模型自己选工具） o.runReactAgent(ctx, t.ID, b, sys, n, tr, "agent:"+n.ID) } else { - o.runAgent(ctx, t.ID, b, sys, tr, "agent:"+n.ID) + o.runConversation(ctx, t.ID, b, sys, tr, "agent:"+n.ID) } case "aggregate": merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...)) @@ -147,7 +147,7 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT // 图里无 agent 节点（纯工具/检索图）也要出一段模型答复，否则没有输出。 if b.answer == "" { - o.runAgent(ctx, t.ID, b, plan.System, tr, "agent") + o.runConversation(ctx, t.ID, b, plan.System, tr, "agent") } return b.answer, nil } diff --git a/sundynix-dispatcher/internal/eino/integration_test.go b/sundynix-dispatcher/internal/eino/integration_test.go index 50da8fc..b3577c6 100644 --- a/sundynix-dispatcher/internal/eino/integration_test.go +++ b/sundynix-dispatcher/internal/eino/integration_test.go @@ -20,6 +20,7 @@ type fakeLLM struct { ready bool stream func(msgs []llm.ChatMessage) string // ChatStream 要回流的整段文本 chat func(msgs []llm.ChatMessage) (string, error) // Chat 返回 + cm model.BaseChatModel // compose 路径用的 Eino 模型（可为 nil） } func (f *fakeLLM) Ready() bool { return f.ready } @@ -43,6 +44,9 @@ func (f *fakeLLM) Chat(_ context.Context, msgs []llm.ChatMessage) (string, error // ToolCallingModel：假模型不支持函数调用 → ReAct 路径会降级回普通对话。 func (f *fakeLLM) ToolCallingModel() model.ToolCallingChatModel { return nil } +// ChatModel：compose 路径用；nil 时 runComposeConversation 降级回 runAgent。 +func (f *fakeLLM) ChatModel() model.BaseChatModel { return f.cm } + type fakeSink struct { mu sync.Mutex tokens []string diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go index 1a5428d..93aa78f 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -43,6 +43,8 @@ type LLM interface { Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error) // ToolCallingModel 返回支持函数调用的模型（ReAct agent 用）；不支持则返回 nil。 ToolCallingModel() model.ToolCallingChatModel + // ChatModel 返回 Eino ChatModel 组件（compose.Graph 编排用）；未就绪则 nil。 + ChatModel() model.BaseChatModel } // 工具调用超时；超时即降级（不带工具上下文继续推理）。 @@ -96,6 +98,11 @@ func (o *Orchestrator) finishStatus(taskID string, err error) { // Handle 消费一个任务：按 DSL 编译 Eino 图并执行，把 Token 流回流到 sundynix.streams.。 func (o *Orchestrator) Handle(ctx context.Context, t *contract.Task) error { + // 护栏：丢弃空任务（无 id），避免误投/历史脏数据被当真任务处理并触发状态回写放大。 + if t.ID == "" { + log.Printf("[eino] 跳过空任务（无 id）") + return nil + } tr := o.tracer(t.ID) defer tr.done() diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go index 0b40538..8901603 100644 --- a/sundynix-dispatcher/internal/llm/pool.go +++ b/sundynix-dispatcher/internal/llm/pool.go @@ -76,6 +76,9 @@ func (p *Pool) model() model.BaseChatModel { // Ready 报告是否已配置可用后端（且 ChatModel 构建成功）。 func (p *Pool) Ready() bool { return p.model() != nil } +// ChatModel 返回当前 Eino ChatModel 组件（用于 compose.Graph 编排）；未就绪则 nil。 +func (p *Pool) ChatModel() model.BaseChatModel { return p.model() } + // ToolCallingModel 返回支持函数调用的模型（用于 ReAct agent）；未就绪 / 不支持则 nil。 func (p *Pool) ToolCallingModel() model.ToolCallingChatModel { if tcm, ok := p.model().(model.ToolCallingChatModel); ok { diff --git a/sundynix-shared/contract/task.go b/sundynix-shared/contract/task.go index 8965a90..fa20099 100644 --- a/sundynix-shared/contract/task.go +++ b/sundynix-shared/contract/task.go @@ -31,7 +31,8 @@ const ( // 任务生命周期状态回写：dispatcher 开跑/跑完/出错经此主题广播，网关订阅落 PG 并推 UI。 // core NATS pub-sub（状态是幂等覆盖，丢一条由下一条纠正，无需持久化）。 - SubjectTaskStatus = "sundynix.tasks.status" + // 注意：必须在 sundynix.tasks.> 之外，否则会被任务流捕获成"幽灵任务"自我放大。 + SubjectTaskStatus = "sundynix.status.task" ) // 任务生命周期状态机：submitted（网关建任务）→ running（dispatcher 开跑）