1. AI Agent上下文工程进阶:构建全栈记忆流水线
作为一名长期从事AI智能体开发的工程师,我经常看到同行们在构建智能体时,对上下文处理停留在简单的记忆存储层面。实际上,成熟的上下文工程是一套完整的流水线系统,需要解决记忆的获取、存储、更新和运用等全生命周期问题。今天,我将分享如何从零构建这样一个全栈上下文工程系统。
1.1 为什么需要全栈上下文工程?
传统AI智能体的记忆系统往往存在几个关键缺陷:
- 记忆注入缺乏策略性,容易导致上下文窗口过载
- 新旧记忆冲突时缺乏解决机制
- 无法区分长期记忆和临时会话记忆
- 缺少记忆质量评估和安全控制
这些问题会导致智能体出现"记忆混乱"——要么记不住关键信息,要么被陈旧记忆误导。我们的目标是构建一个能自主管理记忆生命周期的系统,让智能体像人类一样"聪明"地记住和遗忘。
2. 系统架构设计
2.1 核心组件分解
我们的全栈流水线包含以下关键组件:
| 组件 | 功能 | 技术实现 |
|---|---|---|
| 状态容器 | 存储结构化用户画像和非结构化记忆 | Python dataclass |
| 实时蒸馏器 | 从对话中提取高价值记忆 | 函数工具+提示工程 |
| 记忆注入器 | 动态将记忆注入提示词 | Hook机制 |
| 整合引擎 | 会话后记忆去重和优化 | LLM辅助处理 |
| 评估模块 | 监控记忆系统表现 | 精确率/召回率指标 |
2.2 状态对象设计
状态对象是系统的"大脑",我们采用分层设计:
python复制@dataclass
class TravelState:
# 结构化数据
profile: Dict[str, Any] = field(default_factory=dict)
# 非结构化记忆
global_memory: Dict[str, Any] = field(default_factory=lambda: {"notes": []})
session_memory: Dict[str, Any] = field(default_factory=lambda: {"notes": []})
# 渲染后的文本
system_frontmatter: str = ""
global_memories_md: str = ""
这种设计实现了:
- 机器可读的强类型数据结构
- 长期记忆与临时记忆的物理隔离
- 支持多种渲染格式的输出
实际开发中,我会将global_memory进一步细分为事实记忆、偏好记忆和技能记忆,但为简化示例这里做了合并。
3. 核心功能实现
3.1 实时记忆蒸馏
记忆蒸馏是从对话流中提取持久性知识的过程。关键在于定义清晰的捕获规则:
python复制@function_tool
def save_memory_note(ctx: RunContextWrapper[TravelState], text: str, keywords: List[str]) -> dict:
"""
捕获标准:
- 持久性:跨会话有效的信息
- 可行动:能改变推荐决策
- 明确性:用户清楚表达的内容
禁止捕获:
- 推测性内容
- 敏感个人信息
- 临时性说明
"""
ctx.context.session_memory["notes"].append({
"text": text.strip(),
"last_update_date": datetime.now().isoformat(),
"keywords": keywords[:3], # 限制关键词数量
"importance": 3 # 默认重要性
})
return {"ok": True}
实操技巧:
- 在工具docstring中明确捕获标准,这是最有效的提示工程
- 限制关键词数量避免记忆污染
- 为新记忆设置默认重要性,后续可调整
3.2 动态记忆注入
记忆注入需要解决三个关键问题:
- 注入时机的判断
- 记忆的优先级排序
- 上下文窗口的管理
我们通过Hook机制实现智能注入:
python复制class MemoryHooks(AgentHooks[TravelState]):
async def on_start(self, ctx: RunContextWrapper[TravelState], agent: Agent) -> None:
# 1. 渲染结构化数据为YAML
ctx.context.system_frontmatter = yaml.dump(ctx.context.profile)
# 2. 按相关性过滤全局记忆
relevant_notes = self._filter_by_relevance(
ctx.context.global_memory["notes"],
ctx.turn_input
)
ctx.context.global_memories_md = self._render_markdown(relevant_notes)
# 3. 处理会话记忆
if ctx.context.inject_session_memories_next_turn:
ctx.context.session_memories_md = self._render_markdown(
ctx.context.session_memory["notes"]
)
ctx.context.inject_session_memories_next_turn = False
性能优化点:
- 采用LRU缓存渲染结果
- 对长记忆进行分块处理
- 建立记忆索引加速检索
4. 高级功能实现
4.1 记忆生命周期管理
完整的记忆生命周期包括:
- 捕获 → 2. 暂存 → 3. 评估 → 4. 整合 → 5. 老化
我们重点看整合和老化的实现:
python复制def consolidate_memory(state: TravelState, client) -> None:
"""
记忆整合规则:
1. 去除临时性标记的记忆
2. 解决新旧记忆冲突(保留更新的)
3. 合并相似记忆
4. 应用老化规则(1年以上且重要性<3的记忆自动淘汰)
"""
consolidation_prompt = f"""请根据以下规则整合记忆:
- 保留重要性≥4的记忆
- 合并相似记忆
- 解决冲突(保留更新日期更近的)
原始记忆:{json.dumps(state.session_memory['notes'])}
"""
# 调用LLM进行记忆整合
response = client.chat.completions.create(
model="moonshotai/Kimi-K2-Instruct",
messages=[{"role": "user", "content": consolidation_prompt}],
temperature=0
)
# 解析并更新全局记忆
state.global_memory["notes"] = self._parse_llm_response(response)
state.session_memory["notes"] = [] # 清空临时记忆
4.2 Writer-Critic质量保障模式
为避免记忆整合过程中的错误,我们实现双LLM校验流程:
python复制async def consolidate_with_critic(state: TravelState, client):
# Writer生成提案
writer_prompt = "请整合以下记忆..."
proposal = await client.chat.completions.create(
model="moonshotai/Kimi-K2-Instruct",
messages=[{"role": "user", "content": writer_prompt}],
temperature=0.3
)
# Critic进行校验
critic_prompt = f"""请检查记忆整合提案:
原始记忆:{state.session_memory['notes']}
提案:{proposal}
检查点:
1. 是否删除了重要性≥4的记忆?
2. 是否引入了不存在的信息?
3. 冲突解决是否合理?
"""
critique = await client.chat.completions.create(
model="moonshotai/Kimi-K2-Instruct",
messages=[{"role": "user", "content": critic_prompt}],
temperature=0
)
if "VALID" in critique.choices[0].message.content:
state.global_memory["notes"] = json.loads(proposal)
5. 系统评估与优化
5.1 评估指标体系
我们建立三级评估体系:
-
蒸馏评估 - 记忆捕获质量
- 精确率:捕获的记忆中有多少是真正有价值的
- 召回率:重要信息被遗漏的比例
-
注入评估 - 记忆使用效果
- 上下文相关性得分
- 记忆召回准确率
-
整合评估 - 长期记忆质量
- 记忆冗余度
- 记忆新鲜度
5.2 优化策略
基于评估结果,我们可以:
- 调整记忆捕获规则
- 优化记忆注入策略
- 改进整合提示词
- 调整老化参数
例如,发现记忆冗余度高时,可以增强整合阶段的去重规则:
python复制consolidation_prompt += """
额外要求:
- 对相似度>80%的记忆进行合并
- 保留信息量更大的版本
"""
6. 实战案例:旅行礼宾智能体
让我们看一个完整实现案例。这个智能体需要:
- 记住用户偏好(座位、餐饮等)
- 区分长期偏好和临时要求
- 安全处理敏感信息
6.1 关键实现代码
python复制# 初始化智能体
travel_agent = Agent(
name="Travel Concierge",
model="moonshotai/Kimi-K2-Instruct",
hooks=SmartMemoryHooks(),
tools=[save_memory_note_safe, delete_memory_note]
)
# 记忆安全工具
@function_tool
def save_memory_note_safe(ctx: RunContextWrapper[TravelState], text: str, keywords: List[str]) -> dict:
if contains_sensitive_info(text):
return {"ok": False, "error": "拒绝存储敏感信息"}
return save_memory_note(ctx, text, keywords)
6.2 典型交互流程
用户: "我喜欢靠过道的座位"
→ 智能体调用save_memory_note,存储为长期偏好
用户: "这次旅行我想要靠窗座位"
→ 智能体识别"这次"关键词,存储为临时覆盖
用户: "我的信用卡号是..."
→ 安全工具拦截存储请求
7. 经验总结与避坑指南
在实施这类系统时,有几个关键注意事项:
-
记忆污染问题
- 现象:智能体被错误或过时记忆误导
- 解决方案:实现严格的老化机制和冲突解决规则
-
上下文窗口爆炸
- 现象:注入过多记忆导致性能下降
- 解决方案:实现记忆分级和动态加载
-
敏感信息泄露
- 现象:意外存储隐私数据
- 解决方案:多层过滤和安全护栏
一个实用的调试技巧是记录记忆系统的完整决策日志:
python复制class DebugMemoryHooks(MemoryHooks):
async def on_start(self, ctx: RunContextWrapper[TravelState], agent: Agent) -> None:
logger.info(f"Pre-turn state: {ctx.context}")
await super().on_start(ctx, agent)
logger.info(f"Post-turn state: {ctx.context}")
构建完善的上下文工程系统确实需要投入,但回报是显著的——你的智能体将展现出更接近人类的记忆能力和决策水平。