M13. 运行时控制：steering、follow-up 与事件流#
Agent 一旦开始运行，就不再是“发一条 prompt，等一个字符串”这么简单。模型可能正在流式输出，工具可能正在执行，用户也可能突然补充一个约束。Pi 把这些情况拆成两类队列和一套事件流，让宿主应用能在运行中观察与干预。
1. 两种插队，不是两个同义词#
Steering：纠正当前路线#
steer() 用于在 Agent 工作期间加入一条指导消息。核心 Loop 会在合适的 turn 边界读取 steering 队列，把它放进当前上下文，再决定下一轮怎么继续。
典型场景是用户说：“不要再改这个文件，先检查测试失败原因。”这条消息的语义是改变当前任务的方向。
Follow-up：当前任务结束后的下一件事#
followUp() 更像排在当前工作之后的待办。当当前工具调用和回复链结束，Loop 会检查 follow-up 队列；如果里面还有消息，就启动下一轮。
两者的差别不在函数名字，而在消息进入循环的时间：steering 参与当前运行路线，follow-up 等当前路线自然收束后再接上。把所有用户输入都塞进同一个队列，会让“纠偏”和“追加任务”互相抢语义。
2. 事件流是外部世界看到的时钟#
session.subscribe() 收到的不是一串最终答案，而是 Agent 生命周期的时间线。常用事件可以分为四组：
| 事件 | 用途 |
|---|---|
message_update | 增量渲染 assistant 文本或 thinking |
tool_execution_start/update/end | 展示工具状态、进度和错误 |
turn_start/end、agent_start/end | 统计轮次、结束一次运行 |
queue_update、compaction_start/end | 显示待处理输入和上下文维护 |
UI 应该消费这些事件并生成自己的 projection。不要把每一个 delta 都写成一条数据库消息；它们更适合临时展示。
3. 一个可取消的控制器#
下面的控制器展示三个动作：把事件转成日志、在用户纠偏时调用 steering、在任务结束后排入 follow-up。
import { createAgentSession } from "@earendil-works/pi-coding-agent";
import { getModel } from "@earendil-works/pi-ai/compat";
const session = await createAgentSession({
cwd: process.cwd(),
model: getModel("openai", "gpt-5"),
});
const unsubscribe = session.subscribe((event) => {
switch (event.type) {
case "message_update":
if (event.assistantMessageEvent.type === "text_delta") {
renderTextDelta(event.assistantMessageEvent.delta);
}
break;
case "tool_execution_start":
renderTool(event.toolName, "running");
break;
case "tool_execution_end":
renderTool(event.toolName, event.isError ? "failed" : "done");
break;
case "compaction_start":
renderMaintenance("compacting");
break;
case "compaction_end":
renderMaintenance("ready");
break;
}
});
function onUserCorrection(text: string) {
if (session.isStreaming) return session.steer(text);
return session.prompt(text);
}
function onUserNextTask(text: string) {
if (session.isStreaming) return session.followUp(text);
return session.prompt(text);
}
try {
await session.prompt("Inspect the repository and identify the smallest safe change.");
} finally {
unsubscribe();
session.dispose();
}renderTextDelta、renderTool 和 renderMaintenance 是宿主 UI 的函数。实际产品里还要处理并发调用：按钮事件不能在同一个 session 上无条件地启动第二个 prompt()，否则用户很难判断两次运行的边界。
4. 什么时候应该停止，而不是继续提示#
运行时控制不等于“不断给模型发新消息”。有三种情况更适合停止：
- 用户明确取消当前任务：调用
abort()，并把取消展示为控制流状态，而不是工具失败。 - 预算、超时或风险策略触发：由宿主策略决定停止，不能期待模型自觉节制。
- 工具结果已经满足验收条件：通过
shouldStopAfterTurn或上层状态机结束，而不是再让模型多做一轮。
核心 Agent 还提供 waitForIdle()、prepareNextTurn 和 shouldStopAfterTurn 等扩展点。它们适合把确定性的停止条件放在 Loop 外围；不要在事件监听器里递归调用 prompt()，否则一个 turn_end 可能制造无穷新 turn。
5. 监听器的同步与异步边界#
Pi 的事件体系里有两类常见使用方式：SDK 侧的 session.subscribe() 主要用于接收和投影状态；扩展侧的 pi.on() 可以参与某些生命周期和工具策略。前者适合“不阻塞 Agent 的观察”，后者在需要阻断或等待用户决定时才承担控制职责。
因此，日志、指标和界面刷新应尽快返回；审批、权限和工具前置检查才值得等待。一个监听器若同时做网络上报、写数据库、刷新 UI 和人工确认，任何一个慢操作都会让时序难以解释。
建议把事件先放入一个小型 adapter：
type ProjectionEvent =
| { kind: "text"; delta: string }
| { kind: "tool"; name: string; state: "running" | "done" | "failed" }
| { kind: "lifecycle"; name: string };
function project(event: any): ProjectionEvent | undefined {
if (event.type === "message_update" &&
event.assistantMessageEvent.type === "text_delta") {
return { kind: "text", delta: event.assistantMessageEvent.delta };
}
if (event.type === "tool_execution_start") {
return { kind: "tool", name: event.toolName, state: "running" };
}
if (event.type === "tool_execution_end") {
return {
kind: "tool",
name: event.toolName,
state: event.isError ? "failed" : "done",
};
}
return undefined;
}生产代码应把 any 换成当前 Pi 版本导出的事件联合类型，并为未知事件保留兼容分支。事件协议是适配层，不要把内部字段散落在十几个 UI 组件中。
6. 队列控制的底层语义#
steering 和 follow-up 队列的价值，在于把“现在改变路线”和“当前路线结束后继续”从时间上分离。Loop 在 turn 边界读取 steering，在当前工具链收束后读取 follow-up；队列消息随后成为普通上下文消息，而不是一种永远悬浮在系统外的特殊指令。这个设计既避免并发 prompt() 互相覆盖，也让消息能够被事件和会话层一致地观察、持久化。
7. 资料#
本文依据 Pi SDK 文档、Agent Loop 源码和 Agent 核心源码重写。事件类型和方法签名若发生变化，应以官方当前版本为准。