pi 架构拆解

简体中文

English
日本語
Español
Français
Deutsch
繁體中文

简体中文

English
日本語
Español
Français
Deutsch
繁體中文

Appearance

双层 while 主循环 ​

源码版本v0.73.1

runLoop 是 pi-agent-core 的发动机。外层 while (true) 排空 followUp 队列,内层 while (hasMoreToolCalls || pendingMessages.length > 0) 处理工具调用循环和 steering 插话;每轮调 streamAssistantResponse 拿一条 assistant 消息,有 toolCalls 就走 executeToolCalls 再回内层,没有就退出内层去问 followUp 队列。整个循环不持有状态,所有状态由调用方 (Agent) 传入,事件通过 AgentEventSink 发出。

职责 ​

  1. 事件编排:每个 turn 发 turn_start,assistant 消息发 message_start/message_update/message_end,工具执行发 tool_execution_*,turn 末尾发 turn_end,整个循环结束发 agent_end。见 packages/agent/src/agent-loop.ts:155-246。
  2. 流式拉取 assistant 回复:streamAssistantResponse 调 convertToLlm 转 Message[],用 streamFn(默认 streamSimple)发请求,for await 消费事件并改写 context.messages 末尾的 partial message。见 packages/agent/src/agent-loop.ts:252-345。
  3. steering 注入:内层循环开头检查 pendingMessages,把 steering 消息 push 进 currentContext.messages 再走下一轮 LLM 调用,见 packages/agent/src/agent-loop.ts:180-188。
  4. followUp 接力:内层退出后,外层调 getFollowUpMessages,有消息则塞回 pendingMessages 再进内层,无则 break。见 packages/agent/src/agent-loop.ts:233-243。
  5. 早停:shouldStopAfterTurn 在 turn_end 后被调用,返回 true 时直接发 agent_end 退出,绕过队列轮询,见 packages/agent/src/agent-loop.ts:218-228。

设计动机 ​

为什么是两层 while 而不是一层?因为存在两种「继续」语义。工具调用是「本轮内部」的继续——assistant 发了 toolCalls,执行完要把结果喂回 LLM 让它继续说,这是内层循环。followUp 是「本轮本该结束」后的继续——用户排了队等 agent 干完再处理,这是外层循环。混在一起会让 steering(本轮插话)和 followUp(跨轮接力)的注入时机糊掉。

为什么 steering 在内层而 followUp 在外层?steering 的语义是「在下一轮 LLM 调用前插话」,所以它必须在内层 while 顶部、调 streamAssistantResponse 之前被消费。followUp 是「agent 本来要停了再塞一个任务」,所以要等内层自然退出(hasMoreToolCalls=false 且 steering 空)再检查。

为什么 streamAssistantResponse 直接改 context.messages?因为 partial message 在流式期间就要让 UI 看到,而 LLM 调用又要基于「上一条 assistant 已写入」的 context。把 partial 推进 context 是最简单的做法,done 事件再把 partial 替换成 finalMessage。代价是 context 在流式中处于「半成品」状态,但循环只读不写,不会有竞态。

关键文件 ​

外层排 followUp、内层处理工具调用的骨架:

typescript
// packages/agent/src/agent-loop.ts:168-243
while (true) {
  let hasMoreToolCalls = true;
  while (hasMoreToolCalls || pendingMessages.length > 0) {
    if (!firstTurn) await emit({ type: "turn_start" }); else firstTurn = false;
    // 把 steering 推进 context ...
    const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFn);
    if (message.stopReason === "error" || message.stopReason === "aborted") { /* 发 turn_end/agent_end 返回 */ }
    const toolCalls = message.content.filter((c) => c.type === "toolCall");
    hasMoreToolCalls = false;
    if (toolCalls.length > 0) {
      const executedToolBatch = await executeToolCalls(currentContext, message, config, signal, emit);
      toolResults.push(...executedToolBatch.messages);
      hasMoreToolCalls = !executedToolBatch.terminate;
      // 把 toolResults 推进 context ...
    }
    await emit({ type: "turn_end", message, toolResults });
    if (await config.shouldStopAfterTurn?.(...)) { await emit({ type: "agent_end", ... }); return; }
    pendingMessages = (await config.getSteeringMessages?.()) || [];
  }
  const followUpMessages = (await config.getFollowUpMessages?.()) || [];
  if (followUpMessages.length > 0) { pendingMessages = followUpMessages; continue; }
  break;
}

streamAssistantResponse 在 stream 事件里直接改 context.messages 末尾的 partial message:

typescript
// packages/agent/src/agent-loop.ts:290-316
for await (const event of response) {
  switch (event.type) {
    case "start":
      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":
      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": { /* 替换为 finalMessage,发 message_end,返回 */ }
  }
}

数据流 ​

agent.prompt(text) 进来后的完整路径:

边界与失败 ​

小结 ​

runLoop 用双层 while 把「本轮工具调用循环」和「跨轮 followUp 接力」分开,steering 在内层顶部注入,followUp 在外层末尾接力,错误和 abort 短路退出。循环本身无状态,所有跨轮信息靠 Agent 传入的 context 和 getSteeringMessages/getFollowUpMessages 两个回调。工具调用怎么具体跑,看 工具执行 sequential/parallel;循环的入口封装和状态壳,看 Agent 类与生命周期。

MIT License — pi 架构拆解社区