ReAct Agent 源码拆解：Eino 如何把 Graph 变成 Agent（第64篇-E50）

系列「企业级 AI Agent 实现拆解」E50 篇，Part 10 生产工程篇第八章。上一篇 讲了 Graph 编译期。这篇把 E47–E49 的所有积累用起来：Eino 的 flow/agent/react 是怎么用 Graph 搭出 ReAct 循环的，每个设计决定背后的原因是什么。

读完这篇你会知道

• Eino ReAct Agent 的 Graph 结构：2 个节点 + 2 条分支 + 1 个 Lambda
• State 在 Agent 里的作用：消息历史怎么在多轮之间累积
• StreamToolCallChecker：为什么流式模式需要单独判断"有没有 tool call"
• ToolReturnDirectly：某些工具如何跳过 LLM，直接把结果返给用户
• toolResultCollectorMiddleware：工具结果如何在执行时被"侧路"发送
• MessageModifier vs MessageRewriter 的区别
• NewAgent 里做了什么：从 Config 到一个可调用的 Runnable

ReAct 循环是什么

ReAct（Reason + Act）是最常见的 Agent 模式：

用户输入
↓
LLM 思考（Reason）
├─ 有工具调用 → 执行工具（Act）→ 把结果喂回 LLM → 再次思考
└─ 没有工具调用 → 输出最终答案

这个循环的特点是有环——LLM 完成之后可能回到 LLM 自身。这就是为什么 E48 要有两种 channel 类型：ReAct 用 Pregel 模式，而不是 DAG。

Eino 把这个循环编码成了一个 Graph：

START → ChatModel → [分支] → Tools → [分支] → ChatModel (循环)
↘ END ↘ END (ReturnDirectly)

Graph 结构：5 个组件

// flow/agent/react/react.go

graph := compose.NewGraph[[]*schema.Message, *schema.Message](
compose.WithGenLocalState(func(ctx context.Context) *state {
return &state{Messages: make([]*schema.Message, 0, config.MaxStep+1)}
}),
)

输入是消息列表 []*schema.Message，输出是单条消息 *schema.Message（最终答案）。State 持有完整的对话历史。

节点 1：ChatModel（chat）

graph.AddChatModelNode(nodeKeyModel, chatModel,
compose.WithStatePreHandler(modelPreHandle),
compose.WithNodeName(modelNodeName),
)

modelPreHandle 是这个节点的 preProcessor（E48 讲过）：

modelPreHandle := func(ctx context.Context, input []*schema.Message, state *state) ([]*schema.Message, error) {
state.Messages = append(state.Messages, input…) // 把新消息追加到历史

if config.MessageRewriter != nil {
state.Messages = config.MessageRewriter(ctx, state.Messages) // 先 rewrite
}

if messageModifier == nil {
return state.Messages, nil
}

// modifier 在一个副本上操作，不修改 state
modifiedInput := make([]*schema.Message, len(state.Messages))
copy(modifiedInput, state.Messages)
return messageModifier(ctx, modifiedInput), nil
}

MessageRewriter vs MessageModifier 的区别：

• MessageRewriter：直接修改 state.Messages，改动会持久保留给下一轮。适合压缩历史（比如超出 context window 时删除旧消息）。
• MessageModifier：在副本上操作，不影响 state.Messages。改动只对当前这次 LLM 调用生效。适合"每次都加 System Prompt"但不想让它出现在历史记录里。

节点 2：ToolsNode（tools）

graph.AddToolsNode(nodeKeyTools, toolsNode,
compose.WithStatePreHandler(toolsNodePreHandle),
compose.WithNodeName(toolsNodeName),
)

toolsNodePreHandle：

toolsNodePreHandle := func(ctx context.Context, input *schema.Message, state *state) (*schema.Message, error) {
if input == nil {
return state.Messages[len(state.Messages)–1], nil // HITL resume 时 input 为 nil
}
state.Messages = append(state.Messages, input) // 把 LLM 的 tool call 消息记进历史

// 判断哪个工具需要 ReturnDirectly
state.ReturnDirectlyToolCallID = getReturnDirectlyToolCallID(input, config.ToolReturnDirectly)
return input, nil
}

注意 input == nil 的处理：当 Graph 被 HITL 中断后恢复，节点重跑时没有新输入——它从 State 里取最后一条消息来继续，不会重发工具调用。

分支 1：ChatModel 后分支

