系列「企业级 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. 拿到第一批 nextTasks
3. 进入主循环：
   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 的每个通道里
        把控制信号写入 controls

cm.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 加载 checkpoint
cp := getCheckPointFromStore(ctx, checkPointID)

// 恢复通道状态
cm.loadChannels(cp.Channels)

// 恢复 State
ctx = 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)  // 调 LLM
wait()
  → 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