自定义 Workflow：Agent 编排在 Eino 之上

读完这篇你会知道

一、BuildReActRunnable：把 react.Agent 包进一个外层 Graph

rAgent, 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)return g.Compile(ctx, compose.WithCheckPointStore(cfg.CPStore))

react.NewAgent 构建内部 ReAct 图，ExportGraph() 把它导出为 anyG（compose.AnyGraph）+ 编译选项 nodeOpts。然后 DeepFlux 把这个图当一个节点嵌进外层 Graph，再整体 Compile。

为什么要套一层：compose.WithCheckPointStore 只在最外层 Compile 时生效，内层 react.Agent 的编译不支持直接传检查点存储。套一层才能让 checkpoint 正确绑定。

二、自定义 StreamToolCallChecker

Eino 默认的 checker 叫 firstChunkStreamToolCallChecker——看到第一个非空文字 chunk 就返回 false（判定为"这次不调工具"）。这对 GPT-4o 没问题，但 DeepSeek / Claude 有时会先输出一段文字，再输出 tool_calls。用默认 checker，这些 tool_calls 会被漏掉，ReAct 循环提前退出。

DeepFlux 替换成 allChunksStreamToolCallChecker：

func allChunksStreamToolCallChecker(_ 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 msg != nil && len(msg.ToolCalls) > 0 {return true, 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 是赋值语义：第二次调用会静默覆盖第一次，之前三个编译选项全部丢失。

修复方式：把全部选项合并成一次调用：

if cfg.HITLEnabled {nodeOpts = []compose.GraphAddNodeOpt{compose.WithGraphCompileOptions(compose.WithMaxRunSteps(maxStep),// 从 react 原始选项里复制compose.WithNodeTriggerMode(compose.AnyPredecessor),compose.WithGraphName(react.GraphName),compose.WithInterruptBeforeNodes([]string{"tools"}), // 新增),}}

"tools" 是 react 内部未导出的节点 key（nodeKeyTools），与 react.ToolsNodeName 不同，需要对着源码确认，Eino 升级时要重新验证。

四、ForceNewRun：新 turn 为什么要丢弃旧检查点

同一个 checkPointID（= session UUID）上，Eino 默认行为是：如果检查点存在，就从那个点恢复执行。这对 HITL 恢复是正确的，但对"用户发送新消息开启新 turn"就不对——新消息要从头开始，不应该继续上一次的图状态。

opts := []compose.Option{compose.WithCallbacks(cb),compose.WithCheckPointID(string(sid)),}if !in.IsResume {opts = append(opts, compose.WithForceNewRun()) // 新 turn 丢弃旧 checkpoint}

compose.WithForceNewRun() 告诉 Eino：即使这个 checkPointID 有历史状态，也忽略它，重新开始。这样同一个 session 可以经历多轮对话，每次新消息开启新图执行，HITL 中断时再从中间恢复。

五、resume 路径的 checkpoint 预检

if in.IsResume {if _, found, err := f.cpStore.Get(ctx, string(sid)); err != nil {return nil, nil, err} else if !found {return nil, 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）
• 最大步数限制（MaxStep）

Handle 自己做的部分，Eino 没有覆盖：

会话状态机

idle → running（新 turn）completed → running（多轮续聊，清除上轮 stop_reason）paused → running（HITL 恢复）running → completed / failed / cancelled

状态迁移都在 domain 层（model.Session），Handle 只是驱动它。Eino 不知道 session 存在。

系统提示三层组装

sysContent := h.resolveSystemPrompt(ctx, cfg.SystemPrompt, sess)// 版本化 prompt 模板 + memCtx // 长期记忆召回（memory BC） + fewCtx // few-shot 样例（少样本示例库） + skillCtx // skill 推荐（SKILL.md + trigger evaluator）

再加上可选的第四层：

if hybridCtxEnabled {summaryText := h.loadHybridContext(...)// 长会话摘要压缩（滑动窗口+摘要混合）sysContent += "nn[会话摘要]n" + summaryText}

四层内容全部拼进 system prompt，交给 Eino 的 ReAct 循环使用。

两种预算护栏

墙钟预算：整个 ReAct 循环的时间上限。

if cfg.MaxWallSeconds > 0 {ctx, cancel = context.WithTimeout(ctx, time.Duration(cfg.MaxWallSeconds)*time.Second)defer cancel()}

超时时 ctx.Done() 触发，Stream 返回错误，进入 completeWithReason(StopBudgetWallclock)——不是 failSession，是"带部分结果的安全完成"。

Token 预算：累计 Token 超限后中止循环。

if cfg.MaxTokensBudget > 0 {budgetCtx, budgetCancel := context.WithCancel(ctx)defer budgetCancel()budgetTracker = NewLoopBudgetTracker(cfg.MaxTokensBudget, budgetCancel)ctx = context.WithValue(budgetCtx, port.ContextKeyLLMBudgetProbe{}, budgetTracker.Probe())}

LoopBudgetTracker 挂在 context 里，LLM client 每次收到 usage 数据时调 Probe() 累加。超限时 budgetCancel() 取消 context，ReAct 循环退出。

HITL 审计留痕

if h.audit != nil && interrupt != nil {h.audit.Record(ctx, ..., "hitl.decision_applied", ..., map[string]any{"interrupt":interrupt.ID,"decision": action,// approve / reject"tool": interrupt.ToolCall.Name,"decided_by": dec.DecidedBy,})}

HITL 恢复时，在真正执行之前先写一条不可变审计记录。如果进程在写审计后、执行工具前崩溃，至少留下了"谁批准了什么"的痕迹，稽核时能还原决策链。

异步后处理

完成后触发两件事，都是 fire-and-forget，不阻塞响应：

• 记忆提炼：把这轮对话异步写入 memory BC，下一轮召回
• 长会话摘要：超过 token 阈值时，把历史压缩成摘要存入 context_summaries 表

七、整体数据流

用户 HTTP → RunTurnHandler.Handle ├── 加载/更新 session 状态机（domain 层） ├── 加载 agent_config（cfg.MaxTurns, HITLEnabled...） ├── 组装 system prompt（memory + few-shot + skill + 摘要） ├── 装配 budgetTracker + context deadline │ ├── ReActFactory.StreamTurn（Eino 细节全在这里） │├── buildRunnable（命中/构建 Runnable 缓存） │├── NewAgentCallback（SSE + 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/