系列「企业级 AI Agent 实现拆解」E30 篇也是 Part 8DeepFlux 适配实践的收尾。上一篇讲了五个适配器DeepFlux 如何把 Eino 接进 DDD 架构。这篇把视角提高一层看完整的 ReAct 循环是怎么把这些适配器组装起来的以及 Eino 没有覆盖哪些部分——DeepFlux 自己填补了什么。读完这篇你会知道BuildReActRunnable为什么要自定义StreamToolCallCheckerHITL 中断的接线方式WithInterruptBeforeNodesWithGraphCompileOptions的陷阱ForceNewRun是什么新 turn 为什么要丢弃旧检查点RunTurnHandler.HandleEino 负责的和 DeepFlux 自己做的各是什么系统提示组装memory few-shot skill 三层注入Token 预算和墙钟预算怎么联动HITL 恢复路径的审计留痕一、BuildReActRunnable把 react.Agent 包进一个外层 GraphrAgent,err:react.NewAgent(ctx,react.AgentConfig{ToolCallingModel:chatModel,ToolsConfig:compose.ToolsNodeConfig{Tools:baseTools},MaxStep:maxStep,StreamToolCallChecker:allChunksStreamToolCallChecker,})anyG,nodeOpts:rAgent.ExportGraph()g:compose.NewGraph[[]*schema.Message,*schema.Message]()g.AddGraphNode(react_agent,anyG,nodeOpts...)g.AddEdge(compose.START,react_agent)g.AddEdge(react_agent,compose.END)returng.Compile(ctx,compose.WithCheckPointStore(cfg.CPStore))react.NewAgent构建内部 ReAct 图ExportGraph()把它导出为anyGcompose.AnyGraph 编译选项nodeOpts。然后 DeepFlux 把这个图当一个节点嵌进外层 Graph再整体 Compile。为什么要套一层compose.WithCheckPointStore只在最外层Compile时生效内层react.Agent的编译不支持直接传检查点存储。套一层才能让 checkpoint 正确绑定。二、自定义 StreamToolCallCheckerEino 默认的 checker 叫firstChunkStreamToolCallChecker——看到第一个非空文字 chunk 就返回false判定为这次不调工具。这对 GPT-4o 没问题但 DeepSeek / Claude 有时会先输出一段文字再输出 tool_calls。用默认 checker这些 tool_calls 会被漏掉ReAct 循环提前退出。DeepFlux 替换成allChunksStreamToolCallCheckerfuncallChunksStreamToolCallChecker(_context.Context,sr*schema.StreamReader[*schema.Message])(bool,error){defersr.Close()for{msg,err:sr.Recv()iferrio.EOF{returnfalse,nil}iferr!nil{returnfalse,err}ifmsg!nillen(msg.ToolCalls)0{returntrue,nil// 只要看到工具调用就返回 true}}// 读完所有 chunk 都没有 ToolCalls → false}读完所有 chunk 再判断代价是多一点延迟但保证跨模型行为一致。三、HITL 接线WithGraphCompileOptions 的陷阱react.ExportGraph()返回的nodeOpts里已经打包了 Eino ReAct 内部所需的编译选项WithMaxRunSteps(maxStep) WithNodeTriggerMode(AnyPredecessor) WithGraphName(react.GraphName)加 HITL 时需要追加WithInterruptBeforeNodes([tools])。直觉是append(nodeOpts, newOpt)——但compose.WithGraphCompileOptions是赋值语义第二次调用会静默覆盖第一次之前三个编译选项全部丢失。修复方式把全部选项合并成一次调用ifcfg.HITLEnabled{nodeOpts[]compose.GraphAddNodeOpt{compose.WithGraphCompileOptions(compose.WithMaxRunSteps(maxStep),// 从 react 原始选项里复制compose.WithNodeTriggerMode(compose.AnyPredecessor),compose.WithGraphName(react.GraphName),compose.WithInterruptBeforeNodes([]string{tools}),// 新增),}}tools是 react 内部未导出的节点 keynodeKeyTools与react.ToolsNodeName不同需要对着源码确认Eino 升级时要重新验证。四、ForceNewRun新 turn 为什么要丢弃旧检查点同一个checkPointID session UUID上Eino 默认行为是如果检查点存在就从那个点恢复执行。这对 HITL 恢复是正确的但对用户发送新消息开启新 turn就不对——新消息要从头开始不应该继续上一次的图状态。opts:[]compose.Option{compose.WithCallbacks(cb),compose.WithCheckPointID(string(sid)),}if!in.IsResume{optsappend(opts,compose.WithForceNewRun())// 新 turn 丢弃旧 checkpoint}compose.WithForceNewRun()告诉 Eino即使这个 checkPointID 有历史状态也忽略它重新开始。这样同一个 session 可以经历多轮对话每次新消息开启新图执行HITL 中断时再从中间恢复。五、resume 路径的 checkpoint 预检ifin.IsResume{if_,found,err:f.cpStore.Get(ctx,string(sid));err!nil{returnnil,nil,err}elseif!found{returnnil,nil,port.ErrCheckpointMissing}}Eino 图在 resume 时如果 checkpoint 不存在不会报错——它会静默开启一个全新执行丢失历史对话状态。DeepFlux 在调runnable.Stream之前先检查checkpoint 不存在就 fail fast 返回ErrCheckpointMissing上层能正确提示会话状态已过期而不是让用户看到一个什么都不记得的 Agent。六、Eino 负责什么Handle 自己做什么RunTurnHandler.Handle是调用 Eino 的地方但它做的事比 Eino 多得多Eino 负责的部分ReAct 循环LLM → tool_call 判断 → 工具执行 → 下一轮 LLM流式输出检查点持久化通过CheckPointStore中断WithInterruptBeforeNodes最大步数限制MaxStepHandle自己做的部分Eino 没有覆盖会话状态机idle → running新 turn completed → running多轮续聊清除上轮 stop_reason paused → runningHITL 恢复 running → completed / failed / cancelled状态迁移都在 domain 层model.SessionHandle只是驱动它。Eino 不知道 session 存在。系统提示三层组装sysContent:h.resolveSystemPrompt(ctx,cfg.SystemPrompt,sess)// 版本化 prompt 模板memCtx// 长期记忆召回memory BCfewCtx// few-shot 样例少样本示例库skillCtx// skill 推荐SKILL.md trigger evaluator再加上可选的第四层ifhybridCtxEnabled{summaryText:h.loadHybridContext(...)// 长会话摘要压缩滑动窗口摘要混合sysContent\n\n[会话摘要]\nsummaryText}四层内容全部拼进 system prompt交给 Eino 的 ReAct 循环使用。两种预算护栏墙钟预算整个 ReAct 循环的时间上限。ifcfg.MaxWallSeconds0{ctx,cancelcontext.WithTimeout(ctx,time.Duration(cfg.MaxWallSeconds)*time.Second)defercancel()}超时时ctx.Done()触发Stream返回错误进入completeWithReason(StopBudgetWallclock)——不是 failSession是带部分结果的安全完成。Token 预算累计 Token 超限后中止循环。ifcfg.MaxTokensBudget0{budgetCtx,budgetCancel:context.WithCancel(ctx)deferbudgetCancel()budgetTrackerNewLoopBudgetTracker(cfg.MaxTokensBudget,budgetCancel)ctxcontext.WithValue(budgetCtx,port.ContextKeyLLMBudgetProbe{},budgetTracker.Probe())}LoopBudgetTracker挂在 context 里LLM client 每次收到 usage 数据时调Probe()累加。超限时budgetCancel()取消 contextReAct 循环退出。HITL 审计留痕ifh.audit!nilinterrupt!nil{h.audit.Record(ctx,...,hitl.decision_applied,...,map[string]any{interrupt:interrupt.ID,decision:action,// approve / rejecttool:interrupt.ToolCall.Name,decided_by:dec.DecidedBy,})}HITL 恢复时在真正执行之前先写一条不可变审计记录。如果进程在写审计后、执行工具前崩溃至少留下了谁批准了什么的痕迹稽核时能还原决策链。异步后处理完成后触发两件事都是 fire-and-forget不阻塞响应记忆提炼把这轮对话异步写入 memory BC下一轮召回长会话摘要超过 token 阈值时把历史压缩成摘要存入 context_summaries 表七、整体数据流用户 HTTP → RunTurnHandler.Handle ├── 加载/更新 session 状态机domain 层 ├── 加载 agent_configcfg.MaxTurns, HITLEnabled... ├── 组装 system promptmemory few-shot skill 摘要 ├── 装配 budgetTracker context deadline │ ├── ReActFactory.StreamTurnEino 细节全在这里 │ ├── buildRunnable命中/构建 Runnable 缓存 │ ├── NewAgentCallbackSSE HookRunner │ └── runnable.Stream(inputMsgs, opts...) │ ├── Eino ReAct 循环 │ │ ├── ChatModel.Stream → tokenStreamReader │ │ ├── allChunksStreamToolCallChecker │ │ └── ToolsNode → toolBrokerAdapter.InvokableRun │ └── 返回 (TokenReader, interrupt, err) │ ├── 如果 interrupt → handleEinoInterrupt → session.Pause → SSE 推 interrupt 帧 ├── 如果 TokenReader → consumeEinoStream → 推 turn.delta SSE │ ├── session.Complete() ├── 落库commitFinalAnswer D6 outbox 事务 ├── 审计记录 └── 异步记忆提炼 会话摘要小结BuildReActRunnable解决了三个 Eino 的适配问题自定义 StreamToolCallChecker 兼容先文字后工具的模型、外包一层 Graph 接 CheckPointStore、HITL 时把编译选项合并成一次WithGraphCompileOptions调用避免覆盖。ForceNewRun让同一个 session ID 能区分新 turn和HITL 恢复两种执行路径。checkpoint 预检在 Eino 的静默恢复前 fail fast避免丢失对话历史。RunTurnHandler.Handle是 Eino 之上的业务外壳会话状态机驱动、四层 system prompt 组装、墙钟 Token 双预算、审计留痕、异步后处理。这些都是 Eino 不覆盖的部分也是企业级 Agent 和 demo Agent 之间的真正差距。Part 8 到此收尾。代码来源DeepFlux platform ·server/internal/agent/
