【Claude Code源码剖析】02-Agentic 查询循环与 QueryEngine

这是 Claude Code 最核心的模块。理解了它，就理解了整个系统的运作原理。

一、第一性原理：什么是 Agentic Loop？

传统 Chatbot：

User → LLM → Answer (一次性)

Agentic Loop：

User → LLM → [要用工具] → 执行工具 → [结果] → LLM → [还要用工具] → ... → 最终回答

核心区别：LLM 自主决定是否需要更多信息/操作，循环直到任务完成。

二、两个核心文件

文件	行数	角色
`query.ts`	1729 行	REPL 交互模式的查询循环（generator 函数）
`QueryEngine.ts`	1295 行	SDK/Headless 模式的查询引擎（class）

两者共享相同的核心逻辑，但：

query.ts 使用 Generator (function*)，便于 REPL 逐步渲染
QueryEngine.ts 使用 Class，便于 SDK 编程控制

三、`query()` 函数 — 核心循环解析

// query.ts (简化版核心逻辑)
export async function* query(
  userMessage: UserMessage,
  assistantMessages: Message[],
  systemPrompt: SystemPrompt,
  tools: Tools,
  // ...
): AsyncGenerator<StreamEvent> {

  while (true) {
    // ===== Step 1: 构建 API 请求 =====
    const messages = normalizeMessagesForAPI(allMessages);
    const config = buildQueryConfig(model, tools, systemPrompt);

    // ===== Step 2: 调用 Claude API (Streaming) =====
    const stream = await claudeAPI.stream(messages, config);

    // ===== Step 3: 处理流式响应 =====
    for await (const event of stream) {
      yield event;  // 将事件传给 UI 层渲染

      if (event.type === 'content_block_start' && event.content_block.type === 'tool_use') {
        // 开始收集 tool_use block
      }
    }

    // ===== Step 4: 检查停止原因 =====
    if (stopReason === 'end_turn') {
      break;  // LLM 认为任务完成，退出循环
    }

    if (stopReason === 'tool_use') {
      // ===== Step 5: 执行工具 =====
      const toolResults = await* runTools(toolUseBlocks, context);

      // ===== Step 6: 将 tool_result 加入消息历史 =====
      allMessages.push(...toolResults);

      // 继续循环 → 回到 Step 1
    }

    // ===== Step 7: Token 预算检查 =====
    if (exceedsBudget) break;

    // ===== Step 8: 自动压缩 =====
    if (needsCompaction) {
      await autoCompact(allMessages);
    }
  }
}

四、循环状态机

     ┌─────────────┐
     │  START      │
     └──────┬──────┘
            │
     ┌──────▼──────┐
┌───►│ Call Claude  │
│    │   API       │
│    └──────┬──────┘
│           │
│    ┌──────▼──────┐
│    │  Process    │
│    │  Stream     │◄── yield events (UI渲染)
│    └──────┬──────┘
│           │
│    ┌──────▼──────┐
│    │ stop_reason │
│    │ ?           │
│    └──┬─────┬───┘
│       │     │
│  tool_use  end_turn
│       │     │
│ ┌─────▼───┐ │    ┌──────────┐
│ │ Execute  │ └───►│  DONE    │
│ │ Tools    │      └──────────┘
│ └─────┬───┘
│       │
│ ┌─────▼─────────┐
│ │ Check Budget  │
│ │ Auto-Compact? │
│ └─────┬─────────┘
│       │
└───────┘

五、`QueryEngine` 类 — SDK 模式

export class QueryEngine {
  // 持久化状态 (跨 turn 保持)
  private mutableMessages: Message[];
  private readFileState: FileStateCache;
  private totalUsage: NonNullableUsage;
  private permissionDenials: SDKPermissionDenial[];

  constructor(config: QueryEngineConfig) {
    // 初始化对话状态
  }

  // 提交一条用户消息，开始一个 Turn
  async *submitMessage(userMessage: string): AsyncGenerator<SDKMessage> {
    // 1. 构建系统 prompt
    // 2. 调用 query() generator
    // 3. 转换事件为 SDK 格式
    // 4. yield SDK 事件
  }

  // 获取当前对话历史
  getMessages(): Message[] { ... }

  // 获取累计 token 用量
  getUsage(): NonNullableUsage { ... }
}

QueryEngine 的生命周期

SDK 调用方
    │
    ├─ new QueryEngine(config)         ← 创建实例
    │
    ├─ engine.submitMessage("修复 bug") ← Turn 1
    │   ├─ yield: assistant_message
    │   ├─ yield: tool_use (read file)
    │   ├─ yield: tool_result
    │   ├─ yield: assistant_message
    │   └─ yield: turn_complete
    │
    ├─ engine.submitMessage("再加测试") ← Turn 2
    │   └─ ... (复用之前的消息历史)
    │
    └─ engine.getUsage()               ← 获取统计

六、工具执行机制

6.1 并行 vs 串行

