系列「企业级 AI Agent 实现拆解」E42 篇Part 9 源码深度篇第七章。E40 讲了 ReAct Graph 的拓扑E41 讲了 HITL 中断恢复。这篇拆最底层Graph 编译完之后runner是如何驱动节点一轮一轮执行的。读完这篇你会知道runner是什么graph.Compile()产出的是什么数据结构Graph 的两种执行模式Pregel允许循环vs DAG禁止循环主执行循环的三步提交任务 → 等待完成 → 计算下一批通道系统节点怎么知道自己可以跑了分支节点怎么路由Interrupt 检测是如何嵌入主循环的runner编译产物不是代码生成调用graph.Compile(ctx, opts...)之后Eino 返回的不是生成的代码而是一个runner结构体typerunnerstruct{chanSubscribeTomap[string]*chanCall// 节点名 → 执行描述符successorsmap[string][]string// 节点名 → 下游节点名列表dataPredecessorsmap[string][]stringcontrolPredecessorsmap[string][]stringchanBuilderfunc(isStreambool)map[string]channel// 通道工厂eagerbool// 是否提前提交DAG 模式dagbool// trueDAG, false允许循环(Pregel)checkPointer checkPointerWrapper interruptBeforeNodes[]stringinterruptAfterNodes[]stringoptions graphOptions// maxRunSteps 等runCtxfunc(ctx)ctx// State 注入器}每个chanCall是对某个节点的完整描述typechanCallstruct{action*composableRunnable// 实际执行逻辑ChatModel/ToolsNode/Lambda...writeTo[]string// 输出写进哪些通道writeToBranches[]string// 分支节点需要先跑分支条件函数controls[]string// 控制依赖影响谁的 就绪 判断preProcessor*composableRunnable// modelPreHandle 之类postProcessor*composableRunnable}runner是编译完的图的执行描述只要有runner执行可以随时启动也可以从 checkpoint 恢复。两种执行模式// graph.go 注释// runTypePregel: 支持循环节点激活条件 AnyPredecessor任一前置完成就激活// runTypeDAG: 有向无环节点激活条件 AllPredecessor全部前置完成才激活模式允许循环触发模式用途dag字段Pregel✅AnyPredecessorReAct AgentfalseDAG❌AllPredecessorWorkflow并行分支汇聚trueReAct 必须用 PregelChatModel 有两个前置START 和 ToolsNodeAnyPredecessor让它每次 Tools 跑完就能立刻进入下一轮不需要再等 START。DAG 模式下maxRunSteps不起作用r.dagtrue时主循环不检查步数上限因为无环图一定会终止。主执行循环run(ctx, isStream, input, opts)是入口函数核心是三件事1. 初始化 or 从 checkpoint 恢复 2. 拿到第一批 nextTasks 3. 进入主循环 submit(nextTasks) ↓ completedTasks, canceled tm.wait() ↓ 检测 interrupt ↓ nextTasks calculateNextTasks(completedTasks) ↓ 检测 interruptBeforeNodes ↓ 回到步骤 1loop// 主循环forstep:0;;step{// 超步检查只对 Pregel 有效DAG 跳过if!r.dagstepmaxSteps{returnnil,ErrExceedMaxSteps}// Step 1: 提交当前轮的任务tm.submit(nextTasks)// Step 2: 等待任意一批任务完成completedTasks,canceled,canceledTasks:tm.wait()// Step 3: 检测 interrupt组件内 StatefulInterrupt / 子图中断tempInfo:newInterruptTempInfo()tempInfo.collectCanceledInfo(canceled,canceledTasks,completedTasks)r.resolveInterruptCompletedTasks(tempInfo,completedTasks)iflen(tempInfo.subGraphInterrupts)len(tempInfo.interruptRerunNodes)0{// 有中断 → waitAll 等其他并行任务完成 → 保存 checkpoint → 返回 interruptErrorreturnnil,r.handleInterruptWithSubGraphAndRerunNodes(...)}// Step 4: 把完成的任务输出写进通道判断下一批就绪节点nextTasks,result,isEnd,errr.calculateNextTasks(ctx,completedTasks,...)ifisEnd{returnresult,nil}// Step 5: 检测 interruptBeforeNodes配置级下一批任务执行前ifkeys:getHitKey(nextTasks,r.interruptBeforeNodes);len(keys)0{// waitAll 等其他任务 → 保存 checkpoint → 返回 interruptErrorreturnnil,r.handleInterrupt(...)}}通道系统节点怎么知道自己可以跑了每个节点都有一个通道可以理解为收件箱。节点的所有前置节点输出数据时都往这个收件箱里投一条消息。channelManager维护所有通道的状态// 每轮循环结束时nodeMap,err:cm.updateAndGet(ctx,writeChannelValues,controls)// writeChannelValues: key通道名, value数据// controls: 控制依赖更新影响 AllPredecessor 的就绪判断内部流程resolveCompletedTasks(completedTasks) → 对每个完成的任务 计算 branch哪个下游节点被选中哪些被跳过 把输出 copy 到 writeTo 的每个通道里 把控制信号写入 controls cm.updateValues(values) // 实际写入通道 cm.updateDependencies(deps)// 更新控制依赖计数 cm.getFromReadyChannels() // 返回已经就绪的节点 map就绪条件AnyPredecessorPregel只要有一个前置通道收到数据就返回这个节点AllPredecessorDAG所有前置通道都有数据才返回这个节点当 branch 节点选择路径 A 而跳过 B 时reportSkip会把 B 的通道标记为已跳过避免 B 永远等不到数据。分支路由calculateBranch是在resolveCompletedTasks里被调用的func(r*runner)calculateBranch(ctx,curNodeKey,call,input,isStream,cm)([]string,error){// 调用分支条件函数用户提供的 func(ctx, output) string// 返回被选中的下游节点列表// 同时调用 cm.reportBranch 标记被跳过的节点}ReAct 的 ChatModel 分支条件函数firstChunkStreamToolCallChecker检测 LLM 输出里有没有ToolCalls选中nodeKeyTools或compose.END另一侧自动被reportSkip标记不会饿死等数据Task Managergoroutine 池typetaskManagerstruct{runWrapper runnableCallWrapper// invoke 或 transform流式done*internal.UnboundedChan[*task]// 完成队列无界runningTasksmap[string]*task cancelChchan*time.Duration// 外部取消信号}func(t*taskManager)execute(currentTask*task){deferfunc(){// panic 保护panic 变成 err 存进 task.errifpanicInfo:recover();panicInfo!nil{currentTask.errsafe.NewPanicErr(panicInfo,debug.Stack())}t.done.Send(currentTask)// 完成后入队}()// 初始化节点级 CallbackctxinitNodeCallbacks(currentTask.ctx,currentTask.nodeKey,...)currentTask.output,currentTask.errt.runWrapper(ctx,call.action,input,opts...)}submit(tasks)→ 每个 task 开一个 goroutine 执行wait()→ 从done通道里读若干条完成的 task非阻塞等待一批waitAll()→ 阻塞直到所有运行中的 task 都完成UnboundedChan保证不论多少 task 并发完成done.Send永远不会阻塞。Checkpoint中断时保存了什么handleInterrupt把这些东西存进 checkpointcp:checkpoint{Channels:cm.channels,// 所有通道的当前值Inputs:nextTasks[].input,// 待执行节点的输入恢复时直接调度SkipPreHandler:map[string]bool{...},// 哪些节点恢复时跳过 preProcessorState:deepCopyState(state),// Graph State 的深拷贝JSON 序列化InterruptID2Addr:...,// 中断 ID → 地址 映射InterruptID2State:...,// 中断 ID → 组件 state 映射SubGraphs:map[string]*checkpoint{},// 子图 checkpoint}deepCopyState用 JSON 序列化做深拷贝保证 checkpoint 里的 state 和运行时完全独立。恢复时// 从 ctx 或 store 加载 checkpointcp:getCheckPointFromStore(ctx,checkPointID)// 恢复通道状态cm.loadChannels(cp.Channels)// 恢复 Statectxcontext.WithValue(ctx,stateKey{},internalState{state:cp.State})// 直接生成待执行任务不重新从 START 推导nextTasksrestoreTasks(ctx,cp.Inputs,cp.SkipPreHandler,cp.RerunNodes,...)已完成的节点不会重跑执行从中断点继续。全景一次 ReAct 调用的路径agent.Generate(ctx, messages) ↓ runner.run(ctx, isStreamfalse, inputmessages, opts) ↓ initChannelManager() // 建所有通道 initTaskManager() // 准备 goroutine 调度器 calculateNextTasks([START_task]) → 写入 ChatModel 通道 → nodeMap {ChatModel: messages} → nextTasks [{ChatModel}] --- 循环 step0 --- submit([ChatModel_task]) → go execute(ChatModel_task) // 调 LLM wait() → completedTasks [ChatModel_task{output: msg_with_tool_calls}] resolveInterruptCompletedTasks() // 无中断 calculateNextTasks([ChatModel_task]) → calculateBranch: 检测 tool_calls → 选 Tools → 写入 Tools 通道 → nodeMap {Tools: msg} → nextTasks [{Tools}] --- 循环 step1 --- submit([Tools_task]) → go execute(Tools_task) // 并行跑所有工具 wait() → completedTasks [Tools_task{output: tool_results}] calculateNextTasks([Tools_task]) → 检查 ReturnDirectlyToolCallID → 选 ChatModel → nextTasks [{ChatModel}] --- 循环 step2 --- submit([ChatModel_task]) → go execute(ChatModel_task) // 最终回答 wait() → completedTasks [ChatModel_task{output: final_msg}] calculateNextTasks([ChatModel_task]) → calculateBranch: 无 tool_calls → END → nodeMap {END: final_msg} → isEnd true → return result ✅小结Eino 的 Graph 执行引擎本质是一个通道驱动的超步循环概念实现编译产物runnerstructchanSubscribeTo chanBuilder节点就绪判断channelManager.getFromReadyChannels()AnyPredecessor/AllPredecessor任务调度taskManagergoroutine/panic 保护/UnboundedChan分支路由calculateBranch条件函数 reportSkip中断嵌入每轮 wait 之后检测InterruptSignal命中则handleInterrupt恢复checkpoint 恢复通道 StaterestoreTasks直接调度待执行节点循环、通道、goroutine、三步——把 ReAct 的思考→工具→思考变成了可测量、可中断、可恢复的执行流程。代码来源eino/compose/graph_run.go · eino/compose/graph_manager.go
系列「企业级 AI Agent 实现拆解」E42 篇Part 9 源码深度篇第七章。E40 讲了 ReAct Graph 的拓扑E41 讲了 HITL 中断恢复。这篇拆最底层Graph 编译完之后runner是如何驱动节点一轮一轮执行的。读完这篇你会知道runner是什么graph.Compile()产出的是什么数据结构Graph 的两种执行模式Pregel允许循环vs DAG禁止循环主执行循环的三步提交任务 → 等待完成 → 计算下一批通道系统节点怎么知道自己可以跑了分支节点怎么路由Interrupt 检测是如何嵌入主循环的runner编译产物不是代码生成调用graph.Compile(ctx, opts...)之后Eino 返回的不是生成的代码而是一个runner结构体typerunnerstruct{chanSubscribeTomap[string]*chanCall// 节点名 → 执行描述符successorsmap[string][]string// 节点名 → 下游节点名列表dataPredecessorsmap[string][]stringcontrolPredecessorsmap[string][]stringchanBuilderfunc(isStreambool)map[string]channel// 通道工厂eagerbool// 是否提前提交DAG 模式dagbool// trueDAG, false允许循环(Pregel)checkPointer checkPointerWrapper interruptBeforeNodes[]stringinterruptAfterNodes[]stringoptions graphOptions// maxRunSteps 等runCtxfunc(ctx)ctx// State 注入器}每个chanCall是对某个节点的完整描述typechanCallstruct{action*composableRunnable// 实际执行逻辑ChatModel/ToolsNode/Lambda...writeTo[]string// 输出写进哪些通道writeToBranches[]string// 分支节点需要先跑分支条件函数controls[]string// 控制依赖影响谁的 就绪 判断preProcessor*composableRunnable// modelPreHandle 之类postProcessor*composableRunnable}runner是编译完的图的执行描述只要有runner执行可以随时启动也可以从 checkpoint 恢复。两种执行模式// graph.go 注释// runTypePregel: 支持循环节点激活条件 AnyPredecessor任一前置完成就激活// runTypeDAG: 有向无环节点激活条件 AllPredecessor全部前置完成才激活模式允许循环触发模式用途dag字段Pregel✅AnyPredecessorReAct AgentfalseDAG❌AllPredecessorWorkflow并行分支汇聚trueReAct 必须用 PregelChatModel 有两个前置START 和 ToolsNodeAnyPredecessor让它每次 Tools 跑完就能立刻进入下一轮不需要再等 START。DAG 模式下maxRunSteps不起作用r.dagtrue时主循环不检查步数上限因为无环图一定会终止。主执行循环run(ctx, isStream, input, opts)是入口函数核心是三件事1. 初始化 or 从 checkpoint 恢复 2. 拿到第一批 nextTasks 3. 进入主循环 submit(nextTasks) ↓ completedTasks, canceled tm.wait() ↓ 检测 interrupt ↓ nextTasks calculateNextTasks(completedTasks) ↓ 检测 interruptBeforeNodes ↓ 回到步骤 1loop// 主循环forstep:0;;step{// 超步检查只对 Pregel 有效DAG 跳过if!r.dagstepmaxSteps{returnnil,ErrExceedMaxSteps}// Step 1: 提交当前轮的任务tm.submit(nextTasks)// Step 2: 等待任意一批任务完成completedTasks,canceled,canceledTasks:tm.wait()// Step 3: 检测 interrupt组件内 StatefulInterrupt / 子图中断tempInfo:newInterruptTempInfo()tempInfo.collectCanceledInfo(canceled,canceledTasks,completedTasks)r.resolveInterruptCompletedTasks(tempInfo,completedTasks)iflen(tempInfo.subGraphInterrupts)len(tempInfo.interruptRerunNodes)0{// 有中断 → waitAll 等其他并行任务完成 → 保存 checkpoint → 返回 interruptErrorreturnnil,r.handleInterruptWithSubGraphAndRerunNodes(...)}// Step 4: 把完成的任务输出写进通道判断下一批就绪节点nextTasks,result,isEnd,errr.calculateNextTasks(ctx,completedTasks,...)ifisEnd{returnresult,nil}// Step 5: 检测 interruptBeforeNodes配置级下一批任务执行前ifkeys:getHitKey(nextTasks,r.interruptBeforeNodes);len(keys)0{// waitAll 等其他任务 → 保存 checkpoint → 返回 interruptErrorreturnnil,r.handleInterrupt(...)}}通道系统节点怎么知道自己可以跑了每个节点都有一个通道可以理解为收件箱。节点的所有前置节点输出数据时都往这个收件箱里投一条消息。channelManager维护所有通道的状态// 每轮循环结束时nodeMap,err:cm.updateAndGet(ctx,writeChannelValues,controls)// writeChannelValues: key通道名, value数据// controls: 控制依赖更新影响 AllPredecessor 的就绪判断内部流程resolveCompletedTasks(completedTasks) → 对每个完成的任务 计算 branch哪个下游节点被选中哪些被跳过 把输出 copy 到 writeTo 的每个通道里 把控制信号写入 controls cm.updateValues(values) // 实际写入通道 cm.updateDependencies(deps)// 更新控制依赖计数 cm.getFromReadyChannels() // 返回已经就绪的节点 map就绪条件AnyPredecessorPregel只要有一个前置通道收到数据就返回这个节点AllPredecessorDAG所有前置通道都有数据才返回这个节点当 branch 节点选择路径 A 而跳过 B 时reportSkip会把 B 的通道标记为已跳过避免 B 永远等不到数据。分支路由calculateBranch是在resolveCompletedTasks里被调用的func(r*runner)calculateBranch(ctx,curNodeKey,call,input,isStream,cm)([]string,error){// 调用分支条件函数用户提供的 func(ctx, output) string// 返回被选中的下游节点列表// 同时调用 cm.reportBranch 标记被跳过的节点}ReAct 的 ChatModel 分支条件函数firstChunkStreamToolCallChecker检测 LLM 输出里有没有ToolCalls选中nodeKeyTools或compose.END另一侧自动被reportSkip标记不会饿死等数据Task Managergoroutine 池typetaskManagerstruct{runWrapper runnableCallWrapper// invoke 或 transform流式done*internal.UnboundedChan[*task]// 完成队列无界runningTasksmap[string]*task cancelChchan*time.Duration// 外部取消信号}func(t*taskManager)execute(currentTask*task){deferfunc(){// panic 保护panic 变成 err 存进 task.errifpanicInfo:recover();panicInfo!nil{currentTask.errsafe.NewPanicErr(panicInfo,debug.Stack())}t.done.Send(currentTask)// 完成后入队}()// 初始化节点级 CallbackctxinitNodeCallbacks(currentTask.ctx,currentTask.nodeKey,...)currentTask.output,currentTask.errt.runWrapper(ctx,call.action,input,opts...)}submit(tasks)→ 每个 task 开一个 goroutine 执行wait()→ 从done通道里读若干条完成的 task非阻塞等待一批waitAll()→ 阻塞直到所有运行中的 task 都完成UnboundedChan保证不论多少 task 并发完成done.Send永远不会阻塞。Checkpoint中断时保存了什么handleInterrupt把这些东西存进 checkpointcp:checkpoint{Channels:cm.channels,// 所有通道的当前值Inputs:nextTasks[].input,// 待执行节点的输入恢复时直接调度SkipPreHandler:map[string]bool{...},// 哪些节点恢复时跳过 preProcessorState:deepCopyState(state),// Graph State 的深拷贝JSON 序列化InterruptID2Addr:...,// 中断 ID → 地址 映射InterruptID2State:...,// 中断 ID → 组件 state 映射SubGraphs:map[string]*checkpoint{},// 子图 checkpoint}deepCopyState用 JSON 序列化做深拷贝保证 checkpoint 里的 state 和运行时完全独立。恢复时// 从 ctx 或 store 加载 checkpointcp:getCheckPointFromStore(ctx,checkPointID)// 恢复通道状态cm.loadChannels(cp.Channels)// 恢复 Statectxcontext.WithValue(ctx,stateKey{},internalState{state:cp.State})// 直接生成待执行任务不重新从 START 推导nextTasksrestoreTasks(ctx,cp.Inputs,cp.SkipPreHandler,cp.RerunNodes,...)已完成的节点不会重跑执行从中断点继续。全景一次 ReAct 调用的路径agent.Generate(ctx, messages) ↓ runner.run(ctx, isStreamfalse, inputmessages, opts) ↓ initChannelManager() // 建所有通道 initTaskManager() // 准备 goroutine 调度器 calculateNextTasks([START_task]) → 写入 ChatModel 通道 → nodeMap {ChatModel: messages} → nextTasks [{ChatModel}] --- 循环 step0 --- submit([ChatModel_task]) → go execute(ChatModel_task) // 调 LLM wait() → completedTasks [ChatModel_task{output: msg_with_tool_calls}] resolveInterruptCompletedTasks() // 无中断 calculateNextTasks([ChatModel_task]) → calculateBranch: 检测 tool_calls → 选 Tools → 写入 Tools 通道 → nodeMap {Tools: msg} → nextTasks [{Tools}] --- 循环 step1 --- submit([Tools_task]) → go execute(Tools_task) // 并行跑所有工具 wait() → completedTasks [Tools_task{output: tool_results}] calculateNextTasks([Tools_task]) → 检查 ReturnDirectlyToolCallID → 选 ChatModel → nextTasks [{ChatModel}] --- 循环 step2 --- submit([ChatModel_task]) → go execute(ChatModel_task) // 最终回答 wait() → completedTasks [ChatModel_task{output: final_msg}] calculateNextTasks([ChatModel_task]) → calculateBranch: 无 tool_calls → END → nodeMap {END: final_msg} → isEnd true → return result ✅小结Eino 的 Graph 执行引擎本质是一个通道驱动的超步循环概念实现编译产物runnerstructchanSubscribeTo chanBuilder节点就绪判断channelManager.getFromReadyChannels()AnyPredecessor/AllPredecessor任务调度taskManagergoroutine/panic 保护/UnboundedChan分支路由calculateBranch条件函数 reportSkip中断嵌入每轮 wait 之后检测InterruptSignal命中则handleInterrupt恢复checkpoint 恢复通道 StaterestoreTasks直接调度待执行节点循环、通道、goroutine、三步——把 ReAct 的思考→工具→思考变成了可测量、可中断、可恢复的执行流程。代码来源eino/compose/graph_run.go · eino/compose/graph_manager.go
