30. packages/agent：把流式 LLM 变成可控 Agent

这个模块干什么

packages/agent 把 pi-ai 的一次模型流包装成带状态、工具执行、队列和事件的 Agent 运行时。 【packages/agent/src/agent.ts:165-170】

它始终以 AgentMessage[] 保存应用侧对话，只在调用模型前通过 convertToLlm 变为 Message[]。 【packages/agent/src/types.ts:144-173】

这个包对仓库内其他包的唯一导入目标是 @earendil-works/pi-ai；其余 ./ 导入都留在自身实现中。 【packages/agent/src/types.ts:1-16】

上层可以直接用低层 agentLoop，也可以用会维护状态并等待订阅者的 Agent。 【packages/agent/src/agent-loop.ts:31-53】 【packages/agent/src/agent.ts:522-576】

根入口还把会话、工具、压缩等 Harness 一并导出，因此它既是小型 Agent runtime，也是构建 coding-agent 所需的通用底座。 【packages/agent/src/index.ts:3-50】

核心文件速查

文件	一句话职责	必读优先级
src/types.ts	定义消息、工具、循环配置、状态与事件这一套跨文件契约。 【packages/agent/src/types.ts:28-437】	⭐⭐⭐
src/agent.ts	提供有状态的 Agent API，把循环事件归约为状态并按顺序等待监听器。 【packages/agent/src/agent.ts:171-576】	⭐⭐⭐
src/agent-loop.ts	实现模型调用、工具批处理、转向消息和后续消息的实际循环。 【packages/agent/src/agent-loop.ts:155-275】	⭐⭐⭐
src/stream-fn.ts	存放可选的默认 StreamFn，避免 core 自己绑定 provider。 【packages/agent/src/stream-fn.ts:3-20】	⭐⭐
src/proxy.ts	把服务端 SSE 代理流重新组装成 AssistantMessageEventStream。 【packages/agent/src/proxy.ts:116-232】	⭐⭐
src/index.ts	根包 barrel：导出 Agent、loop、Harness、proxy、类型和压缩能力。 【packages/agent/src/index.ts:1-50】	⭐⭐
src/node.ts	Node 子路径入口；额外导出 NodeExecutionEnv，再转出根入口。 【packages/agent/src/node.ts:1-2】	⭐
src/harness/agent-harness.ts	把 loop 接到 session、资源、provider hooks、压缩与树状会话上。 【packages/agent/src/harness/agent-harness.ts:171-198】	⭐

精读

1. 先读 types.ts：边界不是 Agent，而是这些数据契约

StreamFn 的签名接收 Model<Api>、Context 和 SimpleStreamOptions，返回 AssistantMessageEventStream 或其 Promise。 【packages/agent/src/types.ts:28-32】

这里的关键约束是：请求、模型或运行时失败不能靠 throw/reject 交给 loop，而要在返回流中用协议事件和终态 AssistantMessage 表示。 【packages/agent/src/types.ts:18-27】

这让 loop 可以把 provider 错误和正常流放进同一条 message_start / message_end / turn_end 事件路径。 【packages/agent/src/agent-loop.ts:346-371】

ToolExecutionMode 只有 "sequential" 与 "parallel"；后者的“并行”只指通过前置检查后的执行阶段。 【packages/agent/src/types.ts:34-42】

QueueMode 的 "all" 会一次清空队列，"one-at-a-time" 只取最旧的一条，因此队列语义是配置的一部分而非 UI 细节。 【packages/agent/src/types.ts:44-50】

AgentMessage 是 pi-ai 的 Message 与声明合并后的 CustomAgentMessages 的联合。 【packages/agent/src/types.ts:296-319】

所以应用可以把 UI 通知、会话摘要之类的消息放进同一条 transcript，而不用伪装成 LLM 原生消息。 【packages/agent/src/types.ts:314-319】

但模型只见到 convertToLlm(messages) 的返回值；该回调负责滤掉或投影应用自定义消息。 【packages/agent/src/types.ts:144-173】

AgentContext 很小：系统提示、消息数组和可选工具数组。 【packages/agent/src/types.ts:405-413】

不要把它误解为 AgentState：前者是低层 loop 的一次上下文快照，后者是给宿主读取和修改的长期状态。 【packages/agent/src/types.ts:321-351】 【packages/agent/src/types.ts:405-413】

AgentState.tools 和 AgentState.messages 是访问器；实现可以在赋值时复制数组。 【packages/agent/src/types.ts:321-339】

