> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/context_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/context_engineering_guide/di-san-bu-fen-jin-jie-ji-shu-yu-jia-gou/09_agents/9.5_advanced_agent_context.md). # 9.5 智能体上下文管理的高级主题 本章节深化对智能体系统中复杂上下文问题的讨论，涵盖多智能体协作、长期任务、工具结果管理和记忆系统等实战主题。 ## 9.5.1 多智能体上下文冲突与协调 ### 问题场景 在多智能体系统中，不同的智能体可能有以下冲突： ``` 场景：电商订单处理系统 Agent A (客服助手)的上下文： - 目标：快速回复用户 - 信息：用户想要退货 - 权限：只能提供政策信息 Agent B (订单管理)的上下文： - 目标：防止滥用退货 - 信息：该用户频繁退货 - 权限：可以拒绝退货请求 冲突：两个Agent看到的用户信息不同，可能导致： 1. 重复处理（都试图处理同一个订单） 2. 决策冲突（一个同意，一个拒绝） 3. 循环依赖（Agent A 等待Agent B ，反之亦然） ``` ### 冲突解决机制 ```python from enum import Enum from dataclasses import dataclass from typing import Any, Dict, List, Set, Optional from datetime import datetime import threading class ConflictType(Enum): """多智能体冲突类型""" RESOURCE_CONFLICT = "resource_conflict" # 竞争同一资源 DECISION_CONFLICT = "decision_conflict" # 相互矛盾的决策 INFORMATION_CONFLICT = "information_conflict" # 掌握的信息不一致 AUTHORITY_CONFLICT = "authority_conflict" # 权限冲突 ORDERING_CONFLICT = "ordering_conflict" # 执行顺序冲突 @dataclass class ConflictEvent: """冲突事件记录""" agent_a: str agent_b: str conflict_type: ConflictType resource: str # 冲突涉及的资源 timestamp: str resolved: bool = False resolution_strategy: Optional[str] = None resolution_details: Optional[Dict] = None class ContextConflictResolver: """ 多智能体上下文冲突解决器。 使用仲裁、顺序化和优先级来解决冲突。 """ def __init__(self): self.conflict_log: List[ConflictEvent] = [] self.resource_locks: Dict[str, str] = {} # 资源ID -> 持有者Agent self.priority_matrix: Dict[str, int] = {} # Agent -> 优先级（1-10） self.conflict_mutex = threading.Lock() def detect_potential_conflict( self, agent_a_context: Dict, agent_b_context: Dict, shared_resource: str ) -> Optional[ConflictType]: """ 检测两个Agent的上下文是否可能产生冲突。 检查： 1. 是否竞争同一资源 2. 信息视图是否矛盾 3. 目标是否相反 """ # 检查资源冲突 resources_a = set(agent_a_context.get('accessible_resources', [])) resources_b = set(agent_b_context.get('accessible_resources', [])) if shared_resource in (resources_a & resources_b): # 都能访问这个资源，可能产生竞争 return ConflictType.RESOURCE_CONFLICT # 检查信息冲突 info_a = agent_a_context.get('knowledge_base', {}) info_b = agent_b_context.get('knowledge_base', {}) # 简化：检查关于同一对象的信息是否矛盾 for key in set(info_a.keys()) & set(info_b.keys()): if info_a[key] != info_b[key]: return ConflictType.INFORMATION_CONFLICT # 检查权限冲突 perms_a = set(agent_a_context.get('permissions', [])) perms_b = set(agent_b_context.get('permissions', [])) if 'can_approve' in perms_a and 'can_reject' in perms_b: # 一个能批准，一个能拒绝同一事项 return ConflictType.AUTHORITY_CONFLICT return None def resolve_resource_conflict( self, agent_a: str, agent_b: str, resource: str ) -> str: """ 解决资源竞争冲突。 使用加锁机制确保同时只有一个Agent访问。 """ with self.conflict_mutex: # 检查资源是否已被锁定 if resource in self.resource_locks: current_holder = self.resource_locks[resource] if current_holder != agent_a and current_holder != agent_b: # 资源被第三方持有 return f"WAITING" # 根据优先级分配资源 priority_a = self.priority_matrix.get(agent_a, 5) priority_b = self.priority_matrix.get(agent_b, 5) if priority_a > priority_b: winner = agent_a elif priority_b > priority_a: winner = agent_b else: # 优先级相同，按时间顺序（先请求先得） winner = agent_a # 锁定资源 self.resource_locks[resource] = winner # 记录冲突 conflict = ConflictEvent( agent_a=agent_a, agent_b=agent_b, conflict_type=ConflictType.RESOURCE_CONFLICT, resource=resource, timestamp=datetime.now().isoformat(), resolved=True, resolution_strategy="priority_based_locking", resolution_details={'winner': winner, 'priority_a': priority_a, 'priority_b': priority_b} ) self.conflict_log.append(conflict) return winner def resolve_information_conflict( self, agent_a: str, agent_b: str, attribute: str, value_a: Any, value_b: Any ) -> Any: """ 解决信息冲突。 优先使用更新的或更权威的信息源。 """ # 策略1：时间戳（更新的信息优先） if hasattr(value_a, 'timestamp') and hasattr(value_b, 'timestamp'): newer = value_a if value_a.timestamp > value_b.timestamp else value_b return newer # 策略2：权威性（某些Agent对某些信息更权威） authority_a = self._get_authority_score(agent_a, attribute) authority_b = self._get_authority_score(agent_b, attribute) if authority_a > authority_b: return value_a elif authority_b > authority_a: return value_b # 策略3：共识（需要询问两个Agent，取一致的答案） # 这里简化为取value_a，实际应该是复杂的协商过程 return value_a def synchronize_contexts( self, agents: List[str], shared_attributes: Dict[str, str] # {属性名: 真实值} ) -> Dict[str, Dict]: """ 同步多个Agent的上下文到一致状态。 当冲突无法自动解决时，需要显式同步。 """ synchronized_contexts = {} for agent in agents: # 为每个Agent准备同步后的上下文 sync_context = { 'agent': agent, 'synchronized_attributes': shared_attributes.copy(), 'last_sync_time': datetime.now().isoformat(), } synchronized_contexts[agent] = sync_context return synchronized_contexts def _get_authority_score(self, agent: str, attribute: str) -> float: """ 计算Agent对某个属性的权威性评分。 0 - 1 之间，1表示完全权威。 """ # 这是一个简化的实现，实际应该基于： # - Agent的特化程度 # - 过往准确性记录 # - 数据源的可信度 authority_matrix = { ('customer_service', 'customer_preference'): 0.9, ('customer_service', 'fraud_detection'): 0.3, ('fraud_detection', 'fraud_detection'): 0.95, ('fraud_detection', 'customer_preference'): 0.1, } return authority_matrix.get((agent, attribute), 0.5) ``` ## 9.5.2 长期任务中的上下文漂移问题 ### 问题分析 在长期任务（如跨多天的项目管理或持续监控）中，上下文可能发生“漂移”： ``` 初始任务："为CEO准备季度业绩报告" 第 1 小时: 上下文精准，聚焦于财务数据收集 第 4 小时: 开始包含细枝末节（某个部门的招聘计划） 第 8 小时: 出现明显偏离（讨论员工食堂菜单） 第 16 小时: 完全偏离主题（生成无关的营销文案） 原因： 1. 每一轮交互都添加新信息，久而久之上下文变得嘈杂 2. 早期重要的约束条件逐渐被遗忘 3. 中间步骤的输出可能包含错误，这些错误被纳入后续上下文 4. Token预算压力导致关键信息被压缩或删除 ``` ### 缓解策略 ```python from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple import json class ContextDriftDetector: """检测和防止上下文漂移""" def __init__(self, drift_threshold: float = 0.3): """ drift_threshold: 漂移程度的阈值 - 0: 完全不漂移 - 1: 完全漂移 """ self.drift_threshold = drift_threshold self.initial_goal: Optional[str] = None self.context_history: List[Dict] = [] def set_initial_goal(self, goal: str) -> None: """设置任务初始目标""" self.initial_goal = goal def measure_context_drift( self, current_context: str, step_number: int ) -> Tuple[float, str]: """ 测量当前上下文与初始目标的偏离程度。 使用简化的关键词匹配： - 提取初始目标的关键词 - 计算当前上下文中这些关键词的覆盖率 """ if not self.initial_goal: return 0.0, "No initial goal set" initial_keywords = set(self.initial_goal.lower().split()) current_keywords = set(current_context.lower().split()) # 计算漂移度：初始关键词在当前上下文中的缺失比例 missing_keywords = initial_keywords - current_keywords drift_score = len(missing_keywords) / max(len(initial_keywords), 1) # 记录历史 self.context_history.append({ 'step': step_number, 'drift_score': drift_score, 'timestamp': datetime.now().isoformat(), }) return drift_score, f"Drift score: {drift_score:.2f}" def detect_drift_alert( self, current_context: str, step_number: int ) -> Tuple[bool, Optional[str]]: """ 检测是否需要发出漂移告警。 """ drift_score, _ = self.measure_context_drift(current_context, step_number) if drift_score > self.drift_threshold: alert_msg = ( f"Warning: Context drift detected at step {step_number}. " f"Drift score: {drift_score:.2f}. " f"Consider refocusing the context." ) return True, alert_msg return False, None def get_drift_trend(self) -> Dict: """获取漂移的趋势""" if len(self.context_history) < 2: return {'trend': 'insufficient_data'} recent_drifts = [h['drift_score'] for h in self.context_history[-5:]] # 简单的趋势判断 if all(d < 0.2 for d in recent_drifts): trend = 'stable' elif recent_drifts[-1] > recent_drifts[0]: trend = 'drifting' else: trend = 'improving' return { 'trend': trend, 'current_drift': recent_drifts[-1], 'max_drift': max(recent_drifts), 'avg_drift': sum(recent_drifts) / len(recent_drifts), } class ContextRefocuser: """当检测到漂移时，重新聚焦上下文""" def __init__(self): self.refocus_history = [] def refocus_context( self, original_goal: str, current_context: str, step_number: int ) -> str: """ 重新生成一个聚焦于原始目标的上下文。 策略： 1. 提取原始目标的关键信息 2. 从当前上下文中保留相关部分 3. 删除无关信息 4. 添加进度检查点 """ refocused = f""" 【任务重新聚焦 - 第{step_number}步】 【原始目标】 {original_goal} 【当前进度检查点】 - 当前步骤: {step_number} - 近期完成的关键任务: [待填充] - 下一步目标: [待填充] 【相关上下文】（已过滤无关信息） """ # 简化：保留前500个字符（实际应该更智能的过滤） relevant_context = current_context[:500] refocused += relevant_context refocused += """ 【对齐检查】 请确认： 1. 你仍在追求上述原始目标吗？ 2. 你做的每一步都在帮助实现这个目标吗？ 3. 是否有信息偏离了主题线？ """ # 记录重新聚焦 self.refocus_history.append({ 'step': step_number, 'timestamp': datetime.now().isoformat(), }) return refocused ``` ## 9.5.3 工具调用结果的上下文管理策略 ```python # 场景：智能体需要调用多个工具 agents_task = """ 任务：查找用户John Doe在2024 年 3 月的所有订单 步骤： 1. 调用 search_user API -> 返回用户ID: U12345 2. 调用 get_orders API(U12345, 2024-03) -> 返回订单列表 3. 对每个订单，调用 get_order_details API 4. 生成总结报告 """ # 挑战： # - 中间结果可能很大（如1000个订单的列表） # - 如果全部纳入上下文，会导致Token爆炸 # - 但删除某些结果可能导致后续步骤失败 # - 需要智能的结果摘要和聚合 ``` ### 解决方案 ```python from dataclasses import dataclass from datetime import datetime from enum import Enum from typing import Any, Dict, List, Optional import json class ToolResultProcessingStrategy(Enum): """工具结果处理策略""" KEEP_ALL = "keep_all" # 保留所有结果 SUMMARIZE = "summarize" # 摘要 SAMPLE = "sample" # 采样 AGGREGATE = "aggregate" # 聚合 FILTER_BY_RELEVANCE = "filter_by_relevance" # 按相关性过滤 @dataclass class ToolResult: """工具调用结果""" tool_name: str parameters: Dict result: Any execution_time_ms: float success: bool error: Optional[str] = None timestamp: str = None def __post_init__(self): if not self.timestamp: self.timestamp = datetime.now().isoformat() def size_bytes(self) -> int: """估计结果的大小（字节）""" return len(json.dumps(self.result, default=str).encode()) class ToolResultContextManager: """管理工具结果在上下文中的大小和相关性""" def __init__(self, max_context_size_tokens: int = 4000): self.max_context_size_tokens = max_context_size_tokens self.tool_results: List[ToolResult] = [] def process_tool_result( self, tool_result: ToolResult, strategy: ToolResultProcessingStrategy = ToolResultProcessingStrategy.SUMMARIZE, relevance_query: Optional[str] = None ) -> Dict: """ 处理工具结果，根据策略进行优化。 """ processed = { 'tool_name': tool_result.tool_name, 'original_size_bytes': tool_result.size_bytes(), 'processed_size_bytes': 0, 'strategy_used': strategy.value, 'content': None, } if strategy == ToolResultProcessingStrategy.KEEP_ALL: processed['content'] = tool_result.result processed['processed_size_bytes'] = tool_result.size_bytes() elif strategy == ToolResultProcessingStrategy.SUMMARIZE: processed['content'] = self._summarize_result(tool_result.result) processed['processed_size_bytes'] = len(str(processed['content']).encode()) elif strategy == ToolResultProcessingStrategy.SAMPLE: processed['content'] = self._sample_result(tool_result.result) processed['processed_size_bytes'] = len(str(processed['content']).encode()) elif strategy == ToolResultProcessingStrategy.AGGREGATE: processed['content'] = self._aggregate_result(tool_result.result) processed['processed_size_bytes'] = len(str(processed['content']).encode()) elif strategy == ToolResultProcessingStrategy.FILTER_BY_RELEVANCE: if relevance_query: processed['content'] = self._filter_by_relevance( tool_result.result, relevance_query ) else: processed['content'] = tool_result.result # 计算压缩比 if processed['original_size_bytes'] > 0: compression_ratio = ( 1 - processed['processed_size_bytes'] / processed['original_size_bytes'] ) processed['compression_ratio'] = compression_ratio self.tool_results.append(tool_result) return processed def _summarize_result(self, result: Any) -> Dict: """生成结果的摘要""" if isinstance(result, list): return { 'type': 'list_summary', 'total_count': len(result), 'first_item': result[0] if result else None, 'last_item': result[-1] if result else None, 'sample_count': min(3, len(result)), 'sample_items': result[:3], } elif isinstance(result, dict): return { 'type': 'dict_summary', 'keys': list(result.keys()), 'value_types': {k: type(v).__name__ for k, v in result.items()}, } else: return {'type': 'scalar', 'value': result} def _sample_result(self, result: Any, sample_size: int = 5) -> Any: """从结果中采样""" if isinstance(result, list): if len(result) <= sample_size: return result # 采样策略：头尾各取一些，中间均匀分布 sampled = [] sampled.append(result[0]) # 第一个 if len(result) > 2: sampled.append(result[-1]) # 最后一个 # 中间均匀分布 if len(result) > sample_size: step = len(result) // (sample_size - 2) for i in range(1, len(result) - 1, step): sampled.append(result[i]) return sampled[:sample_size] return result def _aggregate_result(self, result: Any) -> Dict: """聚合结果统计信息""" if isinstance(result, list) and all(isinstance(item, (int, float)) for item in result): return { 'type': 'numeric_aggregate', 'count': len(result), 'sum': sum(result), 'mean': sum(result) / len(result) if result else 0, 'min': min(result) if result else None, 'max': max(result) if result else None, } elif isinstance(result, list) and all(isinstance(item, dict) for item in result): # 对于字典列表，汇总每个键的值 aggregated = {} for key in result[0].keys() if result else []: values = [item[key] for item in result if key in item] if all(isinstance(v, (int, float)) for v in values): aggregated[key] = { 'sum': sum(values), 'mean': sum(values) / len(values), 'count': len(values), } else: aggregated[key] = { 'unique_count': len(set(str(v) for v in values)), 'sample': str(values[0]) if values else None, } return {'type': 'dict_list_aggregate', 'aggregates': aggregated} return {'type': 'non_aggregatable', 'message': 'Cannot aggregate this result type'} def _filter_by_relevance(self, result: Any, query: str) -> Any: """按相关性过滤结果""" if not isinstance(result, list): return result query_terms = set(query.lower().split()) def relevance_score(item: Any) -> float: """计算项目与查询的相关性""" item_str = json.dumps(item, default=str).lower() matching_terms = sum(1 for term in query_terms if term in item_str) return matching_terms / len(query_terms) if query_terms else 0 # 过滤：只保留相关性>0的项目 filtered = [item for item in result if relevance_score(item) > 0] # 按相关性排序 filtered.sort(key=relevance_score, reverse=True) return filtered def build_context_from_results(self) -> str: """从所有工具结果构建上下文""" context = "【工具执行结果】\n" current_size_tokens = 0 for result in self.tool_results: result_text = f"\n【{result.tool_name}】\n" result_text += f"参数: {result.parameters}\n" result_text += f"结果: {str(result.result)[:200]}...\n" # 限制长度 result_text += f"执行时间: {result.execution_time_ms}ms\n" result_size_tokens = len(result_text) // 4 if current_size_tokens + result_size_tokens <= self.max_context_size_tokens: context += result_text current_size_tokens += result_size_tokens else: break return context ``` ## 9.5.4 智能体记忆的持久化与恢复 ```python import sqlite3 import json import re from datetime import datetime, timedelta from pathlib import Path from typing import Any, Dict, List, Optional class AgentMemoryPersistence: """智能体记忆的持久化管理""" def __init__(self, db_path: str = "./agent_memory.db"): self.db_path = Path(db_path) self._init_db() def _init_db(self): """初始化数据库""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() # 创建记忆表 cursor.execute(""" CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, content TEXT NOT NULL, memory_type TEXT NOT NULL, importance REAL, created_at TEXT NOT NULL, last_updated TEXT NOT NULL, access_count INTEGER DEFAULT 0, data_json TEXT ) """) # 创建会话表 cursor.execute(""" CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, created_at TEXT NOT NULL, ended_at TEXT, context_size_tokens INTEGER, final_status TEXT ) """) conn.commit() def save_memory( self, agent_id: str, memory_id: str, content: str, memory_type: str, importance: float = 0.5, data: Optional[Dict[str, Any]] = None ) -> None: """保存单条记忆""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() data_json = json.dumps(data, ensure_ascii=False) if data else None cursor.execute(""" INSERT OR REPLACE INTO memories (id, agent_id, content, memory_type, importance, created_at, last_updated, data_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( memory_id, agent_id, content, memory_type, importance, datetime.now().isoformat(), datetime.now().isoformat(), data_json )) conn.commit() def load_memories( self, agent_id: str, memory_type: Optional[str] = None, min_importance: float = 0.0 ) -> List[Dict]: """加载记忆""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() query = "SELECT * FROM memories WHERE agent_id = ? AND importance >= ?" params = [agent_id, min_importance] if memory_type: query += " AND memory_type = ?" params.append(memory_type) cursor.execute(query, params) rows = cursor.fetchall() memories = [] for row in rows: memory = { 'id': row[0], 'agent_id': row[1], 'content': row[2], 'memory_type': row[3], 'importance': row[4], 'created_at': row[5], 'access_count': row[7], } if row[8]: # JSON 扩展字段，禁止用 pickle 反序列化不可信记忆 try: memory['data'] = json.loads(row[8]) except json.JSONDecodeError: pass memories.append(memory) return memories def save_session_context( self, agent_id: str, session_id: str, context_snapshot: Dict ) -> None: """ 保存整个会话的上下文快照。 用于长期任务的中断和恢复。 """ # 序列化快照 snapshot_json = json.dumps(context_snapshot, default=str) with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO sessions (session_id, agent_id, created_at, context_size_tokens, final_status) VALUES (?, ?, ?, ?, ?) """, ( session_id, agent_id, datetime.now().isoformat(), len(snapshot_json) // 4, # Token估计 'in_progress' )) conn.commit() # 保存详细的上下文为文件（避免BLOB过大） safe_session_id = re.sub(r"[^A-Za-z0-9_.-]", "_", session_id) context_file = self.db_path.parent / f"context_{safe_session_id}.json" with open(context_file, 'w', encoding='utf-8') as f: f.write(snapshot_json) def load_session_context(self, session_id: str) -> Optional[Dict]: """加载保存的会话上下文""" safe_session_id = re.sub(r"[^A-Za-z0-9_.-]", "_", session_id) context_file = self.db_path.parent / f"context_{safe_session_id}.json" if context_file.exists(): with open(context_file, 'r', encoding='utf-8') as f: return json.load(f) return None def cleanup_old_sessions(self, days_to_keep: int = 30) -> int: """清理过期的会话记录""" cutoff_date = (datetime.now() - timedelta(days=days_to_keep)).isoformat() with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute( "DELETE FROM sessions WHERE created_at < ? AND final_status = 'completed'", (cutoff_date,) ) deleted_count = cursor.rowcount conn.commit() return deleted_count ``` ## 9.5.5 智能体推理中的上下文工程视角 Wei 等人在 2026 年的综述 *Agentic Reasoning for Large Language Models*（arXiv: 2601.12538）中提出了智能体推理的三层分类体系：基础推理、自我进化推理和集体多智能体推理。每一层对上下文工程都提出了不同的需求。 ### 上下文内推理与训练后推理 论文区分了两种实现推理能力的路径，这一区分对上下文工程实践有直接影响： * **上下文内推理**（In-Context Reasoning）：通过精心编排的提示词、工具描述和检索结果，在推理时激活模型的推理能力。ReAct、CoT、ToT 等技术均属此类。**上下文工程的核心价值正体现于此**——推理质量高度依赖上下文的信息密度、结构化程度和噪声水平。本书前面章节讨论的 [写入](/context_engineering_guide/di-er-bu-fen-he-xin-ji-shu-yu-ce-le/04_write.md)、[选择](/context_engineering_guide/di-er-bu-fen-he-xin-ji-shu-yu-ce-le/05_select.md)、[压缩](/context_engineering_guide/di-er-bu-fen-he-xin-ji-shu-yu-ce-le/06_compress.md) 和 [隔离](/context_engineering_guide/di-er-bu-fen-he-xin-ji-shu-yu-ce-le/07_isolate.md) 四大策略，本质上都是为上下文内推理服务的。 * **训练后推理**（Post-Training Reasoning）：通过强化学习或监督微调将推理模式固化到模型权重中。这类推理对上下文的依赖相对较低，但仍需要高质量的上下文来触发正确的推理路径。 **工程启示**：对于同一个智能体系统，应根据任务特征混合使用两种路径——高频标准任务使用训练后推理以降低上下文工程复杂度；长尾复杂任务使用上下文内推理以保持灵活性。 ### 自我进化与上下文的动态演进 论文的“自我进化推理”层强调智能体通过反馈和记忆持续改进。从上下文工程角度看，这意味着上下文不再是静态的输入，而是一个随智能体经验积累而动态演进的系统： 1. **经验到上下文的转化**：成功的推理轨迹应被提炼为可复用的上下文模板（如 Few-Shot 示例），失败的轨迹则转化为约束规则注入系统提示词。这与 [9.4 节](/context_engineering_guide/di-san-bu-fen-jin-jie-ji-shu-yu-jia-gou/09_agents/9.4_agent_memory.md) 讨论的记忆机制形成闭环。 2. **上下文质量的反馈信号**：推理成功率本身就是衡量上下文质量的指标。当智能体在某类任务上频繁失败时，应触发上下文管道的自动审查——可能是检索召回率不足、压缩过度丢失了关键信息，或是工具描述不够精确。 ### 多智能体推理的上下文协调 论文将多智能体协作纳入“推理”范畴，认为协调本身是一种分布式推理过程。这与本节 9.5.1「多智能体上下文冲突与协调」讨论的冲突解决机制相呼应，并进一步引出： * **共享上下文层**：参与协作推理的智能体需要一个共享的“事实基础”（Shared Grounding）。上下文工程的挑战在于——既要保证共享信息的一致性，又要为每个智能体保留其专属的角色上下文，防止角色混淆。 * **推理链的上下文开销**：多智能体推理中，每一轮协调都会产生新的中间状态需要注入上下文。随着推理轮次增加，上下文膨胀问题比单智能体场景更为严峻，[第六章](/context_engineering_guide/di-er-bu-fen-he-xin-ji-shu-yu-ce-le/06_compress.md) 讨论的压缩策略在此场景下尤为关键。 这一章节深化了对智能体系统中复杂上下文管理的理解，为生产环境中的多智能体协作提供了实用的解决方案。 ## 9.5.6 案例研究：Claude Code 的上下文管理架构 Claude Code 是 Anthropic 官方的命令行工具，其公开文档展示了现代智能体系统中上下文工程的若干实践。以下分析只把公开文档明确描述的 CLAUDE.md、MCP、工具和记忆机制作为事实；涉及内部函数名、比例和成本数字的内容均为工程化推断，不代表官方实现细节。 ![Claude Code 上下文架构图](https://2119257620-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfnQWa8Cueu67tM389KzJ%2Fuploads%2Fgit-blob-8b322dd98705625c72db6a9d03b62ee6a17d57e3%2Fch09-claude-code-context-architecture.svg?alt=media) 图 9-6：Claude Code 上下文架构图 从工程视角看，Claude Code 的关键不是把所有信息一次性塞进上下文，而是把“可缓存前缀”“项目记忆层级”“按需工具细节”和“压缩恢复回路”拆成不同责任层。 一种可借鉴的系统提示词架构是模块化和可缓存性。可以把系统提示词分解为若干个可独立管理的片段，并用内部边界标记分为两部分： * **静态部分**：包含通用的智能体行为准则、工具使用规范、安全边界等，这部分在整个组织内保持一致，可以跨会话复用。利用供应商提供的 prompt caching 机制，这部分的 cache 创建成本可被多个会话摊销。 * **动态部分**：包含用户特定的约束、会话参数、项目配置等。这部分不能缓存，但占比较小。 在自建系统中，可以把“不可缓存片段”封装成显式函数，并在代码审查中要求说明为什么必须破坏缓存。这样能迫使开发者意识到 cache read 与 cache write 的价格差异；具体倍率应按当前 Anthropic pricing page 和目标模型重算。 ```python # 伪代码：系统提示词架构 def build_system_prompt(user_context, session_config): static_part = SYSTEM_PROMPT_STATIC # 可缓存的稳定前缀 # 缓存标记 prompt = static_part + "\n\n[SYSTEM_PROMPT_DYNAMIC_BOUNDARY]\n\n" # 动态部分，不缓存 dynamic_part = assemble_dynamic_context(user_context, session_config) prompt += dynamic_part return prompt def add_uncached_constraint(prompt, constraint): """添加会破坏缓存的内容""" DANGEROUS_uncachedSystemPromptSection() # 显式警告 return prompt + f"\nCRITICAL: {constraint}" ``` ### 上下文组装的两层策略 公开文档中的项目说明、记忆和工具机制可以抽象为两层上下文组装策略： **第一层：系统上下文** * 收集 Git 仓库状态（当前分支、未提交改动、最近提交信息） * 提取仓库元数据（语言、框架、构建工具、依赖项） * 扫描项目结构和关键文件清单 这一层的结果在每个会话内进行 memoize，避免重复收集同样的静态元数据。 **第二层：用户上下文** * 发现并加载 CLAUDE.md 文件（见下文） * 加载记忆文件和日期上下文 * 整合用户提供的额外信息 ### CLAUDE.md 文件的优先级链发现机制 Claude Code 文档显示，CLAUDE.md 相关说明支持组织级、用户级、项目级和本地级别的上下文管理： 1. **托管策略（Managed Policy）**：组织统一管理的指令 2. **用户级说明（\~/.claude/CLAUDE.md）**：用户主目录下的个人默认上下文 3. **项目级说明**：项目根目录或 `.claude/` 目录下的 CLAUDE.md 4. **本地级说明（CLAUDE.local.md）**：不提交到版本库的本地上下文 加载时需要区分“策略优先级”和“拼接顺序”：官方文档说明 Claude Code 会沿目录树读取 CLAUDE.md / CLAUDE.local.md，并按从外层目录到当前工作目录的顺序拼接；同一目录中本地说明会追加在项目说明之后。也就是说，靠近启动目录的说明更晚进入上下文，但不应假设它们会像配置字典一样覆盖父级内容。 ```python # 伪代码：CLAUDE.md 发现优先级链 def discover_claude_md(start_path): priority_files = [ ("managed", get_org_mdm_memory()), # 最高优先级 ("user", Path.home() / ".claude" / "CLAUDE.md"), ("project", find_in_git_tree([ "CLAUDE.md", ".claude/CLAUDE.md", ".claude/rules/*.md" ])), ("local", Path.cwd() / "CLAUDE.local.md"), # 最低优先级 ] loaded_context = [] # 拼接顺序与官方文档一致：managed → user → project → local， # 即越靠近启动目录的说明越晚进入上下文（出现在更后面） for level, path in priority_files: if path.exists(): loaded_context.append(load_markdown(path)) return "\n\n".join(loaded_context) ``` ### @include 指令与可组合上下文 Claude Code 支持在 CLAUDE.md 文件中使用 `@path/to/import` 形式组合多个外部文件，支持相对路径、绝对路径和用户主目录路径： * `@path/to/file`：相对路径，相对于**包含该导入指令的文件**所在目录解析（而非项目根目录或工作目录） * `@./relative`：同上，显式写出当前目录前缀 * `@~/home`：相对于用户主目录 * `@/absolute`：绝对路径 该机制通过维护一个“已处理文件追踪集合”（processed-file tracking set）和最大包含深度来防止循环包含。当处理导入指令时，系统会： 1. 解析目标文件路径 2. 检查该文件是否已在当前包含链中处理过 3. 如果是，抛出循环依赖错误；否则加载并递归处理 4. 将处理结果合并到当前上下文中 这种设计允许团队构建可复用的上下文库（如通用的编码标准、API 规范等），各项目通过轻量级的 @include 语句组合使用。 ### 令牌预算的精确管理 生产级智能体可以采用精确的令牌预算跟踪。对于并行工具调用（如同一步骤内多个文件操作），系统记录每个工具调用的标识，并在接收结果时根据实际消费的令牌调整预算。 **自动压缩阈值**应按模型窗口、估算误差、工具结果波动和成本预算配置，并保留足够安全余量，确保即使估计略有偏差也不会超出窗口限制。 系统可以实现一个**断路器（circuit breaker）机制**：当连续压缩失败超过预算时（通常表明上下文中存在不可压缩的大型工具结果），系统停止重试，转而采用更激进的策略（如删除较早的工具结果）。这样做的目的是避免每次失败都再次发送完整上下文，造成成本意外放大。 ### 延迟工具加载（Deferred Tool Loading） 为了节省系统提示词中的令牌开销，Claude Code 文档描述了 MCP Tool Search 和 `alwaysLoad` 等机制：常用工具可以预先加载，不常用工具则通过搜索或按需展开减少上下文占用。 例如，代码分析工具可以在初始上下文中只保留短描述，当用户请求代码分析时再加载其完整的参数定义和用法示例。具体节省比例取决于工具数量、schema 大小和触发频率，需要通过线上 token 指标验证。 这类优化体现了现代智能体系统的一个重要方向：把能力、上下文成本和工具可见性作为同一个工程问题来管理，而不是把所有说明和工具 schema 一次性塞进提示词。