跳转至

05 · Agent 与主循环：按执行顺序精读

研究文件：packages/agent/src/agent.ts（613 行）、agent-loop.ts（940 行）、types.ts（529 行），基于 v1.0.0。本文以连续代码段和控制分支解释语义；完整带行号代码见源码浏览附件，不把 import 或空行逐句复述。

5.1 Agent 类与 Loop 为什么分开

Loop 负责「一次运行如何推进」；Agent 类负责长期对象状态、订阅者、队列、取消信号与运行结束。Loop 可以单独消费事件，但 Agent 类提供等待监听器完成的语义屏障。

sequenceDiagram participant App as 应用 participant A as Agent participant L as runAgentLoop participant P as Provider participant T as Tool App->>A: prompt(text) A->>A: 创建 activeRun 与 AbortController A->>L: messages + context snapshot + config L->>A: user message_end L->>P: 规范化上下文后 streamFn P-->>L: 文本与完整 toolCall L->>A: assistant message_end A-->>L: 状态处理与 listeners 完成 L->>T: 校验、beforeToolCall、execute T-->>L: content + details L->>A: tool result message_end L->>P: 下一轮请求 P-->>L: 最终回答 L->>A: agent_end A->>A: 等待 listeners，finishRun A-->>App: prompt Promise 完成

5.2 agent.ts：初始化状态

createMutableAgentState() 复制初始工具和消息数组。如果 messages 没有以 system 开始，createInitialSystemMessage() 用初始 prompt 和工具声明建立 baseline。

systemPrompt 是 getter，通过重放消息求有效值。tools 和 messages 的 setter 复制顶层数组，但 getter 返回当前数组；这不是深度不可变状态，直接 push 仍会修改当前数组。

DEFAULT_MODEL 的 unknown 元数据用于无模型初始状态，不代表自动可工作的 provider。你应主动选择真实 model 并检查找不到的情况。

PendingMessageQueue 的 mode 为 all 或 one-at-a-time。peek() 不消费，drain() 消费被选择的批次。默认一次拿一条，避免一轮里把所有纠正意见同时混在一起。

5.3 prompt() 与 continue()

prompt() 先检查 activeRun，运行中直接再次 prompt 会报错。然后把字符串和图片规范化为 user content，再进入 runPromptMessages()。

continue() 不等于「随便再问一轮」：

  • 空 transcript 或只有 system：拒绝，不消耗队列。
  • 尾部为 assistant：先尝试 steering，再 follow-up；都没有则拒绝。
  • 尾部是可继续的输入：从现有上下文继续。

底层 agentLoopContinue() 自身直接拒绝 assistant tail。Agent 类的队列回退是外层提供的行为。因此把最终 assistant 消息接在尾部后直接调低层继续，会得到不同结果。

为什么这样设计：常规模型协议需要新的输入或工具结果。如果你要主动补一轮，应发送一条 user 消息，或者通过有界 finishTurn 决策安排上下文续跑。

5.4 activeRun 生命周期

runWithLifecycle() 同步设置 activeRun、AbortController 和 isStreaming，避免两个请求同时抢同一个会话。执行器异常时 handleRunFailure() 构造标准 assistant error / aborted message，而不把失败藏起来。

最后 finishRun() 清理 streaming message 和 pending calls，resolve 等待者。abort() 只发信号，工具必须观察信号才能实际停止；同步阻塞或不检查 signal 的外部工具不会因为它存在就自动安全取消。

waitForIdle() 返回 activeRun Promise；agent_end listeners 仍在执行时，isStreaming 保持 true。你可以让 listener 刷日志，调用方得到的完成时刻就包含这些工作。

5.5 processEvents()：状态归约与观察顺序

事件 Agent 的更新
message_start / update 更新 streamingMessage
message_end 清空 partial，追加完整消息
tool_execution_start pendingToolCalls 添加 ID
tool_execution_end 删除 ID
turn_end 如果 assistant 有 errorMessage，记录错误
agent_end 清理 partial，然后仍等待监听器

内部状态更新先发生，再逐个 await listener。工具预检因此能看见已加入历史的 assistant 调用消息。你自己的异步监听器会拖慢循环，应只把需要的屏障工作放这里。

