源码篇第4章双层循环——pi-agent-core

小A在内层传送带处理工具结果，老Z守着外层闸门，等待内层循环结束后放入待处理的信封。

本章导读： 源码路线第 ③ 站（核心层）。从 pi-ai 切到 pi-agent-core 包，读 agent.ts 和 agent-loop.ts。这一站要看的是循环的真面目：模型输出和工具结果怎么一推一拉，把状态一轮轮往前送，以及那个容易被误读的“内外两层循环”到底是怎么回事。

小A读完流适配层，信心满满地跟老Z说：“我懂了。模型流进来，循环就在那儿转：外层管模型轮次，内层管并行跑工具，对吧？”

老Z没直接回答：“你这句话里有几个‘想当然’？”

小A愣住了。

“流适配层只会产出事件；把事件变成上下文中的 assistant message、把 tool call 变成 tool result、再决定是否继续请求的，是 pi-agent-core 的 loop。”老Z说，“你猜的‘外层管模型轮次、内层并行跑工具’就是这一站要拆的误读。答案是否定的，必须按状态变量和分支看。”

本章依据上游 pi 标签 v1.0.0（提交 a13d35a742c6ef8462812a28fbe1d8c8b7431c32）、@earendil-works/pi-agent-core 1.0.0。覆盖 runAgentLoop、runLoop、streamAssistantResponse、executeToolCalls 和 runToolCall；不讨论会话落盘、压缩或 Skills，这些在 v1.0.0 已不属于本包。

在 Agent.continue() 的分支里还要补上一种情况：assistant 结尾且 steering/follow-up 队列皆空时抛错。这一支说明“没有下一步输入”不是静默结束，而是调用方需要处理的错误。

`Agent` 是带队列的宿主包装，不是循环本身

小A还不死心：“那 Agent 类总该就是循环了吧？”

“不是。”老Z说，“packages/agent/src/agent.ts 的 Agent 持有 transcript state、listener 集、steering queue、follow-up queue 和当前 AbortController。它的 prompt() 最终调用 runAgentLoop；continue() 在最后消息不是 assistant 时调用 runAgentLoopContinue。最后消息是 assistant 时，它先取一条 steering，没有再取 follow-up，用取到的消息开一段新 run；两个队列都空才抛错。”

// packages/agent/src/agent.ts，Agent 的队列接口（v1.0.0，节选）
/** Queue a message to be injected after the current assistant turn finishes. */
steer(message: AgentMessage): void {
	this.steeringQueue.enqueue(message);
}

/** Queue a message to run only after the agent would otherwise stop. */
followUp(message: AgentMessage): void {
	this.followUpQueue.enqueue(message);
}

“注意这两个队列的时机不同，不能因为都叫‘用户消息’而合并。”老Z说，“PendingMessageQueue.drain() 还受 QueueMode 控制：all 取全部，否则取一条；默认是 one-at-a-time。drain() 先用 peek() 选出要取的消息再移出，所以宿主也可以用 peekQueuedMessages() 预览下一轮会注入什么，而不消费队列。调用 prompt() 时若已有 active run，源码直接抛错并提示使用 steer() 或 followUp()，避免两个 loop 并发改同一 state。”

API／状态	已确认的作用	不能从中推出
`prompt`	新增用户消息并启动一段 run	可以并行开始多个 run
`continue`	从 user／tool result 结尾继续；assistant 结尾时只消费已排队的 steering／follow-up	assistant 结尾也可无条件继续
`steer`	当前 assistant turn 后、下次请求前注入	立刻打断正在到来的 token
`followUp`	Agent 否则会停止时再注入	每一轮都会插入
`abort`	abort 当前 controller	已经执行的外部副作用自动撤销
`subscribe`	顺序 await listener，传入 signal	listener 是持久化机制

“subscribe 的注释还说明 agent_end 是事件上的末尾，但 Agent 要等该事件的所有 listener settle 后才真正 idle。这个等待关系是观察一致性的一部分；并不等同 session 已经写盘。v1.0.0 的 pi-agent-core 根本不管写盘，那是宿主的职责，例如 coding-agent 的 AgentSession。”

