本文基于 Claude Code 真实源码，深度剖析一个生产级 AI Agent 平台的核心运行时循环是如何设计与实现的。从教学骨架到 1700+ 行的生产代码，揭示 Agent 从"一问一答"进化为"持续决策系统"的全部关键机制。

一、什么是 Agent 主循环

传统 LLM 应用是请求-响应模式：用户问一个问题，模型回一个答案，交互结束。

Agent 则是一个持续决策循环：

用户输入 → 模型推理 → 判断动作类型 → 执行工具 → 结果回写上下文 → 继续下一轮 → ... → 终止

这个循环的本质是：模型不再只是"回答者"，而是一个"决策者"——每一轮它都在决定：是直接回复用户，还是调用工具继续推进任务。

二、教学版最小实现

先看最小骨架如何表达这个思想（来自 src/agt/agent.py）：

def run(self, user_input: str) -> str:
    self.add_message("user", user_input)

    for turn in range(self.max_turns):
        step = self.model_step()

        if step["type"] == "message":
            content = step["content"]
            self.add_message("assistant", content, turn=turn)
            return content

        if step["type"] == "tool_call":
            tool_name = step["tool"]
            tool_input = step.get("input", {})

            if not self.can_use_tool(tool_name):
                self.add_message("tool_result", f"Tool not allowed: {tool_name}", ok=False)
                continue

            result = self.tools[tool_name].call(tool_input)
            self.add_message("tool_result", result.content, ok=result.ok)
            continue

    return "Agent stopped because it reached max_turns."

核心逻辑只有三件事：

1. 调用模型得到决策（model_step()）
2. 如果是 message → 结束循环
3. 如果是 tool_call → 执行工具，把结果写回消息列表，继续循环

这 30 行代码就是 Agent 主循环的本质。接下来看生产级实现如何在这个骨架上叠加真实世界的复杂性。

三、生产级实现架构总览

Claude Code 的主循环实现在 src/query.ts 中，核心函数是 queryLoop()，约 1500 行。它的整体结构如下：

QueryEngine.submitMessage()
    └── query()
        └── queryLoop()   ← 核心循环
            ├── 上下文预算管理（snip / microcompact / autocompact）
            ├── 调用模型（streaming）
            ├── 工具执行（串行 / 并行 / 流式）
            ├── 错误恢复（prompt-too-long / max-output-tokens）
            ├── Stop Hooks 处理
            ├── 附件注入（记忆 / 通知 / Skill 发现）
            └── 状态转移 → 下一轮 or 终止

四、循环状态定义

生产系统必须用显式状态对象来追踪循环进程，而不是散落在局部变量中：

type State = {
  messages: Message[]                          // 当前消息序列
  toolUseContext: ToolUseContext                // 工具执行上下文
  autoCompactTracking: AutoCompactTrackingState // 自动压缩追踪
  maxOutputTokensRecoveryCount: number         // 输出截断恢复计数
  hasAttemptedReactiveCompact: boolean         // 是否已尝试响应式压缩
  maxOutputTokensOverride: number | undefined  // 输出 token 上限覆盖
  pendingToolUseSummary: Promise<...>          // 异步工具摘要
  stopHookActive: boolean | undefined          // stop hook 状态
  turnCount: number                            // 当前轮次
  transition: Continue | undefined             // 上一轮为何继续（用于调试）
}

每一轮结束时，通过构造新的 State 对象并 continue 来进入下一轮。transition 字段记录了为什么上一轮选择继续（如 next_turn、reactive_compact_retry、max_output_tokens_recovery），这让调试和测试能精确断言恢复路径。

五、单轮执行流程详解

5.1 循环入口与参数

export type QueryParams = {
  messages: Message[]              // 消息历史
  systemPrompt: SystemPrompt       // 动态拼装的系统提示词
  userContext: { [k: string]: string }   // 用户上下文
  systemContext: { [k: string]: string } // 系统上下文
  canUseTool: CanUseToolFn         // 权限判断函数
  toolUseContext: ToolUseContext    // 工具执行环境
  fallbackModel?: string           // 备用模型
  maxTurns?: number                // 最大轮次
  taskBudget?: { total: number }   // 任务 token 预算
}