modelPostBranchCondition := func(ctx context.Context, sr *schema.StreamReader[*schema.Message]) (endNode string, err error) {
if isToolCall, err := toolCallChecker(ctx, sr); err != nil {
return "", err
} else if isToolCall {
return nodeKeyTools, nil
}
return compose.END, nil
}

graph.AddBranch(nodeKeyModel,
compose.NewStreamGraphBranch(modelPostBranchCondition,
map[string]bool{nodeKeyTools: true, compose.END: true}))

分支函数接收 流式 的 LLM 输出（StreamReader[*schema.Message]），需要在流没有读完的情况下决定走哪条路。

StreamToolCallChecker：流式模式的难点

非流式情况很简单——msg.ToolCalls 有没有内容一目了然。流式情况下，LLM 输出是逐 Token 推进的，框架必须在不能等完整输出的前提下判断"这次 LLM 是在调工具还是在回答问题"。

Eino 默认的实现：

func firstChunkStreamToolCallChecker(_ 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 // 第一个有内容的 chunk 包含 tool call → 走工具分支
}

if len(msg.Content) == 0 {
continue // 跳过空 chunk（流开头可能有几个空 chunk）
}

return false, nil // 第一个有内容的 chunk 是文本 → 走 END
}
}

关键限制：调用这个 checker 之后，sr 就被消费了（全部读取并 Close）。这意味着：

为什么对 Claude 不适用：OpenAI 的 tool call 出现在流的第一个有内容的 chunk 里（ToolCalls 字段先到）；Claude 的流式输出里，文本内容和 tool call 可能交替出现，第一个 chunk 往往是文本。用默认 checker 就会误判为"直接回答"。如果用 Claude，需要自己实现一个消费完整个流再判断的 checker。

分支 2：Tools 后分支 + ReturnDirectly

这是 ReAct 里比较复杂的一段逻辑：

// Tools 后的分支：决定是回 ChatModel 还是 ReturnDirectly
graph.AddBranch(nodeKeyTools, compose.NewStreamGraphBranch(
func(ctx context.Context, msgsStream *schema.StreamReader[[]*schema.Message]) (endNode string, err error) {
msgsStream.Close() // 不需要读内容，只看 State

err = compose.ProcessState[*state](ctx, func(_ context.Context, state *state) error {
if len(state.ReturnDirectlyToolCallID) > 0 {
endNode = nodeKeyDirectReturn
} else {
endNode = nodeKeyModel // 正常路径：工具结果回 LLM
}
return nil
})
return endNode, err
},
map[string]bool{nodeKeyModel: true, nodeKeyDirectReturn: true},
))

注意 msgsStream.Close() 立刻就被调用了——这个分支函数只需要读 State 里的标志位，不需要工具输出的内容。

ReturnDirectly Lambda 节点

// 从所有工具结果里，找到那个需要直接返回的工具结果
directReturn := func(ctx context.Context, msgs *schema.StreamReader[[]*schema.Message]) (*schema.StreamReader[*schema.Message], error) {
return schema.StreamReaderWithConvert(msgs, func(msgs []*schema.Message) (*schema.Message, error) {
var msg *schema.Message
compose.ProcessState[*state](ctx, func(_ context.Context, state *state) error {
for i := range msgs {
if msgs[i] != nil && msgs[i].ToolCallID == state.ReturnDirectlyToolCallID {
msg = msgs[i]
return nil
}
}
return nil
})
if msg == nil {
return nil, schema.ErrNoValue // 过滤掉不需要直接返回的工具结果
}
return msg, nil
}), nil
}

ErrNoValue（E47 讲过的流过滤哨兵）在这里派上了用场：当工具结果列表里没有匹配的 ToolCallID，返回 ErrNoValue 就自动跳过这条，不会错误输出。

两种触发 ReturnDirectly 的方式

func SetReturnDirectly(ctx context.Context) error {
return compose.ProcessState(ctx, func(ctx context.Context, s *state) error {
s.ReturnDirectlyToolCallID = compose.GetToolCallID(ctx) // 从 ctx 取当前工具的 CallID
return nil
})
}

这让工具本身可以根据自己的执行结果决定"这次要直接返回"，更灵活。

toolResultCollectorMiddleware：工具结果的侧路发送