还有一处变化要先记住：v1.0.0 的 AgentContext 只剩 messages 和 tools 两个字段，没有 systemPrompt。system prompt 和模型可见的工具声明都放进 transcript 里的 role: "system" 消息。Agent 构造时用 pi-ai 的 createInitialSystemMessage 把初始 systemPrompt 和 tools 变成首条 system message；state.systemPrompt 只是从 transcript 回放出来的 getter。

`runAgentLoop` 先建立本次增量

小A想直接看循环：“循环函数进来第一件事干嘛？”

“先建立本次增量。”老Z打开 agent-loop.ts：

低层 runAgentLoop 接收 prompts、已有 context、config、event sink、signal、stream function。它先用 declareToolChanges 处理 prompts，再用结果建立本次 newMessages，用历史 messages 加这批消息构造 currentContext，发出 agent_start、首个 turn_start 和每条消息的 start/end 事件，然后进入 runLoop。

// packages/agent/src/agent-loop.ts，runAgentLoop（v1.0.0，节选）
const initialMessages = declareToolChanges(context, prompts);
const newMessages: AgentMessage[] = [...initialMessages];
const currentContext: AgentContext = {
	...context,
	messages: [...context.messages, ...initialMessages],
};
await emit({ type: "agent_start" });
await emit({ type: "turn_start" });
for (const message of initialMessages) {
	await emit({ type: "message_start", message });
	await emit({ type: "message_end", message });
}
await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn());

“这里有两个列表，别混。”老Z强调，“currentContext.messages 是下一次模型调用可见的完整工作上下文，newMessages 是本次 run 新增加、最后随 agent_end 返回的增量。混淆二者会错误解释为什么 prompts 既被 emit 又会进入模型上下文。”

小A指着第一行：“declareToolChanges 是干嘛的？”

“它比较两份工具清单。”老Z说，“context.tools 是运行时能执行的工具；transcript 里 system message 回放出来的是模型以为自己能调用的工具。两者不一致时，它在待注入消息前插一条 system message，用 toolsAdded／toolsRemoved 声明差异；待注入消息里已有 system message 时，就改写那一条的工具字段。这样回放 transcript 得到的工具集总等于 context.tools。代价是工具变化会作为消息留在历史里，而不是悄悄改一个字段。”

“runAgentLoopContinue 的边界也明确：空 context 会 throw；最后一条是 assistant 也直接 throw。排空队列的逻辑在上一节的 Agent.continue() 里，不在低层函数里。这样不会把 assistant 回答直接接到下一次 provider 请求；真正的 convertToLlm 只在每个 turn 调用时才能验证 provider 接受的消息形式。”

外层和内层循环分别解决什么

小A把那两个 while 圈了出来：“它们各管什么？”

“固定快照中 runLoop 在进入外层前读取一次 getSteeringMessages。外层 while (true) 的注释是‘queued follow-up messages arrive after agent would stop’；内层 while (hasMoreToolCalls || pendingMessages.length > 0) 的注释是‘process tool calls and steering messages’。”老Z说，“所以内层不是你想的并行工具循环，并行只在后面的工具执行辅助函数里发生。”

// packages/agent/src/agent-loop.ts，runLoop（v1.0.0，节选）
while (true) {
	let hasMoreToolCalls = true;
	while (hasMoreToolCalls || pendingMessages.length > 0) {
		// 非首轮：prepareNextTurn，再补取一次 steering，emit turn_start
		// 注入 prepared 与 pending 消息，然后 prepareRequest
		const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFunction);
		// tool result 加回 currentContext.messages 与 newMessages
		const decision = await config.finishTurn?.(lastCompletedTurn, signal);
		await emit({ type: "turn_end", message, toolResults });
		if (decision?.action === "end") { await emit({ type: "agent_end", messages: newMessages }); return; }
		pendingMessages = (await config.getSteeringMessages?.()) || [];
	}
	const followUpMessages = (await config.getFollowUpMessages?.()) || [];
	if (followUpMessages.length > 0) { pendingMessages = followUpMessages; continue; }
	if (explicitContinuation) { explicitContinuation = false; continue; }
	break;
}

正常一次工具轮的状态序列如下：

text