Agent 的实际实现确实对初始数组和后续 setter 使用 slice()，因此“重新赋数组”不会让调用方持有的顶层数组被 runtime 改写。 【packages/agent/src/agent.ts:67-93】

反过来，读取 getter 后再 push() 操作的是当前内部数组，这一点从 getter 直接返回闭包数组可看出。 【packages/agent/src/agent.ts:70-88】

AgentTool 在 pi-ai 的 Tool 上增加 UI label、可选 prepareArguments、异步 execute 与每工具 executionMode。 【packages/agent/src/types.ts:379-403】

工具失败的正式通道是 execute throw；loop 会把错误转成带 isError 的 tool result。 【packages/agent/src/types.ts:388-394】 【packages/agent/src/agent-loop.ts:697-703】

onUpdate 只能在本次 execute() Promise 未结束时使用；工具完成之后的 update 会被 loop 忽略。 【packages/agent/src/types.ts:371-377】 【packages/agent/src/agent-loop.ts:671-705】

beforeToolCall 拿到已验证的参数、原始 tool-call block、请求它的 assistant message 和当前 context。 【packages/agent/src/types.ts:92-102】

它返回 { block: true, reason? } 时不会执行工具，而是制造一个错误结果；这正适合作为权限或扩展拦截点。 【packages/agent/src/types.ts:55-64】 【packages/agent/src/agent-loop.ts:616-663】

afterToolCall 得到执行前的结果与错误标记，可以替换 content、details、isError、usage 或 terminate。 【packages/agent/src/types.ts:66-90】

这些字段是替换而不是深合并，尤其是 content、details、usage。 【packages/agent/src/types.ts:69-77】

shouldStopAfterTurn 在完整 turn_end 之后运行；它返回 true 时会优雅结束，不会取消本轮已完成的模型或工具工作。 【packages/agent/src/types.ts:207-217】 【packages/agent/src/agent-loop.ts:224-257】

prepareNextTurn 可以替换下一轮的 context、model 和 thinking level，而不是直接就下一次 provider request 使用旧快照。 【packages/agent/src/types.ts:132-142】 【packages/agent/src/types.ts:219-226】

AgentEvent 分为 agent、turn、message、tool execution 四组，UI 和 session 都应该以它为唯一观察协议。 【packages/agent/src/types.ts:415-437】

其中 message_update 只用于流式 assistant，附带原始 AssistantMessageEvent 与当时的部分消息。 【packages/agent/src/types.ts:429-433】

agent_end 表示 loop 不会再发事件，不等于 Agent 已空闲；订阅者仍可能在处理这个最后事件。 【packages/agent/src/types.ts:415-421】

2. 再读 agent.ts：Agent 给低层循环加了状态和背压

defaultConvertToLlm 只透传 user、assistant、toolResult 三种 role。 【packages/agent/src/agent.ts:32-36】

如果宿主使用自定义消息而没有传自己的转换器，它们会留在状态 transcript 中，却不会进入模型请求。 【packages/agent/src/agent.ts:32-36】 【packages/agent/src/types.ts:315-319】

构造时没有初始 model、prompt、thinking level、tools 或 messages 也能建立状态：分别落到 unknown model、空字符串、"off"、空数组、空数组。 【packages/agent/src/agent.ts:47-58】 【packages/agent/src/agent.ts:67-93】

AgentOptions.streamFn 是类型上必填的 stream 函数。 【packages/agent/src/agent.ts:96-121】

运行时构造函数仍会在未给 streamFn 时读取 getDefaultStreamFn()，这是给旧编译消费者的兼容分支。 【packages/agent/src/agent.ts:210-231】

PendingMessageQueue.drain() 正是 QueueMode 的落地：all 复制并清空，另一模式切走第一个元素。 【packages/agent/src/agent.ts:123-157】

steer() 把消息放到 steering queue；followUp() 放到只有本次工作自然停止后才会读取的 follow-up queue。 【packages/agent/src/agent.ts:275-303】

subscribe(listener) 返回取消订阅函数，监听器保存在 Set 中。 【packages/agent/src/agent.ts:171-175】 【packages/agent/src/agent.ts:233-246】

真正重要的是 processEvents()：它先归约内部状态，再按注册顺序 await 每个 listener。 【packages/agent/src/agent.ts:522-576】

因此 message_end 的监听器会先看到 state.messages 已经追加了终态消息。 【packages/agent/src/agent.ts:539-542】

tool_execution_start 与 tool_execution_end 每次都会复制 Set 后更新 pendingToolCalls，避免直接暴露可变集合。 【packages/agent/src/agent.ts:544-556】