5.2 上下文预算管理（模型调用前）

在调用模型之前，需要确保消息序列不会超出模型的上下文窗口。这是通过多层压缩策略实现的：

原始消息 → Tool Result 预算裁剪 → Snip 压缩 → Microcompact → Context Collapse → AutoCompact

（1）Tool Result 预算裁剪

工具返回的结果可能非常长（比如一个 grep 命令返回 10000 行），必须先裁剪：

messagesForQuery = await applyToolResultBudget(
  messagesForQuery,
  toolUseContext.contentReplacementState,
  persistReplacements ? records => void recordContentReplacement(...) : undefined,
  exemptTools,  // 某些工具不受限制
)

（2）Snip 压缩

对消息历史中的旧部分进行"剪断"式压缩，释放 token 空间：

const snipResult = snipModule.snipCompactIfNeeded(messagesForQuery)
messagesForQuery = snipResult.messages
snipTokensFreed = snipResult.tokensFreed

（3）Microcompact

更细粒度的压缩——对单个工具调用结果进行内联缩减：

const microcompactResult = await deps.microcompact(messagesForQuery, toolUseContext, querySource)
messagesForQuery = microcompactResult.messages

（4）Context Collapse

按"折叠"逻辑隐藏已完成步骤的冗余细节，保留结构性摘要：

const collapseResult = await contextCollapse.applyCollapsesIfNeeded(
  messagesForQuery, toolUseContext, querySource
)
messagesForQuery = collapseResult.messages

（5）AutoCompact（自动摘要压缩）

当 token 数超过阈值时，调用一个小模型生成历史摘要，替换旧消息：

const { compactionResult } = await deps.autocompact(
  messagesForQuery, toolUseContext, cacheSafeParams, querySource, tracking, snipTokensFreed
)
if (compactionResult) {
  messagesForQuery = buildPostCompactMessages(compactionResult)
}

5.3 模型调用（流式）

上下文准备完毕后，调用模型获取响应。Claude Code 使用流式调用：

for await (const message of deps.callModel({
  messages: prependUserContext(messagesForQuery, userContext),
  systemPrompt: fullSystemPrompt,
  tools: toolUseContext.options.tools,
  signal: toolUseContext.abortController.signal,
  options: {
    model: currentModel,
    fallbackModel,
    maxOutputTokensOverride,
    // ... 其他配置
  },
})) {
  // 处理流式消息
  if (message.type === 'assistant') {
    assistantMessages.push(message)
    // 检测 tool_use block
    const toolBlocks = message.message.content.filter(c => c.type === 'tool_use')
    if (toolBlocks.length > 0) {
      toolUseBlocks.push(...toolBlocks)
      needsFollowUp = true  // 标记需要执行工具后继续循环
    }
  }
}

关键设计点：

• 流式处理允许在模型输出的同时就开始准备工具执行（流式工具执行器）
• needsFollowUp 标志决定循环是否继续
• 支持模型降级（fallback）：当主模型过载时自动切换备用模型

5.4 判断循环去向

模型响应完成后，有两条路径：

路径 A：无工具调用（needsFollowUp = false）→ 准备终止

进入终止前的检查链：

1. Prompt-too-long 恢复（响应式压缩）
2. Max-output-tokens 恢复（注入"继续"指令）
3. Stop Hooks 执行
4. Token Budget 检查
5. 全部通过 → return { reason: 'completed' }

路径 B：有工具调用（needsFollowUp = true）→ 执行工具后继续

进入工具执行，完成后构造新 State 进入下一轮。

5.5 工具执行编排

工具执行不是简单的逐个调用，而是有并发策略：

// 来自 toolOrchestration.ts
export async function* runTools(toolUseBlocks, assistantMessages, canUseTool, toolUseContext) {
  for (const { isConcurrencySafe, blocks } of partitionToolCalls(toolUseBlocks, toolUseContext)) {
    if (isConcurrencySafe) {
      // 只读工具（如 Read、Grep）可以并行执行
      yield* runToolsConcurrently(blocks, ...)
    } else {
      // 有副作用的工具（如 Write、Bash）串行执行
      yield* runToolsSerially(blocks, ...)
    }
  }
}

