34 KiB
逐行精读:agent-loop.ts —— 核心循环(Pi 的心脏)
源文件:
pi-main/packages/agent/src/agent-loop.ts(792 行)前置阅读:01-types-and-protocol.md(types.ts 逐行精读)。 循环只是在操作那些类型——不懂类型就读不懂循环。
文件结构总览
agent-loop.ts (792 行)
├── 公共入口(L1-151)
│ ├── agentLoop() —— 带 prompt 启动
│ ├── agentLoopContinue() —— 不加 prompt 续跑(用于重试)
│ └── createAgentStream() —— EventStream 工厂
├── 核心循环 runLoop()(L155-275)⭐⭐⭐
│ ├── 外层 while:follow-up 循环
│ └── 内层 while:steering + 工具循环
├── streamAssistantResponse()(L281-374)⭐⭐ 边界转换
├── 工具执行(L383-755)
│ ├── failToolCallsFromTruncatedMessage() —— 截断时全部失败
│ ├── executeToolCalls() —— 模式选择
│ ├── executeToolCallsSequential() —— 顺序执行
│ ├── executeToolCallsParallel() —— 并发执行
│ ├── prepareToolCall() —— 验证 + before hook
│ ├── executePreparedToolCall() —— 执行 + 流式更新
│ └── finalizeExecutedToolCall() —— after hook
└── 辅助函数(L757-792)
├── createErrorToolResult()
├── emitToolExecutionEnd()
├── createToolResultMessage()
└── emitToolResultMessage()
一、公共入口(L1-151)
agentLoop —— 带 prompt 启动(L34-118)
// L34-101(签名)
export function agentLoop(
prompts: AgentMessage[],
context: AgentContext,
config: AgentLoopConfig,
signal?: AbortSignal,
streamFn?: StreamFn,
): EventStream<AgentEvent, AgentMessage[]> {
const stream = createAgentStream();
void runAgentLoop(prompts, context, config, async (event) => stream.push(event), signal, streamFn)
.then((messages) => stream.end(messages));
return stream;
}
关键模式:立即返回 EventStream,后台异步推事件。void runAgentLoop(...).then(...) 启动一个不等待的 promise,事件通过 stream.push 推给消费者。这让调用方能边迭代事件边让循环跑——生产者/消费者解耦。
// L103-117(runAgentLoop 主体)
async function runAgentLoop(...): Promise<AgentMessage[]> {
const newMessages: AgentMessage[] = [...prompts]; // 本次 run 产生的所有新消息
const currentContext: AgentContext = {
...context,
messages: [...context.messages, ...prompts], // 上下文 = 原有 + 新 prompt
};
await emit({ type: "agent_start" }); // L109 生命周期开始
await emit({ type: "turn_start" }); // L110
for (const prompt of prompts) { // L111-114 为每个 prompt 发 message 事件
await emit({ type: "message_start", message: prompt });
await emit({ type: "message_end", message: prompt });
}
await runLoop(currentContext, newMessages, config, signal, emit, streamFn); // L116 ← 核心循环
return newMessages; // L117 返回本次产生的新消息
}
newMessages 是只追加的列表——本次 run 产生的所有新消息(prompt + assistant + toolResult + steering + follow-up)。它和 currentContext.messages(完整 transcript)是两个不同的东西。
agentLoopContinue —— 续跑(L120-143)
// L127-133 两个前置校验
if (context.messages.length === 0) throw new Error("Cannot continue: no messages in context");
if (context.messages[last].role === "assistant") throw new Error("Cannot continue from assistant");
为什么最后一条不能是 assistant? 因为继续循环意味着要调 LLM,而 LLM 调用要求最后一条是 user 或 toolResult(Anthropic/OpenAI 的协议约束)。最后一条是 assistant 时"继续"无意义——你应该用 prompt 加新消息。
createAgentStream(L145-150)
function createAgentStream(): EventStream<AgentEvent, AgentMessage[]> {
return new EventStream<AgentEvent, AgentMessage[]>(
(event) => event.type === "agent_end", // 完成检测器
(event) => event.type === "agent_end" ? event.messages : [], // 结果提取器
);
}
EventStream<T, R> 的两个回调:① 判断哪个事件是"完成"事件;② 从完成事件提取最终结果 R。这里完成事件是 agent_end,结果是它的 messages。
二、核心循环 runLoop(L155-275)⭐⭐⭐
这是整个文件最重要的一段。双层循环 + 四个停止条件。
函数签名与初始状态(L155-167)
async function runLoop(
initialContext: AgentContext,
newMessages: AgentMessage[],
initialConfig: AgentLoopConfig,
signal: AbortSignal | undefined,
emit: AgentEventSink,
streamFn?: StreamFn,
): Promise<void> {
let currentContext = initialContext; // 可变:prepareNextTurn 会替换
let config = initialConfig; // 可变:prepareNextTurn 会替换 model/reasoning
let firstTurn = true;
// L167 启动时就检查 steering(用户可能在等待时已输入)
let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || [];
currentContext 和 config 都是 let——prepareNextTurn 可以整体替换它们。这是"每轮重建上下文"的实现基础。
L167 的 getSteeringMessages 在循环开始就调用——用户可能在 agentLoop 返回 EventStream 后、循环真正开始前就输入了 steering 消息。这是为了不丢消息。
双层循环骨架(L170-272)
// L170 外层 while:follow-up 循环
while (true) {
let hasMoreToolCalls = true;
// L174 内层 while:steering + 工具循环
while (hasMoreToolCalls || pendingMessages.length > 0) {
为什么是双层?
- 内层:处理"一个连贯的工作流"——有工具调用就继续,有 steering 就注入。
- 外层:当内层要退出(没工具调用也没 steering 了),再检查 follow-up——"任务做完了,还有别的吗?"有就重启内层。
这种分层让 steering(紧急插话)和 follow-up(后续任务)的语义清晰分离。
内层循环体(L175-260)
① Turn 标记与消息注入(L175-190)
if (!firstTurn) {
await emit({ type: "turn_start" }); // L176 非首轮才发 turn_start
} else {
firstTurn = false; // 首轮的 turn_start 已在 runAgentLoop 发过
}
// L182-190 注入 pendingMessages(steering 或 follow-up)
if (pendingMessages.length > 0) {
for (const message of pendingMessages) {
await emit({ type: "message_start", message });
await emit({ type: "message_end", message });
currentContext.messages.push(message); // 加入 transcript
newMessages.push(message); // 加入本次 run 的新消息
}
pendingMessages = []; // 清空
}
steering/follow-up 消息在调 LLM 之前注入——模型会在下一轮看到它们。
② 流式接收 assistant 响应(L193-200)
const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFn);
newMessages.push(message);
// L196-200 错误/中止:直接结束
if (message.stopReason === "error" || message.stopReason === "aborted") {
await emit({ type: "turn_end", message, toolResults: [] });
await emit({ type: "agent_end", messages: newMessages });
return;
}
这是停止条件①:provider 返回 error/aborted。注意仍然发 turn_end 和 agent_end——保证事件序列完整。
③ 提取并执行工具调用(L202-222)
const toolCalls = message.content.filter((c) => c.type === "toolCall");
const toolResults: ToolResultMessage[] = [];
hasMoreToolCalls = false; // L206 默认假设不继续
if (toolCalls.length > 0) {
// L208-214 关键分支:截断时全部失败
const executedToolBatch =
message.stopReason === "length"
? await failToolCallsFromTruncatedMessage(toolCalls, emit)
: await executeToolCalls(currentContext, message, config, signal, emit);
toolResults.push(...executedToolBatch.messages);
hasMoreToolCalls = !executedToolBatch.terminate; // L216 terminate 决定是否继续
for (const result of toolResults) {
currentContext.messages.push(result); // 工具结果入 transcript
newMessages.push(result);
}
}
length 停止原因的特殊处理(L208-214)——这是停止条件的一个微妙点:
为什么截断时要全部失败?因为流式 tool call 的参数是用增量 JSON 解析(parseStreamingJson)拼出来的。如果输出被 token 限制截断,拼出来的 JSON 可能恰好能解析、能通过 schema 验证,但内容是残缺的——执行它会有不可预测的后果。所以宁可全部报错让模型重新发。
hasMoreToolCalls = !executedToolBatch.terminate(L216)——只有批内所有结果都 terminate: true 才停止(见 shouldTerminateToolBatch,L584-586)。
④ prepareNextTurn + shouldStopAfterTurn(L224-259)
await emit({ type: "turn_end", message, toolResults });
// L226-245 prepareNextTurn:每轮后重建上下文
const nextTurnContext = { message, toolResults, context: currentContext, newMessages };
const nextTurnSnapshot = await config.prepareNextTurn?.(nextTurnContext);
if (nextTurnSnapshot) {
currentContext = nextTurnSnapshot.context ?? currentContext; // 替换上下文
config = {
...config,
model: nextTurnSnapshot.model ?? config.model, // 替换模型
reasoning: nextTurnSnapshot.thinkingLevel === undefined // 替换思考级别
? config.reasoning
: nextTurnSnapshot.thinkingLevel === "off" ? undefined : nextTurnSnapshot.thinkingLevel,
};
}
// L247-257 停止条件②:shouldStopAfterTurn
if (await config.shouldStopAfterTurn?.({ message, toolResults, context: currentContext, newMessages })) {
await emit({ type: "agent_end", messages: newMessages });
return;
}
pendingMessages = (await config.getSteeringMessages?.()) || []; // L259 再 poll steering
} // 内层 while 结束
prepareNextTurn 是 harness 层动态上下文的钩子(L226-245)。harness 在这里:从 session 重载消息、重算系统提示(可含 cwd/git status)、重建工具列表。模型切换、上下文压缩、工具动态加载都通过这个生效。
L247-257 是停止条件②:shouldStopAfterTurn 返回 true。注意它在 steering poll 之前——如果要求停止,就不 poll steering/follow-up 了(L255 直接 return)。用途:上下文快满了提前停。
L259 再次 poll steering——内层循环每次 turn 结束都检查。这就是 steering 的"紧急插话"语义。
外层循环:follow-up(L262-271)
// 内层退出,agent 本来要停了
const followUpMessages = (await config.getFollowUpMessages?.()) || []; // L263
if (followUpMessages.length > 0) {
pendingMessages = followUpMessages; // L266 设为 pending,内层会注入
continue; // L267 重启内层
}
break; // L271 真的没东西了,退出
}
await emit({ type: "agent_end", messages: newMessages }); // L274 最终结束
}
follow-up 只在内层"本来要退出"时才检查——这是它和 steering 的本质区别。steering 是打断(内层每次都 poll),follow-up 是续接(内层要停才问)。
停止条件总结
| # | 条件 | 触发位置 |
|---|---|---|
| ① | provider 返回 error/aborted |
L196 |
| ② | shouldStopAfterTurn 返回 true |
L247 |
| ③ | 没工具调用、没 steering、没 follow-up(自然结束) | 内层退出 + 外层 break |
| ④ | 工具批全部 terminate: true(通过 hasMoreToolCalls=false) |
L216 + 内层退出 |
截断(length)不是停止条件——它只是让工具全部失败,循环继续(模型会看到错误,重发工具调用)。
三、streamAssistantResponse —— 边界转换(L281-374)⭐⭐
这是 AgentMessage↔Message 唯一的翻译点,也是循环和 provider 的接口。
三步转换(L288-314)
async function streamAssistantResponse(...): Promise<AssistantMessage> {
// 第①步:可选的上下文变换(AgentMessage → AgentMessage)
let messages = context.messages;
if (config.transformContext) {
messages = await config.transformContext(messages, signal); // 剪枝/注入
}
// 第②步:转换成 LLM 消息(AgentMessage → Message[])
const llmMessages = await config.convertToLlm(messages);
// 第③步:构建 LLM 上下文
const llmContext: Context = {
systemPrompt: context.systemPrompt,
messages: llmMessages,
tools: context.tools,
};
const streamFunction = streamFn || streamSimple; // L304 可注入自定义 streamFn
// 动态解析 API key(处理过期 token)
const resolvedApiKey =
(config.getApiKey ? await config.getApiKey(config.model.provider) : undefined) || config.apiKey;
// 调 provider
const response = await streamFunction(config.model, llmContext, { ...config, apiKey: resolvedApiKey, signal });
为什么 transformContext 在 convertToLlm 之前? 因为 transformContext 操作 AgentMessage[](应用级,可剪枝自定义消息、注入外部上下文),而 convertToLlm 是格式转换(AgentMessage → Message)。顺序是"先处理应用语义,再转换协议格式"。
流式累加(L316-373)—— 流式消息是"活"的
let partialMessage: AssistantMessage | null = null;
let addedPartial = false; // 标记是否已 push 到 context.messages
for await (const event of response) {
switch (event.type) {
case "start": // L321-326
partialMessage = event.partial;
context.messages.push(partialMessage); // ★ 立即入栈
addedPartial = true;
await emit({ type: "message_start", message: { ...partialMessage } });
break;
case "text_start": case "text_delta": case "text_end":
case "thinking_start": case "thinking_delta": case "thinking_end":
case "toolcall_start": case "toolcall_delta": case "toolcall_end":
// L337-346 所有 delta 事件统一处理
if (partialMessage) {
partialMessage = event.partial;
context.messages[context.messages.length - 1] = partialMessage; // ★ 原地替换最后一个
await emit({ type: "message_update", assistantMessageEvent: event, message: { ...partialMessage } });
}
break;
case "done": case "error": { // L348-361
const finalMessage = await response.result();
if (addedPartial) {
context.messages[last] = finalMessage; // 替换为最终版
} else {
context.messages.push(finalMessage); // 没 push 过就 push
}
if (!addedPartial) {
await emit({ type: "message_start", message: { ...finalMessage } }); // 补发 message_start
}
await emit({ type: "message_end", message: finalMessage });
return finalMessage;
}
}
}
// L365-373 流自然结束(没 done/error 事件)的兜底
const finalMessage = await response.result();
if (addedPartial) context.messages[last] = finalMessage;
else { context.messages.push(finalMessage); await emit({ type: "message_start", message: { ...finalMessage } }); }
await emit({ type: "message_end", message: finalMessage });
return finalMessage;
}
核心设计:流式消息在流式过程中就躺在 transcript 末尾。
start事件 → push 空的 partial 消息- 每个 delta 事件 → 用新 partial 原地替换最后一个元素
done/error→ 用最终消息替换
这让 UI 能实时看到流式输出(通过 message_update 事件),而 transcript 始终保持"当前最新状态"。
addedPartial 标志(L317)处理一个边界情况:如果 provider 没发 start 事件直接发 done(某些 provider 的行为),那 transcript 里还没有 partial,需要补发 message_start。
L365-373 的兜底:如果流自然耗尽(for-await 结束)但没收到 done/error,用 response.result() 取最终消息——保证总有最终消息入栈。这是健壮性设计。
为什么 emit 时用 { ...partialMessage } 浅拷贝?(L325, L343)因为 partial 后续会被替换,emit 出去的应该是那一刻的快照,避免消费者持有的是会被原地修改的引用。
四、工具执行(L383-755)
截断失败(L383-408)
async function failToolCallsFromTruncatedMessage(toolCalls, emit): Promise<ExecutedToolCallBatch> {
const messages: ToolResultMessage[] = [];
for (const toolCall of toolCalls) {
await emit({ type: "tool_execution_start", toolCallId: toolCall.id, toolName: toolCall.name, args: toolCall.arguments });
const finalized: FinalizedToolCallOutcome = {
toolCall,
result: createErrorToolResult(`...arguments may be truncated. Re-issue the tool call...`),
isError: true,
};
await emitToolExecutionEnd(finalized, emit);
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit);
messages.push(toolResultMessage);
}
return { messages, terminate: false }; // 不终止——让模型看到错误重发
}
模拟完整的工具执行事件序列(start → end → message),但结果永远是错误。terminate: false 让循环继续——模型会看到"参数可能截断"的错误,重新发工具调用。
模式选择(L413-428)
async function executeToolCalls(...): Promise<ExecutedToolCallBatch> {
const toolCalls = assistantMessage.content.filter((c) => c.type === "toolCall");
// 只要有一个工具声明了 sequential,整个批次就退化为 sequential
const hasSequentialToolCall = toolCalls.some((tc) =>
currentContext.tools?.find((t) => t.name === tc.name)?.executionMode === "sequential");
if (config.toolExecution === "sequential" || hasSequentialToolCall) {
return executeToolCallsSequential(...);
}
return executeToolCallsParallel(...);
}
保守策略:批内有一个 sequential 工具就全部顺序执行。因为 sequential 通常意味着有副作用(如写文件),并发可能冲突。
顺序执行(L435-489)
async function executeToolCallsSequential(...): Promise<ExecutedToolCallBatch> {
const finalizedCalls: FinalizedToolCallOutcome[] = [];
const messages: ToolResultMessage[] = [];
for (const toolCall of toolCalls) { // 一个一个来
await emit({ type: "tool_execution_start", ... });
const preparation = await prepareToolCall(...); // 验证 + before hook
let finalized: FinalizedToolCallOutcome;
if (preparation.kind === "immediate") { // 验证失败/blocked/未找到 → 直接出结果
finalized = { toolCall, result: preparation.result, isError: preparation.isError };
} else {
const executed = await executePreparedToolCall(...); // 执行
finalized = await finalizeExecutedToolCall(...); // after hook
}
await emitToolExecutionEnd(finalized, emit); // 立即发 end
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit); // 立即发 toolResult message
finalizedCalls.push(finalized);
messages.push(toolResultMessage);
if (signal?.aborted) break; // 中止则跳出
}
return { messages, terminate: shouldTerminateToolBatch(finalizedCalls) };
}
顺序模式的特点:start→execute→end→message 一个工具完整走完才下一个。中止会 break,已完成的工具结果保留。
并发执行(L491-556)—— 精妙之处
async function executeToolCallsParallel(...): Promise<ExecutedToolCallBatch> {
const finalizedCalls: FinalizedToolCallEntry[] = []; // 注意类型是 Entry(可能是函数)
// 第一阶段:顺序 prepare(含 before hook)
for (const toolCall of toolCalls) {
await emit({ type: "tool_execution_start", ... });
const preparation = await prepareToolCall(...);
if (preparation.kind === "immediate") { // immediate 直接 finalize + 发 end
const finalized = { toolCall, result: preparation.result, isError: preparation.isError };
await emitToolExecutionEnd(finalized, emit);
finalizedCalls.push(finalized); // 直接存结果
if (signal?.aborted) break;
continue;
}
// prepared → 存一个 thunk(延迟执行)
finalizedCalls.push(async () => {
const executed = await executePreparedToolCall(preparation, signal, emit);
const finalized = await finalizeExecutedToolCall(...);
await emitToolExecutionEnd(finalized, emit); // end 按完成顺序发
return finalized;
});
if (signal?.aborted) break;
}
// 第二阶段:并发执行所有 thunk
const orderedFinalizedCalls = await Promise.all(
finalizedCalls.map((entry) => (typeof entry === "function" ? entry() : Promise.resolve(entry))),
);
// 第三阶段:toolResult message 按 assistant 源顺序发(不是完成顺序!)
const messages: ToolResultMessage[] = [];
for (const finalized of orderedFinalizedCalls) {
const toolResultMessage = createToolResultMessage(finalized);
await emitToolResultMessage(toolResultMessage, emit);
messages.push(toolResultMessage);
}
return { messages, terminate: shouldTerminateToolBatch(orderedFinalizedCalls) };
}
并发模式的三个阶段:
- 顺序 prepare:所有工具的
prepareToolCall(含 before hook)顺序跑。为什么顺序?因为 before hook 可能 block,且 prepare 可能修改共享状态。 - 并发 execute:prepared 的工具存成 thunk,
Promise.all并发执行。tool_execution_end按完成顺序发(每个 thunk 完成时立即发)。 - 顺序 emit message:
toolResult的message_start/end按源顺序发(orderedFinalizedCalls保持原顺序)。
FinalizedToolCallEntry 是联合类型(L582):
type FinalizedToolCallEntry = FinalizedToolCallOutcome | (() => Promise<FinalizedToolCallOutcome>);
immediate 的存结果对象,prepared 的存 thunk。第二阶段用 typeof entry === "function" 区分。
为什么 end 按完成顺序、message 按源顺序? 因为 tool_execution_end 是给 UI 看的(哪个工具先做完先显示),而 toolResult message 是给下一轮 LLM看的(要和 assistant 的 toolCall 顺序对应,否则模型会困惑)。
prepareToolCall —— 验证 + before hook(L602-666)
async function prepareToolCall(...): Promise<PreparedToolCall | ImmediateToolCallOutcome> {
const tool = currentContext.tools?.find((t) => t.name === toolCall.name);
if (!tool) { // L610 工具未找到
return { kind: "immediate", result: createErrorToolResult(`Tool ${toolCall.name} not found`), isError: true };
}
try {
const preparedToolCall = prepareToolCallArguments(tool, toolCall); // L619 prepareArguments shim
const validatedArgs = validateToolArguments(tool, preparedToolCall); // L620 schema 验证
if (config.beforeToolCall) { // L621 before hook
const beforeResult = await config.beforeToolCall(
{ assistantMessage, toolCall, args: validatedArgs, context: currentContext }, signal);
if (signal?.aborted) return { kind: "immediate", result: createErrorToolResult("Operation aborted"), isError: true };
if (beforeResult?.block) { // L638 block → 错误结果
return { kind: "immediate", result: createErrorToolResult(beforeResult.reason || "Tool execution was blocked"), isError: true };
}
}
if (signal?.aborted) return { kind: "immediate", result: createErrorToolResult("Operation aborted"), isError: true };
return { kind: "prepared", toolCall, tool, args: validatedArgs }; // L653 通过 → 执行
} catch (error) { // L659 验证抛错 → 错误结果
return { kind: "immediate", result: createErrorToolResult(error.message), isError: true };
}
}
三道关卡:① 工具存在性 ② prepareArguments(LLM 输出规整)+ validateToolArguments(schema 验证)③ beforeToolCall hook(权限/安全)。
返回 immediate 的情况:工具未找到、参数验证失败、before hook block、已中止。这些都跳过执行,直接产生错误结果——错误被编码成结果,而非抛出。
executePreparedToolCall —— 执行 + 流式更新(L668-709)
async function executePreparedToolCall(prepared, signal, emit): Promise<ExecutedToolCallOutcome> {
const updateEvents: Promise<void>[] = [];
let acceptingUpdates = true; // 更新开关
try {
const result = await prepared.tool.execute(
prepared.toolCall.id,
prepared.args as never,
signal,
(partialResult) => { // onUpdate 回调
if (!acceptingUpdates) return; // settle 后的调用忽略
updateEvents.push(Promise.resolve(emit({ type: "tool_execution_update", ... partialResult })));
},
);
acceptingUpdates = false; // 执行完成,关闭更新
await Promise.all(updateEvents); // 等待所有 update 事件 flush
return { result, isError: false };
} catch (error) {
acceptingUpdates = false;
await Promise.all(updateEvents); // 错误时也 flush 更新
return { result: createErrorToolResult(error.message), isError: true };
} finally {
acceptingUpdates = false; // 确保 finally 也关闭
}
}
onUpdate 回调的作用域:只在 execute() promise 未 settle 时有效。acceptingUpdates 三处置 false(成功/失败/finally),保证 settle 后的调用被忽略。
updateEvents 收集机制:update 回调里不直接 await emit(回调是同步的),而是把 emit 的 promise 推入数组,execute 完成后统一 Promise.all flush。这避免了回调内的异步竞态。
错误处理(L699-705):工具 execute 抛错 → createErrorToolResult → isError: true。错误编码成结果,绝不抛出。
finalizeExecutedToolCall —— after hook(L711-755)
async function finalizeExecutedToolCall(...): Promise<FinalizedToolCallOutcome> {
let result = executed.result;
let isError = executed.isError;
if (config.afterToolCall) {
try {
const afterResult = await config.afterToolCall(
{ assistantMessage, toolCall: prepared.toolCall, args: prepared.args, result, isError, context: currentContext },
signal);
if (afterResult) {
// 字段级覆盖,无深度合并
result = {
...result,
content: afterResult.content ?? result.content, // 给了就替换
details: afterResult.details ?? result.details,
terminate: afterResult.terminate ?? result.terminate,
};
isError = afterResult.isError ?? isError;
}
} catch (error) { // after hook 抛错 → 错误结果
result = createErrorToolResult(error.message);
isError = true;
}
}
return { toolCall: prepared.toolCall, result, isError };
}
after hook 的字段级覆盖(L736-742):content/details/terminate 用 ??(给了就替换,没给保持原值)。isError 同理。无深度合并——这是有意的,避免意外的部分覆盖导致状态不一致。
用途:脱敏(改写 content)、强制标记错误(isError: true)、审计日志(读取 details)。
五、辅助函数(L757-792)
// L757-762 错误结果工厂
function createErrorToolResult(message: string): AgentToolResult<any> {
return { content: [{ type: "text", text: message }], details: {} };
}
// L774-787 构造 ToolResultMessage
function createToolResultMessage(finalized: FinalizedToolCallOutcome): ToolResultMessage {
return {
role: "toolResult",
toolCallId: finalized.toolCall.id,
toolName: finalized.toolCall.name,
content: finalized.result.content ?? [], // L781 null 归一化(JS 扩展工具可能无 content)
details: finalized.result.details,
...(finalized.result.addedToolNames?.length ? { addedToolNames: finalized.result.addedToolNames } : {}),
isError: finalized.isError,
timestamp: Date.now(),
};
}
L778-781 的 null 归一化:注释说"untyped tools (JS extensions) can return results without content"。因为 JS 扩展工具不走 Rust 的强类型,可能返回 content: null。这里 ?? [] 保证 null 永不进入 session 历史或 provider payload——防御性编程。
addedToolNames 条件展开(L783):只在非空时加入字段,避免序列化出空的 addedToolNames: []。
六、设计模式总结
读完 792 行,提炼出 agent-loop.ts 的核心设计模式:
1. 错误编码而非抛出(贯穿全文)
StreamFn契约:不 throw,错误进 streamprepareToolCall:验证失败返回immediate错误结果executePreparedToolCall:工具抛错 →createErrorToolResultfinalizeExecutedToolCall:after hook 抛错 → 错误结果
所有错误都变成 ToolResultMessage(isError: true),循环永远不会因为工具错误而中断事件序列。
2. 生产者-消费者解耦(L34-49)
agentLoop 立即返回 EventStream,后台 void promise 推事件。消费者 for await 拉事件,生产者跑循环。
3. 流式消息"活性"(L321-361)
partial 消息在流式过程中就躺在 transcript 末尾,delta 原地替换。UI 实时可见,transcript 始终最新。
4. 双层循环分离语义(L170-272)
- 内层:steering(打断)+ 工具调用
- 外层:follow-up(续接) 两种"中途加消息"的语义清晰区分。
5. 边界唯一转换(L288-295)
convertToLlm 是 AgentMessage→Message 的唯一桥梁。transformContext(AgentMessage→AgentMessage)在它之前。
6. 顺序 vs 并发的精巧设计(L435-556)
- sequential:完整流水线一个一个来
- parallel:prepare 顺序、execute 并发、message 按源序
7. 字段级覆盖无深度合并(L736-742)
after hook 的覆盖是"给了就整体替换,没给就保持",避免部分覆盖导致状态不一致。
8. 防御性归一化(L781)
外部来源(JS 扩展)的 content: null 被 ?? [] 归一化,保证内部不变量。
七、纯 std 实现启示
读完循环逻辑,对 focus 实现的关键启示:
-
循环本身是纯的状态机——
runLoop只是while+ 条件判断 + 函数调用,零 I/O。纯 std 完全可实现。 -
StreamFn可以是 trait:trait StreamFn: Send { fn stream(&self, model: &Model, ctx: &Context, opts: &StreamOptions) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>>; }或者更简单——用同步迭代器/channel。核心循环不关心 stream 怎么来。
-
EventStream 可用
mpsc::Receiver替代:let (tx, rx) = mpsc::channel(); // 后台线程:循环跑,tx.send(event) // 主线程:rx.iter() 消费Rust 的 std::sync::mpsc 完全够用。或者用 crossbeam(但那不是 std)。
-
并发工具执行可用
std::thread+JoinHandle:顺序 prepare,spawn 线程并发 execute,join 收集结果。 -
AgentMessage→Message 转换是纯函数——不依赖任何外部库。
结论:agent-loop.ts 的逻辑可以几乎 1:1 移植到纯 std Rust,唯一的妥协是 async(可以用同步阻塞 + 线程替代)和 JSON(手写极简解析器或认栽用 serde)。
至此,Pi 的核心两层(类型 + 循环)已完整解读。配合 00-pi-architecture-overview.md 的全景,你已具备从零实现一个 agent 框架所需的全部知识。