夜雨聆风系列「企业级 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.gograph := 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 在一个副本上操作，不修改 statemodifiedInput := 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 消息记进历史// 判断哪个工具需要 ReturnDirectlystate.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 默认的实现：
funcfirstChunkStreamToolCallChecker(_ 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）。这意味着：
checker 必须 把整个 stream 读完再 Close - 不能
把这个 stream 再转发给下游——在判断完之后，Graph 会在下一轮重新拿 LLM 的输出
为什么对 Claude 不适用：OpenAI 的 tool call 出现在流的第一个有内容的 chunk 里（ToolCalls 字段先到）；Claude 的流式输出里，文本内容和 tool call 可能交替出现，第一个 chunk 往往是文本。用默认 checker 就会误判为"直接回答"。如果用 Claude，需要自己实现一个消费完整个流再判断的 checker。
分支 2：Tools 后分支 + ReturnDirectly
这是 ReAct 里比较复杂的一段逻辑：
// Tools 后的分支：决定是回 ChatModel 还是 ReturnDirectlygraph.AddBranch(nodeKeyTools, compose.NewStreamGraphBranch(func(ctx context.Context, msgsStream *schema.StreamReader[[]*schema.Message]) (endNode string, err error) {msgsStream.Close() // 不需要读内容，只看 Stateerr = 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.Messagecompose.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 的方式
- 配置级
： AgentConfig.ToolReturnDirectly = map[string]struct{}{"final_answer_tool": {}}——某个工具一旦被调用，就直接返回结果 - 运行时
：工具内部调用 react.SetReturnDirectly(ctx)
funcSetReturnDirectly(ctx context.Context) error {return compose.ProcessState(ctx, func(ctx context.Context, s *state) error {s.ReturnDirectlyToolCallID = compose.GetToolCallID(ctx) // 从 ctx 取当前工具的 CallIDreturn nil})}
这让工具本身可以根据自己的执行结果决定"这次要直接返回"，更灵活。
toolResultCollectorMiddleware：工具结果的侧路发送
funcnewToolResultCollectorMiddleware() 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