// toolOrchestration.ts
function partitionToolCalls(toolUseMessages, context): Batch[] {
  // 将工具调用分区:
  // - isConcurrencySafe=true 的连续工具 → 并行批次
  // - isConcurrencySafe=false 的工具    → 单独串行批次
}

核心算法：

工具序列: [FileRead, GrepSearch, FileRead, FileEdit, FileRead]
分区结果:
  Batch 1 (并行): [FileRead, GrepSearch, FileRead]  ← 只读工具可并行
  Batch 2 (串行): [FileEdit]                         ← 写入工具必须串行
  Batch 3 (并行): [FileRead]                         ← 只读恢复并行

每个工具通过 isConcurrencySafe() 方法声明自己是否可并行：

可并行: FileRead, Glob, Grep, WebSearch
不可并行: FileEdit, FileWrite, BashTool (可能有副作用)

6.2 StreamingToolExecutor

// StreamingToolExecutor.ts — 流式工具执行器
class StreamingToolExecutor {
  // 在 LLM 流式输出过程中就开始执行工具（不等 stream 结束）
  addTool(block: ToolUseBlock, assistantMessage: AssistantMessage)
  async *getRemainingResults(): AsyncGenerator<MessageUpdate>
}

优化原理：当 LLM 输出多个 tool_use block 时，不等全部输出完就开始执行前面的工具，减少等待时间。

6.3 最大并发控制

function getMaxToolUseConcurrency(): number {
  return parseInt(process.env.CLAUDE_CODE_MAX_TOOL_USE_CONCURRENCY || '', 10) || 10;
}

默认最多 10 个工具并行执行。

七、系统 Prompt 构建

// 每次 query 调用前构建
const systemPrompt = await fetchSystemPromptParts({
  systemContext: await getSystemContext(),     // git status, 日期等
  userContext: await getUserContext(),          // CLAUDE.md 内容
  customSystemPrompt,                          // 用户自定义
  appendSystemPrompt,                          // 追加内容
  toolDescriptions,                            // 工具描述列表
  mcpInstructions,                             // MCP 服务器说明
});

System Prompt 的分层结构：

[1] 核心系统指令 (constants/prompts.ts)
    ├── 角色定义 ("你是 Claude，一个 AI 编程助手")
    ├── 工具使用规范
    ├── 安全约束
    └── 输出格式要求

[2] 环境上下文 (context.ts)
    ├── 当前日期
    ├── Git 状态 (branch, status, recent commits)
    └── 平台信息

[3] 用户上下文
    ├── CLAUDE.md 文件内容 (项目级配置)
    ├── ~/.claude/CLAUDE.md (全局配置)
    └── 工作目录的 .claude/CLAUDE.md

[4] 工具定义
    ├── 内置工具 schema
    ├── MCP 工具 schema
    └── 插件工具 schema

[5] 附加上下文
    ├── Memory 文件
    └── 用户自定义追加内容

八、Token 预算管理

// query/tokenBudget.ts
export function createBudgetTracker(config) {
  return {
    checkBudget(outputTokens: number): 'continue' | 'stop' {
      if (outputTokens >= maxBudget) return 'stop';
      return 'continue';
    }
  };
}

预算系统用于 SDK 场景，防止无限循环消耗 token：

--max-turns <n>: 限制最大 turn 数
--max-budget <usd>: 限制最大花费
自动检测输出 token 是否超过上下文窗口

九、自动压缩触发

// 在每次 API 调用后检查
const warningState = calculateTokenWarningState(tokenUsage, model);

if (warningState.isAboveAutoCompactThreshold) {
  // 自动压缩对话历史
  const result = await compactConversation(messages, context);
  messages = result.compactedMessages;
}

阈值计算：

有效上下文窗口 = contextWindow - maxOutputTokens(model)
自动压缩阈值 = 有效上下文窗口 - 13,000 tokens (缓冲)
警告阈值 = 有效上下文窗口 - 20,000 tokens

十、错误处理与重试

// services/api/withRetry.ts
// API 调用失败时的重试策略:
// 1. 429 Rate Limit → 指数退避重试
// 2. 500/502/503 → 重试 (可能是临时故障)
// 3. prompt_too_long → 触发压缩后重试
// 4. 401 Unauthorized → 刷新 OAuth token 后重试
// 5. 其他错误 → 上报给用户

十一、Generator 模式的优势

query() 使用 async function* (异步 Generator) 的设计是关键：

// REPL.tsx 中消费 generator
for await (const event of query(message, messages, ...)) {
  switch (event.type) {
    case 'text':
      // 实时渲染文本
      updateUI(event.text);
      break;
    case 'tool_use':
      // 显示工具调用中...
      showToolProgress(event);
      break;
    case 'tool_result':
      // 显示工具结果
      showToolResult(event);
      break;
  }
}

为什么不用回调/EventEmitter？

Generator 保持了 顺序控制流，代码可读性好
调用方可以 暂停/恢复 消费（背压控制）
与 React 的渲染周期完美配合
错误可以通过 try/catch 正常捕获（不像 EventEmitter）