Pi Agent 原理

Appearance

2.4 队列：steer 与 followUp ​

现实中的交互不是"一次提问一次回答"那么干净。用户可能在 Agent 干活中途插嘴（steer），也可能在 Agent 答完之后追加要求（followUp）。pi 用两个队列优雅地处理这两种情况。

先澄清一个误解

steer / followUp 是程序化 API（agent.steer(...) / agent.followUp(...)），不是让用户在终端里二选一的。真实交互式 CLI 里，用户敲的一句话由 AgentSession 根据 Agent 当前状态自动分流（见文末"真实场景"）。这两个方法主要给程序化/自动化场景（脚本、IDE 集成、别的 Agent 驱动）提供精确控制。

两种队列 ​

Agent 内部维护两个 PendingMessageQueue（agent.ts:125）：

队列方法语义注入时机
steering 队列steer(msg)"插队"消息Agent 干活中途，插到下一次模型调用前
followUp 队列followUp(msg)"追加"消息Agent 本来要停的时候，再让它继续
ts
agent.steer("先别改那个文件，改成改这个");
agent.followUp("完成后把改动总结一下");

队列的排空模式 ​

PendingMessageQueue（agent.ts:125）支持两种排空模式 QueueMode：

ts
type QueueMode = "all" | "one-at-a-time";
  • "all"：一到排空点就把队列里所有消息一次性取出。
  • "one-at-a-time"：每次只取最旧的一条，其余留到下一个排空点。

Agent 默认两者都是 "one-at-a-time"（agent.ts:231），可通过 agent.steeringMode / agent.followUpMode 修改。

队列如何被循环消费 ​

回顾 2.1 核心循环，循环在关键节点调用两个"取消息"回调：

ts
// agent.ts:471 —— 注入到 loop 配置
getSteeringMessages: async () => this.steeringQueue.drain(),
getFollowUpMessages: async () => this.followUpQueue.drain(),

在 runLoop 中：

  • steering 在每回合结束时被拉取（agent-loop.ts:259），有消息就把它们设为 pendingMessages，下一回合调用模型前注入。
  • followUp 在内层循环全都结束、Agent 即将停止时被拉取（agent-loop.ts:263），有消息就继续外层循环。

所以：

  • steer() → Agent 正在干活时，你的话会尽快被采纳。
  • followUp() → Agent 答完才处理你的话（适合"总结一下"这类后置指令）。

为什么需要两个队列 ​

因为它俩的时机本质不同：

场景期望
用户看到 Agent 在改错文件，赶紧说"别改那个"希望立刻插进去 → steer
用户说"完成后写个总结"希望等干完再说 → followUp

如果用同一个队列，要么过早（打断当前工作），要么过晚（追加指令被当成插队）。两个队列分别对接循环的两个不同锚点，语义刚好对上。

与 prompt 的关系 ​

  • prompt() 是阻塞式的：运行时若已有 active run 会抛错（agent.ts:347）。
  • steer() / followUp() 是非阻塞的：只入队，不等待。
  • 一个 prompt 运行结束后，continue() 会检查两个队列并处理（agent.ts:367）。

示例时序 ​

t0  agent.prompt("帮我重构 utils")
t1  模型开始干活，调用 read
t2  user.steer("顺便加个注释")        ← 入队
t3  回合结束，循环拉取 steering       ← 取到，下回合注入
t4  模型带着夹带的指令继续干活
t5  模型给出最终答案，内层循环结束
t6  循环拉取 followUp（无）→ 停止
t7  user.followUp("总结一下")
t8  agent.continue() → 处理 followUp，再跑一个回合

真实场景：用户在终端敲的话会被怎么处理 ​

上面 steer / followUp 都是程序化调用。那真实交互式 CLI 里，用户敲进来的一句话怎么办？答案是：AgentSession 根据 Agent 当前状态自动分流，用户不需要自己选。

ts
// 伪代码：AgentSession 收到用户输入后
if (agent.isStreaming) {
	// Agent 正在干活 → 用户插话 → 走 steer（尽快让模型看到）
	agent.steer(message);
} else {
	// Agent 空闲 → 用户开启新对话 → 走 prompt
	await agent.prompt(message);
}
Agent 当前状态用户输入去哪效果
正在干活（isStreaming = true）agent.steer(...)插队，尽快让模型看到
空闲（isStreaming = false）agent.prompt(...)开启一轮新对话

一个完整的真实链路：

用户敲 "帮我重构 utils"
   │  isStreaming=false → prompt
   ▼
Agent 开始干活（isStreaming=true）
   │
   ├─ 用户中途敲 "等等，别动那个文件"
   │    isStreaming=true → steer → 尽快插队
   │
   ├─ 用户再敲 "其实连 README 也一起改"
   │    isStreaming=true → steer（队列里攒着）
   │
   ▼
Agent 干完（isStreaming=false）
   │
   ├─ 用户敲 "总结一下改动"
   │    isStreaming=false → prompt → 新的一轮
   ▼

那 followUp 在真实交互里呢？ 用得较少。交互式用户"等 Agent 干完再追加"的需求，通常就是等 Agent 停了再发一条普通消息（此时 isStreaming = false，自然走 prompt）。followUp 主要给"程序化/自动化"场景用 —— 明确指定"等 Agent 干完再追加"，而不依赖"用户手动等一会儿再输入"这种时序。

一句话：真实交互里是 AgentSession 看 isStreaming 自动分流（干活→steer，空闲→prompt）；followUp 主要给程序化场景用。

小结 ​

  • steer = 插队，followUp = 追加，分别对接循环的不同锚点。
  • 排空模式 all / one-at-a-time 控制取多少。
  • 队列让"中断式"交互成为可能，而不用每次都开启新的 prompt。
真实源码位置
  • PendingMessageQueue：packages/agent/src/agent.ts:125
  • steer / followUp：packages/agent/src/agent.ts:283-290
  • 循环消费点：packages/agent/src/agent-loop.ts:259,263

面试角度：为什么需要两个队列 ​

Q1：为什么需要 steer 和 followUp 两个队列，而不是一个？ 因为两者的时机语义相反：steer 希望"立刻插进当前工作"，followUp 希望"等干完再处理"。它们分别对接循环的两个不同锚点（steer 对接内层回合结束、followUp 对接外层即将停止）。用一个队列要么过早（打断工作）要么过晚（追加指令被当插队），两个队列才能精确表达两种意图。

Q2：为什么默认 one-at-a-time 而不是 all？one-at-a-time 每次只取最旧一条，让 Agent 在排空点之间保持"每回合改变最小"，避免一次塞进一堆消息导致上下文跳变、模型混乱。all 是"一次性全排空"的漩涡模式，适合"攒了一堆一起处理"的场景，但默认保守用单条。

Q3：为什么 steer 的注入锚点在内层循环末尾，而 followUp 的锚点在外层？ 因为内层循环代表"正在处理当前用户请求（含连续工具调用）"，steer 想插进来就该在这个循环里被采纳；而 followUp 想等"所有工作都停"再追加，所以它的检查放在内层全部结束、外层决定是否退出的地方。锚点位置决定了语义。

Q4：prompt 是阻塞的，steer/followUp 是非阻塞的，为什么这样设计？prompt 是"开启一轮新对话"，同一时刻只允许一个 active run（并发 prompt 会互相覆盖 transcript），所以阻塞并抛错。steer/followUp 是"在现有运行中追加"，只入队不等待，因此非阻塞，正好用于交互式场景（用户边看边插话）。

下一步看 Demo：最小 Agent。