第9章 工具调用循环——Agent的行动闭环
想象一位厨师烹饪的完整过程：他先查看菜单（理解用户需求），准备好食材（整理上下文），开始下锅烹饪（调用工具），尝一口味道（获取执行结果），再调整调味料（继续对话），直到菜品完美出锅。这个过程循环往复，直到厨师满意，或者客人喊停。
在 Claude Code 的世界里，query() 函数就是这个掌勺的“厨师”，而工具调用循环，正是这套烹饪流程的核心引擎。这一章，我们来拆解这个状态机是如何设计与实现的，具体包括：while(true) 循环如何撑起整个状态机、State 对象如何管理跨轮次的状态、StreamingToolExecutor 如何实现流式的工具执行、并发控制如何在性能与正确性之间走钢丝、Stop Hooks 的每轮守卫机制，以及上下文压缩如何防止 Token 溢出等问题。
概念讲解
状态机的本质
query() 函数本质上就是一个状态机。它不停循环，直到碰见终止条件才停下来：
while (true) {
// 处理当前状态
// 决定下一步行动
// 更新状态
// 产生输出
}
这种设计的好处非常直观：状态转换的逻辑集中在单个区域，清晰明了；新增状态或转换很方便，扩展性强；而且每个状态和每次转换都能独立测试。
State 对象的设计
State 对象把跨轮次的所有状态都封装在一起：
type State = {
messages: Message[]
toolUseContext: ToolUseContext
autoCompactTracking: AutoCompactTrackingState | undefined
maxOutputTokensRecoveryCount: number
hasAttemptedReactiveCompact: boolean
maxOutputTokensOverride: number | undefined
pendingToolUseSummary: Promise
stopHookActive: boolean | undefined
turnCount: number
transition: Continue | undefined
}
每个字段各司其职：messages 存储对话历史；toolUseContext 管理工具调用的上下文；autoCompactTracking 跟踪自动压缩的情况；maxOutputTokensRecoveryCount 是最大输出 Token 的恢复计数；hasAttemptedReactiveCompact 标记是否尝试过响应式压缩；maxOutputTokensOverride 允许覆盖最大输出 Token 的限制；pendingToolUseSummary 记录待处理的工具使用摘要；stopHookActive 指示停止钩子是否已激活；turnCount 统计轮次；transition 记录转换原因。
流式工具执行
StreamingToolExecutor 实现了流式的工具执行：工具一旦到达，只要条件允许就立即启动；并发安全的工具可以并行运行；最终结果按照工具到达的顺序输出；还能实时输出进度消息。这就像一个全自动流水线：原料不断进来，产品不断出去，中间的加工工序完全可以并行处理。
源码分析
query 函数入口
从 query() 的函数入口看起：
export async function* query(params: QueryParams,): AsyncGenerator<| StreamEvent| RequestStartEvent| Message| TombstoneMessage| ToolUseSummaryMessage, Terminal> {
const consumedCommandUuids: string[] = []
const terminal = yield* queryLoop(params, consumedCommandUuids)
for (const uuid of consumedCommandUuids) {
notifyCommandLifecycle(uuid, 'completed')
}
return terminal
}
这里采用了委托模式：query() 把核心逻辑交给 queryLoop() 来处理。它会记录并通知所有已消费命令的生命周期。AsyncGenerator 这个返回类型意味着它可以逐步生成各种类型的事件和消息。最后返回 Terminal 对象，表示查询结束。
queryLoop 函数
queryLoop() 才是真正的核心状态机循环：
async function* queryLoop(params: QueryParams, consumedCommandUuids: string[],): AsyncGenerator<| StreamEvent| RequestStartEvent| Message| TombstoneMessage| ToolUseSummaryMessage, Terminal> {
const {systemPrompt, userContext, systemContext, canUseTool, fallbackModel, querySource, maxTurns, skipCacheWrite} = params
const deps = params.deps ?? productionDeps()
let state: State = {
messages: params.messages,
toolUseContext: params.toolUseContext,
maxOutputTokensOverride: params.maxOutputTokensOverride,
autoCompactTracking: undefined,
stopHookActive: undefined,
maxOutputTokensRecoveryCount: 0,
hasAttemptedReactiveCompact: false,
turnCount: 1,
pendingToolUseSummary: undefined,
transition: undefined,
}
const budgetTracker = feature('TOKEN_BUDGET') ? createBudgetTracker() : null
let taskBudgetRemaining: number | undefined = undefined
const config = buildQueryConfig()
using pendingMemoryPrefetch = startRelevantMemoryPrefetch(state.messages, state.toolUseContext)
while (true) {
// 循环体...
}
}
初始化阶段完成了以下几件事：先解构出不可变的参数；用 deps 提供所有依赖；创建初始的 State 对象；可选的 Token 预算跟踪器；以及利用 using 语法确保内存预取资源能够及时释放。
主循环结构
主循环的运作模式大致如下：
while (true) {
let { toolUseContext } = state
const {messages, autoCompactTracking, maxOutputTokensRecoveryCount, hasAttemptedReactiveCompact, maxOutputTokensOverride, pendingToolUseSummary, stopHookActive, turnCount} = state
const pendingSkillPrefetch = skillPrefetch?.startSkillDiscoveryPrefetch(null, messages, toolUseContext)
yield { type: 'stream_request_start' }
queryCheckpoint('query_fn_entry')
if (!toolUseContext.agentId) {
headlessProfilerCheckpoint('query_started')
}
const queryTracking = toolUseContext.queryTracking
? { chainId: toolUseContext.queryTracking.chainId, depth: toolUseContext.queryTracking.depth + 1 }
: { chainId: deps.uuid(), depth: 0 }
toolUseContext = {...toolUseContext, queryTracking}
let messagesForQuery = [...getMessagesAfterCompactBoundary(messages)]
// ... 更多处理逻辑
}
每一轮迭代的关键步骤包括：解构 State 对象获取当前状态、启动技能发现预取、产生 stream_request_start 事件、设置性能检查点、初始化或更新查询跟踪信息。
消息准备与压缩
在真正调用 API 之前，消息需要经过多步处理：
let messagesForQuery = [...getMessagesAfterCompactBoundary(messages)]
let tracking = autoCompactTracking
// 应用工具结果预算
messagesForQuery = await applyToolResultBudget(messagesForQuery, toolUseContext.contentReplacementState, persistReplacements ? records => void recordContentReplacement(records, toolUseContext.agentId).catch(logError) : undefined, new Set(toolUseContext.options.tools.filter(t => !Number.isFinite(t.maxResultSizeChars)).map(t => t.name)))
// 应用 snip 压缩
let snipTokensFreed = 0
if (feature('HISTORY_SNIP')) {
queryCheckpoint('query_snip_start')
const snipResult = snipModule!.snipCompactIfNeeded(messagesForQuery)
messagesForQuery = snipResult.messages
snipTokensFreed = snipResult.tokensFreed
if (snipResult.boundaryMessage) {
yield snipResult.boundaryMessage
}
queryCheckpoint('query_snip_end')
}
// 应用微压缩
queryCheckpoint('query_microcompact_start')
const microcompactResult = await deps.microcompact(messagesForQuery, toolUseContext, querySource)
messagesForQuery = microcompactResult.messages
const pendingCacheEdits = feature('CACHED_MICROCOMPACT') ? microcompactResult.compactionInfo?.pendingCacheEdits : undefined
queryCheckpoint('query_microcompact_end')
// 应用上下文折叠
if (feature('CONTEXT_COLLAPSE') && contextCollapse) {
const collapseResult = await contextCollapse.applyCollapsesIfNeeded(messagesForQuery, toolUseContext, querySource)
messagesForQuery = collapseResult.messages
}
压缩的层次非常清晰：首先通过工具结果预算限制单个工具结果的大小；然后用 Snip 压缩移除不重要程度较低的消息；接着用微压缩来压缩重复的文件内容；最后用上下文折叠来处理相似的消息。
自动压缩
自动压缩是防止 Token 溢出的关键机制：
const fullSystemPrompt = asSystemPrompt(appendSystemContext(systemPrompt, systemContext))
queryCheckpoint('query_autocompact_start')
const { compactionResult, consecutiveFailures } = await deps.autocompact(messagesForQuery, toolUseContext, {systemPrompt, userContext, systemContext, toolUseContext, forkContextMessages: messagesForQuery}, querySource, tracking, snipTokensFreed)
queryCheckpoint('query_autocompact_end')
if (compactionResult) {
const {preCompactTokenCount, postCompactTokenCount, truePostCompactTokenCount, compactionUsage} = compactionResult
logEvent('tengu_auto_compact_succeeded', {
originalMessageCount: messages.length,
compactedMessageCount: compactionResult.summaryMessages.length + compactionResult.attachments.length + compactionResult.hookResults.length,
preCompactTokenCount, postCompactTokenCount, truePostCompactTokenCount,
compactionInputTokens: compactionUsage?.input_tokens,
compactionOutputTokens: compactionUsage?.output_tokens,
compactionCacheReadTokens: compactionUsage?.cache_read_input_tokens ?? 0,
compactionCacheCreationTokens: compactionUsage?.cache_creation_input_tokens ?? 0,
compactionTotalTokens: compactionUsage ? compactionUsage.input_tokens + (compactionUsage.cache_creation_input_tokens ?? 0) + (compactionUsage.cache_read_input_tokens ?? 0) + compactionUsage.output_tokens : 0,
queryChainId: queryChainIdForAnalytics,
queryDepth: queryTracking.depth,
})
if (params.taskBudget) {
const preCompactContext = finalContextTokensFromLastResponse(messagesForQuery)
taskBudgetRemaining = Math.max(0, (taskBudgetRemaining ?? params.taskBudget.total) - preCompactContext)
}
tracking = { compacted: true, turnId: deps.uuid(), turnCounter: 0, consecutiveFailures: 0 }
const postCompactMessages = buildPostCompactMessages(compactionResult)
for (const message of postCompactMessages) {
yield message
}
messagesForQuery = postCompactMessages
} else if (consecutiveFailures !== undefined) {
tracking = {...(tracking ?? { compacted: false, turnId: '', turnCounter: 0 }), consecutiveFailures}
}
关键逻辑在于：当消息超过阈值时触发压缩，并记录压缩前后的 Token 数量。如果设定了任务预算，会同步更新预算的剩余量。压缩跟踪状态会被重置，压缩后的消息会替换掉原始消息。
工具执行器初始化
在调用 API 之前，先初始化工具执行器：
const assistantMessages: AssistantMessage[] = []
const toolResults: (UserMessage | AttachmentMessage)[] = []
const toolUseBlocks: ToolUseBlock[] = []
let needsFollowUp = false
queryCheckpoint('query_setup_start')
const useStreamingToolExecution = config.gates.streamingToolExecution
let streamingToolExecutor = useStreamingToolExecution ? new StreamingToolExecutor(toolUseContext.options.tools, canUseTool, toolUseContext) : null
const appState = toolUseContext.getAppState()
const permissionMode = appState.toolPermissionContext.mode
let currentModel = getRuntimeMainLoopModel({ permissionMode, mainLoopModel: toolUseContext.options.mainLoopModel, exceeds200kTokens: permissionMode === 'plan' && doesMostRecentAssistantMessageExceed200k(messagesForQuery)})
queryCheckpoint('query_setup_end')
这里初始化了几个消息容器：assistantMessages、toolResults、toolUseBlocks。根据配置选择是否创建流式工具执行器，最后根据权限模式和 Token 数量选择合适的模型。
API 调用循环
API 调用是在一个嵌套的 while 循环里进行的：
let attemptWithFallback = true
queryCheckpoint('query_api_loop_start')
try {
while (attemptWithFallback) {
attemptWithFallback = false
try {
let streamingFallbackOccured = false
queryCheckpoint('query_api_streaming_start')
for await (const message of deps.callModel({ messages: prependUserContext(messagesForQuery, userContext), systemPrompt: fullSystemPrompt, thinkingConfig: toolUseContext.options.thinkingConfig, tools: toolUseContext.options.tools, signal: toolUseContext.abortController.signal, options: { async getToolPermissionContext() { const appState = toolUseContext.getAppState(); return appState.toolPermissionContext }, model: currentModel, ...(config.gates.fastModeEnabled && { fastMode: appState.fastMode }), toolChoice: undefined, isNonInteractiveSession: toolUseContext.options.isNonInteractiveSession, fallbackModel, onStreamingFallback: () => { streamingFallbackOccured = true }, querySource, agents: toolUseContext.options.agentDefinitions.activeAgents, allowedAgentTypes: toolUseContext.options.agentDefinitions.allowedAgentTypes, hasAppendSystemPrompt: !!toolUseContext.options.appendSystemPrompt, maxOutputTokensOverride , fetchOverride: dumpPromptsFetch, mcpTools: appState.mcp.tools, hasPendingMcpServers: appState.mcp.clients.some(c => c.type === 'pending'), queryTracking, effortValue: appState.effortValue, advisorModel: appState.advisorModel, skipCacheWrite, agentId: toolUseContext.agentId, addNotification: toolUseContext.addNotification, ...(params.taskBudget && { taskBudget: { total: params.taskBudget.total, ...(taskBudgetRemaining !== undefined && { remaining: taskBudgetRemaining }) } }) } })) {
// 处理流式消息...
}
} catch (innerError) {
if (innerError instanceof FallbackTriggeredError && fallbackModel) {
// 处理模型回退...
}
}
}
} catch (error) {
// 处理错误...
}
外层循环负责处理模型回退，内层循环处理流式响应，两者分工非常明确。
流式消息处理
流式消息处理是整个循环的核心：
for await (const message of deps.callModel({...})) {
if (streamingFallbackOccured) {
for (const msg of assistantMessages) {
yield { type: 'tombstone' as const, message: msg }
}
logEvent('tengu_orphaned_messages_tombstoned', { orphanedMessageCount: assistantMessages.length, queryChainId: queryChainIdForAnalytics, queryDepth: queryTracking.depth })
assistantMessages.length = 0
toolResults.length = 0
toolUseBlocks.length = 0
needsFollowUp = false
if (streamingToolExecutor) {
streamingToolExecutor.discard()
streamingToolExecutor = new StreamingToolExecutor(toolUseContext.options.tools, canUseTool, toolUseContext)
}
}
let yieldMessage: typeof message = message
if (message.type === 'assistant') {
let clonedContent: typeof message.message.content | undefined
for (let i = 0; i < message.message.content.length; i++) {
const block = message.message.content[i]!
if (block.type === 'tool_use' && typeof block.input === 'object' && block.input !== null) {
const tool = findToolByName(toolUseContext.options.tools, block.name)
if (tool?.backfillObservableInput) {
const originalInput = block.input as Record
const inputCopy = { ...originalInput }
tool.backfillObservableInput(inputCopy)
const addedFields = Object.keys(inputCopy).some(k => !(k in originalInput))
if (addedFields) {
clonedContent ??= [...message.message.content]
clonedContent[i] = { ...block, input: inputCopy }
}
}
}
}
if (clonedContent) {
yieldMessage = {...message, message: { ...message.message, content: clonedContent }}
}
}
let withheld = false
if (feature('CONTEXT_COLLAPSE')) {
if (contextCollapse?.isWithheldPromptTooLong(message, isPromptTooLongMessage, querySource)) {
withheld = true
}
}
if (reactiveCompact?.isWithheldPromptTooLong(message)) {
withheld = true
}
if (mediaRecoveryEnabled && reactiveCompact?.isWithheldMediaSizeError(message)) {
withheld = true
}
if (isWithheldMaxOutputTokens(message)) {
withheld = true
}
if (!withheld) {
yield yieldMessage
}
if (message.type === 'assistant') {
assistantMessages.push(message)
const msgToolUseBlocks = message.message.content.filter(content => content.type === 'tool_use') as ToolUseBlock[]
if (msgToolUseBlocks.length > 0) {
toolUseBlocks.push(...msgToolUseBlocks)
needsFollowUp = true
}
if (streamingToolExecutor && !toolUseContext.abortController.signal.aborted) {
for (const toolBlock of msgToolUseBlocks) {
streamingToolExecutor.addTool(toolBlock, message)
}
}
}
if (streamingToolExecutor && !toolUseContext.abortController.signal.aborted) {
for (const result of streamingToolExecutor.getCompletedResults()) {
if (result.message) {
yield result.message
toolResults.push(...normalizeMessagesForAPI([result.message], toolUseContext.options.tools).filter(_ => _.type === 'user'))
}
}
}
}
这里的处理逻辑相当丰富：如果发生了流式回退，则清理旧消息和工具执行器；为工具输入回填可观察字段；保留那些可恢复的错误，等待恢复机制后续处理；收集助手消息和工具调用块；把工具调用添加到执行器；最后获取并输出已经完成的工具结果。
StreamingToolExecutor 实现
来看看 StreamingToolExecutor 的具体实现：
export class StreamingToolExecutor {
private tools: TrackedTool[] = []
private toolUseContext: ToolUseContext
private hasErrored = false
private erroredToolDescription = ''
private siblingAbortController: AbortController
private discarded = false
private progressA vailableResolve?: () => void
constructor(private readonly toolDefinitions: Tools, private readonly canUseTool: CanUseToolFn, toolUseContext: ToolUseContext) {
this.toolUseContext = toolUseContext
this.siblingAbortController = createChildAbortController(toolUseContext.abortController)
}
discard(): void {
this.discarded = true
}
addTool(block: ToolUseBlock, assistantMessage: AssistantMessage): void {
const toolDefinition = findToolByName(this.toolDefinitions, block.name)
if (!toolDefinition) {
this.tools.push({
id: block.id, block, assistantMessage,
status: 'completed', isConcurrencySafe: true, pendingProgress: [],
results: [createUserMessage({
content: [{ type: 'tool_result', content: `
toolUseResult: `Error: No such tool a vailable: ${block.name}`,
sourceToolAssistantUUID: assistantMessage.uuid,
})],
})
return
}
const parsedInput = toolDefinition.inputSchema.safeParse(block.input)
const isConcurrencySafe = parsedInput?.success
? (() => { try { return Boolean(toolDefinition.isConcurrencySafe(parsedInput.data)) } catch { return false } })()
: false
this.tools.push({
id: block.id, block, assistantMessage,
status: 'queued', isConcurrencySafe, pendingProgress: [],
})
void this.processQueue()
}
private canExecuteTool(isConcurrencySafe: boolean): boolean {
const executingTools = this.tools.filter(t => t.status === 'executing')
return (executingTools.length === 0 || (isConcurrencySafe && executingTools.every(t => t.isConcurrencySafe)))
}
private async processQueue(): Promise
for (const tool of this.tools) {
if (tool.status !== 'queued') continue
if (this.canExecuteTool(tool.isConcurrencySafe)) {
await this.executeTool(tool)
} else {
if (!tool.isConcurrencySafe) break
}
}
}
}
设计上有几个要点：通过 isConcurrencySafe 标记来控制并发；工具按到达顺序排队处理；如果找不到工具定义，直接返回错误结果；还会创建子中断器，这样就能独立中断工具执行，不会影响到主流程。
设计启示
1. 状态机的威力
while(true) 循环搭配 State 对象，构建了一个清晰的状态机模型。状态转换逻辑集中在一处，新增状态和转换都非常容易，而且每个状态都能独立测试——这套模式在工程实践中确实非常可靠。
2. 流式处理的优雅性
异步生成器让流式处理变得格外优雅。消息到达后立即处理，无需缓冲所有消息，内存效率高，而且随时可以中止流。这种设计在实时性要求高的场景下特别适用。
3. 并发控制的智慧
StreamingToolExecutor 的并发控制设计很巧妙：没有工具在执行时，任何工具都可以执行；如果有工具正在运行，只有并发安全的工具才能并行执行；非并发安全的工具必须独占。这就好比高速公路：多车道可以并行行驶，但某些特殊车辆需要独占一条车道。
4. 错误恢复的层次性
系统在多个层次建立了错误恢复机制：API 流式失败时自动回退；模型不可用时切换到备用模型；Token 超限时自动压缩；长上下文时自动折叠。每个层次专注处理自己的错误类型，形成了一套完整的错误恢复体系。
5. 性能优化的权衡
系统在多个关键点做出了性能权衡：内存和技能预取减少了等待时间；提前压缩减少了 Token 消耗；文件缓存避免了重复读取；并发安全的工具可以并行执行。每个优化决策背后，都是在资源开销和响应速度之间找到平衡点。
思考题
- 设计题：如何为 StreamingToolExecutor 添加优先级支持，让某些工具优先执行？
- 优化题：在什么情况下应该禁用流式工具执行？优缺点分别是什么？
- 扩展题：如何为状态机添加超时机制，防止无限循环？
- 测试题：如何为
query()函数编写集成测试？需要模拟哪些依赖？ - 架构题：状态机和事件驱动架构有什么区别？在什么场景下应该选择哪种方案？
总结
工具调用循环是 Claude Code 的行动闭环。它用 State 对象封装跨轮次的状态，用异步生成器优雅地处理流式响应，通过 isConcurrencySafe 标记控制工具并发，在多个层次实现错误恢复机制，并通过预取、压缩、缓存等手段优化性能。
就像厨师烹饪流程一样——准备食材、下锅烹饪、尝味道、调整、再继续——直到菜品完美出锅。这个循环体现了软件工程中的诸多最佳实践：状态机设计、流式处理、并发控制、错误恢复……理解了它，也就理解了 Claude Code 整个系统的核心交互机制和行动闭环。