分区逻辑：

• 连续的只读工具 → 合并为一批并行执行
• 有副作用的工具 → 单独串行执行
• 最大并发度可通过环境变量控制（默认 10）

此外还支持流式工具执行（StreamingToolExecutor）：在模型还在输出时，已经完成的 tool_use block 就开始执行，不等整个响应结束。

5.6 工具结果回流与附件注入

工具执行完成后，结果必须以结构化方式回写到消息序列：

// 工具结果回流
for await (const update of toolUpdates) {
  if (update.message) {
    yield update.message  // 向外部流输出
    toolResults.push(...normalizeMessagesForAPI([update.message], tools))
  }
}

// 附件注入：记忆、通知、Skill 发现
for await (const attachment of getAttachmentMessages(...)) {
  yield attachment
  toolResults.push(attachment)
}

附件注入是一个重要机制——它让系统能在工具执行后、下一轮模型调用前，注入额外的上下文信息：

• 记忆预取结果：异步加载的相关记忆
• 队列命令：用户在模型执行期间发送的新指令
• Skill 发现：自动发现的相关技能提示

5.7 状态转移进入下一轮

一切就绪后，构造新的 State 对象进入下一轮：

const next: State = {
  messages: [...messagesForQuery, ...assistantMessages, ...toolResults],
  toolUseContext: toolUseContextWithQueryTracking,
  autoCompactTracking: tracking,
  turnCount: nextTurnCount,
  maxOutputTokensRecoveryCount: 0,       // 重置恢复计数
  hasAttemptedReactiveCompact: false,     // 重置压缩尝试标记
  pendingToolUseSummary: nextPendingToolUseSummary,
  transition: { reason: 'next_turn' },
}
state = next
// continue → 回到 while(true) 顶部

六、终止条件详解

循环可能因以下原因终止：

终止原因	触发条件	返回值
completed	模型不再请求工具调用，且通过所有 stop hooks	{ reason: 'completed' }
max_turns	超过最大轮次限制	{ reason: 'max_turns' }
aborted_streaming	用户中断（Ctrl+C），在流式阶段	{ reason: 'aborted_streaming' }
aborted_tools	用户中断，在工具执行阶段	{ reason: 'aborted_tools' }
hook_stopped	Hook 明确阻止继续	{ reason: 'hook_stopped' }
stop_hook_prevented	Stop Hook 中止循环	{ reason: 'stop_hook_prevented' }
blocking_limit	token 数达到硬上限	{ reason: 'blocking_limit' }
prompt_too_long	提示词过长且无法恢复	{ reason: 'prompt_too_long' }
model_error	模型调用异常	{ reason: 'model_error', error }
image_error	图片尺寸/格式错误	{ reason: 'image_error' }

七、错误恢复机制

生产系统不能在遇到错误时直接崩溃，需要内建多层恢复策略。

7.1 Prompt-Too-Long 恢复

当模型返回"提示词过长"错误时：

第一步：尝试 Context Collapse drain（释放已折叠的上下文）
  ↓ 如果仍然过长
第二步：尝试 Reactive Compact（紧急摘要压缩）
  ↓ 如果仍然失败
第三步：向用户显示错误

if (isWithheld413) {
  // 第一步：drain collapsed context
  const drained = contextCollapse.recoverFromOverflow(messagesForQuery, querySource)
  if (drained.committed > 0) {
    state = { ...state, messages: drained.messages, transition: { reason: 'collapse_drain_retry' } }
    continue
  }
}
// 第二步：reactive compact
const compacted = await reactiveCompact.tryReactiveCompact({ ... })
if (compacted) {
  state = { ...state, messages: buildPostCompactMessages(compacted), transition: { reason: 'reactive_compact_retry' } }
  continue
}
// 第三步：无法恢复，终止
yield lastMessage
return { reason: 'prompt_too_long' }

7.2 Max-Output-Tokens 恢复

当模型输出被截断时（达到输出 token 上限）：

第一步：升级 token 上限（8k → 64k）重试
  ↓ 如果仍然被截断
第二步：注入"继续"指令，让模型接续输出（最多 3 次）
  ↓ 如果 3 次仍未完成