SDK 的 public session.subscribe() 是另一套 listener 合约，不应推断它与 Core 一样自动 await Promise。必须先看类型与实现。

自定义 StreamFn 的失败合约

types.ts L20–34 要求请求、模型或运行失败用 AssistantMessageEventStream 的 error / aborted 事件与最终消息表达，而不是直接 throw 或 Promise reject。否则低层流不会自动转成一个可靠的完整失败轮次。

入口 / 回调 它负责什么 调用方必须承担什么
agentLoop() / agentLoopContinue() 推送 EventStream；后台运行 .then(end) 不把它当任意异常的 catch；违约回调 reject 可能让消费流程无法正常终结
runAgentLoop() / runAgentLoopContinue() await sink 与执行 Promise catch unexpected rejection，处理宿主自身失败
Agent 类 runWithLifecycle 捕获执行器异常，生成标准失败消息，finally 清运行状态 完成后检查 errorMessage；前置错误仍可能 throw
Core subscriber 逐个 await，用于必须完成的状态屏障 自己处理可恢复 observer 错误；不要无界等待
transformContext / convertToLlm / getApiKey 等 请求前转换与解析 遵守类型注释中的安全回退合约，不把可能抛错的业务代码随意塞入

Agent.handleRunFailure() 本身还会向 listeners 发失败事件。若同一个坏 listener 再次抛错，不能声称任何 subscriber 失败都被无条件吞掉；finally 会清理运行状态，但宿主仍要 try/catch。这里的“屏障”是一种会等待、也会受故障影响的依赖。

源码依据：StreamFn 合约、低层后台流、Agent 失败处理。

5.6 agent-loop.ts L40–155：四个入口

agentLoop() / agentLoopContinue() 返回 EventStream；runAgentLoop() / runAgentLoopContinue() 接受显式 emit 回调并 await。

runAgentLoop() 用 declareToolChanges() 修正新消息中的工具声明，复制当前 messages 形成这次执行上下文，发 agent_start、turn_start 和初始消息事件，然后进入 runLoop。

newMessages 只包含本次新产生的消息；它不是完整历史。把 agent_end.messages 当完整会话覆盖旧历史，会丢失之前的内容。

5.7 L159–317：两层循环的意义

外层处理「本来要结束，但 follow-up 或显式 continuation 要继续」。内层处理「还有工具结果或 steering 要交给模型」。

每轮依次：

  1. 非首轮调用 prepareNextTurn，允许压缩或调整下一轮状态。
  2. 必要时重新轮询 steering，处理准备阶段新到达的输入。
  3. 声明工具变化，追加 prepared 和 pending messages。
  4. 调 prepareRequest；首轮也执行，用 canonical persisted context 替换当前视图。
  5. 请求模型，收完整 assistant。
  6. error / aborted：仍调用 finishTurn、发 turn_end 和 agent_end，然后硬退出。
  7. 正常消息：执行工具，追加结果。
  8. finishTurn 返回 end 时立即结束，不再取队列；continue 时保证最多补出一个下一轮请求。
  9. 优先处理工具续跑和 steering；自然停止时再取 follow-up。
  10. 没有任何继续理由，结束。
flowchart TD S[输入与准备] --> R[模型响应] R --> E{error 或 aborted?} E -->|是| X[finishTurn 然后 turn_end] X --> Z[agent_end] E -->|否| C{有 toolCall?} C -->|是| LEN{stopReason length?} LEN -->|是| REJECT[整批拒绝: error toolResults] LEN -->|否| T[校验并执行工具] REJECT --> F C -->|否| F[finishTurn] T --> F F --> D{action end?} D -->|是| Z[agent_end] D -->|否| Q{工具续跑或 steering?} Q -->|是| S Q -->|否| U{follow-up 或显式 continue?} U -->|是| S U -->|否| Z

一个容易犯的错误

finishTurn: async () => ({ action: "continue" })

每次都要求继续，会造成无限模型请求。正确做法是带业务条件和轮次/时间预算；实验 Core 用最多 8 轮，并另设 120 秒截止。上限属于宿主策略，不是 Pi 自动保证。

5.8 L329–377：declareToolChanges()

