Pi Agent — 코어 루프 분석

1. 아키텍처 개요

Pi는 2계층 구조로 설계되어 있다:

2. 코어 에이전트 루프 CORE

Pi의 심장부. 이중 while 루프 구조로 되어 있다.

2.1 runLoop() — 메인 루프

export async function runLoop(
  context: AgentContext,     // { systemPrompt, messages, tools }
  config: AgentLoopConfig,   // { convertToLlm, transformContext, beforeToolCall, ... }
  emit: (event: AgentEvent) => Promise<void>,
  signal?: AbortSignal
) {
  const currentContext = { ...context };
  const newMessages: AgentMessage[] = [];

  await emit({ type: "agent_start" });

  // ═══════════════════════════════════════════════════════════════
  // 바깥 루프: follow-up 메시지 처리
  // 에이전트가 "다 했다"고 판단한 후에도 사용자가 추가 지시를 줄 수 있음
  // ═══════════════════════════════════════════════════════════════
  let pendingMessages: AgentMessage[] = [];

  while (true) {

    // ═══════════════════════════════════════════════════════════════
    // 안쪽 루프: tool call + steering 메시지 반복
    // LLM이 도구를 호출하는 한 계속 돌아감
    // ═══════════════════════════════════════════════════════════════
    let hasMoreToolCalls = true;

    while (hasMoreToolCalls || pendingMessages.length > 0) {

      // (1) steering 메시지 주입 — 실행 중에 사용자가 방향을 수정
      if (pendingMessages.length > 0) {
        for (const msg of pendingMessages) {
          currentContext.messages.push(msg);
          newMessages.push(msg);
        }
        pendingMessages = [];
        hasMoreToolCalls = true;  // steering이 오면 무조건 한 번 더 LLM 호출
      }

      // (2) LLM 응답 스트리밍
      await emit({ type: "turn_start" });
      const message = await streamAssistantResponse(
        currentContext, config, emit, signal
      );

      // 어시스턴트 메시지를 컨텍스트에 추가
      currentContext.messages.push(message);
      newMessages.push(message);

      // 에러/중단이면 즉시 종료
      if (message.stopReason === "error" || message.stopReason === "aborted") {
        await emit({ type: "turn_end", message, toolResults: [] });
        await emit({ type: "agent_end", messages: newMessages });
        return;
      }

      // (3) tool call 추출
      const toolCalls = message.content.filter(c => c.type === "toolCall");
      hasMoreToolCalls = toolCalls.length > 0;

      // (4) 도구 실행 → 결과를 컨텍스트에 추가 ← 여기가 피드백 지점
      if (hasMoreToolCalls) {
        const toolResults = await executeToolCalls(
          currentContext, message, config, emit, signal
        );
        for (const result of toolResults) {
          currentContext.messages.push(result);   // ← 피드백! 다음 LLM 호출의 입력이 됨
          newMessages.push(result);
        }
      }

      await emit({ type: "turn_end", message, toolResults });

      // (5) 새 steering 메시지 확인
      pendingMessages = (await config.getSteeringMessages?.()) || [];
    }
    // 안쪽 루프 종료 — 모델이 도구를 호출하지 않았고, steering도 없음

    // ═══════════════════════════════════════════════════════════════
    // follow-up 확인: 에이전트가 멈추려 할 때 추가 메시지가 있는지
    // ═══════════════════════════════════════════════════════════════
    const followUpMessages = (await config.getFollowUpMessages?.()) || [];
    if (followUpMessages.length > 0) {
      pendingMessages = followUpMessages;
      continue;  // 바깥 루프 계속 → 에이전트 재개
    }

    break;  // 완전 종료
  }

  await emit({ type: "agent_end", messages: newMessages });
}

2.2 streamAssistantResponse() — LLM 호출

async function streamAssistantResponse(
  context: AgentContext,
  config: AgentLoopConfig,
  emit: EmitFn,
  signal?: AbortSignal
): Promise<AssistantMessage> {

  // 매 호출마다 전체 메시지 히스토리를 LLM 포맷으로 변환
  let messages = context.messages;

  // transformContext: 호출 전 메시지 변환 훅 (프루닝, 외부 데이터 주입 등)
  if (config.transformContext) {
    messages = await config.transformContext(messages, signal);
  }

  // AgentMessage[] → LLM 프로바이더별 Message[]로 변환
  // 이 변환이 LLM 호출 경계에서만 일어남 → 내부는 항상 AgentMessage
  const llmMessages = await config.convertToLlm(messages);

  const llmContext = {
    systemPrompt: context.systemPrompt,
    messages: llmMessages,
    tools: context.tools?.map(t => ({
      name: t.name,
      description: t.label,
      parameters: t.parameters
    }))
  };

  // 스트리밍 응답 — 실시간 이벤트 발행
  await emit({ type: "message_start" });
  const response = await streamSimple(config.model, llmContext, signal);
  await emit({ type: "message_end", message: response });

  return response;
}