初始 prompts → emit start/end → prepareRequest → assistant stream
  → start/delta 更新 partial assistant → done 得最终 assistant
  → 找 content 内 toolCall
  → 执行／拒绝每个 tool call，产生 ToolResultMessage
  → 将 tool results 追加到 currentContext 与 newMessages
  → finishTurn 给出决策（end／continue／不表态）
  → emit turn_end；决策为 end 则 emit agent_end 并返回
  → 取 steering；有 tool result 或 steering 时内层再走一轮：
      prepareNextTurn 可替换 context、model、thinking 或追加消息
      → emit turn_start → 注入消息 → prepareRequest → 请求模型
  → 否则取 follow-up；有则回到内层
  → 都没有但 finishTurn 要求 continue，则只用当前 context 再请求一次
  → 否则 emit agent_end

“assistant 没有 tool call 时，hasMoreToolCalls 会变为 false；只有 steering 能让内层继续。assistant 之后的 follow-up 则要等内层准备退出才读取。两类队列的语义差别就落在这两个读取位置上，它们不是一张消息优先级表。”

“这一版还多了三个宿主插口，位置各不相同。”老Z说，“finishTurn 在工具结果追加后、turn_end 前调用，返回 { action: "end" } 就结束 run，不再读队列；返回 { action: "continue" } 保证至少再请求一次模型。prepareNextTurn 推迟到下一轮开始时才调用，注释说明它可能很慢，例如做压缩，所以调用后会再补取一次 steering。prepareRequest 在每次请求模型前调用，首轮也调用，此时待注入消息已经追加完。”

插口	调用时机	能改什么	不做什么
`finishTurn`	工具结果追加后、`turn_end` 前	结束 run 或要求再请求一次	不能救回 error／aborted
`prepareNextTurn`	下一轮 `turn_start` 前	context、model、thinking、追加消息	首轮不调用
`prepareRequest`	每次请求模型前	context、model、thinking	不读队列，不追加消息

逐段跟 `streamAssistantResponse`

小A继续追：“上下文是怎么交到模型手里的？”

“看 streamAssistantResponse，它是 AgentMessage[] 与 AI Message[] 的边界。”老Z说，“先可选 transformContext，再 convertToLlm，然后只用转换后的 messages 调 normalizeContext 建请求上下文。system prompt 和工具声明已经在 system message 里，不再单独传。接着刷新可过期的 API key，并调用 injected streamFunction。”

// packages/agent/src/agent-loop.ts，streamAssistantResponse（v1.0.0，节选）
let messages = context.messages;
if (config.transformContext) messages = await config.transformContext(messages, signal);
const llmMessages = await config.convertToLlm(messages);
const llmContext = normalizeContext({ messages: llmMessages });
const resolvedApiKey =
	(config.getApiKey ? await config.getApiKey(config.model.provider) : undefined) || config.apiKey;
const response = await streamFunction(config.model, llmContext, { ...config, apiKey: resolvedApiKey, signal });
const result = async () => Object.assign(await response.result(), { thinkingLevel: config.reasoning ?? "off" });

“normalizeContext 返回的类型是 TranscriptContext，pi-ai 的注释说只有这个函数能产出它，所以未经规范化的 Context 不会意外流到 provider。最后一行把本次请求的 thinking level 记到最终 assistant message 上，不管是哪个 stream function 回答的。”

小A想起一个坑：“那 convertToLlm 要是把 system message 过滤掉了呢？”

“模型就看不到 system prompt 和工具声明。”老Z说，“Agent 的默认 convertToLlm 保留 system、user、assistant、toolResult 四种角色；自己写转换函数的宿主要记得保留 system。”

“这段的收益是 loop 不直接绑定 ModelsImpl：测试或宿主可注入 stream function，且 context 转换在每一 turn 都可以重算。代价是 transform／转换／key resolver 都可能失败；这段函数本身不 catch 它们，错误由调用它的 run 与 Agent 宿主处理。”

“下面是事件消费的核心分支。节选保留了状态改变，删去了全部 delta 类型的枚举。”

// packages/agent/src/agent-loop.ts，streamAssistantResponse（v1.0.0，节选）
case "start":
	partialMessage = event.partial;
	context.messages.push(partialMessage);
	addedPartial = true;
	await emit({ type: "message_start", message: { ...partialMessage } });
	break;