运行时 context.tools 是可执行集合；历史 system 是模型声明集合。函数重放现有声明，对比实现集合，形成增删差量。

如果待发消息已经有 system message，就合并变化到最后一个 system 的工具字段；否则在第一个非 system pending message 前插入更新。原来 pending system 的工具字段不能绕过真实执行集合，最终差量以当前 tools 为准。

这是一个重要一致性约束：不能让模型被告知工具可用，但真正运行时没有实现；也不能忽略工具撤回后的历史状态。

5.9 L382–466：streamAssistantResponse()

顺序为 transformContext → convertToLlm → normalizeContext → 动态凭据解析 → streamFunction。

start 时把 partial 加入当前上下文；delta 更新同一个位置；done / error 时取 response.result() 的完整消息替换 partial，并 emit message_end。无 start 的 provider 也有兜底分支，最终仍会加入完成消息。

不要从 toolcall_delta 开始执行；完整 args 之前的一切只是流式进度。

5.10 L473–510：length 的保护

输出 token 限制可能截掉参数尾部，JSON salvage parser 却勉强得到一个对象。Pi 不执行这一整条 truncated assistant 中的工具，而为每个调用生成错误结果，让模型重发完整参数。

具体例子：本来要修改三个文件，但响应只留下前两个字段。如果框架执行「能解析的部分」，业务动作就不再匹配模型原始意图。Pi 优先拒绝这个不完整批次。

5.11 L514–663：并行与串行

只要一个被调用工具设置 executionMode: "sequential"，本批次全部串行。否则全局默认 parallel。

并行不是先同时做所有操作：预检按调用顺序串行执行，得到允许或立即失败的结果；然后用 Promise.all 启动允许的操作。tool_execution_end 按实际结束时刻发出；最终 toolResult messages 按 assistant 源顺序写入。

assistant: [A, B]
执行耗时: A=40ms, B=1ms
结束事件: B, A
历史结果: A, B

UI 展示用事件完成顺序；模型 transcript 用稳定源顺序。不能把「先完成」误当成「先被请求」。并行执行不表示工具动作彼此没有依赖，需要你自己声明和设计。

5.12 L706–779：prepareToolCall()

查找工具 → 可选 prepareArguments → schema 校验 → beforeToolCall → 检查取消 → 返回 prepared。

未知工具、参数不合法、hook 拒绝、异常都成为 immediate error outcome，不进入 execute。beforeToolCall 接收的是已验证 args，而 toolCall 对象仍保留原始调用块。

拦截器失败会阻断操作，不应默默放行。产品中不要在这里只用字符串包含 rm 的方式实现全部安全策略。

5.13 L805–887：执行与后处理

runToolCall() 复用同一预检/执行/后处理，用于嵌套调用。它不自动发消息或事件，宿主可加自己的包装。

executePreparedToolCall() 接受进展回调；工具 Promise 完成后拒收迟到更新，并等待已经启动的更新任务。工具 throw 或 isError: true 都会进入错误结果。

afterToolCall 的覆盖是逐字段替换，不是深度 merge。替换 content 而不同时提供 structuredContent，会丢弃旧结构化结果，防止文本和结构化数据指向不同事实。

5.14 terminate 不是「一个工具结束整次运行」

shouldTerminateToolBatch() 要求非空 batch 且每个 finalized result 都为 terminate: true。混合批次继续。这个提示只存在运行时，不变成普通 ToolResultMessage 的通用 LLM 字段。

finishTurn 的 end 是另一种轮次级决策；不要把两者合并理解。

5.15 怎么用这些知识写自己的 Agent

  • 业务工具允许并行：默认即可；有冲突的写工具显式 sequential 或用资源锁。
  • 需要知道为什么失败：读取 state.errorMessage 与 final message，不能仅等待 prompt resolve。
  • 有外部状态：prepareRequest 组装一致快照；不要随意插入半个工具批次。
  • 有运行预算：finishTurn 控制轮次，AbortController 控制时间，外层控制请求速率和费用。
  • 要记录可靠完成：Core 等 prompt / waitForIdle；SDK 等 prompt / agent_settled。

源码定位：agent.ts、主循环、工具预检。