第三步：放弃恢复，显示截断的输出

// 升级重试
if (maxOutputTokensOverride === undefined) {
  state = { ...state, maxOutputTokensOverride: ESCALATED_MAX_TOKENS, transition: { reason: 'max_output_tokens_escalate' } }
  continue
}
// 注入继续指令
if (maxOutputTokensRecoveryCount < MAX_OUTPUT_TOKENS_RECOVERY_LIMIT) {
  const recoveryMessage = createUserMessage({
    content: 'Output token limit hit. Resume directly — no apology, no recap...',
    isMeta: true,
  })
  state = { ...state, messages: [...messages, ...assistantMessages, recoveryMessage], maxOutputTokensRecoveryCount: count + 1 }
  continue
}

7.3 模型 Fallback

当主模型过载时自动降级到备用模型：

catch (innerError) {
  if (innerError instanceof FallbackTriggeredError && fallbackModel) {
    currentModel = fallbackModel
    attemptWithFallback = true
    // 清理已有的部分响应
    assistantMessages.length = 0
    toolResults.length = 0
    // 通知用户
    yield createSystemMessage(`Switched to ${fallbackModel} due to high demand`)
    continue  // 用新模型重试
  }
}

八、Stop Hooks 机制

每轮结束前（模型决定不再调用工具时），系统会执行 Stop Hooks：

const stopHookResult = yield* handleStopHooks(
  messagesForQuery, assistantMessages, systemPrompt, userContext, systemContext, toolUseContext, querySource, stopHookActive
)

if (stopHookResult.preventContinuation) {
  return { reason: 'stop_hook_prevented' }
}

if (stopHookResult.blockingErrors.length > 0) {
  // Hook 返回阻断性错误，将错误注入消息后继续循环
  state = { ...state, messages: [...messages, ...assistantMessages, ...blockingErrors], stopHookActive: true }
  continue
}

Stop Hooks 的用途：

• 代码格式检查（lint）
• 安全审计
• 组织规则合规检查
• 自动记忆提取

Hook 可以返回三种结果：

• 通过 → 循环正常终止
• 阻断性错误 → 错误注入消息，模型看到错误后会尝试修复，循环继续
• 阻止继续 → 循环强制终止

九、Token Budget 自动续航

当用户设置了 token budget 时，系统会在模型"主动结束"时检查是否还有预算剩余：

const decision = checkTokenBudget(budgetTracker, agentId, budget, turnOutputTokens)

if (decision.action === 'continue') {
  // 还有预算，注入提示让模型继续
  state = {
    ...state,
    messages: [...messages, ...assistantMessages, createUserMessage({ content: nudgeMessage, isMeta: true })],
    transition: { reason: 'token_budget_continuation' },
  }
  continue
}
// 预算耗尽或收益递减，正常终止

预算检查逻辑：

• 已用 token < 预算 × 90% → 继续（注入 nudge 消息）
• 连续 3+ 次续航且每次增量 < 500 token → 收益递减，停止
• 达到 90% 或无预算 → 正常停止

十、QueryEngine：会话级封装

QueryEngine 是主循环之上的会话管理层，负责：

class QueryEngine {
  private mutableMessages: Message[]    // 完整会话历史
  private abortController: AbortController  // 中断控制
  private totalUsage: NonNullableUsage  // 累计 token 用量
  private readFileState: FileStateCache // 文件读取缓存

  async *submitMessage(prompt) {
    // 1. 处理用户输入（斜杠命令、附件等）
    const { messages, shouldQuery } = await processUserInput(...)
    
    // 2. 持久化用户消息到 transcript
    await recordTranscript(messages)
    
    // 3. 调用核心 query loop
    for await (const message of query({ messages, systemPrompt, ... })) {
      // 4. 记录每条消息到 transcript
      // 5. 追踪 token 用量
      // 6. 检查费用预算
      // 7. 对外 yield SDK 格式消息
    }
    
    // 8. 返回最终结果
    yield { type: 'result', subtype: 'success', ... }
  }
}

十一、完整数据流图