assistant 终态消息带 errorMessage 时，turn_end 会把它写入 state.errorMessage。 【packages/agent/src/agent.ts:558-562】

abort() 只触发当前 run 的 AbortController。 【packages/agent/src/agent.ts:306-314】

它不自行制造 agent_end，后续的结束事件仍由 loop 的模型流与循环控制产生。 【packages/agent/src/agent.ts:311-314】 【packages/agent/src/agent-loop.ts:196-200】

waitForIdle() 等的是 activeRun.promise，而这个 Promise 直到 finishRun() 才 resolve。 【packages/agent/src/agent.ts:316-323】 【packages/agent/src/agent.ts:514-520】

由于 finishRun() 位于 runWithLifecycle() 的 finally，await prompt() 和 waitForIdle() 都包含 listener 的完成时间。 【packages/agent/src/agent.ts:471-493】 【packages/agent/src/agent.ts:569-576】

reset() 清空 transcript、流式消息、pending tools、错误和两个队列，但它没有替当前 run 调用 abort。 【packages/agent/src/agent.ts:325-334】

prompt() 在已有 active run 时直接 throw，调用方应改用 steer()、followUp() 或等待空闲。 【packages/agent/src/agent.ts:336-347】

字符串 prompt 被标准化为带时间戳的 user 消息，图片追加在同一个 content 数组后。 【packages/agent/src/agent.ts:379-396】

continue() 要求最后一条 context message 不是 assistant；如果是 assistant，它会优先抽取已排队的 steering，再抽 follow-up，否则报错。 【packages/agent/src/agent.ts:349-377】

runPromptMessages() 调用 runAgentLoop()，并把每个低层 event 交给 processEvents()。 【packages/agent/src/agent.ts:398-412】

runContinuation() 同样接线到 runAgentLoopContinue()，但不追加新的用户消息。 【packages/agent/src/agent.ts:414-424】

每次开始前 createContextSnapshot() 会复制 messages 和 tools 数组，所以单轮 loop 不会持有状态对象的原数组。 【packages/agent/src/agent.ts:426-432】

createLoopConfig() 把 Agent 的 model、thinking、sessionId、transport、hooks 和队列 drain 函数逐项传给 loop。 【packages/agent/src/agent.ts:434-468】

prepareNextTurnWithContext 优先于旧式 prepareNextTurn，并接收当前 run 的 abort signal。 【packages/agent/src/agent.ts:448-456】

一次 run 开始会设定 isStreaming、清空 partial message 和旧错误。 【packages/agent/src/agent.ts:471-486】

如果 loop 自身 throw，handleRunFailure() 仍补发 assistant 错误消息、turn_end 和 agent_end，从而保持上层事件协议完整。 【packages/agent/src/agent.ts:496-512】

给下一章的 coding-agent 预告：会话层应监听完整 AgentEvent 序列，尤其是 message_end、turn_end、agent_end，并在工具前后挂 beforeToolCall / afterToolCall。 【packages/agent/src/types.ts:422-437】 【packages/agent/src/types.ts:265-286】

现有 packages/coding-agent 的 AgentSession 正是在构造期订阅 agent，并安装这两个工具 hook。 【packages/coding-agent/src/core/agent-session.ts:391-395】 【packages/coding-agent/src/core/agent-session.ts:460-514】

3. 最后读入口：决定什么是公开 API

index.ts 先导出 Agent 与 loop，再导出 AgentHarness。 【packages/agent/src/index.ts:1-6】

根入口还公开 branch summary、compaction、session backends、skills、system prompt、内建工具和 Harness 类型。 【packages/agent/src/index.ts:7-44】

streamProxy 和相关 proxy 类型也从根入口导出。 【packages/agent/src/index.ts:45-46】

setDefaultStreamFn 是公开的，但 getDefaultStreamFn 仅供内部的 Agent 与 loop 兜底使用。 【packages/agent/src/index.ts:47-50】 【packages/agent/src/stream-fn.ts:11-20】

uuidv7 直接从 pi-ai 转出，方便 Harness/session 使用同一 ID 工具。 【packages/agent/src/index.ts:1-3】

node.ts 不改动 API，只在根入口外再添加含 Node 内建依赖的 NodeExecutionEnv。 【packages/agent/src/node.ts:1-2】

这就是 core 能留给浏览器、代理或其他 runtime 使用，而 Node 执行环境改从 @earendil-works/pi-agent-core/node 取得的原因。 【packages/agent/src/node.ts:1-2】 【packages/agent/src/harness/env/nodejs.ts:1-31】

