前言做过 AI 对话应用的人都遇到过这个问题用户昨天说叫我小王今天问我是谁模型一脸茫然。这就是记忆系统要解决的核心问题。在 Deep Research 项目中我设计了一套双层记忆架构——短期记忆负责当前对话上下文长期记忆负责跨会话的用户画像与历史知识。本文完整记录设计思路与关键代码。记忆模型参考人脑记忆机制将 AI 记忆分为三层层级类比职责生命周期后端短期记忆工作记忆当前对话的上下文7天TTLRedis / PostgreSQL / 内存长期-语义记忆百科知识用户画像、偏好、事实永久PostgreSQL / SQLite Milvus长期-情节记忆经历记忆历史任务、对话记录永久PostgreSQL / SQLite Milvus架构图用户提问 │ ▼ MemoryManager.build_personalized_prompt_context() │ ├── 短期记忆最近N条消息 对话摘要 ├── 长期语义记忆Milvus向量检索 → 用户画像/偏好/事实 └── 长期情节记忆Milvus向量检索 → 相似历史任务 │ ▼ 拼接为 [跨会话记忆] 注入 System Prompt │ ▼ Agent 处理后persist_turn() 持久化本轮对话短期记忆存储结构支持三种后端根据环境自动选择def add_short_term_message(self, thread_id, message, user_id, tenant_id): payload {role: human|ai, content: ...} if self.short_term_backend redis: key fma:short:{tenant_id}:{user_id}:{thread_id} self._redis_client.rpush(key, json.dumps(payload)) self._redis_client.expire(key, self.short_term_ttl) elif self.short_term_backend postgres: # INSERT INTO short_term_messages (id, tenant_id, user_id, thread_id, role, content, created_at) ... else: # 内存 dict self.short_term.add_message(thread_id, message)摘要压缩消息数量超阈值时自动触发 LLM 摘要压缩def _compress_pg_thread(self, tenant_id, user_id, thread_id): history self._get_pg_short_term_messages(tenant_id, user_id, thread_id) if len(history) self.short_term_max_messages: return # 还没到阈值 # 保留最近的消息压缩旧消息 split_at len(history) - self.short_term_summary_threshold to_summarize history[:split_at] keep_messages history[split_at:] existing_summary self._get_pg_short_term_summary(...) new_summary self._summarize_text(existing_summary, to_summarize) # 原子替换删旧消息 保留新消息 更新摘要 DELETE ...; INSERT keep_messages; UPSERT summaryLLM 摘要 Prompt 设计def _summarize_text(self, existing_summary, history_slice): prompt ( 你是对话压缩引擎。请在保留事实、偏好、结论、待办和约束的前提下进行递归摘要。\n f已有摘要{existing_summary or 无}\n f新增历史\n{history_text}\n 输出要求100-300字中文结构紧凑。 ) response self._summary_llm.invoke([HumanMessage(contentprompt)]) return str(response.content).strip()关键点摘要不是一次性生成而是递归式——每次合并已有摘要 新增消息类似 git rebase。长期记忆Milvus 向量检索长期记忆的入口是向量检索。用户的提问 embedding 在 Milvus 中找最相似的历史记忆def _search_milvus(self, tenant_id, user_id, query, memory_type, limit5): docs self._milvus_store.similarity_search(query, kmax(limit * 4, 20)) entries [] for doc in docs: metadata doc.metadata or {} # 多租户过滤 if metadata.get(tenant_id) ! tenant_id: continue if metadata.get(user_id) ! user_id: continue # 类型过滤 if memory_type and metadata.get(memory_type) ! memory_type: continue entries.append(MemoryEntry(...)) if len(entries) limit: break return entries注意这里k limit * 4——从 Milvus 多取一些因为多租户和类型过滤会筛掉大量结果。如果只取 limit 个过滤后可能一条不剩。记忆保存def save_fact(self, user_id, fact, category, tenant_id): memory_id str(uuid4()) # 1. 写 PostgreSQL结构化的元数据 全文搜索能力 self._insert_memory_pg(entry, summaryfact[:500]) # 2. 写 Milvus向量语义搜索 self._index_memory_milvus( textfact, metadata{ tenant_id: tenant_id, user_id: user_id, memory_id: memory_id, memory_type: semantic, namespace: ffacts/{category}, }, ) return memory_idPostgreSQL 和 Milvus双写PG 支撑精确匹配和 ILIKE 模糊搜索Milvus 支撑语义相似搜索。任一挂了不影响另一路。自动记忆提取在persist_turn中检测用户是否触发了记住我意图def persist_turn(self, tenant_id, user_id, thread_id, query, answer): # 1. 保存短期记忆 self.add_short_term_messages(...) # 2. 检测记忆触发词 remember_markers [记住, 我叫, 我的偏好, remember, my name is, ...] if any(marker in query.lower() for marker in remember_markers): extracted extract_memory_from_messages([user_message]) # 提取 facts 和 preferences分别存储 for fact in extracted[facts]: self.save_fact(user_id, fact, categoryuser_fact) for pref in extracted[preferences]: self.save_user_profile(user_id, {preferences: [pref]}, mergeTrue)上下文注入每次用户提问前build_personalized_prompt_context组装记忆上下文def build_personalized_prompt_context(self, user_id, thread_id, query, max_memories8): context self.get_context_for_agent(user_id, thread_id, query, max_memories) sections [] if profile_text : context.get(user_profile): sections.append(f## 用户画像\n{profile_text}) if recent_text : context.get(recent_messages): sections.append(f## 最近对话\n{recent_text}) if summary_text : context.get(conversation_summary): sections.append(f## 对话摘要\n{summary_text}) if memory_text : context.get(memory_text): sections.append(memory_text) return \n\n.join(sections)注入到每个节点的 prompt 中def with_memory_context(state, user_prompt): memory_context state.get(memory_context, ).strip() if not memory_context: return user_prompt return f{user_prompt}\n\n[跨会话记忆]\n{memory_context}追踪与调试每次记忆注入后记录完整 trace方便排查为什么给我注入了这段记忆self._last_trace { query: query, memory_count: len(memory_entries), source_count: {milvus: 3, postgres: 1}, # 来自哪个检索源 items: [ {id: xxx, type: semantic, source: milvus, snippet: ...}, ], milvus_raw_hits: [...], # Milvus 原始命中含被过滤的 injected_chars: len(injected), }用户输入/memory-trace即可查看最近一次的注入详情。总结短期记忆消息累积 递归 LLM 摘要压缩多后端自动降级长期记忆PG Milvus 双写向量检索为主、全文搜索兜底自动提取触发词检测 LLM 信息抽取零手动标注可观测性每次注入记录完整 trace排错不靠猜这个架构的可扩展性不错——如果想接入对话推荐、情感分析、知识图谱只需要在新的 Agent 节点中读取memory_context即可。
AI Agent 双层记忆系统：从理论到落地
发布时间：2026/6/14 3:41:33
前言做过 AI 对话应用的人都遇到过这个问题用户昨天说叫我小王今天问我是谁模型一脸茫然。这就是记忆系统要解决的核心问题。在 Deep Research 项目中我设计了一套双层记忆架构——短期记忆负责当前对话上下文长期记忆负责跨会话的用户画像与历史知识。本文完整记录设计思路与关键代码。记忆模型参考人脑记忆机制将 AI 记忆分为三层层级类比职责生命周期后端短期记忆工作记忆当前对话的上下文7天TTLRedis / PostgreSQL / 内存长期-语义记忆百科知识用户画像、偏好、事实永久PostgreSQL / SQLite Milvus长期-情节记忆经历记忆历史任务、对话记录永久PostgreSQL / SQLite Milvus架构图用户提问 │ ▼ MemoryManager.build_personalized_prompt_context() │ ├── 短期记忆最近N条消息 对话摘要 ├── 长期语义记忆Milvus向量检索 → 用户画像/偏好/事实 └── 长期情节记忆Milvus向量检索 → 相似历史任务 │ ▼ 拼接为 [跨会话记忆] 注入 System Prompt │ ▼ Agent 处理后persist_turn() 持久化本轮对话短期记忆存储结构支持三种后端根据环境自动选择def add_short_term_message(self, thread_id, message, user_id, tenant_id): payload {role: human|ai, content: ...} if self.short_term_backend redis: key fma:short:{tenant_id}:{user_id}:{thread_id} self._redis_client.rpush(key, json.dumps(payload)) self._redis_client.expire(key, self.short_term_ttl) elif self.short_term_backend postgres: # INSERT INTO short_term_messages (id, tenant_id, user_id, thread_id, role, content, created_at) ... else: # 内存 dict self.short_term.add_message(thread_id, message)摘要压缩消息数量超阈值时自动触发 LLM 摘要压缩def _compress_pg_thread(self, tenant_id, user_id, thread_id): history self._get_pg_short_term_messages(tenant_id, user_id, thread_id) if len(history) self.short_term_max_messages: return # 还没到阈值 # 保留最近的消息压缩旧消息 split_at len(history) - self.short_term_summary_threshold to_summarize history[:split_at] keep_messages history[split_at:] existing_summary self._get_pg_short_term_summary(...) new_summary self._summarize_text(existing_summary, to_summarize) # 原子替换删旧消息 保留新消息 更新摘要 DELETE ...; INSERT keep_messages; UPSERT summaryLLM 摘要 Prompt 设计def _summarize_text(self, existing_summary, history_slice): prompt ( 你是对话压缩引擎。请在保留事实、偏好、结论、待办和约束的前提下进行递归摘要。\n f已有摘要{existing_summary or 无}\n f新增历史\n{history_text}\n 输出要求100-300字中文结构紧凑。 ) response self._summary_llm.invoke([HumanMessage(contentprompt)]) return str(response.content).strip()关键点摘要不是一次性生成而是递归式——每次合并已有摘要 新增消息类似 git rebase。长期记忆Milvus 向量检索长期记忆的入口是向量检索。用户的提问 embedding 在 Milvus 中找最相似的历史记忆def _search_milvus(self, tenant_id, user_id, query, memory_type, limit5): docs self._milvus_store.similarity_search(query, kmax(limit * 4, 20)) entries [] for doc in docs: metadata doc.metadata or {} # 多租户过滤 if metadata.get(tenant_id) ! tenant_id: continue if metadata.get(user_id) ! user_id: continue # 类型过滤 if memory_type and metadata.get(memory_type) ! memory_type: continue entries.append(MemoryEntry(...)) if len(entries) limit: break return entries注意这里k limit * 4——从 Milvus 多取一些因为多租户和类型过滤会筛掉大量结果。如果只取 limit 个过滤后可能一条不剩。记忆保存def save_fact(self, user_id, fact, category, tenant_id): memory_id str(uuid4()) # 1. 写 PostgreSQL结构化的元数据 全文搜索能力 self._insert_memory_pg(entry, summaryfact[:500]) # 2. 写 Milvus向量语义搜索 self._index_memory_milvus( textfact, metadata{ tenant_id: tenant_id, user_id: user_id, memory_id: memory_id, memory_type: semantic, namespace: ffacts/{category}, }, ) return memory_idPostgreSQL 和 Milvus双写PG 支撑精确匹配和 ILIKE 模糊搜索Milvus 支撑语义相似搜索。任一挂了不影响另一路。自动记忆提取在persist_turn中检测用户是否触发了记住我意图def persist_turn(self, tenant_id, user_id, thread_id, query, answer): # 1. 保存短期记忆 self.add_short_term_messages(...) # 2. 检测记忆触发词 remember_markers [记住, 我叫, 我的偏好, remember, my name is, ...] if any(marker in query.lower() for marker in remember_markers): extracted extract_memory_from_messages([user_message]) # 提取 facts 和 preferences分别存储 for fact in extracted[facts]: self.save_fact(user_id, fact, categoryuser_fact) for pref in extracted[preferences]: self.save_user_profile(user_id, {preferences: [pref]}, mergeTrue)上下文注入每次用户提问前build_personalized_prompt_context组装记忆上下文def build_personalized_prompt_context(self, user_id, thread_id, query, max_memories8): context self.get_context_for_agent(user_id, thread_id, query, max_memories) sections [] if profile_text : context.get(user_profile): sections.append(f## 用户画像\n{profile_text}) if recent_text : context.get(recent_messages): sections.append(f## 最近对话\n{recent_text}) if summary_text : context.get(conversation_summary): sections.append(f## 对话摘要\n{summary_text}) if memory_text : context.get(memory_text): sections.append(memory_text) return \n\n.join(sections)注入到每个节点的 prompt 中def with_memory_context(state, user_prompt): memory_context state.get(memory_context, ).strip() if not memory_context: return user_prompt return f{user_prompt}\n\n[跨会话记忆]\n{memory_context}追踪与调试每次记忆注入后记录完整 trace方便排查为什么给我注入了这段记忆self._last_trace { query: query, memory_count: len(memory_entries), source_count: {milvus: 3, postgres: 1}, # 来自哪个检索源 items: [ {id: xxx, type: semantic, source: milvus, snippet: ...}, ], milvus_raw_hits: [...], # Milvus 原始命中含被过滤的 injected_chars: len(injected), }用户输入/memory-trace即可查看最近一次的注入详情。总结短期记忆消息累积 递归 LLM 摘要压缩多后端自动降级长期记忆PG Milvus 双写向量检索为主、全文搜索兜底自动提取触发词检测 LLM 信息抽取零手动标注可观测性每次注入记录完整 trace排错不靠猜这个架构的可扩展性不错——如果想接入对话推荐、情感分析、知识图谱只需要在新的 Agent 节点中读取memory_context即可。