┌─────────────────────────────────────────────────────────────────────┐
│                        QueryEngine.submitMessage()                     │
│  ┌───────────────────────────────────────────────────────────────┐  │
│  │                        queryLoop()                              │  │
│  │                                                                 │  │
│  │  ┌─────────────┐    ┌──────────────┐    ┌─────────────────┐  │  │
│  │  │ 上下文压缩   │───→│  调用模型     │───→│ 解析模型响应     │  │  │
│  │  │ • snip      │    │  (streaming)  │    │ • message?      │  │  │
│  │  │ • micro     │    │              │    │ • tool_call?    │  │  │
│  │  │ • collapse  │    │              │    │                 │  │  │
│  │  │ • auto      │    │              │    │                 │  │  │
│  │  └─────────────┘    └──────────────┘    └────────┬────────┘  │  │
│  │                                                   │            │  │
│  │                    ┌──────────────────────────────┼────┐       │  │
│  │                    │                              │    │       │  │
│  │                    ▼                              ▼    │       │  │
│  │  ┌─────────────────────┐          ┌────────────────┐ │       │  │
│  │  │ needsFollowUp=false │          │ needsFollowUp  │ │       │  │
│  │  │                     │          │ =true           │ │       │  │
│  │  │ • 413恢复           │          │                │ │       │  │
│  │  │ • max_output恢复    │          │ 执行工具       │ │       │  │
│  │  │ • Stop Hooks        │          │ • 并行/串行    │ │       │  │
│  │  │ • Token Budget      │          │ • 流式执行     │ │       │  │
│  │  │                     │          │                │ │       │  │
│  │  └────────┬────────────┘          └───────┬────────┘ │       │  │
│  │           │                               │          │       │  │
│  │           ▼                               ▼          │       │  │
│  │  ┌─────────────┐              ┌─────────────────┐    │       │  │
│  │  │   终止       │              │ 注入附件/记忆   │    │       │  │
│  │  │ return {...} │              │ 构造新 State    │    │       │  │
│  │  └─────────────┘              │ continue        │────┘       │  │
│  │                               └─────────────────┘            │  │
│  └───────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────┘

十二、关键设计原则总结

原则	实现方式
显式状态	State 类型定义所有循环状态，每轮通过新对象传递
流式处理	AsyncGenerator 贯穿全链路，支持边生成边消费
多层压缩	snip → micro → collapse → auto，渐进式释放 token
优雅恢复	413/截断/降级三类错误都有自动恢复路径
并发工具	只读工具自动并行，有副作用工具保持串行
可观测性	transition 字段、queryCheckpoint、logEvent 全程埋点
可中断	AbortController 支持任意时刻中断，自动清理
可扩展	Hook 机制在关键点注入外部逻辑

十三、从教学版到生产版的演进路径

如果你要从零实现一个 Agent 主循环，建议按以下阶段演进：

阶段 1：最小循环

• while 循环 + model_step() + message/tool_call 分支
• 固定 max_turns 保护

阶段 2：结构化状态

• 定义 State 类型
• 消息模型标准化（role/content/meta）
• 工具结果结构化回流

阶段 3：流式支持

• LLM 调用改为流式
• yield 输出给上层消费
• 支持用户中断（AbortController）

阶段 4：上下文管理

• token 预算估算
• 工具结果裁剪
• 自动摘要压缩

阶段 5：错误恢复

• prompt-too-long 响应式压缩
• max-output-tokens 续航
• 模型降级 fallback

阶段 6：工具编排

• 并发/串行分区
• 流式工具执行
• 权限检查链路

阶段 7：Hook 与扩展

• Stop Hooks（终止前检查）
• Post-sampling Hooks（响应后处理）
• 附件注入（记忆、通知、Skill）

---

参考源码文件

文件	职责
src/query.ts	主循环核心实现（queryLoop）
src/QueryEngine.ts	会话管理层（submitMessage）
src/query/tokenBudget.ts	Token 预算续航逻辑
src/query/stopHooks.ts	Stop Hook 处理
src/services/tools/toolOrchestration.ts	工具并发/串行编排
src/services/compact/autoCompact.ts	自动压缩策略
src/services/compact/reactiveCompact.ts	响应式压缩恢复
src/agt/agent.py	教学版最小实现