PiAgent 13｜运行时控制：Steering、Follow-up 与事件流#

在 Agent 工作期间观察、纠偏、排队和停止，把“用户还在参与”纳入运行时设计

Agent 一旦开始运行，就不再是“发一条 prompt，等一个字符串”。模型可能正在流式输出，工具可能正在执行，用户也可能突然补充约束。Pi 把这些情况拆成两类队列和一套事件流，让宿主应用能在运行中观察与干预。

1. Steering 与 Follow-up 不是同义词#

Steering：改变当前路线#

steer() 用于在 Agent 工作期间加入指导消息：

session.steer("先不要修改文件，先解释测试失败的原因");

它表达的是“当前任务还没结束，但路线需要调整”。Loop 会在合适的 turn 边界读取 steering 队列，把这条消息放进当前上下文，再决定下一轮怎么继续。

Follow-up：当前路线之后的下一件事#

followUp() 更像排在当前工作之后的待办：

session.followUp("完成修复后，再给我一份变更摘要");

它表达的是“当前路线先自然收束，之后再处理新任务”。两者的差别不在函数名字，而在消息进入循环的时间。

2. 队列为什么是运行时语义#

如果所有用户输入都直接追加到 transcript，会出现三种问题：

• Agent 正在执行危险操作时，用户的新约束来不及生效。
• 当前任务尚未完成，后续任务抢占上下文。
• UI 无法解释输入是在纠偏、排队还是已经发送。

因此可以把输入状态表示为：

当前运行
  ├─ steering queue：尽快改变当前路径
  └─ follow-up queue：当前路径结束后继续

队列更新应通过 queue_update 事件对外发布，让界面可以显示待处理输入，也让恢复逻辑知道这些消息是否已经被消费。

3. 扩展中的 sendMessage#

扩展可能需要向当前 Session 注入消息。官方扩展 API 的 pi.sendMessage 支持不同交付方式：

pi.sendMessage(
  { role: "custom", customType: "policy", content: "..." },
  { deliverAs: "steer", triggerTurn: false },
);

deliverAs 决定它进入 steering、follow-up 或下一轮上下文；triggerTurn 决定 Agent 空闲时是否立即触发模型响应。注入消息时要明确它是事实、约束还是任务，不要把扩展内部状态伪装成用户意图。

4. 事件流是外部世界看到的时钟#

session.subscribe() 收到的不是一串最终答案，而是 Agent 生命周期时间线：

agent_start
  → turn_start
  → message_update
  → tool_execution_start/update/end
  → turn_end
  → queue_update
  → agent_end / agent_settled

用户输入、工具执行和模型输出可能交错，事件 ID 与 Session ID 才能让客户端正确更新状态。不要用“最后一条文本到达”推断 Agent 已经结束。

5. Abort 的传播链#

停止一个运行需要沿着资源链传播：

HTTP 连接关闭 / 用户点击停止
  → session.abort()
  → Agent Loop 停止继续请求模型
  → Provider stream 收到 abort signal
  → 正在执行的工具收到 signal
  → 订阅者收到结束或错误状态

工具如果忽略 AbortSignal，就算模型请求停了，数据库查询、文件扫描或子进程仍可能继续。生产工具必须把 signal 传给下游库，或在边界处实现超时和取消检查。

6. 串行输入与并发请求#

一个 Session 同一时间是否允许多个 prompt()，取决于宿主的并发策略。简单服务可以拒绝第二个请求：

if (session.isBusy()) {
  return res.status(429).json({ error: "Agent 正忙" });
}

更复杂的应用可以把消息放入队列，但必须明确它是 steering 还是 follow-up，并保证同一 Session 的 transcript 不被并发写坏。不要仅靠一个全局 busy 变量保护多用户服务；锁的粒度至少应是用户或 Session。

7. 把事件翻译为 SSE#

Web 前端通常不需要全部内部事件，可以建立公共事件协议：

function toClientEvent(event: AgentSessionEvent) {
  if (event.type === "message_update") {
    const update = event.assistantMessageEvent;
    if (update.type === "text_delta") {
      return { type: "text", delta: update.delta };
    }
  }

  if (event.type === "queue_update") {
    return { type: "queue", steering: event.steering.length };
  }

  if (event.type === "agent_settled") {
    return { type: "done" };
  }

  return null;
}

事件投影需要做版本化、脱敏和去重。内部 tool_execution_update 可能很频繁，公共协议可以合并进度；工具参数可能包含敏感信息，不能原样发送给浏览器。

8. 如何判断真正结束#

有三个信号容易被混用：

• turn_end：一轮模型响应结束，可能马上进入下一轮。
• agent_end：一次 Agent 回合结束，重试时可能出现多次。
• agent_settled：Session 层运行最终收束。

对于单个 API 请求，等待 session.prompt() 完成通常是最直接的终局；对于长期连接的 UI，则应该同时监听 settled、错误、取消和连接关闭。

常见误区

• 把 steering 和 follow-up 都当成“追加一条消息”。
• abort() 只停止模型请求，不停止工具和子进程。
• 一个全局 Session 接收多个用户的实时输入。
• 用 agent_end 一次性恢复输入框，忽略自动重试。
• 把完整内部事件原样传给前端，造成字段耦合和信息泄露。

小结

运行时控制解决的是 Agent 已经启动之后的问题：用户如何纠偏，后续任务如何排队，客户端如何观察，断开时如何停止。steer() 管当前路线，followUp() 管后续工作，事件流提供外部时钟，abort() 把停止意图传播到模型和工具。把这四件事设计清楚，Agent 才像一个可操作的系统，而不是一个只能等待结果的黑盒。

源码定位

• AgentSession 控制 API：packages/coding-agent/src/agent-session.ts
• 队列与事件：packages/agent/src/agent.ts、packages/coding-agent/src/agent-session.ts
• 扩展消息注入：官方文档 extensions
• RPC 输入与事件：packages/coding-agent/src/modes/rpc/