这篇先记 Pi Agent 里 agent loop 的设计。

我关心的不是单次 LLM 调用。单次调用能做的事大家都差不多：拼 messages，调模型，拿 response。真正麻烦的是把一次次模型响应、工具执行、用户插话和后续消息组织成一个可以持续运行的 loop。

Agent 难的地方经常在这里：运行过程能不能观察，能不能打断，打断后能不能继续。Pi Agent 这套实现值得看，也是因为它把这些边界都放到了 loop 里。

Loop 主干

Pi Agent 的 loop 在 packages/agent/src/agent-loop.ts，入口是 runLoop。结构上是双层循环。

内层循环处理一轮对话：一次 LLM response，加上这次 response 触发的工具执行。外层循环处理任务延续：如果没有工具调用了，就去看有没有 follow-up message；有就再跑一轮，没有就结束。

Agent 类通过 _runLoop 调 agentLoop 或 agentLoopContinue。loop 运行时会更新内部状态，也会不断发事件。主要机制有这些：

  • transformContext
  • convertToLlm
  • getSteeringMessages
  • getFollowUpMessages
  • 工具执行和中断
  • AgentEvent

出错或 abort 时，loop 直接收尾，发出 agent_end。

这里我觉得最重要的一点是：模型不是中心，loop 才是中心。loop 决定什么时候把消息交给模型，什么时候执行工具，什么时候接用户新的方向，什么时候停下来。

双层循环结构

外层循环负责多轮延续。在没有工具调用、也没有 steering message 时，它会通过 getFollowUpMessages 检查是否有 follow-up message。如果有，就把这些消息放进下一轮；如果没有，整个 agent run 结束。

内层循环负责当前这一轮。它会做几件事：

  • 注入待处理消息，比如 steering message
  • 调 LLM 拿 assistant response
  • 执行工具
  • 收集工具结果
  • 在每轮结束时发出 turn_end

这个拆法很实用。内层循环处理眼前的模型和工具交互，外层循环处理更长一点的任务连续性。这样 agent 不需要把所有事情塞进一次模型调用，也不用把 follow-up 当成最终回答后面的一段尾巴。

消息处理与上下文转换

每次调 LLM 前，Pi Agent 会先处理消息。

第一步是可选的 transformContext。它拿到的是内部的 AgentMessage[]，所以适合做 agent runtime 语义层的事情：裁剪上下文、注入额外信息、重新组织当前要给模型看的内容。

第二步是 convertToLlm。它把 AgentMessage[] 转成 LLM 兼容的 Message[]，顺手过滤 UI 消息，转换自定义类型。

这个顺序不能反过来。transformContext 面对的是 agent 自己的上下文形状，convertToLlm 面对的是模型 API 的输入格式。

换句话说，Pi Agent 没有把上下文当成一坨文本直接塞给模型。它先在 runtime 层整理，再转换成模型能吃的格式。这一点对后面做压缩、插入观察信息、甚至换模型都更友好。

Steering Messages

Pi Agent 实现了一种特殊消息：steering message。

它和普通用户消息不一样，更像运行中的转向信号。用户调用 steer() 后，消息会进队列，用来改变正在运行的 agent 的方向。

它的语义很明确：

  • 当前工具执行完成后投递
  • 跳过剩余工具调用
  • 让 agent 立刻响应新的用户方向

我喜欢这里的边界。它没有选择“任意时刻强杀”。工具可能正在写文件、调外部服务、更新状态，如果在中间直接停，很容易留下半完成状态。

Pi Agent 的做法是等当前工具执行完，再投递 steering message。用户可以打断方向，但 runtime 仍然保留一个清楚的执行边界。

Steering Message 在 Loop 里的位置

Steering message 会在几个位置被检查。

初始化时，loop 先读取 getSteeringMessages()，放进 pendingMessages。如果存在 pending messages，就把它们注入当前消息流，并发出 message_start 和 message_end。

工具执行后，loop 再检查一次 getSteeringMessages()。如果这时有新的 steering message，就跳过剩下的工具调用，把 steering message 放进 pendingMessages，然后回到内层循环。

在没有工具调用的一轮结束后，loop 也会检查 steering message。如果有，就继续内层循环；如果没有，才会去看 follow-up message。

这等于给 steering message 一个更高优先级。follow-up 是自然延续，steering 是用户把方向盘拿回来。

Agent Loop Flow

下面是这套逻辑的完整流程图。

Rendering diagram...

EventStream

Agent loop 会不断发出 AgentEvent。这些事件需要一个能被外部消费的通道，Pi Agent 里的 EventStream 就是在做这件事。

