feat(dispatcher): Eino 采纳 Phase B —— ReAct 智能体 + MCP 工具自主调用
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) <noreply@anthropic.com>
This commit is contained in:
+16
-1
@@ -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)
|
## 0. 现状审计(2026-06-22)
|
||||||
|
|
||||||
实际跑在 `sundynix-dispatcher/internal/` 下:
|
实际跑在 `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 平台"的关键一跳。
|
**目标**:模型能**自主选择并调用** MCP 工具(ReAct),而非只跑画死的工具节点。这是"工作流执行器 → Agent 平台"的关键一跳。
|
||||||
|
|
||||||
|
|||||||
@@ -119,7 +119,12 @@ func (o *Orchestrator) runGraph(ctx context.Context, t *contract.Task, tr *execT
|
|||||||
case "tool":
|
case "tool":
|
||||||
o.execToolNode(ctx, t.ID, n, b, tr)
|
o.execToolNode(ctx, t.ID, n, b, tr)
|
||||||
case "agent":
|
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":
|
case "aggregate":
|
||||||
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
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/cloudwego/eino/components/model"
|
||||||
|
|
||||||
"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"
|
||||||
"github.com/sundynix/sundynix-shared/contract"
|
"github.com/sundynix/sundynix-shared/contract"
|
||||||
@@ -38,6 +40,9 @@ func (f *fakeLLM) Chat(_ context.Context, msgs []llm.ChatMessage) (string, error
|
|||||||
return "", nil
|
return "", nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ToolCallingModel:假模型不支持函数调用 → ReAct 路径会降级回普通对话。
|
||||||
|
func (f *fakeLLM) ToolCallingModel() model.ToolCallingChatModel { return nil }
|
||||||
|
|
||||||
type fakeSink struct {
|
type fakeSink struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
tokens []string
|
tokens []string
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/cloudwego/eino/components/model"
|
||||||
"github.com/cloudwego/eino/schema"
|
"github.com/cloudwego/eino/schema"
|
||||||
|
|
||||||
"github.com/sundynix/sundynix-dispatcher/internal/dsl"
|
"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
|
ChatStream(ctx context.Context, msgs []llm.ChatMessage, onToken func(string)) error
|
||||||
StreamText(ctx context.Context, text string, onToken func([]byte)) error
|
StreamText(ctx context.Context, text string, onToken func([]byte)) error
|
||||||
Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error)
|
Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error)
|
||||||
|
// ToolCallingModel 返回支持函数调用的模型(ReAct agent 用);不支持则返回 nil。
|
||||||
|
ToolCallingModel() model.ToolCallingChatModel
|
||||||
}
|
}
|
||||||
|
|
||||||
// 工具调用超时;超时即降级(不带工具上下文继续推理)。
|
// 工具调用超时;超时即降级(不带工具上下文继续推理)。
|
||||||
|
|||||||
@@ -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())
|
||||||
|
}
|
||||||
@@ -76,6 +76,14 @@ func (p *Pool) model() model.BaseChatModel {
|
|||||||
// Ready 报告是否已配置可用后端(且 ChatModel 构建成功)。
|
// Ready 报告是否已配置可用后端(且 ChatModel 构建成功)。
|
||||||
func (p *Pool) Ready() bool { return p.model() != nil }
|
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 返回当前激活的对话模型名(未配置则空)—— 供服务状态面板展示。
|
// ModelName 返回当前激活的对话模型名(未配置则空)—— 供服务状态面板展示。
|
||||||
func (p *Pool) ModelName() string {
|
func (p *Pool) ModelName() string {
|
||||||
if cfg := p.config(); cfg != nil {
|
if cfg := p.config(); cfg != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user