// text/thinking/toolcall delta：替换 context 的末条并 emit message_update
case "done":
case "error": {
	const finalMessage = await result();
	if (addedPartial) context.messages[context.messages.length - 1] = finalMessage;
	else context.messages.push(finalMessage);
	await emit({ type: "message_end", message: finalMessage });
	return finalMessage;
}

“先放 partial，再以 final message 替换末条，保证后续工具执行看的是已完成的 assistant content。即便没收到 start，done／error 分支仍会 append final message 并补发 message_start；迭代自然结束也会调用 response.result()。这覆盖了 adapter 的不同事件形态，不能据此假定所有 adapter 一定先发 start。”

停止原因先于工具执行

小A又想当然了：“模型说要调工具，就一定执行？”

“不一定。”老Z回到 runLoop，“若 assistant 的 stopReason 是 error 或 aborted，它照常调用 finishTurn，但不看决策，发 turn_end 和 agent_end 后返回，不执行工具。若 content 有 tool calls 且 stop reason 为 length，也不会执行任何一个，而由 failToolCallsFromTruncatedMessage 为每个 call 生成错误结果；注释说明解析器可能把被截断的 JSON 尽力修复，但这些调用仍然不会被执行。”

assistant 状态	有 tool call 时的动作	是否继续内层
正常非 length	`executeToolCalls`	取决于所有结果是否 `terminate: true`，以及 `finishTurn` 是否结束
`length`	每个 call 生成“arguments may be truncated”错误结果	是，让模型重新发完整调用
`error`	不执行工具	否，立即 agent_end
`aborted`	不执行工具	否，立即 agent_end
无 tool call	无 tool result	只可能由 steering 继续

“这至少给出两个中断／边界路径：模型中止停止 loop；输出长度截断保留 tool result 证据但拒绝副作用。把 length 当成‘可安全用已解析参数继续’会违背源码注释。”

`executeToolCalls` 决定顺序或并行

小A绕回开头的疑问：“那并行到底在哪儿？我一开始以为内层就是并行跑工具。”

“回到代码看。”老Z打开 executeToolCalls：

// packages/agent/src/agent-loop.ts，executeToolCalls（v1.0.0，节选）
const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall");
const hasSequentialToolCall = toolCalls.some(
	(tc) => currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential",
);
if (config.toolExecution === "sequential" || hasSequentialToolCall) {
	return executeToolCallsSequential(currentContext, assistantMessage, toolCalls, config, signal, emit);
}
return executeToolCallsParallel(currentContext, assistantMessage, toolCalls, config, signal, emit);

“它先搜出 assistant content 的所有 toolCall，并行与顺序的分岔就在这里：只要全局 toolExecution 为 sequential，或其中一个被找到的工具声明 sequential，整个 batch 走顺序路径；否则走并行路径。”

“两条路径都先发每个 tool_execution_start，并经过 prepareToolCall。准备会检查工具存在、预处理参数、validateToolArguments、可选 beforeToolCall hook 与 abort；任何异常、找不到工具、被 hook block 或 abort 都变成立即 error result，而不是抛掉整个 batch。”

“并行路径先按 assistant 中的顺序收集立即结果或 async thunk，随后 Promise.all 并发执行 prepared thunk；每个 thunk 开始执行前会再查一次 signal，已经 abort 就直接给出 Operation aborted 错误结果，不调用工具。它们却按照 orderedFinalizedCalls 的原始顺序再生成 tool-result messages。这个选择的收益是独立工具可以同时运行而上下文顺序稳定；代价是共享资源工具必须显式标 executionMode: "sequential" 或自行串行化。”

“executePreparedToolCall 通过注入的 update sink 把 tool 的 progress callback 转为 tool_execution_update，等待已发出的 update event 后才返回。工具正常返回但结果自带 isError: true 时，这个标记会保留；其 catch 把工具抛出的异常包成 error result；beforeToolCall block 时还可以带 terminate: true；afterToolCall hook 还能修改结果，hook 自己抛错则也生成错误结果。最后 shouldTerminateToolBatch 只有在 batch 非空且所有 result 都 terminate === true 时才停止下一次模型请求。”

