ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

Agent 记忆系统设计:短时会话记忆与长时语义记忆的隔离

Agent 记忆系统设计:短时会话记忆与长时语义记忆的隔离 Agent 记忆系统设计短时会话记忆与长时语义记忆的隔离在自主智能体Autonomous Agents与复杂对话系统的设计中“记忆系统Memory Architecture”是决定 Agent 是否具备真正拟人化智能与长期持续进化能力的核心中枢。然而很多初学者在设计 Agent 记忆时常常写出极其粗暴的**“大杂烩记忆Monolithic Memory Dumping”**把用户在 3 个月前的历史提问、50 轮前的闲聊废话、以及当前这一轮的即时任务指令全部一股脑拼成一个长达 15,000 Token 的巨型 Prompt 喂给大模型。这种设计在生产环境中会瞬间引发毁灭性的**“记忆污染与注意力崩溃”**短时工作记忆被历史陈旧噪声淹没Agent 正在执行当前的代码重构任务却被 20 轮前讨论的“午饭吃什么”严重分散注意力导致工具调用参数频繁出现幻觉长时记忆无法跨会话泛化沉淀用户在 Session A 中明确强调过自己的编程偏好“我使用的是 Python 3.11所有的异步代码必须使用 asyncio.timeout”当开启全新的 Session B 时Agent 却瞬间“失忆”一切又必须从头交代上下文窗口与算力成本爆炸。认知心理学与现代 AI 架构给出的标准解法是“短时工作记忆Short-term Working Memory与长时语义记忆Long-term Semantic Memory的严格物理隔离与动态按需召回”。双层记忆系统的底层架构拓扑[ 用户当前新提问: 帮我为这个下单函数写一个带超时保护的异步装饰器 ] | v ------------------------- 阶段一: 长时语义记忆智能检索 (Long-term Semantic Recall) ------------------------- | 1. 从用户专属的向量记忆库 (Milvus / Qdrant) 中做相似度近邻检索 | | 2. 命中 2 条跨越几个月沉淀下的高价值用户画像与偏好记忆 (Long-term Memories): | | - 偏好 A: 用户技术栈环境为 Python 3.11要求超时控制使用现代 asyncio.timeout | | - 偏好 B: 用户所有业务函数必须记录 TraceID 审计日志 | ------------------------------------------------------------------------------------------------------------- | v ------------------------- 阶段二: 短时工作记忆滑动窗口精炼 (Short-term Working Memory) ----------------------- | 1. 从 Redis / PostgresSaver 中仅读取【当前会话 (Current Thread)】最近 3~5 轮即时对话上下文 (Sliding Window) | | 2. 严格丢弃 5 轮之前的过期临时细节仅保留与当前任务直接相关的输入输出与变量 | ------------------------------------------------------------------------------------------------------------- | v ------------------------- 阶段三: 结构化 Prompt 上下文清晰分层注入 ------------------------- | 【长期偏好记忆 (Long-term)】: 遵循 Python 3.11 asyncio.timeout 与 TraceID 规范 | | 【即时会话上下文 (Short-term)】: 最近 3 轮下单函数讨论代码 | | 【当前用户指令】: 帮我为这个下单函数写一个带超时保护的异步装饰器 | ------------------------------------------------------------------------------------------- | v [ 大模型生成 100% 契合用户长期习惯、且聚焦当前核心任务的完美代码! ]Python LangGraph 生产级双层记忆系统完整实现import asyncio from typing import TypedDict, List, Dict, Any, Optional from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 1. 结构化状态定义 class DualMemoryAgentState(TypedDict): user_id: str thread_id: str messages: List[BaseMessage] # 短时工作记忆 (由 LangGraph Checkpointer 管理) long_term_profile_memories: List[str] # 动态拉取的长时语义记忆 final_response: str class LongTermMemoryStore: 模拟基于向量数据库构建的长时语义记忆仓库 def __init__(self): # user_id - list of memories with embeddings self._user_memories: Dict[str, List[str]] { user_888: [ 【编码偏好】用户使用的是 Python 3.11超时控制坚决使用 asyncio.timeout 上下文管理器, 【架构约束】所有接口响应时间不得超过 500ms严格禁止使用同步 requests 库 ] } async def asearch_relevant_memories(self, user_id: str, current_query: str) - List[str]: 通过向量相似度检索与当前提问相关的长时记忆 await asyncio.sleep(0.01) # 模拟 10ms 向量检索 # 匹配到该用户的长期技术偏好 return self._user_memories.get(user_id, []) async def aadd_long_term_fact(self, user_id: str, new_fact: str): 当用户在对话中透露了长期偏好时沉淀入长时记忆库 if user_id not in self._user_memories: self._user_memories[user_id] [] self._user_memories[user_id].append(new_fact) print(f [长时记忆沉淀] 成功归档事实: {new_fact} 至用户 [{user_id}] 长期记忆库) # 实例化全局长时记忆库 long_term_memory_store LongTermMemoryStore() # 2. 节点逻辑 async def recall_long_term_memory_node(state: DualMemoryAgentState): 节点一检索长时语义记忆 user_id state[user_id] last_user_query state[messages][-1].content memories await long_term_memory_store.asearch_relevant_memories(user_id, last_user_query) print(f [长时记忆唤醒] 为用户 [{user_id}] 召回了 {len(memories)} 条长期语义偏好) return {long_term_profile_memories: memories} async def generate_response_node(state: DualMemoryAgentState): 节点二精炼组合两层记忆并生成答案 # 1. 提取长时记忆 long_term_text \n.join(state.get(long_term_profile_memories, [])) # 2. 提取短时记忆强制使用滑动窗口截取最近 4 条消息绝不让历史消息无限膨胀 short_term_messages state[messages][-4:] # 3. 分层组装 System Instruction system_prompt f你是一个智能编程架构助手。 【用户的长期偏好与硬性约束】 {long_term_text} 请严格遵守上述长期约束并结合最近的上下文回答用户问题。 full_payload [SystemMessage(contentsystem_prompt)] short_term_messages # 模拟大模型推理 mock_reply 已为您生成代码使用 async with asyncio.timeout(2.0): 严格遵循了您的 Python 3.11 长期编码规范。 print(f [Agent 输出]: {mock_reply}) return { messages: [AIMessage(contentmock_reply)], final_response: mock_reply } # 3. 编排 LangGraph 双层记忆状态图 def build_dual_memory_agent(): workflow StateGraph(DualMemoryAgentState) workflow.add_node(recall_memory, recall_long_term_memory_node) workflow.add_node(generate, generate_response_node) workflow.set_entry_point(recall_memory) workflow.add_edge(recall_memory, generate) workflow.add_edge(generate, END) # 短时记忆由内存 Checkpointer 托管 (绑定 thread_id) return workflow.compile(checkpointerMemorySaver())生产治理三大黄金军规短时记忆必须实施“滑动窗口Sliding Window或摘要淘汰Summarizer”messages列表随着轮次增加会快速耗尽 Token 窗口。每隔 10 轮对话必须触发一次后台轻量模型将历史对话压缩为一段 200 字的精炼摘要长时记忆必须具备“事实萃取器Fact Extractor”不是用户说的每一句话都要存入长时向量库。只有包含明确事实如“我叫李雷”、“我喜欢使用 TypeScript”、“我的服务器 IP 是 10.0.1.2”的句子才由独立 Agent 提取为结构化 Fact 存入长时库支持用户自主遗忘与隐私清除Right to be Forgotten长时语义记忆必须支持通过DELETE /v1/memories/{fact_id}接口进行物理删除保障合规与用户隐私。总结记忆的秩序是智能的基石。“短时工作记忆用 Checkpointer 滑动窗口管理即时交互长时语义记忆用向量数据库实现跨会话的精准偏好沉淀与动态唤醒”双层隔离的记忆架构让你的 Agent 既拥有洞悉全局的长期智慧又保持了面对即时任务时如手术刀般的极致敏锐。
返回列表