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 メッセージを 1 件取り、toolCalls があれば executeToolCalls に回してまた内側へ、無ければ内側を抜けて followUp キューを問い合わせる。ループ全体は状態を持たず、すべての状態は呼び出し側 (Agent) が渡し、イベントは AgentEventSink 経由で送られる。

責務 ​

  1. イベント編成:各ターンで turn_start、assistant メッセージで message_start/message_update/message_end、ツール実行で tool_execution_*、ターン末尾で 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 メッセージを currentContext.messages に push してから次の 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 参照。

設計動機 ​

なぜ 1 層ではなく 2 層の while なのか? 「続き」の意味が 2 種類あるからだ。ツール呼び出しは「本ターン内」の続き——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 に push ...
    const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFn);
    if (message.stopReason === "error" || message.stopReason === "aborted") { /* turn_end/agent_end 送って return */ }
    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 に push ...
    }
    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 送って return */ }
  }
}

データフロー ​

agent.prompt(text) が入ってきた後の完全なパス:

境界と失敗 ​

  • エラーショートサーキット:stopReason が "error" か "aborted" の時、直接 turn_end+agent_end を送って return する。ツールを実行せず、steering/followUp も問わない。packages/agent/src/agent-loop.ts:194-198 参照。
  • transformContext 契約:transformContext が throw するとループが中断し正常なイベント列が生まれない。そのため型契約は「throw せず fallback を返す」を要求する。packages/agent/src/agent-loop.ts:260-263 と 型契約 参照。
  • steering 初回スキップ:runPromptMessages は runAgentLoop 呼び出し時に skipInitialSteeringPoll を渡せて、prompt 直後に steering に奪われるのを防ぐ。packages/agent/src/agent.ts:374-388 参照。
  • continue のロール事前検証:runAgentLoopContinue/agentLoopContinue はどちらも runLoop 呼び出し前に末尾メッセージが assistant でないことをチェックし、違反なら throw する。packages/agent/src/agent-loop.ts:127-133 参照。
  • partial は newMessages に入れない:partial message は次ターンの LLM コンテキストのために currentContext.messages に書き込まれるが、newMessages (呼び出し側に返す本ターンの新規分) には message_end 後にだけ完全な finalMessage を push する。packages/agent/src/agent-loop.ts:191-192 参照。

小ねた ​

runLoop は二重 while で「本ターンのツール呼び出しループ」と「ターン越え followUp リレー」を分ける。steering は内側先頭で注入、followUp は外側末尾でリレー、エラーと abort はショートサーキットで抜ける。ループ自体はステートレスで、ターン越えの情報はすべて Agent が渡す context と getSteeringMessages/getFollowUpMessages の 2 つのコールバックに頼る。ツール呼び出しが具体的にどう走るかは ツール実行 sequential/parallel、ループの入口ラッパーと状態の殻は Agent クラスとライフサイクル 参照。