Claude Code Agent四种运行模式详解：从单次查询到钩子驱动

在实际 AI 应用开发中Claude Code Agent 的自主执行能力让开发者能够将复杂任务交给 AI 代理完成而理解其内部运行机制是构建可靠生产级应用的关键。Claude Code Agent 的核心执行循环遵循“评估-行动-反馈”的模式但根据不同的使用场景和需求开发者可以选择四种不同的运行方式来控制代理的行为。1. 理解 Claude Code Agent 的基本执行循环Claude Code Agent 的执行循环是其自主能力的核心机制。这个循环基于一个简单的概念AI 代理接收任务提示评估当前状态调用工具执行操作接收执行结果然后重复这个过程直到任务完成。1.1 循环的基本工作流程每个代理会话都遵循相同的周期接收提示Claude 接收你的提示同时加载系统提示、工具定义和对话历史。SDK 会产生一个 SystemMessage子类型为 init包含会话元数据。评估并响应Claude 评估当前状态并确定如何继续。它可能用文本响应、请求一个或多个工具调用或两者都有。SDK 产生一个 AssistantMessage包含文本和任何工具调用请求。执行工具SDK 运行每个请求的工具并收集结果。每组工具结果反馈给 Claude 以做出下一个决定。重复执行步骤 2 和 3 作为一个循环重复每个完整循环称为一个轮次。Claude 继续调用工具并处理结果直到产生没有工具调用的响应。返回结果SDK 产生最终的 AssistantMessage包含文本响应无工具调用然后是 ResultMessage包含最终文本、令牌使用、成本和会话 ID。1.2 轮次与消息类型的关系轮次是循环内的一个完整往返Claude 产生包含工具调用的输出SDK 执行这些工具结果自动反馈给 Claude。这发生在不将控制权交回给你的代码的情况下。考虑一个实际例子提示修复 auth.ts 中的失败测试的完整会话轮次 1Claude 调用 Bash 运行npm testSDK 执行命令并返回测试失败结果轮次 2Claude 在 auth.ts 和 auth.test.ts 上调用 ReadSDK 返回文件内容轮次 3Claude 调用 Edit 修复 auth.ts然后调用 Bash 重新运行测试最后轮次Claude 产生仅包含文本的响应修复了认证错误所有测试现在都通过了SDK 在循环过程中会产生五种核心消息类型消息类型触发时机关键字段SystemMessage会话生命周期事件subtype: init, compact_boundary, informationalAssistantMessage每个 Claude 响应后包含文本内容块和工具调用块UserMessage每个工具执行后包含发送回 Claude 的工具结果内容StreamEvent启用部分消息时包含原始 API 流事件ResultMessage代理循环结束时包含最终结果、成本和使用情况2. 第一种方式单次查询模式单次查询模式是最简单的运行方式适合执行明确、范围有限的任务。在这种模式下你提供一个提示代理运行直到完成任务或达到限制然后返回最终结果。2.1 基本单次查询实现Python 实现示例import asyncio from claude_agent_sdk import query, ResultMessage async def simple_query_example(): try: async for message in query( prompt列出当前目录下的所有 TypeScript 文件, options{ allowed_tools: [Glob], max_turns: 5, effort: low } ): if isinstance(message, ResultMessage): if message.subtype success: print(f结果: {message.result}) print(f成本: ${message.total_cost_usd:.4f}) else: print(f任务中断: {message.subtype}) except Exception as error: print(f会话错误: {error}) asyncio.run(simple_query_example())TypeScript 实现示例import { query } from anthropic-ai/claude-agent-sdk; async function simpleQueryExample() { try { for await (const message of query({ prompt: 列出当前目录下的所有 TypeScript 文件, options: { allowedTools: [Glob], maxTurns: 5, effort: low } })) { if (message.type result) { if (message.subtype success) { console.log(结果: ${message.result}); console.log(成本: $${message.total_cost_usd.toFixed(4)}); } else { console.log(任务中断: ${message.subtype}); } } } } catch (error) { console.log(会话错误: ${error}); } } simpleQueryExample();2.2 单次查询模式的关键配置参数单次查询模式下以下几个参数对控制代理行为至关重要参数类型默认值说明max_turns / maxTurnsnumber无限制最大工具使用轮次max_budget_usd / maxBudgetUsdnumber无限制停止前的最大成本effortstring模型默认推理努力级别low, medium, high, xhigh, maxallowed_tools / allowedToolsstring[]所有工具自动批准的工具列表permission_mode / permissionModestringdefault权限控制模式努力级别的选择策略low适合文件查找、目录列表等简单任务medium常规编辑和标准任务high代码重构、调试等复杂任务xhigh深度编码和代理任务推荐用于 Fable 5、Opus 4.7 和 Sonnet 5max需要深度分析的多步骤问题2.3 单次查询的错误处理单次查询在遇到错误时会抛出异常因此需要适当的错误处理async def robust_single_query(prompt: str, max_retries: int 3): for attempt in range(max_retries): try: async for message in query(promptprompt, options{ max_turns: 20, max_budget_usd: 1.0 }): if isinstance(message, ResultMessage): if message.subtype success: return message.result elif message.subtype error_max_turns: # 轮次不足可以调整策略重试 print(轮次不足尝试更精确的提示) prompt f{prompt} 请用更少的步骤完成 elif message.subtype error_max_budget_usd: print(预算超限尝试使用成本更低的模型) # 可以在这里切换模型或调整努力级别 break except Exception as e: print(f尝试 {attempt 1} 失败: {e}) if attempt max_retries - 1: raise3. 第二种方式交互式流式会话交互式流式会话模式允许你在代理运行过程中实时监控进度并在需要时提供额外输入。这种模式适合需要人工干预的复杂任务。3.1 建立流式会话连接Python 流式会话实现import asyncio from claude_agent_sdk import create_session, AssistantMessage, UserMessage async def interactive_session(): # 创建会话 async with create_session( prompt帮我分析这个代码库的结构, options{ allowed_tools: [Read, Glob, Grep], include_partial_messages: True # 启用实时流式传输 } ) as session: async for message in session.stream(): # 实时处理消息 if message.type assistant: print(fClaude: {message.content}) elif message.type user: # 工具执行结果 print(f工具结果: {message.content}) elif hasattr(message, type) and message.type stream_event: # 实时流式输出 if hasattr(message, delta): print(message.delta, end, flushTrue) # 在特定条件下提供额外输入 if (isinstance(message, AssistantMessage) and 需要更多信息 in str(message.content)): user_input input(\n你的回复: ) await session.send_message(user_input) # 运行交互式会话 asyncio.run(interactive_session())3.2 实时监控与控制在流式会话中你可以实时监控代理的决策过程并适时干预class InteractiveAgentController { private session: any; private isPaused false; private pendingActions: string[] []; async startSession(prompt: string) { this.session await createSession({ prompt, options: { includePartialMessages: true, permissionMode: default // 需要人工批准工具使用 } }); this.monitorSession(); } private async monitorSession() { for await (const message of this.session.stream()) { switch (message.type) { case assistant: await this.handleAssistantMessage(message); break; case tool_call_request: await this.handleToolCallRequest(message); break; case stream_event: this.handleStreamEvent(message); break; } } } private async handleToolCallRequest(message: any) { console.log(工具调用请求: ${message.toolName}); console.log(参数: ${JSON.stringify(message.parameters)}); // 等待用户批准 const approved await this.requestApproval(); if (approved) { await this.session.approveToolCall(message.callId); } else { await this.session.rejectToolCall(message.callId, 用户拒绝执行); } } private async requestApproval(): Promiseboolean { // 实现用户批准逻辑 return true; // 简化示例 } }3.3 会话状态管理与恢复流式会话支持状态保存和恢复这对于长时间运行的任务非常重要import json from claude_agent_sdk import resume_session class SessionManager: def __init__(self, storage_path: str sessions.json): self.storage_path storage_path self.sessions self.load_sessions() def save_session(self, session_id: str, metadata: dict): self.sessions[session_id] { metadata: metadata, last_updated: datetime.now().isoformat() } self.persist_sessions() async def resume_session(self, session_id: str, additional_prompt: str None): if session_id not in self.sessions: raise ValueError(f会话 {session_id} 不存在) async with resume_session(session_id) as session: if additional_prompt: await session.send_message(additional_prompt) # 继续处理消息流 async for message in session.stream(): yield message4. 第三种方式受控权限模式受控权限模式通过精细的工具权限控制确保代理在安全边界内运行。这种模式适合在生产环境或敏感项目中部署 AI 代理。4.1 权限模式详解Claude Code Agent 提供多种权限控制模式权限模式适用场景安全级别自动化程度default交互式应用高低需要人工批准acceptEdits开发环境中中自动批准文件操作plan规划阶段高低只读模式dontAsk严格环境最高高完全基于规则auto智能判断中高高AI 自动分类bypassPermissions隔离环境低最高无权限检查4.2 实现细粒度工具控制from claude_agent_sdk import query, ClaudeAgentOptions from typing import List, Dict class SecureAgentController: def __init__(self, project_rules: Dict): self.project_rules project_rules self.allowed_patterns self.load_allowed_patterns() async def run_with_permissions(self, prompt: str): options ClaudeAgentOptions( allowed_toolsself.get_allowed_tools(), disallowed_toolsself.get_disallowed_tools(), permission_modedontAsk, # 基于规则的完全控制 hooks{ pre_tool_use: self.validate_tool_use, post_tool_use: self.audit_tool_result } ) async for message in query(promptprompt, optionsoptions): if message.type result: return self.handle_result(message) def validate_tool_use(self, tool_call): 预工具使用验证 tool_name tool_call.tool_name parameters tool_call.parameters # 检查工具是否允许 if tool_name not in self.get_allowed_tools(): return {allowed: False, reason: 工具不在允许列表中} # 检查参数安全性 if tool_name Bash: return self.validate_bash_command(parameters.get(command, )) return {allowed: True} def validate_bash_command(self, command: str) - Dict: 验证 Bash 命令安全性 dangerous_patterns [rm -rf, chmod 777, /dev/null] for pattern in dangerous_patterns: if pattern in command: return {allowed: False, reason: f检测到危险命令模式: {pattern}} # 检查是否匹配允许的模式 for allowed_pattern in self.allowed_patterns: if allowed_pattern in command: return {allowed: True} return {allowed: False, reason: 命令不匹配任何允许的模式}4.3 基于上下文的动态权限更高级的权限控制可以根据任务上下文动态调整interface ContextAwarePermission { toolName: string; conditions: PermissionCondition[]; action: allow | deny | prompt; } class ContextAwarePermissionManager { private permissions: ContextAwarePermission[] []; private currentContext: AgentContext {}; async evaluateToolPermission(toolCall: ToolCall): PromisePermissionResult { const relevantPermissions this.permissions.filter( p p.toolName toolCall.toolName ); for (const permission of relevantPermissions) { const conditionsMet await this.checkConditions( permission.conditions, this.currentContext ); if (conditionsMet) { return { allowed: permission.action allow, requiresApproval: permission.action prompt, reason: 匹配规则: ${permission.conditions} }; } } // 默认拒绝 return { allowed: false, reason: 无匹配的权限规则 }; } updateContext(newContext: PartialAgentContext) { this.currentContext { ...this.currentContext, ...newContext }; } }5. 第四种方式钩子驱动扩展模式钩子驱动扩展模式通过事件钩子深度定制代理行为适合需要复杂业务逻辑集成的高级用例。5.1 核心钩子类型与使用场景Claude Code Agent 提供多种钩子类型用于在循环的关键节点插入自定义逻辑钩子类型触发时机典型用途PreToolUse工具执行前参数验证、输入转换、权限检查PostToolUse工具返回后结果处理、副作用触发、日志记录UserPromptSubmit发送提示时上下文注入、提示增强、安全检查Stop代理完成时结果验证、状态保存、清理操作SubagentStart/Stop子代理生成/完成任务跟踪、资源管理、结果聚合PreCompact上下文压缩前数据存档、摘要定制、状态保存5.2 实现复杂的钩子逻辑from claude_agent_sdk import ClaudeAgentOptions from datetime import datetime import json class AdvancedHookManager: def __init__(self): self.execution_log [] self.subagent_tracker {} def get_hooks(self): return { pre_tool_use: self.pre_tool_use, post_tool_use: self.post_tool_use, user_prompt_submit: self.enhance_prompt, subagent_start: self.track_subagent, subagent_stop: self.aggregate_subagent_result, pre_compact: self.archive_before_compaction } async def pre_tool_use(self, tool_call, context): 工具执行前钩子 log_entry { timestamp: datetime.now().isoformat(), tool: tool_call.tool_name, parameters: tool_call.parameters, context: context } self.execution_log.append(log_entry) # 业务逻辑检查资源使用 if await self.exceeds_resource_limits(): return { allowed: False, reason: 资源使用超限 } return {allowed: True} async def enhance_prompt(self, prompt, context): 提示提交时增强 # 注入项目特定上下文 enhanced_prompt f {prompt} 项目上下文: - 代码规范: 使用 TypeScript 严格模式 - 测试框架: Jest - 数据库: PostgreSQL - 当前环境: {context.get(environment, development)} 请遵循项目的最佳实践。 return enhanced_prompt async def archive_before_compaction(self, trigger, messages): 压缩前存档完整对话 if trigger auto: # 自动触发压缩 archive_path farchives/compaction_{datetime.now().strftime(%Y%m%d_%H%M%S)}.json with open(archive_path, w) as f: json.dump({ timestamp: datetime.now().isoformat(), trigger: trigger, messages: messages }, f, indent2) print(f对话已存档到: {archive_path})5.3 构建生产级钩子系统对于企业级应用需要更健壮的钩子管理系统interface HookSystemConfig { hooks: HookDefinition[]; errorHandling: ErrorHandlingStrategy; performance: PerformanceConfig; } class EnterpriseHookSystem { private hookRegistry: Mapstring, HookExecutor[] new Map(); private metricsCollector: MetricsCollector; registerHook(hookPoint: string, executor: HookExecutor, priority: number 0) { if (!this.hookRegistry.has(hookPoint)) { this.hookRegistry.set(hookPoint, []); } const executors this.hookRegistry.get(hookPoint)!; executors.push({ executor, priority }); // 按优先级排序 executors.sort((a, b) b.priority - a.priority); } async executeHooks(hookPoint: string, data: any): PromiseHookResult { const startTime Date.now(); const executors this.hookRegistry.get(hookPoint) || []; let result: HookResult { shouldContinue: true }; for (const { executor } of executors) { try { const hookResult await executor(data, result); result this.mergeHookResults(result, hookResult); // 如果钩子要求停止执行 if (result.shouldContinue false) { break; } } catch (error) { this.metricsCollector.recordHookError(hookPoint, error); // 根据错误处理策略决定是否继续 if (this.config.errorHandling fail-fast) { throw error; } // 否则记录错误但继续执行其他钩子 } } this.metricsCollector.recordHookExecution(hookPoint, Date.now() - startTime); return result; } }6. 四种运行方式的对比与选型指南在实际项目中选择合适的运行方式需要考虑多个因素。以下是四种方式的详细对比6.1 技术特性对比特性单次查询交互式流式受控权限钩子驱动复杂度低中中高高控制粒度粗中细最细实时性无高中中高安全性中依赖实现高依赖实现适用场景简单任务复杂交互生产环境企业集成6.2 性能与资源考量不同运行方式对系统资源的需求差异显著class PerformanceAnalyzer: def analyze_mode_performance(self, mode: str, task_complexity: str) - Dict: 分析不同模式的性能特征 base_metrics { 单次查询: {内存占用: 低, CPU使用: 中, 网络流量: 低}, 交互式流式: {内存占用: 中, CPU使用: 高, 网络流量: 高}, 受控权限: {内存占用: 中, CPU使用: 中, 网络流量: 中}, 钩子驱动: {内存占用: 高, CPU使用: 高, 网络流量: 中} } complexity_multipliers { simple: 1.0, medium: 1.5, complex: 2.5 } metrics base_metrics.get(mode, {}) multiplier complexity_multipliers.get(task_complexity, 1.0) return {k: f{v} × {multiplier} for k, v in metrics.items()}6.3 实际项目选型建议根据项目需求选择最合适的运行方式选择单次查询模式当任务明确且范围有限不需要实时监控资源约束严格简单的自动化脚本选择交互式流式模式当任务复杂需要人工指导需要实时了解代理决策过程开发调试阶段教育或演示场景选择受控权限模式当生产环境部署处理敏感数据或操作需要严格的访问控制合规性要求高的场景选择钩子驱动模式当需要深度业务逻辑集成有复杂的预处理或后处理需求企业级系统集成需要高度定制化的行为6.4 混合模式策略在实际项目中经常需要组合使用多种模式class HybridAgentOrchestrator { private modes: AgentMode[] []; async executeComplexWorkflow(workflow: WorkflowDefinition) { for (const step of workflow.steps) { const mode this.selectModeForStep(step); await this.executeStep(step, mode); } } private selectModeForStep(step: WorkflowStep): AgentMode { if (step.requiresHumanInput) { return interactive; } else if (step.involvesSensitiveOperations) { return controlled; } else if (step.hasComplexPreprocessing) { return hook-driven; } else { return single-query; } } private async executeStep(step: WorkflowStep, mode: AgentMode) { switch (mode) { case single-query: return await this.executeSingleQuery(step); case interactive: return await this.executeInteractive(step); case controlled: return await this.executeControlled(step); case hook-driven: return await this.executeWithHooks(step); } } }7. 生产环境部署的最佳实践将 Claude Code Agent 部署到生产环境需要特别注意稳定性、安全性和可观测性。7.1 配置管理与环境隔离# config/production.yaml claude_agent: model: claude-sonnet-5 max_turns: 50 max_budget_usd: 5.0 effort: high permissions: mode: dontAsk allowed_tools: - Read - Glob - Grep - Bash(npm *) - Bash(git *) hooks: pre_tool_use: security.validate_tool post_tool_use: audit.record_tool_result monitoring: enable_metrics: true log_level: INFO cost_tracking: true7.2 错误处理与重试机制健全的错误处理是生产系统的关键from tenacity import retry, stop_after_attempt, wait_exponential import logging logger logging.getLogger(__name__) class ProductionAgentClient: def __init__(self, config: Dict): self.config config self.metrics MetricsCollector() retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) async def execute_with_fallback(self, prompt: str, primary_mode: str, fallback_mode: str): 带降级策略的执行 try: start_time time.time() result await self.execute_in_mode(prompt, primary_mode) self.metrics.record_success(primary_mode, time.time() - start_time) return result except AgentError as e: self.metrics.record_failure(primary_mode, str(e)) logger.warning(f主模式 {primary_mode} 失败: {e}, 尝试降级到 {fallback_mode}) if self.should_fallback(e): try: result await self.execute_in_mode(prompt, fallback_mode) self.metrics.record_fallback_success(fallback_mode) return result except AgentError as fallback_error: logger.error(f降级模式也失败: {fallback_error}) raise ProductionAgentError(所有执行模式均失败) from fallback_error else: raise def should_fallback(self, error: AgentError) - bool: 判断是否应该降级 non_fallback_errors [ permission_denied, invalid_configuration, authentication_failed ] return error.code not in non_fallback_errors7.3 监控与可观测性完整的监控体系应该覆盖所有关键指标interface AgentMetrics { executionCount: number; successRate: number; averageCost: number; averageDuration: number; errorBreakdown: Mapstring, number; toolUsage: Mapstring, number; } class ProductionMonitoring { private metrics: AgentMetrics { executionCount: 0, successRate: 0, averageCost: 0, averageDuration: 0, errorBreakdown: new Map(), toolUsage: new Map() }; recordExecutionStart() { this.metrics.executionCount; } recordExecutionSuccess(duration: number, cost: number) { this.updateSuccessRate(true); this.updateAverages(duration, cost); } recordExecutionFailure(error: AgentError, duration: number) { this.updateSuccessRate(false); this.recordError(error); this.updateAverages(duration, 0); } recordToolUsage(toolName: string) { const currentCount this.metrics.toolUsage.get(toolName) || 0; this.metrics.toolUsage.set(toolName, currentCount 1); } private updateSuccessRate(success: boolean) { const total this.metrics.executionCount; const currentSuccesses this.metrics.successRate * (total - 1); this.metrics.successRate (currentSuccesses (success ? 1 : 0)) / total; } }7.4 安全加固措施生产环境必须实施严格的安全控制class SecurityHardenedAgent: def __init__(self, security_config: SecurityConfig): self.validator SecurityValidator(security_config) self.auditor SecurityAuditor() self.limiter RateLimiter() async def secure_execute(self, prompt: str, user_context: UserContext): # 1. 输入验证 if not self.validator.validate_input(prompt): raise SecurityError(输入验证失败) # 2. 速率限制 if not await self.limiter.check_limit(user_context): raise RateLimitError(请求频率超限) # 3. 上下文安全检查 safe_prompt self.sanitize_prompt(prompt, user_context) # 4. 安全执行 async with self.create_secure_session(safe_prompt) as session: async for message in session.stream(): # 5. 输出过滤 filtered_message self.filter_output(message) yield filtered_message # 6. 实时审计 await self.auditor.audit_message(message, user_context) def sanitize_prompt(self, prompt: str, user_context: UserContext) - str: 提示词安全处理 # 移除敏感信息 sanitized self.remove_sensitive_data(prompt) # 添加安全边界 secured_prompt f 安全执行上下文: - 用户: {user_context.user_id} - 权限: {user_context.permissions} - 时间: {datetime.now().isoformat()} 任务: {sanitized} 安全约束: - 禁止访问系统文件 - 禁止执行危险命令 - 遵守数据保护政策 return secured_prompt通过理解 Claude Code Agent 的四种运行方式及其适用场景开发者可以根据具体需求选择合适的模式构建出既强大又安全的 AI 代理应用。在实际项目中往往需要根据任务复杂度、安全要求和资源约束灵活选择和组合这些模式。