从源代码导入看，Agent 的统一模型、消息、stream、工具参数校验和 UUID 都来自 pi-ai。 【packages/agent/src/types.ts:1-15】 【packages/agent/src/agent-loop.ts:6-12】 【packages/agent/src/index.ts:1-3】

因此 packages/agent 不挑 Anthropic、OpenAI 或其他 provider；它只依赖已经统一好的 pi-ai 抽象。 【packages/agent/src/types.ts:28-32】 【packages/agent/src/agent-loop.ts:294-312】

ThinkingLevel 的 "xhigh" 与 "max" 不该无条件发给所有模型，具体支持性要由 pi-ai 的模型 metadata 判断。 【packages/agent/src/types.ts:289-294】

AgentOptions 还透传 sessionId、thinking budgets、transport 和最大 provider retry delay；这些都作用于未来请求。 【packages/agent/src/agent.ts:96-121】 【packages/agent/src/agent.ts:434-459】

sessionId 的意义是让支持缓存的 provider 识别同一会话，不是 Agent 自己保存 transcript 的主键。 【packages/agent/src/agent.ts:199-208】 【packages/agent/src/agent.ts:434-445】

订阅返回的函数只删除对应 listener；它不会清理 agent state 或终止正在进行的 run。 【packages/agent/src/agent.ts:243-246】

对 listener 而言，抛错会从 processEvents() 冒出，再由 runWithLifecycle() 转换成失败 assistant 事件序列。 【packages/agent/src/agent.ts:487-512】 【packages/agent/src/agent.ts:569-576】

这意味着 listener 本身也应把可恢复 UI/持久化故障处理好，避免把正常模型输出改写成整个 run 的 error。 【packages/agent/src/agent.ts:496-512】

Agent 不自行执行 provider SDK；它只把保存的 streamFunction 和构造好的 config 传入低层 loop。 【packages/agent/src/agent.ts:398-410】

因此替换模型 transport、代理或 provider 实现时，通常替换 StreamFn，而不是重写队列和工具循环。 【packages/agent/src/types.ts:18-32】 【packages/agent/src/agent.ts:177-208】

数据流

一次 await agent.prompt("...") 先经过 normalizePromptInput()，生成 user 的 text/image content 与时间戳。 【packages/agent/src/agent.ts:336-347】 【packages/agent/src/agent.ts:379-396】

随后 runPromptMessages() 创建 state snapshot 和 loop config，并把 loop 的每个事件交给 processEvents()。 【packages/agent/src/agent.ts:398-432】

processEvents() 先更新 streamingMessage、messages、pending tools 或错误，再依次 await 所有 listener。 【packages/agent/src/agent.ts:529-576】

所以 session/UI 的典型顺序是：message_start(user) → message_end(user) → assistant 的 message_start / 多个 message_update / message_end。 【packages/agent/src/agent-loop.ts:109-116】 【packages/agent/src/agent-loop.ts:317-371】

assistant 有 tool call 时，后续会收到 tool_execution_start、可选 tool_execution_update、tool_execution_end，再收到每条 toolResult 的 start/end。 【packages/agent/src/agent-loop.ts:444-486】 【packages/agent/src/agent-loop.ts:671-705】 【packages/agent/src/agent-loop.ts:773-792】

当这一轮结束，loop 发 turn_end；当无工具、steering 与 follow-up 消息可继续时，才发 agent_end。 【packages/agent/src/agent-loop.ts:224-274】

Agent 在 agent_end listener 全部结束后才由 finishRun() 把 isStreaming 置回 false 并解析 waitForIdle()。 【packages/agent/src/agent.ts:471-493】 【packages/agent/src/agent.ts:514-520】

自测

1. 为什么 AgentMessage 不能直接假定等于发给模型的 Message？ 【packages/agent/src/types.ts:144-173】

2. message_end listener 执行时，为什么可以安全读取刚写入的 agent.state.messages？ 【packages/agent/src/agent.ts:539-576】

3. steer() 与 followUp() 分别在哪个循环停靠点被 drain？ 【packages/agent/src/agent.ts:275-303】 【packages/agent/src/agent-loop.ts:259-274】

4. 为什么 agent_end 出现后 waitForIdle() 仍可能尚未 resolve？ 【packages/agent/src/types.ts:415-421】 【packages/agent/src/agent.ts:514-576】

5. 想让 Agent 默认使用宿主的 stream 实现，应该设置哪个 API，而不是在 core 中加入 provider？ 【packages/agent/src/stream-fn.ts:5-20】