它是一个泛型异步事件流。核心想法很简单：把 push-based 的事件源适配成 pull-based 的 AsyncIterable，同时保留一个 Promise 用来拿最终结果。

核心数据结构

EventStream 里最重要的状态大概是这些：

queue[]              // 已推入但尚未被消费的事件缓冲
waiting[]            // 消费者在等待时注册的 resolve 回调
done                 // 流是否已关闭
finalResultPromise   // 流结束后的最终值

重点是 queue 和 waiting。它们处理的是两种时序。

  • 生产者比消费者快：事件先进 queue
  • 消费者比生产者快：消费者挂到 waiting

这样生产者和消费者不用踩同一个节奏。

Push 和 Pull 的协调

LLM 流式输出、工具执行进度、agent 状态更新，本质上都是 push-based：事件源什么时候有数据，就什么时候推过来。

onChunk(data) {
  stream.push(data);
}

onDone(result) {
  stream.end(result);
}

消费端更自然的写法是 pull-based：

for await (const event of stream) {
  updateUI(event);
}

这两个节奏天然对不上。

生产者：push - push - push -------- push - end
消费者：------------ read - read -------- read

EventStream 在中间加了一层缓冲。

LLM / Tool / Agent Runtime
        |
        | push()
        v
    EventStream
   queue + waiting
        |
        | for await
        v
   UI / Logger / Caller

push() 和 [Symbol.asyncIterator] 之间的协调可以简化成这样：

生产者 push()
- 有 waiting：直接唤醒消费者
- 没有 waiting：事件进入 queue

消费者 for await
- queue 有值：直接 yield
- queue 空且未 done：进入 waiting
- done：return 结束迭代

这就是它避免读写速度不一致时卡住的方式。

双重结果通道

EventStream 同时支持两种消费方式：

  • for await (const event of stream)：逐个处理中间事件
  • await stream.result()：只拿最终状态

这正好对应 agent 产品里的两类需求。

UI 需要中间事件，比如 message update、tool execution update、turn end。调用方有时候只关心这次 run 最后是成功、失败，还是被 abort。

把这两条通道放在同一个对象里，比只返回一个 Promise 更适合 agent runtime。

AsyncIterable

AsyncIterable 是 JavaScript / TypeScript 的内置协议。对象只要实现 [Symbol.asyncIterator]()，就可以被 for await...of 消费。

最简定义是这样：

interface AsyncIterable<T> {
  [Symbol.asyncIterator](): AsyncIterator<T>;
}

interface AsyncIterator<T> {
  next(): Promise<{ value: T; done: boolean }>;
}

for await...of 本质上是这段逻辑的语法糖：

const iterator = stream[Symbol.asyncIterator]();

while (true) {
  const { value, done } = await iterator.next();
  if (done) break;
  handle(value);
}

回到 EventStream，它的异步迭代器大概是这个形状：

async *[Symbol.asyncIterator](): AsyncIterator<T> {
  while (true) {
    if (this.queue.length > 0) {
      yield this.queue.shift()!;
    } else if (this.done) {
      return;
    } else {
      const result = await new Promise(...);
      if (result.done) return;
      yield result.value;
    }
  }
}

async function* 是异步生成器。每个 yield 对应一次 next() 返回的值，return 会让 done 变成 true，for await 循环也就结束。

从 agent loop 的角度看，EventStream 不是 UI 附属品。它是 runtime 可观察性的底层结构。没有它，agent 只能给一个最终答案；有了它，运行过程才能变成连续、可消费、可调试的事件序列。

我喜欢这个设计的地方

Pi Agent 的 loop 把几件事分得比较清楚：模型调用、上下文转换、工具执行、用户中途转向。

这些东西一旦混在一起，agent 很快会变成一段难推理的异步流程。Pi Agent 用 pendingMessages、steering queue、follow-up queue 和事件流，把它们放在不同的控制点上。

这里最值得借鉴的是 steering message 的语义。它不是普通消息追加，它是 runtime 级别的中断信号。用户可以在 agent 运行中重新拿回方向，同时不破坏当前工具执行的边界。

一个可用的 agent loop，不能只让模型不断往下想。它还要回答这些问题：

  • 当前轮什么时候结束？
  • 工具执行之后还要不要继续？
  • 用户插话排在哪里？
  • follow-up 和 steering 谁优先？
  • error 或 abort 时事件流怎么收尾？

Pi Agent 把这些问题显式放进了 loop。对我来说，这比更长上下文或者更复杂 prompt 更重要。

References