工具边界	来源	loop 可见的结果	后续效果
工具名不存在	`prepareToolCall`	error tool result	仍交模型决定下一步
参数无效／before hook block	准备阶段	error tool result	不执行工具
signal aborted	准备、顺序循环或并行 thunk 开始前	error result 或停止后续顺序项	不承诺撤销已启动工具
tool.execute 抛错	执行阶段 catch	error tool result	仍形成上下文证据
所有 result terminate	`shouldTerminateToolBatch`	`terminate: true`	内层不因工具继续

小A：“要是一个工具想调另一个工具呢？比如脚本工具里批量调 read。”

“v1.0.0 为此导出了 runToolCall。”老Z说，“它走和模型发起的调用相同的步骤：参数预处理、schema 校验、beforeToolCall、执行、afterToolCall。它不发事件，也不往 transcript 加消息；工具失败都以 isError: true 返回，不 reject。这样权限检查这类 hook 对嵌套调用同样生效。”

收益、代价与未覆盖范围

基于源码的推断：将低层 loop 的 stream function、消息转换、hook、回合插口和队列都做成注入项，使同一循环可服务不同宿主；事件又让 UI／持久化层观察每个状态变化。代价是状态分散于 context、newMessages、队列和 listener，扩展实现得认真处理取消、写入顺序和 listener 失败。

本章没有覆盖宿主怎样持久化 message_end，也没有证明任何 tool 的幂等性或取消能力。工具调用的实际副作用取决于具体 tool；并行执行不是安全保证。源码篇第5章调度台——pi-coding-agent会跟着这些事件进入产品层，看 AgentSession 怎样把它们接到会话文件、压缩和 Skills。

小结

模型流进来之后，循环是这样转起来的：runAgentLoop 建立本次增量和工作上下文；runLoop 的内层处理工具结果与 steering，外层只在原本停止后才读取 follow-up。streamAssistantResponse 用 partial 占位、最终消息替换来消费流，保证模型还没结束时，上下文里不会出现半截的 assistant 消息；executeToolCalls 依配置和工具声明选择顺序或并行，把多个失败转换为可见的 tool result，不让异常静默吞掉。

边界上，length、aborted、error 都不是普通成功结束，各自映射不同的恢复路径；被截断消息里的 tool call 不会进入 executeToolCalls，这是防止半截调用被执行的硬边界。

要点
外层轮次和内层工具循环不是一回事：内层处理“这一次模型响应引发的所有工具结果”，外层才处理“下一轮该不该继续”。把两层混成一个循环，是理解 pi-mono 循环设计最常见的误读。

源码走查

在 packages/agent/src/agent.ts 阅读 PendingMessageQueue.drain、steer、followUp；分别把 mode 设想为 all 与默认值，写下每次 drain 的消息数。
从 runAgentLoop 读到 runLoop；跟踪同一 prompt 如何同时进入 currentContext.messages 与 newMessages，并读 declareToolChanges，说明它在什么条件下插入一条新的 system message。
在 runLoop 给内外两个 while 标注各自继续条件；验证 follow-up 只在内层退出后读取，steering 在每个 turn 尾部读取；再标出 finishTurn、prepareNextTurn、prepareRequest 三个调用点。
逐行读 streamAssistantResponse 的 start、done／error 与循环结束后 result 分支；确认没有 start 时也会把 final message 写入 context。
搜索 failToolCallsFromTruncatedMessage；核对 length 的 tool call 不会进入 executeToolCalls。
比较 executeToolCallsSequential 与 executeToolCallsParallel；确认 parallel 用 Promise.all 执行 prepared thunk，却按原 call 顺序追加 tool-result message。
读 runToolCall；对照 executeToolCalls，列出它跳过了哪两件事（事件与 transcript 消息）。

源码篇第4章 双层循环——pi-agent-core ​

Agent 是带队列的宿主包装，不是循环本身 ​

runAgentLoop 先建立本次增量 ​

外层和内层循环分别解决什么 ​

逐段跟 streamAssistantResponse ​

停止原因先于工具执行 ​

executeToolCalls 决定顺序或并行 ​

收益、代价与未覆盖范围 ​

小结 ​

源码走查 ​