2.3 executeToolCalls() — 도구 실행

async function executeToolCalls(
  context: AgentContext,
  message: AssistantMessage,
  config: AgentLoopConfig,
  emit: EmitFn,
  signal?: AbortSignal
): Promise<ToolResultMessage[]> {

  const toolCalls = message.content.filter(c => c.type === "toolCall");
  const results: ToolResultMessage[] = [];

  // 도구별 실행 모드 확인: parallel(기본) vs sequential
  const allParallel = toolCalls.every(tc => {
    const tool = context.tools?.find(t => t.name === tc.name);
    return tool?.executionMode !== "sequential";
  });

  if (allParallel && toolCalls.length > 1) {
    // 병렬 실행 — 모든 도구가 parallel 모드일 때
    const promises = toolCalls.map(tc => executeSingleTool(tc, context, config, emit, signal));
    results.push(...await Promise.all(promises));
  } else {
    // 순차 실행 — 하나라도 sequential이면 전체 순차
    for (const tc of toolCalls) {
      results.push(await executeSingleTool(tc, context, config, emit, signal));
    }
  }

  return results;
}

async function executeSingleTool(toolCall, context, config, emit, signal) {

  // (1) 도구 찾기
  const tool = context.tools?.find(t => t.name === toolCall.name);
  if (!tool) return errorResult(`Unknown tool: ${toolCall.name}`);

  // (2) 인자 검증 + 정규화
  const args = validateToolArguments(toolCall.args, tool.parameters);
  const prepared = tool.prepareArguments?.(args) ?? args;

  // (3) beforeToolCall 훅 — 차단 가능
  const beforeResult = await config.beforeToolCall?.(toolCall.id, tool.name, prepared);
  if (beforeResult?.block) {
    return blockedResult(toolCall.id, beforeResult.message);
  }

  // (4) 실제 실행
  await emit({ type: "tool_execution_start", toolCall });
  const result = await tool.execute(toolCall.id, prepared, signal, onUpdate);
  await emit({ type: "tool_execution_end", toolCall, result });

  // (5) afterToolCall 훅 — 결과 수정 가능
  const afterResult = await config.afterToolCall?.(toolCall.id, tool.name, result);
  if (afterResult) {
    result.content = afterResult.content ?? result.content;
    result.isError = afterResult.isError ?? result.isError;
  }

  return toToolResultMessage(toolCall.id, result);
}

3. Agent 클래스 — 상태 관리와 메시지 큐

export class Agent {
  private context: AgentContext;
  private subscribers: Set<(event: AgentEvent) => void>;
  private abortController: AbortController | null;

  // ─── 두 종류의 메시지 큐 ───
  private steeringQueue: AgentMessage[];  // 실행 중 주입
  private followUpQueue: AgentMessage[];  // 종료 시점 주입
  private queueMode: "all" | "one-at-a-time";

  // ─── steer(): 에이전트 실행 도중 방향 수정 ───
  // 사용자가 "아, 그거 말고 이걸 해봐"라고 할 때
  steer(messages: AgentMessage[]) {
    this.steeringQueue.push(...messages);
  }

  // ─── followUp(): 에이전트가 완료했다고 판단한 후 추가 지시 ───
  // 사용자가 "좋아, 이제 테스트도 작성해줘"라고 할 때
  followUp(messages: AgentMessage[]) {
    this.followUpQueue.push(...messages);
  }

  // ─── prompt(): 새 대화 시작 ───
  async prompt(messages: AgentMessage[]) {
    if (this.isRunning) throw new Error("Already running");

    this.context.messages.push(...messages);
    this.abortController = new AbortController();

    await runLoop(
      this.context,
      {
        ...this.config,
        // runLoop가 매 턴마다 이 콜백들을 호출
        getSteeringMessages: () => this.drainQueue(this.steeringQueue),
        getFollowUpMessages: () => this.drainQueue(this.followUpQueue),
      },
      this.emit.bind(this),
      this.abortController.signal
    );
  }

