731 lines
34 KiB
Markdown
731 lines
34 KiB
Markdown
# 逐行精读：`agent-loop.ts` —— 核心循环（Pi 的心脏）
|
||
|
||
> 源文件：`pi-main/packages/agent/src/agent-loop.ts`（792 行）
|
||
>
|
||
> 前置阅读：[01-types-and-protocol.md](./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）
|
||
|
||
```ts
|
||
// 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` 推给消费者。这让调用方能边迭代事件边让循环跑——**生产者/消费者解耦**。
|
||
|
||
```ts
|
||
// 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）
|
||
|
||
```ts
|
||
// 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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
// 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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
// 内层退出，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）
|
||
|
||
```ts
|
||
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）—— 流式消息是"活"的
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）—— 精妙之处
|
||
|
||
```ts
|
||
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) };
|
||
}
|
||
```
|
||
|
||
**并发模式的三个阶段**：
|
||
|
||
1. **顺序 prepare**：所有工具的 `prepareToolCall`（含 before hook）顺序跑。为什么顺序？因为 before hook 可能 block，且 prepare 可能修改共享状态。
|
||
2. **并发 execute**：prepared 的工具存成 thunk，`Promise.all` 并发执行。`tool_execution_end` 按**完成顺序**发（每个 thunk 完成时立即发）。
|
||
3. **顺序 emit message**：`toolResult` 的 `message_start/end` 按**源顺序**发（`orderedFinalizedCalls` 保持原顺序）。
|
||
|
||
**`FinalizedToolCallEntry` 是联合类型**（L582）：
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
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）
|
||
|
||
```ts
|
||
// 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，错误进 stream
|
||
- `prepareToolCall`：验证失败返回 `immediate` 错误结果
|
||
- `executePreparedToolCall`：工具抛错 → `createErrorToolResult`
|
||
- `finalizeExecutedToolCall`：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 实现的关键启示：
|
||
|
||
1. **循环本身是纯的状态机**——`runLoop` 只是 `while` + 条件判断 + 函数调用，零 I/O。纯 std 完全可实现。
|
||
|
||
2. **`StreamFn` 可以是 trait**：
|
||
```rust
|
||
trait StreamFn: Send {
|
||
fn stream(&self, model: &Model, ctx: &Context, opts: &StreamOptions)
|
||
-> Pin<Box<dyn Stream<Item = StreamEvent> + Send>>;
|
||
}
|
||
```
|
||
或者更简单——用同步迭代器/channel。核心循环不关心 stream 怎么来。
|
||
|
||
3. **EventStream 可用 `mpsc::Receiver` 替代**：
|
||
```rust
|
||
let (tx, rx) = mpsc::channel();
|
||
// 后台线程：循环跑，tx.send(event)
|
||
// 主线程：rx.iter() 消费
|
||
```
|
||
Rust 的 std::sync::mpsc 完全够用。或者用 crossbeam（但那不是 std）。
|
||
|
||
4. **并发工具执行可用 `std::thread` + `JoinHandle`**：顺序 prepare，spawn 线程并发 execute，join 收集结果。
|
||
|
||
5. **AgentMessage→Message 转换是纯函数**——不依赖任何外部库。
|
||
|
||
**结论**：agent-loop.ts 的逻辑可以几乎 1:1 移植到纯 std Rust，唯一的妥协是 async（可以用同步阻塞 + 线程替代）和 JSON（手写极简解析器或认栽用 serde）。
|
||
|
||
---
|
||
|
||
至此，Pi 的核心两层（类型 + 循环）已完整解读。配合 [00-pi-architecture-overview.md](./00-pi-architecture-overview.md) 的全景，你已具备从零实现一个 agent 框架所需的全部知识。
|