Graph 执行引擎源码:Eino 的节点调度是怎么工作的(第56篇-E42)

系列「企业级 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 结构体：

type runner struct {    chanSubscribeTo map[string]*chanCall  // 节点名 → 执行描述符    successors      map[string][]string   // 节点名 → 下游节点名列表    dataPredecessors    map[string][]string    controlPredecessors map[string][]string    chanBuilder     func(isStream bool) map[string]channel  // 通道工厂    eager           bool   // 是否提前提交（DAG 模式）    dag             bool   // true=DAG, false=允许循环(Pregel)    checkPointer    checkPointerWrapper    interruptBeforeNodes []string    interruptAfterNodes  []string    options         graphOptions  // maxRunSteps 等    runCtx          func(ctx) ctx // State 注入器}

每个 chanCall 是对某个节点的完整描述：

type chanCall struct {    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	✅	AnyPredecessor	ReAct Agent	`false`
DAG	❌	AllPredecessor	Workflow（并行分支汇聚）	`true`

ReAct 必须用 Pregel：ChatModel 有两个前置（START 和 ToolsNode），AnyPredecessor 让它每次 Tools 跑完就能立刻进入下一轮，不需要再等 START。

DAG 模式下 maxRunSteps 不起作用（r.dag=true 时主循环不检查步数上限），因为无环图一定会终止。

主执行循环

run(ctx, isStream, input, opts) 是入口函数，核心是三件事：

1. 初始化 or 从 checkpoint 恢复2. 拿到第一批 nextTasks3. 进入主循环：   submit(nextTasks)       ↓   completedTasks, canceled = tm.wait()       ↓   检测 interrupt       ↓   nextTasks = calculateNextTasks(completedTasks)       ↓   检测 interruptBeforeNodes       ↓   回到步骤 1（loop）

// 主循环for step := 0; ; step++ {    // 超步检查（只对 Pregel 有效，DAG 跳过）    if !r.dag && step >= maxSteps {        return nil, 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)    if len(tempInfo.subGraphInterrupts) + len(tempInfo.interruptRerunNodes) > 0 {        // 有中断 → waitAll 等其他并行任务完成 → 保存 checkpoint → 返回 interruptError        return nil, r.handleInterruptWithSubGraphAndRerunNodes(...)    }    // Step 4: 把完成的任务输出写进通道，判断下一批就绪节点    nextTasks, result, isEnd, err = r.calculateNextTasks(ctx, completedTasks, ...)    if isEnd { return result, nil }    // Step 5: 检测 interruptBeforeNodes（配置级，下一批任务执行前）    if keys := getHitKey(nextTasks, r.interruptBeforeNodes); len(keys) > 0 {        // waitAll 等其他任务 → 保存 checkpoint → 返回 interruptError        return nil, r.handleInterrupt(...)    }}

通道系统：节点怎么知道自己可以跑了

每个节点都有一个"通道"，可以理解为收件箱。节点的所有前置节点输出数据时，都往这个收件箱里投一条消息。

channelManager 维护所有通道的状态：

// 每轮循环结束时：nodeMap, err := cm.updateAndGet(ctx, writeChannelValues, controls)// writeChannelValues: key=通道名, value=数据// controls: 控制依赖更新（影响 AllPredecessor 的就绪判断）

内部流程：

resolveCompletedTasks(completedTasks)    → 对每个完成的任务：        计算 branch（哪个下游节点被选中，哪些被跳过）        把输出 copy 到 writeTo 的每个通道里        把控制信号写入 controlscm.updateValues(values)    // 实际写入通道cm.updateDependencies(deps)// 更新控制依赖计数cm.getFromReadyChannels()  // 返回已经就绪的节点 map

就绪条件：

AnyPredecessor
（Pregel）：只要有一个前置通道收到数据，就返回这个节点
AllPredecessor
（DAG）：所有前置通道都有数据，才返回这个节点

当 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 Manager：goroutine 池

type taskManager struct {    runWrapper runnableCallWrapper          // invoke 或 transform（流式）    done       *internal.UnboundedChan[*task]  // 完成队列（无界）    runningTasks map[string]*task    cancelCh   chan *time.Duration          // 外部取消信号}func(t *taskManager) execute(currentTask *task) {    defer func() {        // panic 保护：panic 变成 err 存进 task.err        if panicInfo := recover(); panicInfo != nil {            currentTask.err = safe.NewPanicErr(panicInfo, debug.Stack())        }        t.done.Send(currentTask)  // 完成后入队    }()    // 初始化节点级 Callback    ctx = initNodeCallbacks(currentTask.ctx, currentTask.nodeKey, ...)    currentTask.output, currentTask.err = t.runWrapper(ctx, call.action, input, opts...)}

submit(tasks)
→ 每个 task 开一个 goroutine 执行
wait()
→ 从 done 通道里读若干条完成的 task（非阻塞等待一批）
waitAll()
→ 阻塞直到所有运行中的 task 都完成

UnboundedChan 保证不论多少 task 并发完成，done.Send 永远不会阻塞。

Checkpoint：中断时保存了什么

handleInterrupt 把这些东西存进 checkpoint：

cp := &checkpoint{    Channels:        cm.channels,          // 所有通道的当前值    Inputs:          nextTasks[].input,     // 待执行节点的输入（恢复时直接调度）    SkipPreHandler:  map[string]bool{...},  // 哪些节点恢复时跳过 preProcessor    State:           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)// 恢复 Statectx = context.WithValue(ctx, stateKey{}, &internalState{state: cp.State})// 直接生成待执行任务（不重新从 START 推导）nextTasks = restoreTasks(ctx, cp.Inputs, cp.SkipPreHandler, cp.RerunNodes, ...)

已完成的节点不会重跑，执行从中断点继续。

全景：一次 ReAct 调用的路径

agent.Generate(ctx, messages)  ↓runner.run(ctx, isStream=false, input=messages, opts)  ↓initChannelManager()  // 建所有通道initTaskManager()     // 准备 goroutine 调度器calculateNextTasks([START_task])  → 写入 ChatModel 通道 → nodeMap = {ChatModel: messages}  → nextTasks = [{ChatModel}]--- 循环 step=0 ---submit([ChatModel_task])  → go execute(ChatModel_task)  // 调 LLMwait()  → completedTasks = [ChatModel_task{output: msg_with_tool_calls}]resolveInterruptCompletedTasks()  // 无中断calculateNextTasks([ChatModel_task])  → calculateBranch: 检测 tool_calls → 选 Tools  → 写入 Tools 通道 → nodeMap = {Tools: msg}  → nextTasks = [{Tools}]--- 循环 step=1 ---submit([Tools_task])  → go execute(Tools_task)  // 并行跑所有工具wait()  → completedTasks = [Tools_task{output: tool_results}]calculateNextTasks([Tools_task])  → 检查 ReturnDirectlyToolCallID == "" → 选 ChatModel  → nextTasks = [{ChatModel}]--- 循环 step=2 ---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 执行引擎本质是一个通道驱动的超步循环：

概念	实现
编译产物	`runner` struct（chanSubscribeTo + chanBuilder）
节点就绪判断	`channelManager.getFromReadyChannels()` （AnyPredecessor/AllPredecessor）
任务调度	`taskManager` （goroutine/panic 保护/UnboundedChan）
分支路由	`calculateBranch` （条件函数 + reportSkip）
中断嵌入	每轮 wait 之后检测 `InterruptSignal`，命中则 `handleInterrupt`
恢复	checkpoint 恢复通道 + State，`restoreTasks` 直接调度待执行节点

循环、通道、goroutine、三步——把 ReAct 的"思考→工具→思考"变成了可测量、可中断、可恢复的执行流程。

代码来源：eino/compose/graph_run.go · eino/compose/graph_manager.go