  // ─── continue(): 중단 후 재개 (컴팩션 후 등) ───
  async continue() {
    // prompt()와 동일하지만 메시지를 추가하지 않고 기존 컨텍스트로 재개
    await runLoop(this.context, this.config, this.emit.bind(this));
  }

  // ─── 큐 드레인 ───
  private drainQueue(queue: AgentMessage[]): AgentMessage[] {
    if (this.queueMode === "all") {
      const all = [...queue];
      queue.length = 0;
      return all;
    }
    // one-at-a-time: 하나만 꺼내서 처리
    return queue.length > 0 ? [queue.shift()!] : [];
  }
}

4. 컨텍스트 압축 MEMORY

4.1 압축 트리거 판단

// 압축이 필요한지 판단
export function shouldCompact(
  contextTokens: number,
  contextWindow: number,
  settings: CompactionSettings
): boolean {
  const reserveTokens = settings.reserveTokens ?? 16384;

  // 컨텍스트 토큰 > (윈도우 크기 - 예약분) 이면 압축
  // 예: 128K 윈도우에서 111,616 토큰 이상이면 트리거
  return contextTokens > contextWindow - reserveTokens;
}

// 토큰 추정: chars / 4 휴리스틱, 실제 usage 데이터가 있으면 우선 사용
export function estimateContextTokens(
  messages: AgentMessage[],
  lastKnownUsage?: { promptTokens: number }
): number {
  if (lastKnownUsage) {
    // 마지막 API 응답의 실제 토큰 수 + 그 이후 추가된 메시지의 추정치
    const trailingTokens = estimateTrailingTokens(messages, lastKnownUsage);
    return lastKnownUsage.promptTokens + trailingTokens;
  }
  // 실제 데이터 없으면 전체를 chars/4로 추정
  return Math.ceil(totalChars(messages) / 4);
}

4.2 컷 포인트 결정 알고리즘

export function findCutPoint(
  messages: AgentMessage[],
  settings: CompactionSettings
): { cutIndex: number; recentMessages: AgentMessage[] } {

  const keepRecentTokens = settings.keepRecentTokens ?? 20000;
  let accumulatedTokens = 0;

  // 뒤에서부터 역순으로 토큰 누적
  for (let i = messages.length - 1; i >= 0; i--) {
    accumulatedTokens += estimateMessageTokens(messages[i]);

    if (accumulatedTokens >= keepRecentTokens) {
      // 유효한 컷 포인트인지 확인
      // user, assistant, custom 메시지에서만 자를 수 있음
      // tool result 중간에서는 절대 자르지 않음 (tool call과 결과가 분리되면 안 됨)
      const cutIndex = findValidCutPoint(messages, i);
      return {
        cutIndex,
        recentMessages: messages.slice(cutIndex)
      };
    }
  }

  // 전체가 keepRecentTokens 이내면 자르지 않음
  return { cutIndex: 0, recentMessages: messages };
}

4.3 LLM 기반 요약 생성

const SUMMARIZATION_PROMPT = `
Summarize the conversation into a structured checkpoint.
Use this EXACT format:

## Goal
[What the user wants to achieve]

## Constraints & Preferences
[Any restrictions, preferences, or requirements mentioned]

## Progress
### Done
- [Completed items]
### In Progress
- [Current work]
### Blocked
- [Blocking issues]

## Key Decisions
- [Important decisions made and their rationale]

## Next Steps
- [What should happen next]

## Critical Context
- [Anything else essential for continuing the work]
`;

// 반복적 요약 업데이트: 이전 요약 + 새 메시지 → 병합된 요약
const UPDATE_SUMMARIZATION_PROMPT = `
Here is the PREVIOUS summary checkpoint:
---
{previousSummary}
---

And here are the NEW messages since that checkpoint.
UPDATE the summary by merging new information.
Keep the same structure. Remove items that are no longer relevant.
`;

export async function generateSummary(
  messagesToSummarize: AgentMessage[],
  previousSummary: string | null,
  model: string
): Promise<string> {

  const prompt = previousSummary
    ? UPDATE_SUMMARIZATION_PROMPT.replace("{previousSummary}", previousSummary)
    : SUMMARIZATION_PROMPT;

  // 파일 추적: 요약할 메시지들에서 읽은/수정한 파일 목록 추출
  const fileOps = extractFileOperations(messagesToSummarize);

  const summary = await llm.complete(prompt, messagesToSummarize);

  // 파일 목록을 요약에 첨부 (이전 compaction의 파일 목록도 누적)
  return summary + "\n\n## Files\n" + formatFileOps(fileOps);
}

