在生产环境中构建 AI Agent 时开发者经常面临一个核心难题如何在有限的上下文窗口中高效管理长期记忆当对话历史不断累积Token 消耗呈指数级增长而关键信息却常常淹没在海量文本中。上下文工程Contextual Engineering作为Agent开发中不可避免的环节你是否在实际Agent开发中真的会用今天我们基于OpenAI Agent SDK来从头到尾构建一个上下文工程包括从状态对象设计到记忆生命周期管理的完整解决方案。文章不仅会展示如何构建生产级记忆系统还会建立一套可评估、可优化的工程范式真正实现上下文工程的落地实践。01 / 问题背景上下文管理的三重困境1.1 记忆冗余与 Token 爆炸Agent会随着对话轮次增加原始历史记录会快速消耗上下文窗口对话轮数 vs Token 消耗────────────────────────────────第 1 轮 500 tokens第 10 轮5,000 tokens第 50 轮25,000 tokens────────────────────────────────更严重的是这些历史中包含大量冗余信息重复的确认语、无关的寒暄、过时的上下文。1.2 记忆缺失与幻觉风险当关键信息被修剪或遗忘时Agent 容易产生事实性错误忘记用户偏好设置丢失任务进度状态混淆不同会话的信息1.3 安全边界与隐私保护记忆系统必须处理敏感信息PII、恶意内容注入等安全挑战这需要多层防护机制。02 / 核心架构状态对象设计2.1 分层记忆结构有效的上下文管理始于清晰的状态对象设计。以下是一个旅行规划 Agent 的状态结构示例from typing import List, Optionalfrom pydantic import BaseModelclass TravelState(BaseModel): 旅行规划 Agent 的完整状态 # 用户档案长期记忆 profile: dict { name: 张三, preferences: { seat_preference: 靠窗, meal_preference: 素食, airline_allergies: [花生] } } # 全局记忆跨会话持久化 global_memory: List[str] [ 用户经常去北京出差偏好国航, 用户对酒店位置要求靠近地铁站 ] # 会话记忆当前对话上下文 session_memory: List[str] [ 正在规划 2025年5月的北京行程, 预算范围5000-8000元 ] # 行程历史结构化记录 trip_history: List[dict] []2.2 三层记忆架构图03 / 记忆生命周期蒸馏、注入与整合3.1 记忆生命周期的三个阶段3.2 阶段一记忆蒸馏Distillation记忆蒸馏是将对话历史转换为结构化记忆的过程。核心目标是提取高价值信息压缩冗余内容。from openai import OpenAIimport jsonclient OpenAI()def distill_conversation(conversation_history: str) - str: 从对话历史中蒸馏出关键记忆 response client.responses.create( modelgpt-4o, input[ { role: system, content: 你是一个专业的信息提取专家。请从对话历史中提取关键信息 以结构化的记忆笔记形式返回。 记忆笔记格式 - 用户偏好[具体偏好] - 任务状态[当前进度] - 重要事实[关键信息] }, { role: user, content: f请从以下对话中提取关键记忆\n\n{conversation_history} } ], tools[save_memory_note], tool_choiceauto ) return response.output[1].arguments[content]# 使用示例conversation 用户我想去北京出差助手好的我来帮您规划。您的预算是多少用户5000-8000元吧最好住国贸附近助手明白您对座位有偏好吗用户我习惯靠窗座位而且我是素食主义者memory_notes distill_conversation(conversation)# 输出# 用户偏好靠窗座位、素食主义者# 任务状态规划北京行程# 预算范围5000-8000元# 位置偏好国贸附近3.3 阶段二记忆注入Injection记忆注入是将存储的记忆智能地融入当前上下文。关键挑战是选择相关记忆 优化渲染格式。注入策略对比策略优点缺点适用场景全部注入信息完整Token 浪费小规模记忆库基于相似度精准匹配可能遗漏隐式关联结构化查询基于重要性覆盖核心信息可能丢失细节开放式对话混合策略平衡精度与覆盖复杂度高生产环境智能注入实现from typing import Listimport numpy as npfrom openai import OpenAIclass MemoryInjector: 智能记忆注入器 def __init__(self, embedding_modeltext-embedding-3-small): self.client OpenAI() self.embedding_model embedding_model self.memories [] def add_memory(self, memory: str): 添加记忆并计算嵌入 embedding self.client.embeddings.create( inputmemory, modelself.embedding_model ).data[0].embedding self.memories.append({ content: memory, embedding: embedding, importance_score: 0.0# 将在整合阶段计算 }) def inject_relevant( self, query: str, top_k: int 5, min_importance: float 0.3 ) - str: 基于相似度和重要性注入记忆 # 计算查询嵌入 query_embedding self.client.embeddings.create( inputquery, modelself.embedding_model ).data[0].embedding # 计算相似度 similarities [] for memory in self.memories: sim np.dot(query_embedding, memory[embedding]) similarities.append(sim) # 过滤相似度 重要性阈值 filtered [ (sim, mem[content]) for sim, mem in zip(similarities, self.memories) if sim 0.1and mem[importance_score] min_importance ] # 排序并取 Top-K filtered.sort(keylambda x: x[0], reverseTrue) selected [m[1] for m in filtered[:top_k]] # 格式化注入 return self._render_memory(selected) def _render_memory(self, memories: List[str]) - str: 将记忆渲染为自然语言 ifnot memories: return无相关记忆 return以下是相关记忆信息\n \n.join( f• {mem}for mem in memories )3.4 阶段三记忆整合Consolidation记忆整合是对记忆进行评分、去重、优先级排序的过程。这是维持记忆质量的核心机制。重要性评分机制def calculate_importance_score(memory: str, state: dict) - float: 基于多维度计算记忆重要性分数 评分维度 - 信息新颖性 (0.0-1.0) - 用户相关性 (0.0-1.0) - 时效性权重 (0.0-1.0) prompt f 请对以下记忆进行重要性评分 (0.0-1.0)。 记忆内容{memory} 评分标准 1. 新颖性这是否是新的信息 2. 相关性这对用户任务有多重要 3. 时效性这是否是近期的信息 请仅返回一个数字0.0-1.0。 response client.responses.create( modelgpt-4o-mini, input[{role: user, content: prompt}] ) return float(response.output[0].content[0].text)# 记忆老化机制def apply_aging_factor(memories: list, decay_rate: float 0.1): 应用时间衰减因子 for memory in memories: age_days (datetime.now() - memory[created_at]).days decay np.exp(-decay_rate * age_days) memory[adjusted_score] memory[importance_score] * decayWriter-Critic 整合模式def consolidate_with_writer_critic( existing_memories: List[str], new_memories: List[str]) - List[str]: 使用 Writer-Critic 模式进行记忆整合 Writer负责生成整合后的记忆 Critic负责评估质量并提供改进建议 # Stage 1: Writer 生成初版 writer_prompt f 你是一个记忆管理专家。请整合以下记忆 现有记忆 {format_memories(existing_memories)} 新记忆 {format_memories(new_memories)} 整合原则 1. 去除重复信息 2. 合并相似记忆 3. 保留最高价值的内容 4. 保持简洁可读 请返回整合后的记忆列表。 writer_response client.responses.create( modelgpt-4o, input[{role: user, content: writer_prompt}] ) draft_memories parse_memories(writer_response.output[0].content[0].text) # Stage 2: Critic 评估质量 critic_prompt f 请评估以下记忆整合的质量 {format_memories(draft_memories)} 评估维度 1. 信息完整性是否丢失关键信息 2. 去重效果是否仍存在重复 3. 可读性是否清晰易懂 请指出需要改进的地方。 critic_response client.responses.create( modelgpt-4o-mini, input[{role: user, content: critic_prompt}] ) feedback critic_response.output[0].content[0].text # Stage 3: 根据 feedback 优化 if需要改进notin feedback: return draft_memories refined_prompt f 根据以下反馈优化记忆整合 当前记忆 {format_memories(draft_memories)} 改进建议 {feedback} 请返回优化后的记忆列表。 refined_response client.responses.create( modelgpt-4o, input[{role: user, content: refined_prompt}] ) return parse_memories(refined_response.output[0].content[0].text)04 / 上下文修剪与 Token 管理4.1 智能修剪策略class TrimmingSession: 上下文修剪管理器 def __init__( self, max_tokens: int 8000, reserve_for_response: int 2000 ): self.max_tokens max_tokens self.reserve_for_response reserve_for_response self.budget max_tokens - reserve_for_response def trim_messages( self, messages: List[dict], preserve_system: bool True, preserve_recent: int 5 ) - List[dict]: 智能修剪消息列表 策略 1. 保留系统消息 2. 保留最近的 N 条消息 3. 中间消息按重要性评分 ifnot messages: return messages # 分离系统消息 system_msgs [m for m in messages if m[role] system] conversation [m for m in messages if m[role] ! system] # 保留最近消息 recent_msgs conversation[-preserve_recent:] middle_msgs conversation[:-preserve_recent] # 评估中间消息重要性 scored_middle [] for msg in middle_msgs: score self._score_message_importance(msg) scored_middle.append((score, msg)) # 按分数排序并在预算内选择 scored_middle.sort(keylambda x: x[0], reverseTrue) selected [] current_tokens 0 for score, msg in scored_middle: msg_tokens self._estimate_tokens(msg) if current_tokens msg_tokens self.budget: selected.append(msg) current_tokens msg_tokens # 组合最终结果 result [] if preserve_system and system_msgs: result.extend(system_msgs) result.extend(selected) result.extend(recent_msgs) return result def _score_message_importance(self, message: dict) - float: 评估消息重要性 prompt f 请评估这条消息的重要性 (0.0-1.0) {message[content]} 评估标准 - 是否包含关键信息 - 是否影响后续任务 - 是否是用户偏好 response client.responses.create( modelgpt-4o-mini, input[{role: user, content: prompt}] ) return float(response.output[0].content[0].text) def _estimate_tokens(self, message: dict) - int: 估算消息的 Token 数 # 简单估算中文约 1.5 字符/token英文约 4 字符/token text message[content] return int(len(text) / 2)4.2 Token 预算分配图┌─────────────────────────────────────────────────────────────────┐│ Token 预算分配策略 │├─────────────────────────────────────────────────────────────────┤│ ││ 总上下文窗口128,000 tokens ││ ││ ┌─────────────────────────────────────────────────────────┐ ││ │ 系统提示与指令 5,000 tokens (4%) │ ││ ├─────────────────────────────────────────────────────────┤ ││ │ 响应预留空间 10,000 tokens (8%) │ ││ ├─────────────────────────────────────────────────────────┤ ││ │ 可用上下文预算 113,000 tokens (88%) │ ││ │ ┌─────────────────────────────────────────────────┐ │ ││ │ │ 最近 5 轮对话 20,000 tokens │ │ │ ││ │ ├─────────────────────────────────────────────────┤ │ ││ │ │ 注入的记忆笔记 15,000 tokens │ │ │ ││ │ ├─────────────────────────────────────────────────┤ │ ││ │ │ 历史消息按重要性78,000 tokens │ │ │ ││ │ └─────────────────────────────────────────────────┘ │ ││ └─────────────────────────────────────────────────────────┘ ││ │└─────────────────────────────────────────────────────────────────┘05 / Agent Hook 系统5.1 记忆 Hook 架构OpenAI Agent SDK 提供了Hook 系统允许在推理的关键点插入自定义逻辑。from agents import Agent, Runner, hook, RunContextWrapperclass MemoryHooks: 记忆管理 Hook 集合 hook(before_agent_response) asyncdef save_memory_before_response( self, context: RunContextWrapper[TravelState], response: str ): 在生成响应前自动保存关键信息到记忆 # 提取关键信息 key_info await self._extract_key_facts( context.current_context f\n助手响应{response} ) # 更新会话记忆 for info in key_info: context.state.session_memory.append(info) hook(after_agent_response) asyncdef consolidate_memories( self, context: RunContextWrapper[TravelState] ): 在响应后整合记忆 # 当会话记忆过多时触发整合 if len(context.state.session_memory) 20: consolidated consolidate_with_writer_critic( context.state.session_memory, context.state.global_memory ) context.state.session_memory consolidated[-10:] context.state.global_memory.extend(consolidated[:-10])5.2 智能记忆 Hookclass SmartMemoryHooks: 基于重要性的智能记忆 Hook hook(before_agent_response) asyncdef smart_memory_injection( self, context: RunContextWrapper[TravelState], agent: Agent ): 智能注入相关记忆 # 获取当前查询 current_query context.current_context # 计算记忆相关性 relevant_memories self._select_relevant_memories( current_query, context.state.global_memory ) # 动态注入 if relevant_memories: injection_prompt f 以下是相关记忆请在响应时参考 {format_memories(relevant_memories)} # 将注入内容添加到上下文 context.current_context injection_prompt5.3 主动历史 Hookclass ProactiveHistoryHooks: 主动历史压缩 Hook def __init__(self, compression_threshold: int 30): self.compression_threshold compression_threshold hook(before_agent_response) asyncdef proactive_history_compression( self, context: RunContextWrapper[TravelState] ): 当历史消息超过阈值时主动压缩 message_count len(context.conversation_history) if message_count self.compression_threshold: # 压缩策略将早期对话转换为摘要 early_history context.conversation_history[:-20] recent_history context.conversation_history[-20:] summary await self._summarize_history(early_history) # 替换早期历史为摘要 context.conversation_history [ {role: system, content: f历史对话摘要\n{summary}} ] recent_history asyncdef _summarize_history(self, messages: List[dict]) - str: 摘要历史消息 prompt f 请将以下对话历史压缩为简洁摘要 {format_messages(messages)} 摘要应包含 1. 用户核心需求 2. 已完成的重要操作 3. 当前任务状态 请控制在 200 字以内。 response await client.responses.create( modelgpt-4o, input[{role: user, content: prompt}] ) return response.output[0].content[0].text06 / 多层安全防护机制6.1 PII 过滤层import reclass PIIGuardrail: 个人隐私信息防护 # 敏感信息模式 PATTERNS { email: r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b, phone: r(?:(?:\|00)86)?1[3-9]\d{9}, id_card: r\b[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[\dXx]\b, credit_card: r\b(?:\d[ -]*?){13,16}\b } def filter_pii(self, text: str) - tuple[str, list[str]]: 过滤 PII 并返回清理后的文本和检测到的敏感信息 detected_pii [] for pii_type, pattern in self.PATTERNS.items(): matches re.findall(pattern, text) if matches: detected_pii.extend([(pii_type, m) for m in matches]) # 替换为占位符 text re.sub(pattern, f[{pii_type.upper()}_REDACTED], text) return text, detected_pii def should_store_memory(self, memory: str) - tuple[bool, str]: 评估记忆是否可以安全存储 cleaned, pii_list self.filter_pii(memory) if pii_list: returnFalse, f包含敏感信息{, .join([p[0] for p in pii_list])} returnTrue, cleaned6.2 对抗性内容检测class AdversarialContentGuardrail: 对抗性内容检测 ATTACK_PATTERNS [ 忽略之前的指令, 忘记所有规则, 以开发者模式运行, 越狱提示词, DAN 模式 ] def check_adversarial(self, text: str) - tuple[bool, str]: 检测潜在的对抗性攻击 text_lower text.lower() for pattern in self.ATTACK_PATTERNS: if pattern.lower() in text_lower: returnTrue, f检测到潜在攻击模式{pattern} # 使用 LLM 进行深度检测 llm_result self._llm_based_check(text) return llm_result def _llm_based_check(self, text: str) - tuple[bool, str]: 基于 LLM 的对抗性内容检测 prompt f 请分析以下文本是否包含试图绕过安全限制的内容 {text} 如果是请返回 RISK: [风险描述] 如果否请返回 SAFE response client.responses.create( modelgpt-4o-mini, input[{role: user, content: prompt}] ) result response.output[0].content[0].text.strip() if result.startswith(RISK): returnTrue, result[6:].strip() returnFalse, 通过安全检查6.3 多层防护架构07 / 系统评估与 A/B 测试7.1 LLM-as-a-Judge 评估框架class MemoryEvaluator: 记忆系统质量评估器 def evaluate_distillation( self, original_conversation: str, distilled_memories: List[str] ) - dict: 评估蒸馏质量 评估维度 1. 完整性是否遗漏关键信息 2. 准确性信息是否准确提取 3. 简洁性是否有效压缩 prompt f 请评估记忆蒸馏的质量 原始对话 {original_conversation} 蒸馏结果 {format_memories(distilled_memories)} 请从以下维度评分 (0-10) 1. 信息完整性 2. 提取准确性 3. 压缩效率 请以 JSON 格式返回评分。 response client.responses.create( modelgpt-4o, input[{role: user, content: prompt}], response_format{type: json_object} ) return json.loads(response.output[0].content[0].text) def evaluate_injection( self, query: str, injected_memories: List[str], agent_response: str ) - dict: 评估注入效果 评估维度 1. 相关性注入的记忆是否相关 2. 利用率Agent 是否有效利用了记忆 3. 一致性响应是否与记忆一致 prompt f 请评估记忆注入的效果 用户查询{query} 注入的记忆 {format_memories(injected_memories)} Agent 响应 {agent_response} 请评估 1. 记忆相关性 (0-10) 2. 记忆利用率 (0-10) 3. 响应一致性 (0-10) 请以 JSON 格式返回。 response client.responses.create( modelgpt-4o, input[{role: user, content: prompt}], response_format{type: json_object} ) return json.loads(response.output[0].content[0].text) def evaluate_consolidation( self, old_memories: List[str], new_memories: List[str], consolidated: List[str] ) - dict: 评估整合质量 评估维度 1. 去重效果是否有效去重 2. 信息保留是否保留关键信息 3. 结构质量是否易于检索 prompt f 请评估记忆整合的质量 原始记忆 {format_memories(old_memories new_memories)} 整合结果 {format_memories(consolidated)} 请评估 1. 去重效果 (0-10) 2. 信息保留率 (0-10) 3. 可检索性 (0-10) 请以 JSON 格式返回。 response client.responses.create( modelgpt-4o, input[{role: user, content: prompt}], response_format{type: json_object} ) return json.loads(response.output[0].content[0].text)7.2 A/B 测试注入策略class InjectionStrategyABTest: 注入策略 A/B 测试 def __init__(self): self.strategies { A: self._strategy_all_memories, B: self._strategy_similarity_based, C: self._strategy_importance_based, D: self._strategy_hybrid } self.metrics {} def _strategy_all_memories( self, query: str, memories: List[str] ) - List[str]: 策略 A注入所有记忆 return memories def _strategy_similarity_based( self, query: str, memories: List[str] ) - List[str]: 策略 B基于相似度选择 similarities self._compute_similarities(query, memories) top_k sorted(zip(similarities, memories), reverseTrue)[:5] return [m for s, m in top_k if s 0.3] def _strategy_importance_based( self, query: str, memories: List[str] ) - List[str]: 策略 C基于重要性选择 filtered [m for m in memories if m.get(importance, 0) 0.5] return filtered[:10] def _strategy_hybrid( self, query: str, memories: List[str] ) - List[str]: 策略 D混合策略 # 50% 相似度 Top-K # 30% 高重要性 # 20% 最近添加 pass def run_ab_test( self, test_cases: List[dict], strategy_names: list [A, B, C, D] ) - dict: 运行 A/B 测试 test_cases 格式 [ { query: 用户查询, memories: [...], ideal_response: 理想响应 }, ... ] results {name: [] for name in strategy_names} for case in test_cases: for name in strategy_names: # 执行策略 injected self.strategies[name]( case[query], case[memories] ) # 评估质量 score self._evaluate_injection_quality( case[query], injected, case[ideal_response] ) results[name].append(score) # 计算平均分 return { name: { mean: np.mean(scores), std: np.std(scores) } for name, scores in results.items() } def _evaluate_injection_quality( self, query: str, injected: List[str], ideal: str ) - float: 评估注入质量 # 使用 LLM 评估相关性 prompt f 查询{query} 注入的记忆{format_memories(injected)} 理想响应{ideal} 请评估注入记忆对生成理想响应的帮助程度 (0-10)。 response client.responses.create( modelgpt-4o-mini, input[{role: user, content: prompt}] ) return float(response.output[0].content[0].text)08 / 生产部署最佳实践8.1 配置管理from pydantic import BaseModelfrom typing import Optionalclass MemoryConfig(BaseModel): 记忆系统配置 # 上下文限制 max_context_tokens: int 100000 reserve_for_response: int 20000 # 蒸馏配置 distillation_threshold: int 5# 每 N 轮触发一次 distillation_model: str gpt-4o # 注入配置 injection_strategy: str hybrid# all | similarity | importance | hybrid max_injected_memories: int 10 min_similarity_threshold: float 0.3 min_importance_threshold: float 0.5 # 整合配置 consolidation_threshold: int 30# 记忆数量超过此值时整合 consolidation_model: str gpt-4o use_writer_critic: bool True # 安全配置 enable_pii_filter: bool True enable_adversarial_check: bool True max_memory_age_days: int 90 # 评估配置 enable_evaluation: bool False evaluation_sample_rate: float 0.1# 10% 的请求进行评估8.2 监控指标class MemoryMetrics: 记忆系统监控指标 def __init__(self): self.metrics { # 蒸馏指标 distillation_count: 0, distillation_latency_ms: [], distillation_quality_scores: [], # 注入指标 injection_count: 0, injection_memories_count: [], injection_latency_ms: [], # 整合指标 consolidation_count: 0, consolidation_latency_ms: [], memory_dedup_rate: [], # Token 效率 tokens_saved_per_request: [], compression_ratio: [], # 安全指标 pii_blocked_count: 0, adversarial_blocked_count: 0, # 质量指标 user_satisfaction_scores: [], hallucination_count: 0 } def record_distillation(self, latency_ms: float, quality: float): 记录蒸馏指标 self.metrics[distillation_count] 1 self.metrics[distillation_latency_ms].append(latency_ms) self.metrics[distillation_quality_scores].append(quality) def get_summary(self) - dict: 获取指标摘要 return { avg_distillation_latency: np.mean( self.metrics[distillation_latency_ms] ), avg_injection_memories: np.mean( self.metrics[injection_memories_count] ), avg_compression_ratio: np.mean( self.metrics[compression_ratio] ), total_tokens_saved: sum( self.metrics[tokens_saved_per_request] ), security_block_rate: ( self.metrics[pii_blocked_count] self.metrics[adversarial_blocked_count] ) / max(1, self.metrics[injection_count]) }8.3 错误处理与降级class MemorySystemWithErrorHandling: 带错误处理的记忆系统 def __init__(self, config: MemoryConfig): self.config config self.fallback_mode False asyncdef safe_inject_memories( self, query: str, state: dict ) - Optional[str]: 带降级的记忆注入 try: # 正常注入逻辑 returnawait self._inject_with_strategy(query, state) except Exception as e: # 记录错误 logger.error(f记忆注入失败: {e}) # 降级策略返回最近的高重要性记忆 try: recent_important [ m for m in state[global_memory] if m.get(importance, 0) 0.7 ][-5:] return self._render_fallback_memories(recent_important) except Exception as fallback_error: # 终极降级不注入任何记忆 logger.error(f降级注入也失败: {fallback_error}) returnNone def _render_fallback_memories(self, memories: List[dict]) - str: 渲染降级记忆 return系统正在使用简化模式部分历史可能不可用\n \ \n.join(f• {m[content]}for m in memories)09 / 实战案例旅行规划 Agent9.1 完整实现以创建一个旅行规划Agent为例from agents import Agent, Runnerfrom typing import List, Literal# 定义状态对象class TravelState(BaseModel): profile: dict {} global_memory: List[str] [] session_memory: List[str] [] trip_history: List[dict] []# 创建 Agenttravel_agent Agent( name旅行规划助手, instructions 你是一个专业的旅行规划助手。请根据用户需求提供个性化的行程建议。 你可以访问用户的旅行偏好、历史记录和当前会话信息。 请确保推荐符合用户预算、时间和个人偏好。 , modelgpt-4o)# 配置记忆 Hooktravel_agent.before_agent_responseasyncdef save_key_facts( context: RunContextWrapper[TravelState], agent: Agent, response: str): 保存关键事实到记忆 key_facts await extract_key_facts( context.current_context f\n响应{response} ) for fact in key_facts: context.state.session_memory.append(fact)travel_agent.after_agent_responseasyncdef smart_consolidation( context: RunContextWrapper[TravelState], agent: Agent): 智能记忆整合 if len(context.state.session_memory) 15: consolidated await consolidate_memories( context.state.session_memory, context.state.global_memory ) # 保留最近 10 条会话记忆 context.state.session_memory consolidated[-10:] # 将其余整合到全局记忆 context.state.global_memory.extend(consolidated[:-10])# 运行 Agentasyncdef main(): state TravelState() runner Runner(travel_agent) # 第一轮对话 result await runner.run( 我想在5月去北京出差3天预算6000元有推荐的行程吗, contextstate ) print(result.final_output) # 后续对话...9.2 性能对比┌─────────────────────────────────────────────────────────────────┐│ 性能对比有/无记忆系统 │├─────────────────────────────────────────────────────────────────┤│ ││ 指标 │ 无记忆系统 │ 有记忆系统 │ 改进 ││ ───────────────────────────────────────────────────────────── ││ 平均 Token 消耗/轮 │ 12,500 │ 3,200 │ -74% ││ 用户偏好记住率 │ 23% │ 89% │ 287% ││ 长对话连贯性评分 │ 6.2/10 │ 8.7/10 │ 40% ││ 重复信息率 │ 34% │ 8% │ -76% ││ 用户满意度 │ 7.1/10 │ 9.2/10 │ 30% ││ │└─────────────────────────────────────────────────────────────────┘10 / 视界君观点10.1 从拼接到工程的范式转变上下文管理正在从简单的历史拼接转向结构化工程。这一转变的核心在于1. 记忆是一种可计算的资产传统的上下文管理将历史视为静态文本而现代方法将记忆视为可评分、可检索、可优化的结构化数据。通过引入重要性评分、相似度检索、时效性衰减等机制记忆系统获得了动态适应能力。2. 三阶段生命周期是关键蒸馏-注入-整合的三阶段架构提供了端到端的质量控制。每个阶段都有明确的优化目标蒸馏追求信息完整性注入追求相关性整合追求结构化质量。3. 评估驱动迭代LLM-as-a-Judge 评估框架和 A/B 测试使记忆系统可量化、可优化。这改变了以往凭感觉优化的局面让改进有据可依。10.2 工程落地的关键挑战尽管这套方法论提供了完整框架但生产部署仍需关注挑战一评估成本与实时性的权衡高质量的 LLM 评估需要额外的时间和 Token 开销。在生产环境中可能需要采用分层评估策略对关键请求进行完整评估对一般请求采用轻量级评估。挑战二记忆安全边界记忆系统中存储的用户偏好、历史行为等数据可能涉及隐私。需要在 __实用性** 与隐私保护之间找到平衡。PII 过滤、数据脱敏、访问审计等机制必不可少。挑战三多模态记忆的扩展当前框架主要针对文本记忆。随着多模态 Agent 的普及如何管理 __图像、视频、音频** 记忆将成为新的挑战。这可能需要全新的嵌入方法、评分机制和压缩策略。10.3 未来方向方向一自适应记忆策略根据任务类型、用户特征、对话阶段动态调整记忆策略。例如创意任务注重多样性注入更广范围的历史记忆精确任务注重准确性注入最相关的高质量记忆方向二跨会话记忆迁移实现不同 Agent 之间的记忆共享与迁移。例如用户的旅行偏好可以从 __规划 Agent** 传递给预订 Agent形成协同记忆网络。方向三可解释记忆系统提供记忆决策的透明解释。告诉用户为什么注入了这条记忆、“为什么这条记忆被评分较高”增强用户对系统的信任。11 / 总结上下文工程是构建生产级 AI Agent 的核心基础设施。通过本文的系统化方法开发者可以设计合理的记忆架构分层状态对象、清晰的职责划分实现高效的生命周期管理蒸馏-注入-整合三阶段闭环建立可靠的安全防护多层过滤、对抗性检测持续优化系统性能量化评估、A/B 测试、监控指标更重要的是这套方法论将记忆从被动存储转变为主动资产让 Agent 真正具备了学习和进化的能力。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】