pi-agent-core:agent 循环与工具调用
30 秒导读:
@earendil-works/agent这个包,是 Pi 项目名所指的那颗「心脏」——一个与 provider 无关的 agent 运行时循环。你喂给它一段 prompt、一组工具、一个 model,它就反复地「让模型说话 → 执行模型点名的工具 → 把结果喂回去」,直到模型不再要工具为止;期间还允许你中途插话(steering)和排队追问(follow-up)。 本章只讲这颗心脏本身:纯循环 + 工具编排 + 事件流。会话/压缩/系统提示见 03 章, 具体工具实现见 04 章,底层统一 LLM 层见 01 章。
1. 这是什么(零基础也能懂)
一句话定义: 一段「让 LLM 反复调用工具直到完成任务」的循环代码,加上一个有状态的门面类 Agent。
它解决什么问题。 单次调用 LLM 只能得到一段文字或一个「我想调用 read_file」的请求——它自己没有手,
执行不了。要让模型真正「干活」,你得写一个循环:把模型的工具请求真正执行掉,再把结果塞回对话,让它接着想
下一步。这段循环 90% 的项目都在重复造轮子,而且容易写错(并发、中止、错误处理、插话)。agent 包把这段
循环抽成一个可复用、provider 无关、事件驱动的库。
给谁用。 想自己搭一个编码 agent / 自动化 agent 的开发者。Pi 的编码 agent(04 章)就是在这颗心脏外面 套了工具集、会话、CLI。
它能做什么:
- 驱动「助手回复 ↔ 工具执行」的多轮循环,直到模型停止要工具。
- 工具批次可串行或并行执行,且可按工具粒度覆盖。
- 支持三种「插话」时机:steering(干活途中插)、follow-up(本要停下时追问)、以及一批钩子回调。
- 全程发事件流(agent/turn/message/tool 各级生命周期),UI 可实时渲染。
- 与 provider 解耦:真正调模型的那一步是一个可替换的
streamFn(默认走 01 章的streamSimple, 也可换成streamProxy走自己的服务器)。
用起来什么样。 门面类 Agent 让最小用法只有几行(下面是示意,非源码):
// 示意,非源码 —— 展示 Agent 门面最小用法
const agent = new Agent({
initialState: { systemPrompt, model, tools }, // 系统提示 + 模型 + 工具
});
agent.subscribe((event) => render(event)); // 订阅事件流做 UI
await agent.prompt("把 src 下的测试都跑一遍"); // 发一句话,循环自动转起来
// 干活途中想补一句 → agent.steer(...); 想等它停下再问 → agent.followUp(...)
一句话直觉。 把它想成一台传送带:模型放上一个「我要调 X 工具」的零件,循环把零件加工(执行工具)、 再放回传送带让模型接着看;传送带一圈圈转,直到模型说「我不需要更多零件了」。你可以在传送带转的时候往上 丢新零件(steering),也可以等它空转时再补料(follow-up)。
2. 顶层全景(它大概怎么转)
整个包围绕一个数据类型和一个双层循环展开。
2.1 部件一句话职责
| 部件 | 干什么 | 在哪 |
|---|---|---|
Agent 门面类 | 有状态封装:持有 transcript、发 事件、管两个插话队列、暴露 prompt/continue/steer/followUp | agent.ts:166 |
runLoop | 无状态的纯循环核心:双层 while,编排一切 | agent-loop.ts:155 |
agentLoop / agentLoopContinue | 两个入口:带新 prompt 起循环 / 从现有 context 续跑 | agent-loop.ts:31,64 |
streamAssistantResponse | 一次「让模型说话」:AgentMessage[]→Message[] 的转换边界就在这 | agent-loop.ts:275 |
executeToolCalls | 执行一批工具调用(串行 / 并行) | agent-loop.ts:373 |
EventStream | 通用异步事件流,循环把事件 push 进去、消费者 for await 出来 | packages/ai/src/utils/event-stream.ts:4 |
streamFn(默认 streamSimple,或 streamProxy) | 真正调 provider 的可替换函数 | types.ts:27、proxy.ts:116 |
2.2 数据是怎么流的
关键设计:循环内部一路用 AgentMessage,只有在调模型的前一刻才转成 provider 认识的 Message。
这让 app 能往对话里塞自定义消息(UI 通知、状态卡)而不污染发给模型的内容。
怎么读下图:从上到下是一次 prompt 的主干;右侧标了「谁在这一步插得进话」。
agent.prompt("...")
│
┌─────────▼──────────────────────────────────────────┐
│ runLoop 外层循环(follow-up 层) │
│ ┌──────────────────────────────────────────────┐ │
│ │ 内层循环(工具批次 + steering 层) │ │
│ │ │ │
│ │ ① 注入 pendingMessages ───────◄── steering │ │
│ │ ② streamAssistantResponse │ │
│ │ AgentMessage[] ──convertToLlm──► Message[]│ │
│ │ └──► streamFn ──► provider(01 章) │ │
│ │ ③ 过滤出 toolCall │ │
│ │ ④ executeToolCalls(串行/并行)→ toolResults │ │
│ │ ⑤ turn_end / prepareNextTurn │ │
│ │ ⑥ 还有工具? 或 有 steering? ──是──► 回 ① │ │
│ └──────────────────┬────────────────────────────┘ │
│ 否(本要停) │ │
│ ⑦ getFollowUpMessages ──有──► 设为 pending,回内层 ◄─ follow-up
│ 无 │ │
└─────────────▼───────────────────────────────────────┘
agent_end
主线走一遍(高层): 发 prompt → 进内层循环 → 模型说话 → 若点名工具就执行、结果塞回 → 只要还有工具或有人
steering 就继续转 → 内层空了看有没有 follow-up → 有就再进内层,没有就 agent_end 收尾。
3. 核心数据类型(循环搬运的都是这些)
先认清类型,后面的循环才好懂。全部在 packages/agent/src/types.ts。
3.1 AgentMessage:循环的「货物」
// types.ts:314
export type AgentMessage = Message | CustomAgentMessages[keyof CustomAgentMessages];
AgentMessage = provider 的标准 Message(user / assistant / toolResult)并上 app 自定义消息。app 通过
TypeScript 的声明合并往 CustomAgentMessages 接口里加自己的消息类型(types.ts:305)。这就是「循环内一路
用 AgentMessage」的意义:通知、状态卡都能进 transcript,但发给模型前会被 convertToLlm 过滤掉(见 §5.1)。
3.2 AgentTool / AgentToolResult:工具长什么样
AgentTool(types.ts:371)在 01 章的 Tool(名字 + TypeBox 参数 schema)之上,加了:
| 字段 | 作用 |
|---|---|
label | UI 显示用的人类可读名 |
execute(id, params, signal?, onUpdate?) | 真正干活;失败要 throw,不要把错误编码进 content |
prepareArguments? | schema 校验前对原始 args 的兼容性修正 |
executionMode? | 按工具覆盖串行/并行(见 §6) |
execute 返回 AgentToolResult<T>(types.ts:350):content(回给模型的文字/图像)+ details(给日志/UI
的结构化数据)+ 可选 terminate(提示这批工具跑完就该停,见 §6.3)。onUpdate 回调让长工具流式吐进度。
3.3 AgentContext / AgentLoopConfig:一次运行的两半
AgentContext(types.ts:397):运行的数据——systemPrompt+messages+tools。AgentLoopConfig(types.ts:140):运行的行为——model、那些回调钩子、convertToLlm、toolExecution等。它extends SimpleStreamOptions(温度、maxTokens 等直接透传给 provider)。
3.4 AgentEvent:循环对外广播的一切
AgentEvent(types.ts:413)是一个可辨识联合,按四级生命周期分组:
| 级别 | 事件 |
|---|---|
| agent | agent_start / agent_end(带最终 messages) |
| turn(一次助手回复 + 其工具) | turn_start / turn_end(带 message 和 toolResults) |
| message | message_start / message_update(仅流式助手消息)/ message_end |
| tool | tool_execution_start / tool_execution_update / tool_execution_end |
3.5 三个小枚举
ThinkingLevel(types.ts:289):off/minimal/low/medium/high/xhigh,推理档位。ToolExecutionMode(types.ts:41):"sequential" | "parallel",工具批次执行策略。QueueMode(types.ts:49):"all" | "one-at-a-time",排队消息一次注入几条(见 §7)。
4. 循环结构:双层 while 与 firstTurn / pendingMessages
这是全章的骨。所有编排都在 runLoop(agent-loop.ts:155)里。
4.1 两个入口:起新 prompt vs 续跑
两个公开入口都只是给 runLoop 铺好初始 context,再 .then 把最终 newMessages 塞进 EventStream:
agentLoop(prompts, ...)(agent-loop.ts:31→runAgentLoop:95):把新 prompt 拼进 context,先发agent_start+turn_start,再为每条 prompt 发message_start/end(agent-loop.ts:109-114)。newMessages以[...prompts]起步(:103)——它是本次调用的返回值,只累积「这次新产生的消息」。agentLoopContinue(context, ...)(agent-loop.ts:64→runAgentLoopContinue:120):不加新消息,直接从 现有 context 续跑(用于重试)。它先断言:context 非空、且最后一条不是 assistant 消息 (agent-loop.ts:70-77)——因为 provider 要求请求以 user/toolResult 结尾,否则会被拒。
4.2 外层循环 = follow-up 层,内层循环 = 工具 + steering 层
runLoop 的两个 while 各管一件事:
外层 while (true) ← 处理「 本该停下,但有 follow-up 排队」
内层 while (hasMoreToolCalls ← 只要还有工具要跑
|| pendingMessages>0) ← 或者有 steering 消息待注入
…一个 turn…
// 内层退出后:
followUp 有? → 设 pending,continue 回内层
没有 → break 出外层 → agent_end
内层的续跑条件(agent-loop.ts:174):hasMoreToolCalls || pendingMessages.length > 0。前者来自「上一
turn 模型是否点了工具」,后者来自「是否有 steering 消息要插」。只要任一为真,就再转一圈。
外层(agent-loop.ts:170)只在内层彻底空转后才起作用:去问 getFollowUpMessages,有就把它设成
pendingMessages 并 continue 重进内层(:257-261),没有就 break(:265)、发 agent_end(:268)。
4.3 firstTurn:别把开场的 turn_start 发两遍
入口 runAgentLoop 已经发过一次 turn_start 了。runLoop 里 firstTurn 标志(agent-loop.ts:165)保证
第一圈不再重发 turn_start,之后每圈开头才发(:175-179)。这是个小而关键的去重,避免 UI 收到重复的
turn 边界。
4.4 pendingMessages 是 steering 的落点
pendingMessages 在循环启动前就先 poll 一次 steering(agent-loop.ts:167,注释:用户可能在等待时已经打字)。
每圈开头,若 pendingMessages 非空,就逐条发 message_start/end 并推进 context 和 newMessages
(:182-190),然后清空。每个 turn 跑完再 poll 一次(:253)。这样「插话」总是在下一次模型说话之前落地。
5. 一次 turn 的生命周期
一个 turn = 一次助手回复 + 它点名的工具全部执行完。拆开看四步。
5.1 streamAssistantResponse:AgentMessage → Message[] 的唯一转换边界
streamAssistantResponse(agent-loop.ts:275)是「让模型说一次话」的地方,也是整个包唯一做类型转换的点:
- 若配了
transformContext,先在AgentMessage层做变换(如压缩/剪枝,agent-loop.ts:284)。 convertToLlm(messages)(agent-loop.ts:289)把AgentMessage[]转成 provider 的Message[]—— UI-only 的自定义消息在这被过滤掉。默认实现只留 user/assistant/toolResult 三种(agent.ts:31defaultConvertToLlm)。- 拼出
Context(systemPrompt + 转换后的 messages + tools,:292),解析 API key(为过期 token 每次动态取,:301),调streamFn(默认streamSimple,:298)。 - 消费返回的事件流:
start时把 partial 消息推进 context,后续 delta 事件原地替换并发message_update(:313-340);done/error时用response.result()取最终消息、发message_end、返回(:342-355)。
契约(重要):
convertToLlm和transformContext绝不能 throw——注释明确要求返回安全兜底值 (types.ts:169,191)。抛异常会打断底层循环、破坏正常事件序列。
5.2 过滤工具调用
拿到最终助手消息后,循环先看它是不是失败:stopReason === "error" || "aborted" 就发 turn_end + agent_end
直接 退出(agent-loop.ts:196-200)。否则从 message.content 里过滤出 type === "toolCall" 的块
(:203)——这些就是模型点名要调的工具。
5.3 executeToolCalls:串行 / 并行 / terminate
有工具就交给 executeToolCalls(agent-loop.ts:208)。返回的 terminate 决定 hasMoreToolCalls——
hasMoreToolCalls = !terminate(:210),即工具批次要求终止时,内层不再因「还有工具」而继续。执行完的
toolResults 逐条推进 context 和 newMessages(:212-215),再发 turn_end(:218)。执行细节见 §6。
5.4 turn 收尾的三个钩子
一个 turn 的 turn_end 发完后,循环依次问三个可选回调(全部要求不 throw):
| 钩子 | 时机 | 作用 | 位置 |
|---|---|---|---|
prepareNextTurn | turn_end 后、决定是否再请求前 | 返回替换的 context / model / thinkingLevel,影响下一 turn | agent-loop.ts:226 |
shouldStopAfterTurn | 紧接其后 | 返回 true 就发 agent_end 退出,不再 poll steering/follow-up | :242 |
getSteeringMessages | 若没停 | 取本 turn 期间攒下的 steering 消息,设成下圈的 pendingMessages | :253 |
prepareNextTurn 的合并很细:thinkingLevel === "off" 会把 reasoning 置 undefined,否则原样替换
(agent-loop.ts:232-238)。shouldStopAfterTurn 是「优雅停」的正道——比如上下文快满时主动停(见 03 章压缩)。
6. 工具执行的内部机制
executeToolCalls(agent-loop.ts:373)先决定用哪种模式,再走对应实现。
6.1 模式选择:一票否决制
executeToolCalls
│
├─ config.toolExecution === "sequential" ? ─┐
├─ 或 任一工具 executionMode === "sequential" ?
│ （hasSequentialToolCall） │
│ ▼
│ executeToolCallsSequential
└─ 否则 ──────────────────────► executeToolCallsParallel
关键:只要批次里有一个工具声明了 executionMode: "sequential",整批就走串行(agent-loop.ts:381-386
hasSequentialToolCall)。这是「一票否决」——某个危险工具(比如写文件)可以强制整批不并发。
6.2 一个工具调用的三段式:prepare → execute → finalize
无论串并行,单个工具都走同样三步:
prepareToolCall(agent-loop.ts:562):找工具(找不到→立即错误结果:570),跑prepareArguments兼容修正,validateToolArguments按 TypeBox schema 校验(:580),再跑beforeToolCall钩子——钩子返回{block:true}就拦截、产出错误 结果(:598-604)。任何异常都被兜成 immediate 错误结果,不会冒泡打断循环。executePreparedToolCall(agent-loop.ts:628):真正tool.execute(...),把onUpdate回调接成tool_execution_update事件;工具 throw 会被兜成错误结果(:659-665)。finalizeExecutedToolCall(agent-loop.ts:671):跑afterToolCall钩子,按字段覆盖 result 的content/details/terminate/isError(:695-702,无深合并,整字段替换)。
串行(executeToolCallsSequential:395):for 循环里一个个 prepare→execute→finalize,每个跑完立刻发
tool_execution_end + 造 toolResult 消息;signal.aborted 时提前 break(:440)。
并行(executeToolCallsParallel:451):先顺序 preflight(prepare + emit start),把需要执行的包成
闭包塞进数组,再 Promise.all 并发跑(:502-504)。注意事件时序:tool_execution_end 按完成顺序发,
而 toolResult 消息按助手源顺序发(:505-510)——保证喂回模型的消息顺序稳定、可复现。
6.3 terminate:全票才停
shouldTerminateToolBatch(agent-loop.ts:544):当且仅当批次非空、且每个 finalized 结果都
terminate === true 才返回 true。也就是说「早停」是全票制——只要有一个工具没喊 停,循环就继续。这防止
单个工具误伤整批。
6.4 错误的编码方式
失败不靠异常冒泡,而是产出一条 toolResult 消息塞回对话:createErrorToolResult(:716)造 content,
createToolResultMessage(:733)带 isError 标志。模型能看到错误文本、自行纠错重试。这是 agent 循环
「容错」的核心手法——错误是数据,不是控制流中断。
7. Agent 门面类:把纯循环变成有状态对象
runLoop 是无状态的;真正给 app 用的是 Agent 类(agent.ts:166),它在纯循环外面包了状态与队列。
7.1 它多管了什么
| 职责 | 实现 |
|---|---|
| 持有 transcript / model / tools 等可变状态 | MutableAgentState,tools/messages 用 accessor,赋值时 .slice() 拷贝(agent.ts:76-87) |
| 两个插话队列 | steeringQueue / followUpQueue, 均为 PendingMessageQueue(:169,118) |
| 对外 API | prompt / continue / steer / followUp / abort / waitForIdle / reset |
| 把 loop 事件 reduce 进状态并转发订阅者 | processEvents(:509) |
| 生命周期管理 | runWithLifecycle(:451)开 AbortController、置 isStreaming、兜异常 |
7.2 steering vs follow-up 两个队列
steer(msg)(agent.ts:264)入steeringQueue:干活途中插,内层循环每 turn 后 drain。followUp(msg)(agent.ts:269)入followUpQueue:等 agent 本要停下时才 drain(外层循环)。
QueueMode 控制一次 drain 几条:"all" 全倒,"one-at-a-time" 只倒最旧一条、其余留到下个 drain 点
(PendingMessageQueue.drain:134)。两个队列默认都是 "one-at-a-time"(agent.ts:212-213)。Agent
把队列包成 getSteeringMessages/getFollowUpMessages 传给 createLoopConfig(agent.ts:440-447),纯循环
只认这两个回调、不知道队列的存在——干净的解耦。
7.3 门面如何接住失败
runWithLifecycle(agent.ts:451)里 executor 抛出时走 handleRunFailure(:476):造一条 stopReason
为 aborted(若 signal 已 abort)或 error 的空助手消息,补发 message_start/end + turn_end + agent_end
(:488-491)。所以即便底层炸了,订阅者也总能收到一套完整的收尾事件。finishRun(:494)最后清 streaming
状态、resolve waitForIdle 的 promise。
7.4 一个并发保护
prompt(agent.ts:327)开头就查 activeRun,已在跑就抛错,提示改用 steer()/followUp()(:328-332)。
continue(:338)同理。这保证同一时刻只有一个 run 在动,插话必须走队列。
8. 事件流机制与中止
8.1 EventStream:push / for-await 的桥
EventStream<T, R>(packages/ai/src/utils/event-stream.ts:4)是个通用异步流:生产者 push 事件、消费者
for await 消费;构造时给两个谓词——「哪个事件算完成」和「从完成事件里抽最终结果」。agent 用它把
agent_end 认作完成、抽出 messages(agent-loop.ts:145 createAgentStream)。同一个类也被 01 章的
AssistantMessageEventStream 和 proxy 复用。
8.2 emit:事件先 reduce 再等监听器
Agent.processEvents(agent.ts:509)是事件的汇聚点:先按事件类型更新内部状态(streamingMessage、
pendingToolCalls 增删、message_end 时把消息推进 transcript、turn_end 时记 errorMessage),再按订阅
顺序 await 每个监听器(:553-555)。因为是 await 的,agent_end 的监听器也算进本次 run 的「结算」——
状态注释特别强调:agent 要等 agent_end 监听器全 settle 后才真正 idle(types.ts:340,411)。
8.3 中止与 stopReason
中止有两条路汇到同一处:
- 模型侧:助手消息回来时
stopReason已是aborted/error→ 循环发turn_end+agent_end退出 (agent-loop.ts:196-200)。 - signal 侧:
agent.abort()(agent.ts:301)触发 AbortController;prepareToolCall里signal.aborted会把工具变成 immediate 的 "Operation aborted" 错误结果(agent-loop.ts:591-596,606-611), 串行执行也会提前 break(:440)。
两者都不靠抛异常穿透循环,而是收敛成明确的终止事件序列——这是整个包对「可预测收尾」的坚持。
8.4 tool 参数校验:validateToolArguments
validateToolArguments(packages/ai/src/utils/validation.ts:292)在执行前把工具 args 按 TypeBox schema
Value.Convert 强转再 Check,失败就抛出带逐字段路径的详细错误(:315-323)。循环在 prepareToolCall
里调它(agent-loop.ts:580),抛错被就地兜成错误 toolResult——于是「模型给了不合法参数」也变成模型能读到、
能自我纠正的一条对话消息,而不是崩溃。
9. 巧妙之处(可借鉴)
- 单一转换边界。 全程用
AgentMessage,只在streamAssistantResponse里convertToLlm一次 (agent-loop.ts:289)。app 想加 UI-only 消息,改convertToLlm过滤即可,循环其余部分零改动。 - 纯循环 + 门面的分层。
runLoop无状态、只认回调;Agent持状态、管队列。测试可直接喂runLoop假回调,不需要整个Agent。 - 错误即数据。 工具失败、参数非法、被拦截、被中止——全部收敛成
toolResult消息或明确的终止事件, 从不让异常穿透循环(agent-loop.ts:619-625,659-665)。 - 一票否决的并发 + 全票的终止。 串行由「任一工具要求」触发,终止由「所有工具要求」触发——两个方向的
阈值刻意相反,各自防的是不同的误伤(
:381,544)。 - 并发但顺序稳定。 并行执行,却让 toolResult 消息按助手源顺序回灌(
:505-510),保证重放/缓存一致。
10. 边界与局限(诚实)
- 本包不管会话持久化、上下文压缩、系统提示组装、Skills——那些在
harness/,是 03 章的内容。 - 本包不含任何具体工具实现(读文件、跑 bash 等)——那是 04 章
pi-coding-agent的事。本包只定义AgentTool接口和执行编排。 - 不真正调 provider——真正的网络/协议在 01 章;本包通过可替换的
streamFn解耦。 agentLoopContinue有前置约束:最后一条消息必须能转成 user/toolResult,否则 provider 拒绝——而这 无法在此校验,因为convertToLlm每 turn 才调一次(agent-loop.ts:64-77注释明说)。- 回调契约靠约定,不靠类型:
convertToLlm/transformContext/各钩子「不得 throw」是文档契约 (types.ts:169等),类型系统不强制;违反会打断底层循环。
11. 横向对比(同 shelf)
同为「agent 运行时」,Pi 的这颗心脏最鲜明的取舍是provider 无关 + 事件驱动 + AgentMessage 抽象:它不像 某些编码 agent 把循环和工具、会话焊死,而是把循环剥成一个独立包,工具/会话/CLI 都在外面按需组装(见 04 章)。「steering(途中插话)vs follow-up(停下追问)」这对显式区分,以及 「串行一票否决 / 终止全票」这类阈值设计,是它相对朴素 while 循环的精细处。底层 LLM 抽象见 01 章,UI 侧的差分渲染见 05 章。
12. 代码地图(导航索引)
按符号名 grep 比按行号更抗漂移。
| 主题 | 文件 | 符号 |
|---|---|---|
| 双层循环核心 | packages/agent/src/agent-loop.ts | runLoop |
| 起新 prompt 入口 | packages/agent/src/agent-loop.ts | agentLoop / runAgentLoop |
| 从现有 context 续跑 | packages/agent/src/agent-loop.ts | agentLoopContinue / runAgentLoopContinue |
| 让模型说话 + 转换边界 | packages/agent/src/agent-loop.ts | streamAssistantResponse |
| 工具批次执行分派 | packages/agent/src/agent-loop.ts | executeToolCalls |
| 串行 / 并行执行 | packages/agent/src/agent-loop.ts | executeToolCallsSequential / executeToolCallsParallel |
| 工具三段式 | packages/agent/src/agent-loop.ts | prepareToolCall / executePreparedToolCall / finalizeExecutedToolCall |
| 全票终止判定 | packages/agent/src/agent-loop.ts | shouldTerminateToolBatch |
| 有状态门面类 | packages/agent/src/agent.ts | Agent |
| 默认转换器 | packages/agent/src/agent.ts | defaultConvertToLlm |
| 插话队列 | packages/agent/src/agent.ts | PendingMessageQueue |
| 事件 reduce + 转发 | packages/agent/src/agent.ts | processEvents |
| 生命周期 / 失败兜底 | packages/agent/src/agent.ts | runWithLifecycle / handleRunFailure |
| 核心类型 | packages/agent/src/types.ts | AgentMessage / AgentTool / AgentContext / AgentLoopConfig / AgentEvent |
| 代理 streamFn | packages/agent/src/proxy.ts | streamProxy |
| 通用事件流 | packages/ai/src/utils/event-stream.ts | EventStream |
| 工具参数校验 | packages/ai/src/utils/validation.ts | validateToolArguments |