From fa2e2ebeeb71c66503c4ce0875fb435bc6763d99 Mon Sep 17 00:00:00 2001 From: Blizzard Date: Mon, 22 Jun 2026 17:35:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(dispatcher):=20Eino=20=E9=87=87=E7=BA=B3?= =?UTF-8?q?=20Phase=20B=20=E2=80=94=E2=80=94=20ReAct=20=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E4=BD=93=20+=20MCP=20=E5=B7=A5=E5=85=B7=E8=87=AA=E4=B8=BB?= =?UTF-8?q?=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit agent 节点带 autonomous:true 时走 ReAct（flow/agent/react），模型在 推理-工具循环里自行决定调哪些 MCP 工具，而非只跑画死的工具节点。 - 新增 eino/react_agent.go：mcpTool 把 MCP 工具（NATS）适配成 components/tool.InvokableTool；agentTools 暴露 wiki_search + recall_user_memory（context 参数 user_id 服务端注入，不暴露给模型）； runReactAgent 流式回流答复，工具调用经适配器落 ExecEvent 轨迹。 - LLM 接口 + *llm.Pool 增 ToolCallingModel()（openai 组件实现 model.ToolCallingChatModel）；fakeLLM 同步桩（返回 nil → 降级普通对话）。 - graph.go：agent 节点按 autonomous 开关分流 ReAct / 普通；自主 agent 不预注入画像（让其经 recall_user_memory 工具按需自取）。 关键修复：默认 StreamToolCallChecker 只看首个流片段，deepseek 常先吐 文本再给 tool call 致漏判 → 改 streamHasToolCall 扫整段流（命中率 ~0→7/7）。 验收：自主工具调用 7/7 命中，args={} 证明注入生效，完整闭环跑通； make test-go 全绿。已知 eval 看不到工具结果会误判 agent（Phase C 修）。 Co-Authored-By: Claude Opus 4.8 (1M context) --- EINO_ADOPTION.md | 17 +- sundynix-dispatcher/internal/eino/graph.go | 7 +- .../internal/eino/integration_test.go | 5 + .../internal/eino/orchestrator.go | 3 + .../internal/eino/react_agent.go | 191 ++++++++++++++++++ sundynix-dispatcher/internal/llm/pool.go | 8 + 6 files changed, 229 insertions(+), 2 deletions(-) create mode 100644 sundynix-dispatcher/internal/eino/react_agent.go diff --git a/EINO_ADOPTION.md b/EINO_ADOPTION.md index 1fcf3c8..2895ff4 100644 --- a/EINO_ADOPTION.md +++ b/EINO_ADOPTION.md @@ -5,6 +5,15 @@ --- +## 进度 + +- [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 D · 状态化执行**：任务生命周期 FSM / HITL 中断恢复 / 多智能体 + +--- + ## 0. 现状审计（2026-06-22） 实际跑在 `sundynix-dispatcher/internal/` 下： @@ -81,7 +90,13 @@ github.com/cloudwego/eino-ext/... # ⚠️ 官方组件实现（open --- -## Phase B · 质变：函数调用 + ADK 单智能体 〔P1〕 +## Phase B · 质变：函数调用 + ADK 单智能体 〔P1〕✅ 已完成 + +> 落地：新增 `eino/react_agent.go`——`mcpTool` 把 MCP 工具（NATS）适配成 `tool.InvokableTool`；`agent` 节点带 `autonomous:true` 时走 `flow/agent/react`（首批工具：`wiki_search`、`recall_user_memory`，context 参数 user_id 服务端注入、不暴露给模型）。验收：自主工具调用 **7/7 命中**，args={} 证明注入生效，完整闭环（模型→tool→MCP→观察→答复）。 +> +> **两个工程发现**： +> 1. **`StreamToolCallChecker` 必须扫整段流**：默认只看首个流片段，deepseek 常先吐文本再给 tool call → 漏判致不调工具。已用 `streamHasToolCall` 扫全流修复（命中率 ~0 → 7/7）。 +> 2. **eval 看不到工具结果 → 误判**：LLM 评委只见 query+answer，agent 用召回记忆作答会被判"虚构"。Phase C 用 callbacks 把工具上下文喂给 eval 修正。 **目标**：模型能**自主选择并调用** MCP 工具（ReAct），而非只跑画死的工具节点。这是"工作流执行器 → Agent 平台"的关键一跳。 diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go index 38f2a04..5b95d86 100644 --- a/sundynix-dispatcher/internal/eino/graph.go +++ b/sundynix-dispatcher/internal/eino/graph.go @@ -119,7 +119,12 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT case "tool": o.execToolNode(ctx, t.ID, n, b, tr) case "agent": - o.runAgent(ctx, t.ID, b, firstNonEmpty(cstr(n.Config, "system"), plan.System), tr, "agent:"+n.ID) + sys := firstNonEmpty(cstr(n.Config, "system"), plan.System) + 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) + } case "aggregate": merged := aggregate(cstr(n.Config, "strategy"), append(append([]string{}, b.refs...), b.toolOut...)) b.refs, b.toolOut = merged, nil diff --git a/sundynix-dispatcher/internal/eino/integration_test.go b/sundynix-dispatcher/internal/eino/integration_test.go index ba0b58f..50da8fc 100644 --- a/sundynix-dispatcher/internal/eino/integration_test.go +++ b/sundynix-dispatcher/internal/eino/integration_test.go @@ -7,6 +7,8 @@ import ( "sync" "testing" + "github.com/cloudwego/eino/components/model" + "github.com/sundynix/sundynix-dispatcher/internal/harness" "github.com/sundynix/sundynix-dispatcher/internal/llm" "github.com/sundynix/sundynix-shared/contract" @@ -38,6 +40,9 @@ func (f *fakeLLM) Chat(_ context.Context, msgs []llm.ChatMessage) (string, error return "", nil } +// ToolCallingModel：假模型不支持函数调用 → ReAct 路径会降级回普通对话。 +func (f *fakeLLM) ToolCallingModel() model.ToolCallingChatModel { return nil } + 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 591fdd1..5e6e197 100644 --- a/sundynix-dispatcher/internal/eino/orchestrator.go +++ b/sundynix-dispatcher/internal/eino/orchestrator.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/schema" "github.com/sundynix/sundynix-dispatcher/internal/dsl" @@ -34,6 +35,8 @@ type LLM interface { ChatStream(ctx context.Context, msgs []llm.ChatMessage, onToken func(string)) error StreamText(ctx context.Context, text string, onToken func([]byte)) error Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error) + // ToolCallingModel 返回支持函数调用的模型（ReAct agent 用）；不支持则返回 nil。 + ToolCallingModel() model.ToolCallingChatModel } // 工具调用超时；超时即降级（不带工具上下文继续推理）。 diff --git a/sundynix-dispatcher/internal/eino/react_agent.go b/sundynix-dispatcher/internal/eino/react_agent.go new file mode 100644 index 0000000..7e8f922 --- /dev/null +++ b/sundynix-dispatcher/internal/eino/react_agent.go @@ -0,0 +1,191 @@ +package eino + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log" + "time" + + "github.com/cloudwego/eino/components/tool" + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/flow/agent/react" + "github.com/cloudwego/eino/schema" + + "github.com/sundynix/sundynix-dispatcher/internal/dsl" + "github.com/sundynix/sundynix-dispatcher/internal/harness" + "github.com/sundynix/sundynix-shared/contract" +) + +// reactMaxStep 限制 ReAct 推理-工具循环的步数上限（控成本/时延）。 +const reactMaxStep = 8 + +// streamHasToolCall 扫描模型整段流输出，任一片段含 tool call 即判定为工具调用。 +// 比默认"只看首片段"鲁棒（兼容先吐文本/思考再给 tool call 的模型，如 deepseek）。 +// 契约要求：返回前必须 Close 传入的流。 +func streamHasToolCall(_ context.Context, sr *schema.StreamReader[*schema.Message]) (bool, error) { + defer sr.Close() + for { + msg, err := sr.Recv() + if err == io.EOF { + return false, nil + } + if err != nil { + return false, err + } + if len(msg.ToolCalls) > 0 { + return true, nil + } + } +} + +// mcpTool 把一个 MCP 工具（NATS 那头）适配成 Eino InvokableTool： +// 模型给的参数(JSON) + 运行时注入的绑定参数(uid/kb，不暴露给模型) 合并后经 NATS 调 MCP。 +type mcpTool struct { + info *schema.ToolInfo + mcpName string // MCP 侧真实工具名（可与 info.Name 不同） + subject func(string) string // contract.ToolSubjectGo/Py + bind map[string]any // 运行时注入参数 + caller ToolCaller + taskID string + tr *execTracer +} + +func (m *mcpTool) Info(_ context.Context) (*schema.ToolInfo, error) { return m.info, nil } + +// InvokableRun 执行一次工具调用：合并参数 → NATS 调 MCP → 返回观察给模型； +// 调用本身落一条 ExecEvent 轨迹（"模型自己调了哪个工具"在观测里可见）。 +func (m *mcpTool) InvokableRun(ctx context.Context, argsJSON string, _ ...tool.Option) (string, error) { + args := map[string]any{} + if argsJSON != "" { + _ = json.Unmarshal([]byte(argsJSON), &args) + } + for k, v := range m.bind { + args[k] = v + } + log.Printf("[react] 模型自主调用工具 %s (mcp=%s) task=%s args=%s", m.info.Name, m.mcpName, m.taskID, truncate(argsJSON, 120)) + end := m.tr.span("tool:"+m.mcpName, "tool", "模型自主调用 "+m.info.Name) + cctx, cancel := context.WithTimeout(ctx, toolCallTimeout) + defer cancel() + res, err := m.caller.CallTool(cctx, m.subject(m.mcpName), &contract.ToolCall{Tool: m.mcpName, TaskID: m.taskID, Args: args}) + if err != nil { + end("调用失败", err) + return "工具调用失败：" + err.Error(), nil // 作为观察返回，不中断 ReAct + } + if res == nil || !res.OK { + msg := "工具无结果" + if res != nil && res.Error != "" { + msg = res.Error + } + end(msg, nil) + return msg, nil + } + end("入参 "+truncate(argsJSON, 120)+" → "+truncate(res.Content, 160), nil) + return res.Content, nil +} + +// agentTools 构建 ReAct 可用的工具集（按当前任务上下文绑定 uid/kb）。 +// 工具名是给"模型看"的语义名；mcpName 是实际 NATS 调用名。context 参数（user_id）服务端注入，不暴露给模型。 +func (o *Orchestrator) agentTools(b *board, taskID string, tr *execTracer) []tool.BaseTool { + if o.tools == nil { + return nil + } + kbBind := map[string]any{} + if b.kb != "" { + kbBind["kb"] = b.kb + } + return []tool.BaseTool{ + &mcpTool{ + mcpName: "wiki_search", + subject: contract.ToolSubjectGo, + caller: o.tools, taskID: taskID, tr: tr, + bind: kbBind, + info: &schema.ToolInfo{ + Name: "wiki_search", + Desc: "检索知识库，返回与查询最相关的资料片段。需要外部知识/事实依据时调用。", + ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{ + "q": {Type: schema.String, Desc: "检索查询语句", Required: true}, + }), + }, + }, + &mcpTool{ + mcpName: "memory_get", + subject: contract.ToolSubjectGo, + caller: o.tools, taskID: taskID, tr: tr, + bind: map[string]any{"user_id": b.uid}, + info: &schema.ToolInfo{ + Name: "recall_user_memory", + Desc: "召回当前用户的长期画像与偏好（称呼/职业/回答偏好等）。需要个性化、了解“我是谁”时调用。", + ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{}), + }, + }, + } +} + +// runReactAgent 执行带"自主工具"的 agent 节点：模型在 ReAct 循环里自行决定调哪些 MCP 工具。 +// 模型不支持函数调用 / 无工具时降级回普通 runAgent。最终答复流式回流；工具调用由适配器落轨迹。 +func (o *Orchestrator) runReactAgent(ctx context.Context, taskID string, b *board, system string, n dsl.Node, tr *execTracer, node string) { + tcm := o.pool.ToolCallingModel() + tools := o.agentTools(b, taskID, tr) + if tcm == nil || len(tools) == 0 { + tr.info(node, "system", "ReAct 降级", "模型不支持函数调用或无可用工具，退回普通对话") + o.runAgent(ctx, taskID, b, system, tr, node) + return + } + + ag, err := react.NewAgent(ctx, &react.AgentConfig{ + ToolCallingModel: tcm, + ToolsConfig: compose.ToolsNodeConfig{Tools: tools}, + MaxStep: reactMaxStep, + // 默认检查器只看首个流片段；deepseek 等常先吐文本再给 tool call → 漏判。 + // 改成扫描整段流，任一片段含 tool call 即判定为工具调用。 + StreamToolCallChecker: streamHasToolCall, + }) + if err != nil { + tr.emit(node, "model", "error", "构建 ReAct 智能体", err.Error(), 0) + o.runAgent(ctx, taskID, b, system, tr, node) + return + } + + rc := &RunCtx{ + UserID: b.uid, SessionID: b.sid, + System: firstNonEmpty(system, defaultAgentSystem), + Query: b.query, + // 自主 agent 不预注入画像：让它经 recall_user_memory 工具按需自取（否则模型直接答、不调工具）。 + Profile: "", + History: b.history, + ToolOut: append(append([]string{}, b.toolOut...), b.refs...), + } + msgs, _ := buildMessages(ctx, rc) + + tr.emit(node, "model", "start", "ReAct 智能体（自主调工具）", fmt.Sprintf("%d 个工具可用", len(tools)), 0) + t0 := time.Now() + sr, err := ag.Stream(ctx, msgs) + if err != nil { + tr.emit(node, "model", "error", "ReAct 智能体", 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", "ReAct 智能体", 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.emit(node, "model", "end", "ReAct 智能体", + fmt.Sprintf("%d 段输出 / %d 字", chunks, len([]rune(b.answer))), time.Since(t0).Milliseconds()) +} diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go index 8197f31..0b40538 100644 --- a/sundynix-dispatcher/internal/llm/pool.go +++ b/sundynix-dispatcher/internal/llm/pool.go @@ -76,6 +76,14 @@ func (p *Pool) model() model.BaseChatModel { // Ready 报告是否已配置可用后端（且 ChatModel 构建成功）。 func (p *Pool) Ready() bool { return p.model() != nil } +// ToolCallingModel 返回支持函数调用的模型（用于 ReAct agent）；未就绪 / 不支持则 nil。 +func (p *Pool) ToolCallingModel() model.ToolCallingChatModel { + if tcm, ok := p.model().(model.ToolCallingChatModel); ok { + return tcm + } + return nil +} + // ModelName 返回当前激活的对话模型名（未配置则空）—— 供服务状态面板展示。 func (p *Pool) ModelName() string { if cfg := p.config(); cfg != nil {