AI Agent 上下文管理实战:从基础到高级的完整指南
Karpathy 和 Shopify CEO Tobi 都在 2025 年开始强调"上下文工程"这个概念——这比 Prompt 工程更根本。本文将为你揭示 AI Agent 上下文管理的核心秘密。
一、上下文管理的本质:LLM 的无状态性
1.1 核心洞察:输入决定输出
LLM 本质上是一个无状态函数——给定输入,产生输出。这意味着:
# LLM 的数学本质output = f(input)# 其中 input = 上下文窗口# output = 下一个 token 的概率分布关键结论:要获得最好的输出,就要给它最好的输入。而最好的输入,就是精心设计的上下文。
1.2 上下文包含什么?
完整的上下文窗口包括:
系统指令 - Agent 的角色、目标、行为规范 用户输入 - 当前请求或问题 历史对话 - 之前的对话轮次 工具调用结果 - 之前执行的结果 外部数据 - RAG 检索的相关信息 结构化指令 - 输出格式要求 错误信息 - 之前的执行错误 状态信息 - 当前执行状态
二、标准做法 vs 自定义上下文
2.1 标准消息格式的局限性
大多数 LLM 客户端使用标准消息格式:
messages = [ {"role": "system", "content": "你是一个助手..."}, {"role": "user", "content": "你好"}, {"role": "assistant", "content": "你好!"}, {"role": "tool", "content": "工具调用结果..."}]问题:
Token 效率低:每个消息都有角色标签 信息密度低:结构化数据需要大量描述 灵活性差:难以自定义格式
2.2 自定义上下文格式的优势
更高效的做法是构建自己的上下文格式,用 XML 标签把所有信息打包进一条 user 消息:
context = """<slack_message> From: @alex Channel: #deployments Text: 能否部署后端 v1.2.3 到生产环境?</slack_message><list_git_tags> intent: "list_git_tags"</list_git_tags><list_git_tags_result> tags: - name: "v1.2.3" commit: "abc123" date: "2024-03-15T10:00:00Z"</list_git_tags_result>下一步应该做什么?"""优势:
信息密度高:用最少的 token 传递最多的信息 结构清晰:LLM 更容易理解结构化数据 可扩展:轻松添加新的信息类型 可过滤:方便移除或压缩不必要的信息
三、上下文管理的四个层次
3.1 第一层:基础上下文管理
classBasicContextManager:"""基础上下文管理器:简单的消息队列"""def__init__(self, max_tokens=4000):self.max_tokens = max_tokensself.messages = []self.token_count = 0defadd_message(self, role: str, content: str):"""添加消息到上下文""" message = {"role": role, "content": content}self.messages.append(message)self.token_count += self._estimate_tokens(content)# 如果超出限制,移除最旧的消息whileself.token_count > self.max_tokens andlen(self.messages) > 1: removed = self.messages.pop(0)self.token_count -= self._estimate_tokens(removed["content"])defget_context(self) -> List[Dict]:"""获取当前上下文"""returnself.messages.copy()3.2 第二层:智能上下文压缩
classSmartContextManager:"""智能上下文管理器:自动压缩和优化"""def__init__(self, compression_ratio=0.3):self.compression_ratio = compression_ratioself.full_history = [] # 完整历史self.compressed_context = [] # 压缩后的上下文defcompress_context(self, current_query: str) -> str:""" 智能压缩上下文,保留关键信息 """# 1. 提取与当前查询最相关的历史 relevant_history = self._extract_relevant_history(current_query)# 2. 压缩冗余内容 compressed = self._compress_redundant_content(relevant_history)# 3. 结构化表示 structured = self._structure_context(compressed, current_query)return structureddef_extract_relevant_history(self, query: str) -> List[str]:"""提取与查询最相关的历史对话"""# 使用语义相似度或关键词匹配 scored_history = []for item inself.full_history: score = self._calculate_relevance(item, query) scored_history.append((score, item))# 按相关性排序并选择前N个 scored_history.sort(reverse=True) selected_count = int(len(scored_history) * self.compression_ratio)return [item for _, item in scored_history[:selected_count]]3.3 第三层:分层上下文架构
classHierarchicalContextManager:"""分层上下文管理器:短期/长期/工作记忆"""def__init__(self):self.short_term = [] # 短期记忆:当前会话self.long_term = [] # 长期记忆:重要信息self.working = [] # 工作记忆:临时缓存defupdate_context(self, new_info: Dict, importance: float):""" 根据重要性更新不同层次的记忆 """# 短期记忆:总是添加self.short_term.append(new_info)# 如果重要性高,添加到长期记忆if importance > 0.7:self._add_to_long_term(new_info)# 工作记忆:临时存储中间结果self.working.append(new_info)# 定期清理self._cleanup_memory()def_add_to_long_term(self, info: Dict):"""添加到长期记忆,避免重复"""# 检查是否已存在类似信息for existing inself.long_term:ifself._is_similar(info, existing):# 合并或更新现有信息self._merge_info(existing, info)return# 添加新信息self.long_term.append(info)3.4 第四层:动态上下文优化
classDynamicContextOptimizer:"""动态上下文优化器:基于反馈自动调整"""def__init__(self):self.optimization_history = []self.current_strategy = "balanced"defoptimize_context(self, context: str, feedback: Dict) -> str:""" 基于反馈优化上下文 """# 分析反馈 analysis = self._analyze_feedback(feedback)# 选择优化策略 strategy = self._select_strategy(analysis)# 应用优化 optimized = self._apply_strategy(context, strategy)# 记录优化历史self.optimization_history.append({"timestamp": time.time(),"strategy": strategy,"feedback": feedback,"improvement": analysis.get("improvement", 0) })return optimizeddef_select_strategy(self, analysis: Dict) -> str:"""根据分析结果选择优化策略"""if analysis.get("token_inefficient", False):return"compress"elif analysis.get("information_sparse", False):return"enrich"elif analysis.get("structure_poor", False):return"structure"else:return"balanced"四、上下文工程的核心技术
4.1 Token 效率优化
策略一:信息压缩算法
defcompress_information(text: str, target_ratio: float = 0.5) -> str:""" 压缩文本信息,保留核心含义 """# 1. 提取关键句子 sentences = text.split('. ') key_sentences = extract_key_sentences(sentences, target_ratio)# 2. 移除冗余词汇 compressed = remove_redundant_words(' '.join(key_sentences))# 3. 使用缩写和简写 abbreviated = use_abbreviations(compressed)return abbreviateddefextract_key_sentences(sentences: List[str], ratio: float) -> List[str]:"""提取关键句子"""# 基于 TF-IDF 或语义重要性评分 scored = [(sentence_importance(s), s) for s in sentences] scored.sort(reverse=True) keep_count = max(1, int(len(sentences) * ratio))return [s for _, s in scored[:keep_count]]策略二:结构化数据表示
defstructure_data_for_llm(data: Dict) -> str:""" 将数据转换为 LLM 友好的结构化格式 """# 原始 JSON(低效)# {"name": "John", "age": 30, "city": "New York"}# 优化格式(高效) structured = f"""<person>name: Johnage: 30city: New York</person>"""return structured4.2 错误信息处理
关键原则:错误信息需要被 LLM 理解,但不能让 Agent"原地打转"。
defformat_error_for_context(error: Exception, attempt: int) -> str:""" 格式化错误信息,便于 LLM 理解和修复 """if attempt == 1:# 第一次错误:详细描述returnf"""<error>type: {type(error).__name__}message: {str(error)}suggestion: 请检查参数是否正确</error>"""elif attempt == 2:# 第二次错误:简化描述returnf"""<error_retry>hint: 参数格式可能有问题</error_retry>"""else:# 第三次错误:最小化信息return"<need_human_help/>"4.3 上下文窗口的动态调整
classDynamicContextWindow:"""动态上下文窗口:根据任务复杂度调整"""def__init__(self, base_size=4000):self.base_size = base_sizeself.current_size = base_sizeself.adjustment_history = []defadjust_for_task(self, task_complexity: float, available_tokens: int) -> int:""" 根据任务复杂度调整上下文窗口大小 """if task_complexity < 0.3:# 简单任务:小窗口 new_size = int(self.base_size * 0.5)elif task_complexity < 0.7:# 中等任务:标准窗口 new_size = self.base_sizeelse:# 复杂任务:大窗口 new_size = min(self.base_size * 2, available_tokens)# 记录调整self.adjustment_history.append({"timestamp": time.time(),"complexity": task_complexity,"old_size": self.current_size,"new_size": new_size })self.current_size = new_sizereturn new_size五、实战案例:客服 AI Agent 的上下文管理
5.1 问题场景
某电商客服 AI Agent 需要处理:
用户咨询历史(可能长达 30 轮对话) 订单信息、物流状态、产品详情 公司政策、促销活动 之前的处理记录和解决方案
5.2 解决方案设计
classCustomerServiceContextManager:"""客服 AI Agent 专用上下文管理器"""def__init__(self, user_id: str):self.user_id = user_idself.conversation_history = self._load_history(user_id)self.user_profile = self._load_profile(user_id)self.order_info = self._load_orders(user_id)self.policy_info = self._load_policies()defbuild_context(self, user_query: str) -> str:"""构建客服对话的完整上下文""" context_parts = []# 1. 系统指令 context_parts.append(self._build_system_prompt())# 2. 用户画像摘要 context_parts.append(self._summarize_user_profile())# 3. 相关订单信息 context_parts.append(self._extract_relevant_orders(user_query))# 4. 最近对话摘要(非完整历史) context_parts.append(self._summarize_recent_chat(limit=5))# 5. 相关政策摘要 context_parts.append(self._extract_relevant_policies(user_query))# 6. 当前查询 context_parts.append(f"<current_query>{user_query}</current_query>")# 7. 响应格式要求 context_parts.append(self._response_format_instruction())return"\n\n".join(context_parts)def_summarize_recent_chat(self, limit: int = 5) -> str:"""摘要最近对话,而非完整历史""" recent = self.conversation_history[-limit:] ifself.conversation_history else []ifnot recent:return"<chat_history>这是第一次对话</chat_history>"# 提取关键信息摘要 summary = []for msg in recent:if msg["role"] == "user": summary.append(f"用户: {self._summarize_text(msg['content'])}")else: summary.append(f"客服: {self._summarize_text(msg['content'])}")returnf"<chat_history>\n" + "\n".join(summary) + "\n</chat_history>"5.3 优化效果对比
六、高级技巧:上下文工程的黑科技
6.1 上下文窗口的"热区域"技术
classHotZoneContextManager:"""热区域上下文管理器:重要信息放在最佳位置"""def__init__(self):# LLM 对上下文不同位置的关注度不同# 通常:开头 > 结尾 > 中间self.hot_zones = {"beginning": 0.1, # 前10% - 最高关注度"end": 0.1, # 后10% - 高关注度"middle": 0.8# 中间80% - 标准关注度 }defplace_important_info(self, context: List[str], important_items: List[str]) -> List[str]:"""将重要信息放在热区域""" total_len = len(context) beginning_idx = int(total_len * self.hot_zones["beginning"]) end_idx = int(total_len * (1 - self.hot_zones["end"]))# 在开头插入最重要的信息for item in important_items[:2]: context.insert(beginning_idx, item) beginning_idx += 1# 在结尾插入次重要信息for item in important_items[2:4]: context.insert(end_idx, item) end_idx += 1# 其余放在中间for item in important_items[4:]: middle_idx = (beginning_idx + end_idx) // 2 context.insert(middle_idx, item)return context6.2 上下文信息的"衰减"机制
classDecayingContextManager:"""衰减上下文管理器:旧信息逐渐降低权重"""def__init__(self, decay_rate=0.1):self.decay_rate = decay_rate # 每轮衰减10%self.context_items = [] # (content, weight, timestamp)defadd_context(self, content: str, importance: float = 1.0):"""添加上下文信息"""self.context_items.append({"content": content,"weight": importance,"timestamp": time.time() })defdecay_weights(self):"""衰减旧信息的权重""" current_time = time.time()for item inself.context_items:# 计算时间衰减 age_hours = (current_time - item["timestamp"]) / 3600 time_decay = math.exp(-self.decay_rate * age_hours)# 应用衰减 item["weight"] *= time_decay# 如果权重太低,标记为可移除if item["weight"] < 0.1: item["removable"] = Truedefget_effective_context(self, min_weight=0.3) -> str:"""获取有效上下文(权重高于阈值)"""self.decay_weights() effective_items = [ item for item inself.context_itemsif item["weight"] >= min_weight ]# 按权重排序 effective_items.sort(key=lambda x: x["weight"], reverse=True)# 构建上下文 context_parts = []for item in effective_items:# 根据权重调整表示方式if item["weight"] > 0.8: context_parts.append(f"【重要】{item['content']}")elif item["weight"] > 0.5: context_parts.append(item['content'])else: context_parts.append(f"(参考){item['content']}")return"\n".join(context_parts)6.3 上下文信息的"关联图"技术
classGraphBasedContextManager:"""基于图的上下文管理器:维护信息关联关系"""def__init__(self):self.nodes = {} # 信息节点self.edges = {} # 关联关系self.node_counter = 0defadd_information(self, info: str, related_to: List[int] = None):"""添加信息节点""" node_id = self.node_counterself.node_counter += 1self.nodes[node_id] = {"id": node_id,"content": info,"importance": 1.0,"timestamp": time.time(),"access_count": 0 }# 建立关联关系if related_to:for related_id in related_to:if related_id inself.nodes:self._add_edge(node_id, related_id)return node_iddef_add_edge(self, node1: int, node2: int):"""添加关联边""" key = (min(node1, node2), max(node1, node2))if key notinself.edges:self.edges[key] = {"strength": 1.0}else:self.edges[key]["strength"] += 0.1defget_relevant_context(self, query: str, max_nodes: int = 10) -> str:"""获取与查询相关的上下文"""# 找到与查询最相关的节点 query_node_id = self.add_information(f"查询: {query}")# 计算相关性 relevant_nodes = self._find_relevant_nodes(query_node_id, max_nodes)# 构建上下文 context_parts = []for node_id in relevant_nodes: node = self.nodes[node_id]# 更新访问计数 node["access_count"] += 1# 根据重要性格式化if node["importance"] > 0.8: prefix = "★ "elif node["importance"] > 0.5: prefix = "• "else: prefix = " " context_parts.append(f"{prefix}{node['content']}")return"\n".join(context_parts)七、最佳实践总结
7.1 上下文管理的黄金法则
信息密度优先:用最少的 token 传递最多的信息 结构化表示:LLM 更容易理解结构化数据 动态调整:根据任务复杂度调整上下文窗口 错误友好:格式化错误信息便于 LLM 理解 定期清理:移除过时或低价值信息
7.2 实施路线图
阶段一:基础实现
实现基本的上下文队列管理 添加简单的 token 计数和截断 使用标准消息格式
阶段二:智能优化
实现上下文压缩算法 添加信息重要性评分 引入结构化数据表示
阶段三:高级功能
实现分层记忆系统 添加动态调整机制 引入关联图管理
阶段四:生产就绪
添加监控和日志 实现性能优化 建立自动化测试
7.3 未来展望
随着 LLM 技术的不断发展,上下文管理也将面临新的挑战和机遇:
超长上下文:如何处理 100K+ token 的上下文窗口 多模态上下文:文本、图像、音频的融合管理 实时上下文:流式数据的实时处理和压缩 联邦上下文:跨设备、跨用户的上下文共享 隐私保护上下文:在保护隐私的前提下管理敏感信息
结语
上下文工程是 AI Agent 开发中最被低估但最关键的技术之一。正如 Karpathy 所说:"上下文工程 > Prompt 工程"。
掌握上下文管理,意味着你能够:
显著提升性能:减少 token 使用,加快响应速度 大幅提高准确率:给 LLM 提供最相关的信息 有效控制成本:优化 API 调用费用 构建更智能的 Agent:支持更复杂的任务和对话
记住,最好的上下文管理是让用户感受不到它的存在,却享受到极致的体验。
夜雨聆风