在实际 AI 应用开发中Claude Code 提供了一套完整的工具链来构建智能代理系统。这套系统涉及多个核心概念MCPModel Context Protocol用于标准化模型与外部工具的交互SubAgent 用于处理特定子任务Agent Skill 封装可复用的能力单元Hook 机制用于拦截和处理数据流而图片处理、上下文管理和后台任务则是实际项目中必须面对的工程挑战。理解这些组件如何协同工作是构建稳定、可扩展 AI 应用的关键。本文将围绕 Claude Code 的核心架构从基础概念到实战配置完整介绍如何搭建一个具备复杂任务处理能力的智能代理系统。重点会放在 MCP 协议的配置与集成、SubAgent 的创建与调度、Agent Skill 的设计与组合、Hook 机制的安全实现以及图片、上下文和后台任务这些高频场景的处理方案。1. 理解 Claude Code 的核心组件与协作关系在开始配置和编码之前需要先理清 Claude Code 系统中各个核心组件的职责和边界。很多配置问题本质上是对组件关系理解不清导致的。1.1 MCPModel Context Protocol的角色与价值MCP 是 Claude Code 体系中最重要的基础设施之一。它定义了一套标准协议让 AI 模型能够安全、可控地调用外部工具和资源。可以把 MCP 理解为模型与外部世界之间的“安全网关”。在实际项目中MCP 主要解决三个问题工具标准化不同工具数据库、API、文件系统通过 MCP Server 提供统一的调用接口权限控制MCP 可以限制模型只能访问授权的资源和操作会话管理MCP 维护工具调用的上下文确保多次调用间的状态一致性一个典型的 MCP 配置包含 Server 和 Client 两部分。Server 负责具体工具的实现Client 负责与模型交互。1.2 SubAgent 与 Agent Skill 的区别与联系这是最容易混淆的两个概念。简单来说SubAgent是一个完整的、可独立运行的代理实例有自己独立的配置、上下文和任务处理能力Agent Skill是封装好的能力单元可以被多个 Agent 复用类似于编程中的函数或模块举例说明在一个电商客服系统中你可能有一个“订单查询 SubAgent”专门处理订单相关问题这个 SubAgent 会使用“数据库查询 Skill”、“用户验证 Skill”等多个技能模块。关键区别在于SubAgent 有独立的任务循环和状态管理Agent Skill 是无状态的专注于单一能力的实现一个 SubAgent 可以组合多个 Agent Skill同一个 Agent Skill 可以被不同 SubAgent 共享使用1.3 Hook 机制的工作原理与适用场景Hook钩子是 Claude Code 中的拦截器机制允许在特定执行点插入自定义逻辑。常见的 Hook 类型包括预处理 Hook在请求到达模型前进行数据清洗、验证或增强后处理 Hook在模型响应后对结果进行格式化、过滤或记录错误处理 Hook在异常发生时进行统一处理和恢复Hook 的正确使用可以大幅提升系统的可观测性和可靠性。比如通过 Hook 记录所有模型调用日志或者对敏感信息进行自动脱敏。2. 环境准备与 Claude Code 基础配置开始实战前需要先搭建基础的开发环境。Claude Code 对运行环境有特定要求版本不匹配是常见问题来源。2.1 系统要求与依赖检查Claude Code 通常需要以下环境支持组件最低版本推荐版本验证命令Python3.83.9python --versionNode.js16.018.0node --versionDocker20.024.0docker --version内存8GB16GB-安装基础依赖包# 创建虚拟环境 python -m venv claude_code_env source claude_code_env/bin/activate # Linux/Mac # claude_code_env\Scripts\activate # Windows # 安装核心包 pip install anthropic-claude-code pip install mcp-protocol pip install agent-skill-kit2.2 Claude Code 项目结构规划合理的项目结构是后续扩展的基础。推荐按功能模块划分目录claude-code-project/ ├── config/ │ ├── mcp_servers.yaml # MCP 服务器配置 │ ├── agents.yaml # Agent 定义配置 │ └── hooks.yaml # Hook 配置 ├── src/ │ ├── mcp_servers/ # 自定义 MCP 服务器 │ ├── agents/ # SubAgent 实现 │ ├── skills/ # Agent Skill 模块 │ └── hooks/ # Hook 实现 ├── data/ # 数据文件 ├── logs/ # 日志文件 └── tests/ # 测试用例这种结构的好处是配置与代码分离功能模块清晰便于团队协作和自动化部署。2.3 基础配置文件详解创建核心配置文件config/base.yaml# Claude Code 基础配置 claude_code: version: 1.0 environment: development # 模型配置 model: name: claude-3-sonnet-20240229 temperature: 0.7 max_tokens: 4096 # 日志配置 logging: level: INFO format: %(asctime)s - %(name)s - %(levelname)s - %(message)s file_path: ./logs/claude_code.log # 超时配置秒 timeouts: mcp_request: 30 agent_response: 60 background_task: 300这个基础配置为后续各个组件提供了默认参数避免在每个子配置中重复定义。3. MCP 服务器的配置与集成实战MCP 是 Claude Code 与外部系统交互的桥梁正确的 MCP 配置是整个系统稳定运行的前提。3.1 内置 MCP 服务器的启用与配置Claude Code 提供了一些内置的 MCP 服务器开箱即用。在config/mcp_servers.yaml中配置# 内置 MCP 服务器配置 builtin_servers: # 文件系统 MCP filesystem: enabled: true root_path: ./data/files read_only: false allowed_extensions: [.txt, .json, .csv, .md] # 网络请求 MCP http_client: enabled: true timeout: 30 allowed_domains: [api.example.com, data.example.org] rate_limit: 100 # 每分钟最大请求数 # 数据库 MCP需要额外配置 database: enabled: false # 生产环境需要配置具体连接信息启用内置服务器后需要在代码中初始化from claude_code.mcp import BuiltinMCPServerManager class MCPManager: def __init__(self, config_path): self.config self.load_config(config_path) self.server_manager BuiltinMCPServerManager() def start_servers(self): 启动所有启用的 MCP 服务器 enabled_servers [ name for name, config in self.config[builtin_servers].items() if config.get(enabled, False) ] for server_name in enabled_servers: server_config self.config[builtin_servers][server_name] try: self.server_manager.start_server(server_name, server_config) print(fMCP Server {server_name} 启动成功) except Exception as e: print(fMCP Server {server_name} 启动失败: {e}) def get_client(self, server_name): 获取指定 MCP 服务器的客户端 return self.server_manager.get_client(server_name)3.2 自定义 MCP 服务器的开发指南当内置服务器无法满足需求时需要开发自定义 MCP 服务器。以下是一个数据库查询 MCP 服务器的完整示例# src/mcp_servers/database_server.py import json import logging from typing import List, Dict, Any from mcp import MCPServer, Tool, TextContent class DatabaseMCPServer(MCPServer): def __init__(self, db_config: Dict[str, Any]): super().__init__(database-server) self.db_config db_config self.logger logging.getLogger(database_mcp) self.setup_tools() def setup_tools(self): 注册 MCP 工具 self.register_tool( Tool( namequery_database, description执行 SQL 查询语句, input_schema{ type: object, properties: { sql: {type: string, description: SQL 查询语句}, parameters: {type: object, description: 查询参数} }, required: [sql] } ) ) self.register_tool( Tool( nameexecute_sql, description执行 SQL 更新操作INSERT/UPDATE/DELETE, input_schema{ type: object, properties: { sql: {type: string, description: SQL 语句}, parameters: {type: object, description: 参数} }, required: [sql] } ) ) async def handle_query_database(self, sql: str, parameters: Dict None) - List[TextContent]: 处理数据库查询请求 try: # 实际项目中这里会连接真实的数据库 # 示例中返回模拟数据 self.logger.info(f执行查询: {sql}) # 参数化查询防止 SQL 注入 if parameters: sql self._apply_parameters(sql, parameters) # 模拟查询结果 results [ {id: 1, name: 示例用户, email: userexample.com}, {id: 2, name: 测试用户, email: testexample.com} ] return [TextContent(typetext, textjson.dumps(results, ensure_asciiFalse))] except Exception as e: self.logger.error(f数据库查询失败: {e}) return [TextContent(typetext, textf查询失败: {str(e)})] def _apply_parameters(self, sql: str, parameters: Dict) - str: 简单的参数替换生产环境应使用预编译语句 for key, value in parameters.items(): placeholder f:{key} sql sql.replace(placeholder, str(value)) return sql配置自定义服务器# config/mcp_servers.yaml custom_servers: database: enabled: true module: src.mcp_servers.database_server class: DatabaseMCPServer config: host: localhost port: 5432 database: myapp username: claude_user password: encrypted_password3.3 MCP 客户端集成与错误处理MCP 客户端负责与服务器通信需要妥善处理网络异常和超时# src/mcp_clients/managed_client.py import asyncio from typing import Optional, Dict, Any from mcp import MCPClient class ManagedMCPClient: def __init__(self, server_name: str, server_config: Dict[str, Any]): self.server_name server_name self.server_config server_config self.client: Optional[MCPClient] None self.max_retries 3 self.retry_delay 1.0 async def connect(self): 连接 MCP 服务器支持重试机制 for attempt in range(self.max_retries): try: self.client await MCPClient.create( self.server_config[transport], self.server_config[config] ) print(fMCP Client {self.server_name} 连接成功) return except Exception as e: if attempt self.max_retries - 1: raise ConnectionError(f连接 {self.server_name} 失败: {e}) print(f连接尝试 {attempt 1} 失败{self.retry_delay}秒后重试: {e}) await asyncio.sleep(self.retry_delay) async def call_tool(self, tool_name: str, arguments: Dict[str, Any], timeout: int 30): 调用 MCP 工具支持超时控制 if not self.client: raise RuntimeError(客户端未连接) try: return await asyncio.wait_for( self.client.call_tool(tool_name, arguments), timeouttimeout ) except asyncio.TimeoutError: raise TimeoutError(fMCP 工具调用超时: {tool_name}) except Exception as e: raise RuntimeError(fMCP 工具调用失败: {tool_name}, 错误: {e}) async def close(self): 关闭客户端连接 if self.client: await self.client.close()4. SubAgent 的创建与任务调度SubAgent 是 Claude Code 系统中承担具体任务的执行单元良好的 SubAgent 设计直接影响系统性能。4.1 SubAgent 基础架构设计一个完整的 SubAgent 应该包含以下组件# src/agents/base_agent.py from abc import ABC, abstractmethod from typing import Dict, Any, List, Optional from dataclasses import dataclass dataclass class AgentContext: Agent 执行上下文 session_id: str user_id: Optional[str] None metadata: Dict[str, Any] None class BaseSubAgent(ABC): SubAgent 基类 def __init__(self, agent_id: str, config: Dict[str, Any]): self.agent_id agent_id self.config config self.is_running False self.task_queue asyncio.Queue() self.current_context: Optional[AgentContext] None abstractmethod async def initialize(self): 初始化 Agent 资源 pass abstractmethod async def process_task(self, task_data: Dict[str, Any], context: AgentContext) - Dict[str, Any]: 处理具体任务 pass abstractmethod async def cleanup(self): 清理资源 pass async def run(self): Agent 主循环 self.is_running True await self.initialize() while self.is_running: try: task, context await self.task_queue.get() result await self.process_task(task, context) # 处理结果回调或存储 self.task_queue.task_done() except Exception as e: print(fAgent {self.agent_id} 任务处理异常: {e}) def submit_task(self, task_data: Dict[str, Any], context: AgentContext): 提交任务到队列 self.task_queue.put_nowait((task_data, context))4.2 专用 SubAgent 实现示例以下是一个专门处理数据分析任务的 SubAgent# src/agents/data_analysis_agent.py import pandas as pd import numpy as np from typing import Dict, Any, List from .base_agent import BaseSubAgent, AgentContext class DataAnalysisAgent(BaseSubAgent): 数据分析专用 SubAgent def __init__(self, agent_id: str, config: Dict[str, Any]): super().__init__(agent_id, config) self.supported_operations [statistics, filter, aggregate, visualize] self.data_cache {} async def initialize(self): 初始化数据分析环境 print(f数据分析 Agent {self.agent_id} 初始化完成) # 可以在这里加载预训练模型或常用数据集 async def process_task(self, task_data: Dict[str, Any], context: AgentContext) - Dict[str, Any]: 处理数据分析任务 operation task_data.get(operation) data task_data.get(data) if operation not in self.supported_operations: return {error: f不支持的操作: {operation}} try: if operation statistics: result await self._calculate_statistics(data) elif operation filter: result await self._filter_data(data, task_data.get(conditions)) elif operation aggregate: result await self._aggregate_data(data, task_data.get(group_by)) elif operation visualize: result await self._generate_visualization(data, task_data.get(chart_type)) return {success: True, operation: operation, result: result} except Exception as e: return {success: False, error: str(e)} async def _calculate_statistics(self, data: List[Dict]) - Dict[str, Any]: 计算数据统计信息 df pd.DataFrame(data) return { count: len(df), mean: df.select_dtypes(include[np.number]).mean().to_dict(), std: df.select_dtypes(include[np.number]).std().to_dict(), min: df.select_dtypes(include[np.number]).min().to_dict(), max: df.select_dtypes(include[np.number]).max().to_dict() } async def _filter_data(self, data: List[Dict], conditions: Dict) - List[Dict]: 根据条件过滤数据 # 简化实现实际项目需要更复杂的条件解析 filtered [] for item in data: if all(item.get(k) v for k, v in conditions.items()): filtered.append(item) return filtered async def cleanup(self): 清理缓存数据 self.data_cache.clear() print(f数据分析 Agent {self.agent_id} 资源清理完成)4.3 SubAgent 调度器与负载均衡多个 SubAgent 需要统一的调度管理# src/agents/agent_scheduler.py import asyncio from typing import Dict, List, Optional from enum import Enum class AgentStatus(Enum): IDLE idle BUSY busy OFFLINE offline class AgentScheduler: SubAgent 调度器 def __init__(self): self.agents: Dict[str, Dict] {} # agent_id - agent_info self.task_router {} # task_type - agent_type def register_agent(self, agent_id: str, agent_type: str, agent_instance, capacity: int 10): 注册 SubAgent self.agents[agent_id] { instance: agent_instance, type: agent_type, status: AgentStatus.IDLE, capacity: capacity, current_tasks: 0, last_heartbeat: asyncio.get_event_loop().time() } def register_task_route(self, task_type: str, agent_type: str): 注册任务路由规则 self.task_router[task_type] agent_type async def dispatch_task(self, task_type: str, task_data: Dict, context: Dict) - Optional[str]: 分发任务到合适的 SubAgent if task_type not in self.task_router: return None target_agent_type self.task_router[task_type] suitable_agents [ agent_id for agent_id, info in self.agents.items() if info[type] target_agent_type and info[status] ! AgentStatus.OFFLINE and info[current_tasks] info[capacity] ] if not suitable_agents: return None # 简单负载均衡选择任务最少的 Agent best_agent min(suitable_agents, keylambda aid: self.agents[aid][current_tasks]) self.agents[best_agent][current_tasks] 1 self.agents[best_agent][status] AgentStatus.BUSY # 实际提交任务 await self.agents[best_agent][instance].submit_task(task_data, context) return best_agent async def update_agent_status(self, agent_id: str, status: AgentStatus): 更新 Agent 状态 if agent_id in self.agents: self.agents[agent_id][status] status if status AgentStatus.IDLE: self.agents[agent_id][current_tasks] 05. Agent Skill 的设计与组合应用Agent Skill 是能力复用的关键良好的 Skill 设计可以大幅提升开发效率。5.1 Skill 接口规范与实现模板定义统一的 Skill 接口# src/skills/base_skill.py from abc import ABC, abstractmethod from typing import Dict, Any, List class BaseSkill(ABC): Skill 基类 def __init__(self, skill_name: str, version: str 1.0): self.skill_name skill_name self.version version self.required_configs [] # 必须的配置项 abstractmethod async def execute(self, parameters: Dict[str, Any], context: Dict[str, Any]) - Dict[str, Any]: 执行 Skill 功能 pass abstractmethod def get_description(self) - str: 返回 Skill 描述 pass def validate_parameters(self, parameters: Dict[str, Any]) - bool: 验证输入参数 return True # 子类可重写具体验证逻辑 def get_usage_examples(self) - List[Dict[str, Any]]: 返回使用示例 return []5.2 常用 Skill 实现示例以下是一些实用的 Agent Skill 实现# src/skills/data_processing_skill.py import json import re from typing import Dict, Any, List from .base_skill import BaseSkill class DataProcessingSkill(BaseSkill): 数据处理 Skill def __init__(self): super().__init__(data_processing, 1.0) self.supported_operations [clean, transform, validate, extract] async def execute(self, parameters: Dict[str, Any], context: Dict[str, Any]) - Dict[str, Any]: operation parameters.get(operation) data parameters.get(data) if operation not in self.supported_operations: return {error: f不支持的操作: {operation}} try: if operation clean: result self._clean_data(data, parameters.get(rules)) elif operation transform: result self._transform_data(data, parameters.get(mapping)) elif operation validate: result self._validate_data(data, parameters.get(schema)) elif operation extract: result self._extract_info(data, parameters.get(patterns)) return {success: True, result: result} except Exception as e: return {success: False, error: str(e)} def _clean_data(self, data: Any, rules: List[Dict]) - Any: 数据清洗 if isinstance(data, str): for rule in rules or []: if rule[type] remove_whitespace: data re.sub(r\s, , data).strip() elif rule[type] remove_special_chars: data re.sub(r[^\w\s], , data) return data def get_description(self) - str: return 提供数据清洗、转换、验证和提取等数据处理功能 def get_usage_examples(self) - List[Dict[str, Any]]: return [ { operation: clean, data: Hello, World! , rules: [{type: remove_whitespace}] } ] # src/skills/api_integration_skill.py import aiohttp from typing import Dict, Any from .base_skill import BaseSkill class APIIntegrationSkill(BaseSkill): API 集成 Skill def __init__(self): super().__init__(api_integration, 1.0) self.required_configs [base_url, timeout] async def execute(self, parameters: Dict[str, Any], context: Dict[str, Any]) - Dict[str, Any]: method parameters.get(method, GET).upper() endpoint parameters.get(endpoint, ) data parameters.get(data) if not endpoint: return {error: 必须提供 endpoint 参数} try: async with aiohttp.ClientSession() as session: url f{self.config[base_url]}/{endpoint} async with session.request( methodmethod, urlurl, jsondata, timeoutaiohttp.ClientTimeout(totalself.config[timeout]) ) as response: if response.status 200: result await response.json() return {success: True, data: result} else: return {success: False, error: fHTTP {response.status}} except Exception as e: return {success: False, error: str(e)} def get_description(self) - str: return 提供统一的 HTTP API 调用能力支持 GET/POST/PUT/DELETE 等方法5.3 Skill 组合与依赖管理多个 Skill 可以组合成更复杂的能力# src/skills/skill_composer.py from typing import Dict, List, Any from .base_skill import BaseSkill class SkillComposer: Skill 组合器 def __init__(self): self.registered_skills: Dict[str, BaseSkill] {} self.skill_dependencies: Dict[str, List[str]] {} # skill_name - 依赖的skills def register_skill(self, skill: BaseSkill): 注册 Skill self.registered_skills[skill.skill_name] skill def define_composite_skill(self, name: str, skill_sequence: List[Dict]): 定义组合 Skill # 存储组合逻辑实际执行时按顺序调用 self.skill_dependencies[name] skill_sequence async def execute_composite_skill(self, composite_name: str, initial_parameters: Dict[str, Any], context: Dict[str, Any]) - Dict[str, Any]: 执行组合 Skill if composite_name not in self.skill_dependencies: return {error: f未定义的组合 Skill: {composite_name}} current_result initial_parameters skill_sequence self.skill_dependencies[composite_name] for step in skill_sequence: skill_name step[skill] if skill_name not in self.registered_skills: return {error: f未注册的 Skill: {skill_name}} # 将上一步的结果作为下一步的输入 step_parameters {**step.get(parameters, {}), **current_result} skill_instance self.registered_skills[skill_name] result await skill_instance.execute(step_parameters, context) if not result.get(success, False): return {error: f步骤 {skill_name} 执行失败, details: result} current_result result return {success: True, final_result: current_result}6. Hook 机制的实现与安全考量Hook 是系统可扩展性的关键但不当使用可能引入安全风险。6.1 核心 Hook 点与实现方案Claude Code 的主要 Hook 点包括# src/hooks/hook_manager.py from enum import Enum from typing import Dict, Any, List, Callable, Optional import inspect class HookPoint(Enum): BEFORE_MODEL_CALL before_model_call AFTER_MODEL_CALL after_model_call BEFORE_MCP_CALL before_mcp_call AFTER_MCP_CALL after_mcp_call ON_ERROR on_error ON_TASK_START on_task_start ON_TASK_COMPLETE on_task_complete class HookManager: Hook 管理器 def __init__(self): self.hooks: Dict[HookPoint, List[Callable]] { point: [] for point in HookPoint } def register_hook(self, hook_point: HookPoint, hook_func: Callable, priority: int 5): 注册 Hook 函数 if not inspect.iscoroutinefunction(hook_func): raise ValueError(Hook 函数必须是异步函数) self.hooks[hook_point].append((priority, hook_func)) # 按优先级排序 self.hooks[hook_point].sort(keylambda x: x[0]) async def execute_hooks(self, hook_point: HookPoint, context: Dict[str, Any]) - Optional[Dict[str, Any]]: 执行指定 Hook 点的所有函数 if hook_point not in self.hooks: return context current_context context.copy() for priority, hook_func in self.hooks[hook_point]: try: result await hook_func(current_context) if result is not None: current_context.update(result) except Exception as e: print(fHook 执行失败 {hook_point}.{hook_func.__name__}: {e}) # 继续执行其他 Hook不阻断主流程 return current_context6.2 安全 Hook 的实现示例安全相关的 Hook 需要特别注意# src/hooks/security_hooks.py import re from typing import Dict, Any class SecurityHooks: 安全相关 Hook staticmethod async def sanitize_input(context: Dict[str, Any]) - Dict[str, Any]: 输入清洗 Hook if user_input in context: user_input context[user_input] # 移除潜在的恶意脚本 sanitized re.sub(rscript\b[^]*(?:(?!\/script)[^]*)*\/script, , user_input) # 移除危险 HTML 标签 sanitized re.sub(r(?:iframe|object|embed)[^]*, , sanitized) context[user_input] sanitized return context staticmethod async def validate_mcp_parameters(context: Dict[str, Any]) - Dict[str, Any]: MCP 参数验证 Hook if mcp_call in context and parameters in context[mcp_call]: parameters context[mcp_call][parameters] # 检查参数大小防止过大请求 param_size len(str(parameters)) if param_size 1024 * 1024: # 1MB raise ValueError(MCP 参数过大) # 检查危险操作根据具体 MCP 定制 if context[mcp_call][tool] file_write: allowed_paths [/tmp/, ./data/] path parameters.get(path, ) if not any(path.startswith(allowed) for allowed in allowed_paths): raise ValueError(f文件写入路径不在允许范围内: {path}) return context staticmethod async def log_sensitive_operations(context: Dict[str, Any]) - Dict[str, Any]: 敏感操作日志 Hook sensitive_keywords [password, token, key, secret] if mcp_call in context: tool_name context[mcp_call].get(tool, ) parameters context[mcp_call].get(parameters, {}) # 检查是否涉及敏感操作 if any(keyword in tool_name for keyword in sensitive_keywords): # 记录日志但脱敏参数 logged_params { k: ***REDACTED*** if any(sk in k.lower() for sk in sensitive_keywords) else v for k, v in parameters.items() } print(f敏感操作记录: {tool_name}, 参数: {logged_params}) return context6.3 Hook 执行顺序与错误处理复杂的 Hook 系统需要明确的执行顺序和错误处理策略# src/hooks/advanced_hook_manager.py from typing import Dict, Any, List, Tuple from enum import Enum class HookExecutionPolicy(Enum): STOP_ON_FIRST_ERROR stop_on_first_error CONTINUE_ON_ERROR continue_on_error STOP_ON_CRITICAL_ERROR stop_on_critical_error class AdvancedHookManager(HookManager): 增强的 Hook 管理器 def __init__(self): super().__init__() self.hook_policies: Dict[HookPoint, HookExecutionPolicy] { HookPoint.BEFORE_MODEL_CALL: HookExecutionPolicy.STOP_ON_CRITICAL_ERROR, HookPoint.BEFORE_MCP_CALL: HookExecutionPolicy.STOP_ON_FIRST_ERROR, HookPoint.ON_ERROR: HookExecutionPolicy.CONTINUE_ON_ERROR } async def execute_hooks_with_policy(self, hook_point: HookPoint, context: Dict[str, Any]) - Tuple[Dict[str, Any], List[Exception]]: 根据策略执行 Hook返回结果和错误列表 policy self.hook_policies.get(hook_point, HookExecutionPolicy.CONTINUE_ON_ERROR) errors [] current_context context.copy() for priority, hook_func in self.hooks[hook_point]: try: result await hook_func(current_context) if result is not None: current_context.update(result) except Exception as e: errors.append(e) print(fHook 执行错误 {hook_point}.{hook_func.__name__}: {e}) if policy HookExecutionPolicy.STOP_ON_FIRST_ERROR: break elif policy HookExecutionPolicy.STOP_ON_CRITICAL_ERROR and self._is_critical_error(e): break # CONTINUE_ON_ERROR 策略继续执行 return current_context, errors def _is_critical_error(self, error: Exception) - bool: 判断是否为关键错误 critical_errors [ PermissionDenied, SecurityViolation, Timeout, ConnectionError ] return any(keyword in str(error) for keyword in critical_errors)7. 图片处理与上下文管理实战图片处理和长上下文管理是 AI 应用中的常见挑战需要专门的解决方案。7.1 图片上传、处理与存储方案完整的图片处理流程# src/features/image_processor.py import base64 import io from PIL import Image from typing import Dict, Any, Optional import aiofiles class ImageProcessor: 图片处理器 def __init__(self, config: Dict[str, Any]): self.config config self.supported_formats [JPEG, PNG, GIF, BMP] self.max_file_size config.get(max_file_size,