pi 架构拆解

Deutsch

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

Deutsch

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

Appearance

Doppelte while-Hauptschleife ​

源码版本v0.73.1

runLoop ist der Motor von pi-agent-core. Die äußere while (true) leert die followUp-Warteschlange, die innere while (hasMoreToolCalls || pendingMessages.length > 0) verarbeitet den Werkzeugaufrufzyklus und steering-Einwürfe; jede Runde ruft streamAssistantResponse auf, um eine assistant-Nachricht zu holen, bei toolCalls geht es über executeToolCalls zurück in die innere Schleife, sonst wird sie verlassen und die followUp-Warteschlange befragt. Die gesamte Schleife hält keinen Zustand, der gesamte Zustand wird vom Aufrufer (Agent) übergeben, Events werden über AgentEventSink gesendet.

Verantwortung ​

  1. Event-Orchestrierung: Jeder turn sendet turn_start, assistant-Nachrichten senden message_start/message_update/message_end, Werkzeugausführung sendet tool_execution_*, am turn-Ende turn_end, am Ende der gesamten Schleife agent_end. Siehe packages/agent/src/agent-loop.ts:155-246.
  2. Streaming der Assistant-Antwort: streamAssistantResponse ruft convertToLlm auf, um in Message[] zu konvertieren, sendet mit streamFn (default streamSimple) die Anfrage, konsumiert mit for await die Events und überschreibt die partial message am Ende von context.messages. Siehe packages/agent/src/agent-loop.ts:252-345.
  3. steering-Injektion: Am Anfang der inneren Schleife wird pendingMessages geprüft, steering-Nachrichten werden in currentContext.messages gepusht, bevor der nächste LLM-Aufruf läuft, siehe packages/agent/src/agent-loop.ts:180-188.
  4. followUp-Übergabe: Nach Verlassen der inneren Schleife ruft die äußere getFollowUpMessages auf, bei Nachrichten werden sie zurück in pendingMessages gesteckt und die innere Schleife erneut betreten, sonst break. Siehe packages/agent/src/agent-loop.ts:233-243.
  5. Frühzeitiger Stopp: shouldStopAfterTurn wird nach turn_end aufgerufen, bei true direkt agent_end senden und beenden, ohne die Warteschlange abzufragen, siehe packages/agent/src/agent-loop.ts:218-228.

Entwurfsmotivation ​

Warum zwei while-Schleifen statt einer? Weil es zwei "Weiter"-Semantiken gibt. Werkzeugaufrufe sind ein "Innerhalb-dieser-Runde"-Weiter - der assistant hat toolCalls gesendet, nach der Ausführung werden die Ergebnisse an den LLM zurückgefüttert, damit er weiterredet, das ist die innere Schleife. followUp ist ein Weiter nach "diese Runde wäre eigentlich zu Ende" - der Nutzer hat etwas eingereiht und wartet, bis der Agent fertig ist, das ist die äußere Schleife. Beides zusammen mischt steering (Einwurf in dieser Runde) und followUp (Übergabe über Runden hinweg) zeitlich unklar.

Warum steering innen und followUp außen? steering bedeutet semantisch "vor dem nächsten LLM-Aufruf einwerfen", also muss es oben in der inneren while konsumiert werden, bevor streamAssistantResponse aufgerufen wird. followUp bedeutet "der Agent wollte eigentlich anhalten, noch eine Aufgabe hinzufügen", also wird erst nach natürlichem Verlassen der inneren Schleife (hasMoreToolCalls=false und steering leer) geprüft.

Warum schreibt streamAssistantResponse direkt context.messages? Weil die partial message schon während des Streamings von der UI gesehen werden soll, aber der LLM-Aufruf auf dem context "letzte assistant ist schon geschrieben" basieren muss. Die partial in den context zu schieben ist die einfachste Lösung, das done-Event ersetzt die partial durch finalMessage. Der Preis ist, dass der context während des Streamings halbfertig ist, aber die Schleife liest nur, schreibt nicht, keine Race Condition.

Wichtige Dateien ​

Das Gerüst: außen followUp abfragen, innen Werkzeugaufrufe bearbeiten:

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 in context pushen ...
    const message = await streamAssistantResponse(currentContext, config, signal, emit, streamFn);
    if (message.stopReason === "error" || message.stopReason === "aborted") { /* turn_end/agent_end senden und 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 in context pushen ...
    }
    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 überschreibt in den stream-Events direkt die partial message am Ende von context.messages:

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": { /* ersetzen durch finalMessage, message_end senden, return */ }
  }
}

Datenfluss ​

Vollständiger Pfad, nachdem agent.prompt(text) reinkommt:

Grenzen und Fehler ​

  • Fehler-Kurzschluss: Wenn stopReason "error" oder "aborted" ist, wird direkt turn_end+agent_end gesendet und zurückgekehrt, keine Werkzeugausführung, keine steering/followUp-Abfrage, siehe packages/agent/src/agent-loop.ts:194-198.
  • transformContext-Vertrag: Wenn transformContext wirft, wird die Schleife ohne normale Event-Sequenz abgebrochen, deshalb verlangt der Typvertrag, dass es nicht wirft, sondern einen fallback zurückgibt, siehe packages/agent/src/agent-loop.ts:260-263 und Typvertrag.
  • steering erste Runde ueberspringen: runPromptMessages kann beim Aufruf von runAgentLoop skipInitialSteeringPoll übergeben, um zu vermeiden, dass ein gerade gesendeter prompt von steering unterbrochen wird, siehe packages/agent/src/agent.ts:374-388.
  • continue Rollen-Vorpruefung: runAgentLoopContinue/agentLoopContinue prüfen vor dem Aufruf von runLoop, dass die letzte Nachricht nicht assistant ist, sonst wird ein Fehler geworfen, siehe packages/agent/src/agent-loop.ts:127-133.
  • partial nicht in newMessages: Die partial message wird in currentContext.messages für den LLM-Kontext der nächsten Runde geschrieben, aber newMessages (das an den Aufrufer zurückgegebene Neue dieser Runde) schiebt nur nach message_end die vollständige finalMessage, siehe packages/agent/src/agent-loop.ts:191-192.

Zusammenfassung ​

runLoop trennt mit der doppelten while-Schleife den "Werkzeugaufrufzyklus dieser Runde" von der "followUp-Übergabe über Runden hinweg": steering wird oben in der inneren Schleife injiziert, followUp am Ende der äußeren Schleife übergeben, Fehler und abort verlassen kurzgeschlossen. Die Schleife selbst ist zustandslos, alle rundengreifenden Informationen kommen über den context und die beiden Callbacks getSteeringMessages/getFollowUpMessages, die von Agent übergeben werden. Wie Werkzeugaufrufe konkret laufen, siehe Werkzeugausführung sequential/parallel; für die Eingangsverpackung und Zustandshülle der Schleife siehe Agent-Klasse und Lebenszyklus.