AI Agent 主循环（Runtime Loop）底层实现深度解析

本文基于 Claude Code 真实源码深度剖析一个生产级 AI Agent 平台的核心运行时循环是如何设计与实现的。从教学骨架到 1700 行的生产代码揭示 Agent 从一问一答进化为持续决策系统的全部关键机制。一、什么是 Agent 主循环传统 LLM 应用是请求-响应模式用户问一个问题模型回一个答案交互结束。Agent 则是一个持续决策循环用户输入 → 模型推理 → 判断动作类型 → 执行工具 → 结果回写上下文 → 继续下一轮 → ... → 终止这个循环的本质是模型不再只是回答者而是一个决策者——每一轮它都在决定是直接回复用户还是调用工具继续推进任务。二、教学版最小实现先看最小骨架如何表达这个思想来自src/agt/agent.pydefrun(self,user_input:str)-str:self.add_message(user,user_input)forturninrange(self.max_turns):stepself.model_step()ifstep[type]message:contentstep[content]self.add_message(assistant,content,turnturn)returncontentifstep[type]tool_call:tool_namestep[tool]tool_inputstep.get(input,{})ifnotself.can_use_tool(tool_name):self.add_message(tool_result,fTool not allowed:{tool_name},okFalse)continueresultself.tools[tool_name].call(tool_input)self.add_message(tool_result,result.content,okresult.ok)continuereturnAgent stopped because it reached max_turns.核心逻辑只有三件事调用模型得到决策model_step()如果是 message → 结束循环如果是 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 终止四、循环状态定义生产系统必须用显式状态对象来追踪循环进程而不是散落在局部变量中typeState{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 循环入口与参数exporttypeQueryParams{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 → AutoCompact1Tool Result 预算裁剪工具返回的结果可能非常长比如一个 grep 命令返回 10000 行必须先裁剪messagesForQueryawaitapplyToolResultBudget(messagesForQuery,toolUseContext.contentReplacementState,persistReplacements?recordsvoidrecordContentReplacement(...):undefined,exemptTools,// 某些工具不受限制)2Snip 压缩对消息历史中的旧部分进行剪断式压缩释放 token 空间constsnipResultsnipModule.snipCompactIfNeeded(messagesForQuery)messagesForQuerysnipResult.messages snipTokensFreedsnipResult.tokensFreed3Microcompact更细粒度的压缩——对单个工具调用结果进行内联缩减constmicrocompactResultawaitdeps.microcompact(messagesForQuery,toolUseContext,querySource)messagesForQuerymicrocompactResult.messages4Context Collapse按折叠逻辑隐藏已完成步骤的冗余细节保留结构性摘要constcollapseResultawaitcontextCollapse.applyCollapsesIfNeeded(messagesForQuery,toolUseContext,querySource)messagesForQuerycollapseResult.messages5AutoCompact自动摘要压缩当 token 数超过阈值时调用一个小模型生成历史摘要替换旧消息const{compactionResult}awaitdeps.autocompact(messagesForQuery,toolUseContext,cacheSafeParams,querySource,tracking,snipTokensFreed)if(compactionResult){messagesForQuerybuildPostCompactMessages(compactionResult)}5.3 模型调用流式上下文准备完毕后调用模型获取响应。Claude Code 使用流式调用forawait(constmessageofdeps.callModel({messages:prependUserContext(messagesForQuery,userContext),systemPrompt:fullSystemPrompt,tools:toolUseContext.options.tools,signal:toolUseContext.abortController.signal,options:{model:currentModel,fallbackModel,maxOutputTokensOverride,// ... 其他配置},})){// 处理流式消息if(message.typeassistant){assistantMessages.push(message)// 检测 tool_use blockconsttoolBlocksmessage.message.content.filter(cc.typetool_use)if(toolBlocks.length0){toolUseBlocks.push(...toolBlocks)needsFollowUptrue// 标记需要执行工具后继续循环}}}关键设计点流式处理允许在模型输出的同时就开始准备工具执行流式工具执行器needsFollowUp标志决定循环是否继续支持模型降级fallback当主模型过载时自动切换备用模型5.4 判断循环去向模型响应完成后有两条路径路径 A无工具调用needsFollowUp false→ 准备终止进入终止前的检查链Prompt-too-long 恢复响应式压缩Max-output-tokens 恢复注入继续指令Stop Hooks 执行Token Budget 检查全部通过 →return { reason: completed }路径 B有工具调用needsFollowUp true→ 执行工具后继续进入工具执行完成后构造新 State 进入下一轮。5.5 工具执行编排工具执行不是简单的逐个调用而是有并发策略// 来自 toolOrchestration.tsexportasyncfunction*runTools(toolUseBlocks,assistantMessages,canUseTool,toolUseContext){for(const{isConcurrencySafe,blocks}ofpartitionToolCalls(toolUseBlocks,toolUseContext)){if(isConcurrencySafe){// 只读工具如 Read、Grep可以并行执行yield*runToolsConcurrently(blocks,...)}else{// 有副作用的工具如 Write、Bash串行执行yield*runToolsSerially(blocks,...)}}}分区逻辑连续的只读工具 → 合并为一批并行执行有副作用的工具 → 单独串行执行最大并发度可通过环境变量控制默认 10此外还支持流式工具执行StreamingToolExecutor在模型还在输出时已经完成的 tool_use block 就开始执行不等整个响应结束。5.6 工具结果回流与附件注入工具执行完成后结果必须以结构化方式回写到消息序列// 工具结果回流forawait(constupdateoftoolUpdates){if(update.message){yieldupdate.message// 向外部流输出toolResults.push(...normalizeMessagesForAPI([update.message],tools))}}// 附件注入记忆、通知、Skill 发现forawait(constattachmentofgetAttachmentMessages(...)){yieldattachment toolResults.push(attachment)}附件注入是一个重要机制——它让系统能在工具执行后、下一轮模型调用前注入额外的上下文信息记忆预取结果异步加载的相关记忆队列命令用户在模型执行期间发送的新指令Skill 发现自动发现的相关技能提示5.7 状态转移进入下一轮一切就绪后构造新的 State 对象进入下一轮constnext:State{messages:[...messagesForQuery,...assistantMessages,...toolResults],toolUseContext:toolUseContextWithQueryTracking,autoCompactTracking:tracking,turnCount:nextTurnCount,maxOutputTokensRecoveryCount:0,// 重置恢复计数hasAttemptedReactiveCompact:false,// 重置压缩尝试标记pendingToolUseSummary:nextPendingToolUseSummary,transition:{reason:next_turn},}statenext// continue → 回到 while(true) 顶部六、终止条件详解循环可能因以下原因终止终止原因触发条件返回值completed模型不再请求工具调用且通过所有 stop hooks{ reason: completed }max_turns超过最大轮次限制{ reason: max_turns }aborted_streaming用户中断CtrlC在流式阶段{ reason: aborted_streaming }aborted_tools用户中断在工具执行阶段{ reason: aborted_tools }hook_stoppedHook 明确阻止继续{ reason: hook_stopped }stop_hook_preventedStop Hook 中止循环{ reason: stop_hook_prevented }blocking_limittoken 数达到硬上限{ 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 contextconstdrainedcontextCollapse.recoverFromOverflow(messagesForQuery,querySource)if(drained.committed0){state{...state,messages:drained.messages,transition:{reason:collapse_drain_retry}}continue}}// 第二步reactive compactconstcompactedawaitreactiveCompact.tryReactiveCompact({...})if(compacted){state{...state,messages:buildPostCompactMessages(compacted),transition:{reason:reactive_compact_retry}}continue}// 第三步无法恢复终止yieldlastMessagereturn{reason:prompt_too_long}7.2 Max-Output-Tokens 恢复当模型输出被截断时达到输出 token 上限第一步升级 token 上限8k → 64k重试 ↓ 如果仍然被截断 第二步注入继续指令让模型接续输出最多 3 次 ↓ 如果 3 次仍未完成 第三步放弃恢复显示截断的输出// 升级重试if(maxOutputTokensOverrideundefined){state{...state,maxOutputTokensOverride:ESCALATED_MAX_TOKENS,transition:{reason:max_output_tokens_escalate}}continue}// 注入继续指令if(maxOutputTokensRecoveryCountMAX_OUTPUT_TOKENS_RECOVERY_LIMIT){constrecoveryMessagecreateUserMessage({content:Output token limit hit. Resume directly — no apology, no recap...,isMeta:true,})state{...state,messages:[...messages,...assistantMessages,recoveryMessage],maxOutputTokensRecoveryCount:count1}continue}7.3 模型 Fallback当主模型过载时自动降级到备用模型catch(innerError){if(innerErrorinstanceofFallbackTriggeredErrorfallbackModel){currentModelfallbackModel attemptWithFallbacktrue// 清理已有的部分响应assistantMessages.length0toolResults.length0// 通知用户yieldcreateSystemMessage(Switched to${fallbackModel}due to high demand)continue// 用新模型重试}}八、Stop Hooks 机制每轮结束前模型决定不再调用工具时系统会执行 Stop HooksconststopHookResultyield*handleStopHooks(messagesForQuery,assistantMessages,systemPrompt,userContext,systemContext,toolUseContext,querySource,stopHookActive)if(stopHookResult.preventContinuation){return{reason:stop_hook_prevented}}if(stopHookResult.blockingErrors.length0){// Hook 返回阻断性错误将错误注入消息后继续循环state{...state,messages:[...messages,...assistantMessages,...blockingErrors],stopHookActive:true}continue}Stop Hooks 的用途代码格式检查lint安全审计组织规则合规检查自动记忆提取Hook 可以返回三种结果通过→ 循环正常终止阻断性错误→ 错误注入消息模型看到错误后会尝试修复循环继续阻止继续→ 循环强制终止九、Token Budget 自动续航当用户设置了 token budget 时系统会在模型主动结束时检查是否还有预算剩余constdecisioncheckTokenBudget(budgetTracker,agentId,budget,turnOutputTokens)if(decision.actioncontinue){// 还有预算注入提示让模型继续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是主循环之上的会话管理层负责classQueryEngine{privatemutableMessages:Message[]// 完整会话历史privateabortController:AbortController// 中断控制privatetotalUsage:NonNullableUsage// 累计 token 用量privatereadFileState:FileStateCache// 文件读取缓存async*submitMessage(prompt){// 1. 处理用户输入斜杠命令、附件等const{messages,shouldQuery}awaitprocessUserInput(...)// 2. 持久化用户消息到 transcriptawaitrecordTranscript(messages)// 3. 调用核心 query loopforawait(constmessageofquery({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 │ │ │ │ │ │ │ │ │ └─────────────┘ └──────────────┘ └────────┬────────┘ │ │ │ │ │ │ │ │ │ ┌──────────────────────────────┼────┐ │ │ │ │ │ │ │ │ │ │ │ ▼ ▼ │ │ │ │ │ ┌─────────────────────┐ ┌────────────────┐ │ │ │ │ │ │ needsFollowUpfalse │ │ 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工具编排并发/串行分区流式工具执行权限检查链路阶段 7Hook 与扩展Stop Hooks终止前检查Post-sampling Hooks响应后处理附件注入记忆、通知、Skill参考源码文件文件职责src/query.ts主循环核心实现queryLoopsrc/QueryEngine.ts会话管理层submitMessagesrc/query/tokenBudget.tsToken 预算续航逻辑src/query/stopHooks.tsStop Hook 处理src/services/tools/toolOrchestration.ts工具并发/串行编排src/services/compact/autoCompact.ts自动压缩策略src/services/compact/reactiveCompact.ts响应式压缩恢复src/agt/agent.py教学版最小实现