5. 도구 설계 TOOLS

5.1 AgentTool 인터페이스

interface AgentTool<TParameters, TDetails> {
  name: string;
  label: string;                           // LLM에 보여줄 설명
  parameters: TSchema;                     // TypeBox 스키마

  execute(
    toolCallId: string,
    params: Static<TParameters>,
    signal?: AbortSignal,
    onUpdate?: (update: ToolUpdate) => void  // 스트리밍 업데이트
  ): Promise<AgentToolResult<TDetails>>;

  prepareArguments?: (args: unknown) => Static<TParameters>;  // 인자 정규화
  executionMode?: "sequential" | "parallel";                // 기본: parallel
}

interface AgentToolResult<T> {
  content: (TextContent | ImageContent)[];  // LLM에 전달 (토큰 절약)
  details: T;                               // UI 전용 (풍부한 정보)
  isError?: boolean;
}

5.2 4개 핵심 도구

5.3 Bash 도구 구현 (대표 예시)

// tools/bash.ts — 가장 복잡한 도구

interface BashOperations {
  // 플러그 가능한 실행 백엔드 — 로컬 셸 또는 SSH
  spawn(command: string, options: SpawnOptions): ChildProcess;
  kill(pid: number): void;
}

export const bashTool: ToolDefinition = {
  name: "bash",
  parameters: Type.Object({
    command: Type.String({ description: "The bash command to execute" }),
    timeout: Type.Optional(Type.Number({ description: "Timeout in ms" }))
  }),

  async execute(toolCallId, { command, timeout }, signal, onUpdate) {
    const process = ops.spawn(command, { timeout: timeout ?? 120000 });

    // 스트리밍 출력 — onUpdate로 실시간 전달
    let output = "";
    process.stdout.on("data", (chunk) => {
      output += chunk;
      onUpdate?.({ type: "partial", content: chunk });
    });

    await process.wait();

    // Tail truncation: 마지막 200줄 또는 32KB만 LLM에 전달
    const truncated = tailTruncate(output, {
      maxLines: 200,
      maxBytes: 32768
    });

    // 전체 출력은 임시 파일에 보존 (사용자가 나중에 확인 가능)
    if (truncated.wasTruncated) {
      const tmpPath = await saveTempFile(output);
      truncated.content += `\n[Full output saved to ${tmpPath}]`;
    }

    return {
      content: [{ type: "text", text: truncated.content }],
      details: { exitCode: process.exitCode, fullOutput: output }
    };
  }
};

6. 세션 관리와 에러 복구

class AgentSession {

  async prompt(messages: AgentMessage[]) {
    await this.agent.prompt(messages);
    await this.waitForRetry();  // 재시도가 진행 중이면 대기
  }

  // ─── 에이전트 이벤트 핸들러 ───
  private async _processAgentEvent(event: AgentEvent) {
    if (event.type === "agent_end") {

      // (1) Retryable 에러? → 지수 백오프 재시도
      if (isRetryableError(event)) {
        await this._handleRetryableError(event);
        return;
      }

      // (2) Context overflow? → 에러 메시지 제거 + 컴팩션 + 재개
      if (isContextOverflow(event)) {
        // 실패한 응답을 메시지 히스토리에서 제거
        this.removeLastErrorMessage();
        await this.compact();          // LLM 요약으로 압축
        await this.agent.continue();   // 루프 재개
        return;
      }

      // (3) 정상 종료 → threshold 확인 후 자동 컴팩션
      await this._checkCompaction();
    }
  }

  // ─── 자동 컴팩션 체크 ───
  private async _checkCompaction() {
    const tokens = estimateContextTokens(this.agent.messages, this.lastUsage);
    if (shouldCompact(tokens, this.contextWindow, this.settings)) {
      await this.compact();
      // 컴팩션 후 재시도하지 않음 — 다음 사용자 입력을 기다림
    }
  }

  // ─── Retryable 에러 처리 ───
  private async _handleRetryableError(event: AgentEvent) {
    const delay = Math.min(1000 * Math.pow(2, this.retryCount), 30000);
    await sleep(delay);
    this.retryCount++;
    await this.agent.continue();
  }
}

7. 종합 인사이트