func newToolResultCollectorMiddleware() compose.ToolMiddleware {
return compose.ToolMiddleware{
Invokable: func(next compose.InvokableToolEndpoint) compose.InvokableToolEndpoint {
return func(ctx context.Context, input *compose.ToolInput) (*compose.ToolOutput, error) {
senders := getToolResultSendersFromCtx(ctx)
output, err := next(ctx, input) // 先执行原始工具
if err != nil {
return nil, err
}
if senders != nil && senders.sender != nil {
senders.sender(input.Name, input.CallID, output.Result) // 侧路发送
}
return output, nil
}
},
// Streamable / Enhanced 版本类似，流式的话先 Copy(2)
Streamable: func(next compose.StreamableToolEndpoint) compose.StreamableToolEndpoint {
return func(ctx context.Context, input *compose.ToolInput) (*compose.StreamToolOutput, error) {
senders := getToolResultSendersFromCtx(ctx)
output, err := next(ctx, input)
if err != nil {
return nil, err
}
if senders != nil && senders.streamSender != nil {
streams := output.Result.Copy(2) // fan-out：一份给 sender，一份继续
senders.streamSender(input.Name, input.CallID, streams[0])
output.Result = streams[1]
}
return output, nil
}
},
}
}

这个中间件自动被塞在所有工具的最前面：

config.ToolsConfig.ToolCallMiddlewares = append(
[]compose.ToolMiddleware{newToolResultCollectorMiddleware()},
config.ToolsConfig.ToolCallMiddlewares…,
)

getToolResultSendersFromCtx(ctx) 从 context 里取 sender——调用方可以在 ctx 里注入 sender，这样工具结果就能实时推送到 SSE 流（DeepFlux 就是这么做的：工具执行时实时推送进度给前端，不用等所有工具跑完）。

流式工具的 Copy(2) 正是 E47 讲过的 fan-out：一份推给 sender，一份继续走 Graph 正常流程。

NewAgent 组装全流程

NewAgent(ctx, config)
│
├─ genToolInfos → 收集所有工具的 ToolInfo（名字、描述、schema）
├─ ChatModelWithTools → 把工具 schema 注入 LLM 配置
├─ NewToolNode → 创建 ToolsNode
│
├─ compose.NewGraph → 带 State 的 Graph（state = 消息历史）
├─ AddChatModelNode → 含 modelPreHandle（累积消息历史）
├─ AddEdge(START, chat)
├─ AddToolsNode → 含 toolsNodePreHandle（记录工具调用消息）
├─ AddBranch(chat, …)→ StreamToolCallChecker 决定 tools or END
├─ buildReturnDirectly
│ ├─ AddLambdaNode("direct_return", …) → 过滤出需要直接返回的工具结果
│ ├─ AddBranch(tools, …) → 看 State 决定 chatModel or direct_return
│ └─ AddEdge(direct_return, END)
│
└─ graph.Compile(ctx,
WithMaxRunSteps(config.MaxStep),
WithNodeTriggerMode(AnyPredecessor), ← Pregel 模式（有环）
WithGraphName(graphName))

AnyPredecessor 是 Pregel 模式（E48 里讲过），这里显式声明了——ReAct 的循环边（tools → chat）必须用 Pregel，DAG 会因为有环而报错。

最终 Graph 结构图

┌──────────────────────────────────────────┐
│ Pregel Graph │
│ │
START ──► │ ChatModel ──► [branch] │
│ │ tool calls │
│ ▼ │
│ ToolsNode ──► [branch] │
│ │ ReturnDirectly
│ ▼ │
│ direct_return ──►│ END
│ │ normal │
│ └──────────────► ChatModel │
└──────────────────────────────────────────┘
▲
cycle

完整调用示例

agent, err := react.NewAgent(ctx, &react.AgentConfig{
ToolCallingModel: llm,
ToolsConfig: compose.ToolsNodeConfig{
Tools: []tool.BaseTool{searchTool, calcTool},
},
MessageModifier: react.NewPersonaModifier("You are a helpful assistant."),
MaxStep: 20,
ToolReturnDirectly: map[string]struct{}{
"final_answer": {}, // 这个工具被调用后直接返回，不再进 LLM
},
})

// 非流式
msg, err := agent.Generate(ctx, []*schema.Message{
schema.UserMessage("帮我搜索一下 Eino 的 GitHub star 数"),
})

// 流式
stream, err := agent.Stream(ctx, []*schema.Message{
schema.UserMessage("…"),
})
for {
chunk, err := stream.Recv()
if err == io.EOF { break }
fmt.Print(chunk.Content)
}
defer stream.Close()

小结

Eino ReAct Agent 不是凭空写的——它是在 E47/E48/E49 讲过的所有机制上搭起来的：

390 行代码实现了一个生产可用的 ReAct Agent——因为基础设施都在 Graph 引擎里，业务逻辑只需要描述"怎么连"。

代码来源：eino/flow/agent/react/react.go