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 類別與生命週期。