在构建复杂AI智能体时你是否遇到过这样的困扰智能体在长对话中忘记关键信息或者将不同用户、不同会话的记忆混淆这背后是智能体“记忆系统”的缺失或设计不当。一个健壮的记忆系统是智能体从“单轮应答机”进化为“持续协作伙伴”的核心。本文将深入拆解Agent记忆系统的底层原理从零开始构建一个具备长短时记忆能力的智能体并提供可直接复用的完整工程案例彻底解决上下文丢失和记忆错乱两大痛点。本文适合对AI智能体开发有初步了解希望深入其内部机制并实现定制化记忆功能的开发者。通过本文你将掌握记忆系统的核心概念、主流架构、代码实现以及生产级的最佳实践。1. 智能体记忆系统从概念到价值1.1 什么是智能体记忆在AI智能体Agent的语境中记忆Memory并非指生物大脑的生理功能而是指智能体在运行过程中存储、检索和利用历史交互信息如对话、观察、行动结果的能力。它是智能体实现持续性、连贯性和个性化交互的基石。一个没有记忆的智能体每次交互都是孤立的。它无法记住用户的姓名、偏好、上一次对话的上下文也无法从历史错误中学习。这就像每次与一个患有“瞬时失忆症”的助手交谈体验极差。而一个拥有良好记忆系统的智能体则能像一位贴心的私人助理理解上下文提供连贯的服务。1.2 记忆系统的核心价值与挑战核心价值维持对话连贯性记住多轮对话的上下文使回答自然衔接。实现个性化服务记忆用户偏好、习惯和历史请求提供定制化响应。支持复杂任务分解在完成多步骤任务时记忆已完成步骤和中间结果。实现持续学习与优化从历史交互中总结经验调整自身行为策略。主要挑战上下文长度限制大语言模型LLM有固定的上下文窗口如4K、8K、128K tokens无法无限存储历史。记忆检索效率如何从海量历史信息中快速、准确地找到与当前问题最相关的记忆记忆的抽象与压缩如何将冗长的原始对话提炼成简洁、有用的知识片段记忆的隔离与安全如何确保不同用户、不同会话之间的记忆不会相互污染如何保护用户隐私1.3 长短时记忆仿生学的启发借鉴人类的记忆模型智能体的记忆系统通常也分为短期记忆和长期记忆。短期记忆Short-term Memory, STM也称为工作记忆或上下文记忆。它容量有限存储当前会话中最近、最活跃的信息。其核心作用是直接为LLM生成回答提供即时上下文。当对话超出模型窗口时最旧的短期记忆会被“遗忘”移出上下文。长期记忆Long-term Memory, LTM容量理论上无限用于持久化存储重要的、结构化的知识。这些记忆不会直接全部塞进上下文而是通过“检索”机制在需要时被动态地、有选择性地召回并注入短期记忆上下文中。一个高效的记忆系统关键在于实现短期记忆与长期记忆的协同工作短期记忆处理即时交互长期记忆存储持久知识并通过智能检索在两者之间建立桥梁。2. 环境准备与核心工具选型在开始代码实战前我们需要搭建开发环境并选择合适的技术栈。本教程将以Python为主要语言构建一个基于本地向量数据库的智能体记忆系统。2.1 基础环境操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。本文示例在 macOS/Linux 环境下演示。Python版本 3.8。推荐使用 3.9 或 3.10 以获得更好的兼容性。包管理工具pip或conda。代码编辑器/IDEVS Code, PyCharm 等任选。2.2 核心库与工具我们将使用以下开源库它们是目前构建AI智能体生态中的热门选择LangChain / LangGraph提供了构建智能体所需的大量组件和编排框架。我们将使用其Memory模块和Agent工具作为基础。但请注意本文重点在于理解原理和自建核心记忆模块。大语言模型LLM接入openai库用于调用GPT系列API或litellm用于统一接口调用多种模型。为简化演示我们可能使用模拟响应。向量数据库Vector Database用于实现长期记忆的存储和检索。我们将使用轻量级、无需服务器的ChromaDB。文本嵌入模型Embedding Model用于将文本转换为向量。我们将使用Sentence Transformers库中的开源模型如all-MiniLM-L6-v2它可以在本地运行。记忆数据结构使用Pydantic来定义结构化的记忆数据模型。2.3 项目初始化与依赖安装首先创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir agent-memory-system cd agent-memory-system # 创建虚拟环境 (以venv为例) python -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # venv\Scripts\activate # 创建requirements.txt文件并安装核心依赖 cat requirements.txt EOF langchain0.1.0 langchain-community0.0.10 chromadb0.4.22 sentence-transformers2.2.2 pydantic2.5.0 openai1.12.0 # 可选如需调用OpenAI API litellm1.30.2 # 可选用于多模型统一接口 EOF pip install -r requirements.txt安装完成后你的基础环境就准备好了。接下来我们将从原理出发逐步构建记忆系统的各个模块。3. 记忆系统核心原理与模块拆解一个完整的记忆系统通常包含以下几个核心模块我们将逐一实现。3.1 记忆的数据结构记忆单元Memory Entity记忆不是一堆杂乱无章的文本。我们需要为其设计一个结构化的数据模型。使用Pydantic可以方便地定义和验证。# file: memory_entities.py from datetime import datetime from typing import Optional, List, Dict, Any from pydantic import BaseModel, Field from enum import Enum class MemoryType(str, Enum): 记忆类型枚举 OBSERVATION observation # 观察用户输入、系统输出等 REFLECTION reflection # 反思智能体对事件的总结、学习 FACT fact # 事实用户提供的静态信息 PREFERENCE preference # 偏好用户喜好 PLAN plan # 计划任务分解步骤 class MemoryEntity(BaseModel): 记忆单元基础模型 id: str Field(default_factorylambda: str(datetime.utcnow().timestamp())) content: str # 记忆的文本内容 memory_type: MemoryType # 记忆类型 embedding: Optional[List[float]] None # 文本内容的向量表示 metadata: Dict[str, Any] Field(default_factorydict) # 元数据如来源、重要性、时间戳、会话ID、用户ID等 created_at: datetime Field(default_factorydatetime.utcnow) last_accessed_at: Optional[datetime] None # 最后访问时间用于记忆衰减或重要性计算 access_count: int 0 # 访问次数 class Config: arbitrary_types_allowed True def to_dict_for_storage(self): 转换为适合向量数据库存储的字典格式 data self.dict() # 确保embedding是列表 if self.embedding: data[embedding] self.embedding # 处理datetime对象转换为字符串 data[created_at] self.created_at.isoformat() if self.last_accessed_at: data[last_accessed_at] self.last_accessed_at.isoformat() return data这个MemoryEntity类定义了记忆的“原子结构”。metadata字段非常关键它可以存储session_id和user_id这是实现记忆隔离解决记忆错乱的基础。3.2 短期记忆管理滑动上下文窗口短期记忆的核心是管理一个固定大小的上下文窗口。当新对话加入时最旧的对话被移出。但简单的“先进先出”可能丢失重要信息因此需要一些策略。# file: short_term_memory.py from collections import deque from typing import List, Deque from memory_entities import MemoryEntity class ShortTermMemory: 短期记忆上下文窗口管理器 def __init__(self, max_tokens: int 2000, tokenizerNone): 初始化短期记忆。 :param max_tokens: 上下文窗口最大token数 :param tokenizer: 用于计算token数量的分词器如果为None则使用简单字数估算 self.max_tokens max_tokens self.tokenizer tokenizer self.memories: Deque[MemoryEntity] deque() self.current_tokens 0 def add(self, memory: MemoryEntity): 添加一条新记忆到短期记忆 memory_tokens self._estimate_tokens(memory.content) # 检查加入后是否会超出限制 while self.current_tokens memory_tokens self.max_tokens and self.memories: # 移除最旧的记忆 removed self.memories.popleft() removed_tokens self._estimate_tokens(removed.content) self.current_tokens - removed_tokens # 添加新记忆 self.memories.append(memory) self.current_tokens memory_tokens def get_context(self, as_string: bool True) - str: 获取当前短期记忆的上下文内容 if not self.memories: return if as_string: # 按时间顺序拼接记忆内容 context_parts [] for mem in self.memories: # 可以根据memory_type添加前缀如“用户说”、“系统回答” prefix self._get_prefix_for_type(mem.memory_type) context_parts.append(f{prefix}{mem.content}) return \n.join(context_parts) else: # 返回记忆对象列表 return list(self.memories) def _estimate_tokens(self, text: str) - int: 估算文本的token数量简化版 if self.tokenizer: return len(self.tokenizer.encode(text)) else: # 简单估算英文约1 token4字符中文约1 token2字符。这里取平均值。 # 这是一个非常粗略的估算生产环境应使用准确的分词器。 return len(text) // 3 def _get_prefix_for_type(self, mem_type): 根据记忆类型返回上下文中的前缀 prefixes { observation: Observation: , reflection: Reflection: , fact: Fact: , preference: User prefers: , plan: Plan: } return prefixes.get(mem_type.value if hasattr(mem_type, value) else mem_type, ) def clear(self): 清空短期记忆例如开始新会话时 self.memories.clear() self.current_tokens 0这个短期记忆管理器实现了基本的滑动窗口。在生产环境中_estimate_tokens函数需要替换为对应LLM的准确分词器如tiktokenfor OpenAI。3.3 长期记忆存储与检索向量数据库长期记忆的核心是将记忆的向量化表示存储到向量数据库中并支持基于语义相似度的检索。# file: long_term_memory.py import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer from typing import List, Optional, Dict, Any from memory_entities import MemoryEntity, MemoryType import uuid class LongTermMemory: 长期记忆管理器基于ChromaDB向量数据库 def __init__(self, persist_directory: str ./chroma_db, embedding_model_name: str all-MiniLM-L6-v2): 初始化长期记忆。 :param persist_directory: ChromaDB数据持久化目录 :param embedding_model_name: Sentence Transformers模型名称 self.embedding_model SentenceTransformer(embedding_model_name) self.client chromadb.PersistentClient(pathpersist_directory, settingsSettings(anonymized_telemetryFalse)) # 创建或获取集合collection。可以为不同用户/会话创建不同集合这里使用一个通用集合并靠metadata过滤。 self.collection self.client.get_or_create_collection(nameagent_memories) def _generate_embedding(self, text: str) - List[float]: 生成文本的向量嵌入 return self.embedding_model.encode(text).tolist() def store(self, memory: MemoryEntity): 存储一条记忆到长期记忆 # 生成向量嵌入 if not memory.embedding: memory.embedding self._generate_embedding(memory.content) # 准备存储数据 memory_id memory.id or str(uuid.uuid4()) embedding memory.embedding metadata memory.metadata.copy() # 确保metadata中的类型是字符串 metadata[memory_type] memory.memory_type.value if hasattr(memory.memory_type, value) else str(memory.memory_type) metadata[created_at] memory.created_at.isoformat() # 存储到ChromaDB self.collection.add( documents[memory.content], embeddings[embedding], metadatas[metadata], ids[memory_id] ) print(f[LTM] Stored memory: {memory.content[:50]}... (ID: {memory_id})) def retrieve(self, query: str, filter_conditions: Optional[Dict] None, top_k: int 5) - List[MemoryEntity]: 从长期记忆中检索与查询最相关的记忆。 :param query: 查询文本 :param filter_conditions: 过滤条件如 {user_id: alice, memory_type: fact} :param top_k: 返回最相关的K条记忆 :return: 记忆实体列表 # 生成查询的向量 query_embedding self._generate_embedding(query) # 执行查询 results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, wherefilter_conditions # ChromaDB支持通过metadata过滤 ) # 将查询结果转换回MemoryEntity对象 retrieved_memories [] if results[documents]: for i in range(len(results[documents][0])): doc results[documents][0][i] meta results[metadatas][0][i] mem_id results[ids][0][i] # 重建MemoryEntity对象注意embedding在查询结果中不返回这里设为None # 需要从metadata中解析出memory_type mem_type_str meta.get(memory_type, observation) try: mem_type MemoryType(mem_type_str) except ValueError: mem_type MemoryType.OBSERVATION # 默认类型 memory MemoryEntity( idmem_id, contentdoc, memory_typemem_type, embeddingNone, # 查询结果不包含原始向量 metadata{k: v for k, v in meta.items() if k not in [memory_type, created_at]}, # created_at需要从字符串解析此处简化处理 ) retrieved_memories.append(memory) return retrieved_memories def search_by_metadata(self, filter_conditions: Dict) - List[MemoryEntity]: 直接通过元数据过滤查询记忆非向量检索 # ChromaDB的get方法支持where过滤 results self.collection.get(wherefilter_conditions) memories [] for i in range(len(results[ids])): memory MemoryEntity( idresults[ids][i], contentresults[documents][i], memory_typeMemoryType(results[metadatas][i].get(memory_type, observation)), metadata{k: v for k, v in results[metadatas][i].items() if k ! memory_type}, ) memories.append(memory) return memories这个LongTermMemory类封装了向量数据库的存储和检索操作。关键点在于store方法将记忆内容向量化后存入数据库。retrieve方法根据查询文本的语义找到最相关的历史记忆。filter_conditions参数至关重要它允许我们根据user_id和session_id进行过滤确保用户A的记忆不会被用户B检索到从而解决“记忆错乱”。3.4 记忆的融合与上下文构建这是记忆系统的“大脑”负责协调短期记忆和长期记忆为LLM构建最终的提示词上下文。# file: memory_fusion.py from typing import List from short_term_memory import ShortTermMemory from long_term_memory import LongTermMemory from memory_entities import MemoryEntity class MemoryFusion: 记忆融合器整合短期和长期记忆构建最终上下文 def __init__(self, short_term_memory: ShortTermMemory, long_term_memory: LongTermMemory): self.stm short_term_memory self.ltm long_term_memory def build_context(self, current_query: str, user_id: str, session_id: str, max_context_tokens: int 3000) - str: 构建用于LLM生成的完整上下文。 策略 1. 获取当前的短期记忆最近的对话。 2. 从长期记忆中检索与当前查询和会话相关的记忆。 3. 将两者融合并确保不超过token限制。 # 1. 获取短期记忆上下文 stm_context self.stm.get_context(as_stringTrue) stm_tokens self.stm._estimate_tokens(stm_context) # 注意这里使用了内部方法实际应封装 # 2. 从长期记忆中检索相关记忆 # 构建过滤条件只检索当前用户和会话的记忆避免信息泄露 filter_conditions {user_id: user_id, session_id: session_id} retrieved_memories self.ltm.retrieve( querycurrent_query, filter_conditionsfilter_conditions, top_k5 # 检索最相关的5条长期记忆 ) # 将检索到的长期记忆转换为文本 ltm_context_parts [] for mem in retrieved_memories: # 标记为长期记忆并附带类型 prefix f[LTM - {mem.memory_type.value.upper()}] ltm_context_parts.append(f{prefix}{mem.content}) ltm_context \n.join(ltm_context_parts) ltm_tokens self.stm._estimate_tokens(ltm_context) if ltm_context else 0 # 3. 融合策略如果总token数超限优先保留短期记忆对长期记忆进行裁剪 total_tokens stm_tokens ltm_tokens available_for_ltm max_context_tokens - stm_tokens if total_tokens max_context_tokens and ltm_tokens 0: # 简化裁剪策略按记忆的重要性如访问次数、新鲜度排序后截断。 # 这里假设retrieved_memories已按相关性排序由向量数据库保证我们简单按顺序拼接直到token限额。 adjusted_ltm_parts [] current_ltm_tokens 0 for part in ltm_context_parts: part_tokens self.stm._estimate_tokens(part) if current_ltm_tokens part_tokens available_for_ltm: adjusted_ltm_parts.append(part) current_ltm_tokens part_tokens else: break ltm_context \n.join(adjusted_ltm_parts) # 4. 组装最终上下文 final_context_parts [] if ltm_context: final_context_parts.append( Relevant Long-term Memories ) final_context_parts.append(ltm_context) if stm_context: final_context_parts.append( Recent Conversation (Short-term) ) final_context_parts.append(stm_context) final_context \n\n.join(final_context_parts) print(f[MemoryFusion] Context built. STM tokens: {stm_tokens}, LTM tokens used: {self.stm._estimate_tokens(ltm_context)}) return final_context def reflect_and_store(self, conversation_summary: str, user_id: str, session_id: str): 反思与存储在一段对话结束后或达到某个触发条件时 将短期记忆中的重要信息总结、提炼并存储到长期记忆。 # 这是一个简化的示例直接将总结作为一条“反思”类型的记忆存储。 # 更高级的策略可以使用另一个LLM来分析和提取关键事实、偏好等。 reflection_memory MemoryEntity( contentconversation_summary, memory_typeMemoryType.REFLECTION, metadata{ user_id: user_id, session_id: session_id, source: auto_reflection } ) self.ltm.store(reflection_memory) print(f[MemoryFusion] Stored reflection to LTM: {conversation_summary[:80]}...)MemoryFusion类是系统的调度中心。build_context方法展示了核心的融合逻辑根据当前查询动态地从长期记忆中召回相关记忆并与短期记忆拼接形成送给LLM的最终提示词。reflect_and_store方法则模拟了“记忆固化”的过程将重要的短期记忆转化为长期记忆。4. 完整实战构建一个具备记忆的对话智能体现在我们将上述模块组合起来创建一个完整的、具备记忆功能的对话智能体。4.1 项目结构agent-memory-system/ ├── memory_entities.py # 记忆数据结构 ├── short_term_memory.py # 短期记忆管理 ├── long_term_memory.py # 长期记忆存储与检索 ├── memory_fusion.py # 记忆融合器 ├── agent_with_memory.py # 主智能体类 ├── requirements.txt # 依赖 └── main.py # 运行示例4.2 主智能体类实现这个智能体将模拟一个“个人图书推荐助手”它能记住用户喜欢的书籍类型和之前提过的书。# file: agent_with_memory.py import json from typing import Dict, Any, Optional from memory_entities import MemoryEntity, MemoryType from short_term_memory import ShortTermMemory from long_term_memory import LongTermMemory from memory_fusion import MemoryFusion class BookRecommendationAgent: 一个具备记忆功能的图书推荐智能体 def __init__(self, user_id: str default_user): self.user_id user_id self.session_id fsession_{id(self)} # 简单生成会话ID # 初始化记忆组件 self.stm ShortTermMemory(max_tokens1500) self.ltm LongTermMemory(persist_directoryf./chroma_db_{user_id}) self.memory_fusion MemoryFusion(self.stm, self.ltm) # 模拟的LLM实际项目中替换为真实的API调用 self.llm_provider self._mock_llm def process_query(self, user_input: str) - str: 处理用户输入生成有记忆的回复 # 1. 将用户输入作为观察存入短期记忆 user_memory MemoryEntity( contentuser_input, memory_typeMemoryType.OBSERVATION, metadata{ user_id: self.user_id, session_id: self.session_id, role: user } ) self.stm.add(user_memory) # 2. 构建融合了记忆的上下文 context self.memory_fusion.build_context( current_queryuser_input, user_idself.user_id, session_idself.session_id, max_context_tokens2500 ) # 3. 从用户输入中提取可能的事实或偏好存入长期记忆简单规则匹配 self._extract_and_store_facts(user_input) # 4. 准备LLM的提示词 system_prompt 你是一个专业的图书推荐助手拥有与用户对话的记忆。 请根据下面的“相关长期记忆”和“近期对话”来理解上下文并回应用户当前的问题。 你的回答应自然、连贯并充分利用已有的记忆信息。 full_prompt f{system_prompt}\n\n{context}\n\n用户最新问题{user_input}\n助手 # 5. 调用LLM生成回复 llm_response self.llm_provider(full_prompt) # 6. 将助手的回复也存入短期记忆 assistant_memory MemoryEntity( contentllm_response, memory_typeMemoryType.OBSERVATION, metadata{ user_id: self.user_id, session_id: self.session_id, role: assistant } ) self.stm.add(assistant_memory) # 7. 模拟每3轮对话后进行一次简单的反思总结 if len(self.stm.memories) % 6 0: # 每3轮用户输入因为记忆包含用户和助手 self._trigger_reflection() return llm_response def _extract_and_store_facts(self, user_input: str): 一个简单的规则用于从用户输入中提取事实如偏好并存入长期记忆 # 这是一个非常简化的示例。生产环境应使用更复杂的NLP或LLM来提取。 keywords_to_fact { 喜欢: preference, 讨厌: preference, 爱看: preference, 读过: fact, 作者是: fact, 类型是: fact } for keyword, mem_type in keywords_to_fact.items(): if keyword in user_input: # 简单地将整句作为事实存储 fact_memory MemoryEntity( contentuser_input, memory_typeMemoryType.FACT if mem_type fact else MemoryType.PREFERENCE, metadata{ user_id: self.user_id, session_id: self.session_id, extracted_keyword: keyword, source: rule_based_extraction } ) self.ltm.store(fact_memory) print(f[Agent] Extracted and stored a {mem_type}: {user_input[:60]}...) break # 只提取一个关键词 def _trigger_reflection(self): 触发反思总结近期对话并存入长期记忆 # 获取最近几轮对话作为总结材料 recent_conversation self.stm.get_context(as_stringTrue) # 简化直接取最后200个字符作为总结 summary recent_conversation[-200:] if len(recent_conversation) 200 else recent_conversation if summary: self.memory_fusion.reflect_and_store( conversation_summaryfConversation snippet: {summary}, user_idself.user_id, session_idself.session_id ) def _mock_llm(self, prompt: str) - str: 模拟LLM的响应。在实际应用中这里应调用OpenAI、通义千问等API。 # 这是一个非常简单的模拟仅用于演示流程。 # 它“看到”了prompt中的记忆上下文并做出相应回复。 if 科幻 in prompt and 推荐 in prompt: return 根据我们的对话历史您提到喜欢科幻小说。我为您推荐《三体》系列它是一部非常经典的科幻作品。 elif 历史 in prompt: return 您之前对历史类书籍感兴趣。推荐《万历十五年》它以独特视角解读明朝历史。 else: return 这是一个基于记忆的回复。我看到您之前讨论过一些书籍偏好。请问您今天想找什么类型的书呢 def get_memory_stats(self) - Dict[str, Any]: 获取当前记忆状态统计 return { short_term_memory_count: len(self.stm.memories), short_term_memory_tokens: self.stm.current_tokens, long_term_memory_collection_size: self.ltm.collection.count() }4.3 运行与演示创建一个主程序来演示智能体的记忆能力。# file: main.py from agent_with_memory import BookRecommendationAgent import time def main(): print( 启动具备记忆的图书推荐智能体 \n) # 为用户Alice创建一个智能体实例 agent_alice BookRecommendationAgent(user_idalice) # 模拟多轮对话 conversations [ 你好我喜欢读科幻小说。, 尤其是那些关于太空探索的。, 你能推荐一些吗, 我还对历史书有点兴趣。, 关于明朝历史的有什么好书, 对了我讨厌言情小说。 ] for i, user_input in enumerate(conversations, 1): print(f\n--- 第{i}轮对话 ---) print(f用户: {user_input}) response agent_alice.process_query(user_input) print(f助手: {response}) # 显示一些记忆状态 stats agent_alice.get_memory_stats() print(f[状态] 短期记忆条数: {stats[short_term_memory_count]}, 长期记忆条数: {stats[long_term_memory_collection_size]}) time.sleep(0.5) # 稍微延迟方便观察 print(\n 演示记忆检索 ) # 演示长期记忆的检索功能 print(\n1. 检索与‘科幻’相关的长期记忆) memories agent_alice.ltm.retrieve(query科幻, filter_conditions{user_id: alice}) for mem in memories: print(f - {mem.content} (类型: {mem.memory_type.value})) print(\n2. 检索与‘讨厌’相关的长期记忆) memories agent_alice.ltm.retrieve(query讨厌, filter_conditions{user_id: alice}) for mem in memories: print(f - {mem.content} (类型: {mem.memory_type.value})) print(\n 演示记忆隔离为新用户Bob创建智能体 ) agent_bob BookRecommendationAgent(user_idbob) response_bob agent_bob.process_query(我喜欢读武侠小说。) print(f用户Bob: 我喜欢读武侠小说。) print(f助手给Bob: {response_bob}) # 检查Bob能否看到Alice的记忆 print(\n尝试用Bob的智能体检索‘科幻’记忆应该为空) memories_bob agent_bob.ltm.retrieve(query科幻, filter_conditions{user_id: bob}) # 过滤条件为bob print(f 检索到 {len(memories_bob)} 条记忆。) print(\n 演示完成 ) if __name__ __main__: main()4.4 运行结果与说明运行python main.py你将看到类似以下的输出模拟LLM的回复可能略有不同 启动具备记忆的图书推荐智能体 --- 第1轮对话 --- 用户: 你好我喜欢读科幻小说。 [Agent] Extracted and stored a preference: 你好我喜欢读科幻小说。... 助手: 这是一个基于记忆的回复。我看到您之前讨论过一些书籍偏好。请问您今天想找什么类型的书呢 [状态] 短期记忆条数: 2, 长期记忆条数: 1 --- 第2轮对话 --- 用户: 尤其是那些关于太空探索的。 助手: 这是一个基于记忆的回复。我看到您之前讨论过一些书籍偏好。请问您今天想找什么类型的书呢 [状态] 短期记忆条数: 4, 长期记忆条数: 1 ... --- 第4轮对话 --- 用户: 我还对历史书有点兴趣。 [Agent] Extracted and stored a preference: 我还对历史书有点兴趣。... 助手: 您之前对历史类书籍感兴趣。推荐《万历十五年》它以独特视角解读明朝历史。 [状态] 短期记忆条数: 8, 长期记忆条数: 2 ... 演示记忆检索 1. 检索与‘科幻’相关的长期记忆 - 你好我喜欢读科幻小说。 (类型: preference) - 我还对历史书有点兴趣。 (类型: preference) 2. 检索与‘讨厌’相关的长期记忆 - 对了我讨厌言情小说。 (类型: preference) 演示记忆隔离为新用户Bob创建智能体 用户Bob: 我喜欢读武侠小说。 [Agent] Extracted and stored a preference: 我喜欢读武侠小说。... 助手给Bob: 这是一个基于记忆的回复。我看到您之前讨论过一些书籍偏好。请问您今天想找什么类型的书呢 尝试用Bob的智能体检索‘科幻’记忆应该为空 检索到 0 条记忆。 演示完成 结果分析记忆提取与存储智能体成功从对话中提取了“喜欢科幻”、“对历史感兴趣”、“讨厌言情”等偏好并存储为PREFERENCE类型的长期记忆。上下文感知回复在第4轮当用户提到历史书时助手的模拟回复引用了“您之前对历史类书籍感兴趣”这表明它利用了记忆上下文尽管我们的模拟LLM很简单。记忆检索演示显示可以根据语义如“科幻”、“讨厌”检索到相关的长期记忆。记忆隔离为用户Bob创建的新智能体其长期记忆数据库是独立的通过不同的user_id和持久化目录实现。Bob检索不到Alice的记忆有效解决了“记忆错乱”。5. 常见问题与排查思路在实际开发和应用中你可能会遇到以下问题问题现象常见原因解决思路记忆检索不准确1. 嵌入模型不适合领域。2. 检索时未正确设置过滤条件user_id。3. 检索返回数量top_k设置不当。1. 针对领域微调嵌入模型或选择更合适的预训练模型如text-embedding-ada-002。2. 检查retrieve方法的filter_conditions参数是否准确包含了用户/会话ID。3. 调整top_k值太小可能遗漏太大会引入噪声。上下文Token超限1. 短期记忆窗口max_tokens设置过大。2. 检索回的长期记忆过多未做裁剪。3. 单条记忆内容过长。1. 根据所用LLM的上下文窗口合理设置max_tokens预留部分空间给系统提示词和用户查询。2. 在MemoryFusion.build_context中实现更智能的裁剪策略如按重要性评分排序。3. 在存储长期记忆前对长文本进行分块或摘要。不同用户记忆混淆1. 存储记忆时未添加user_id到metadata。2. 检索时未传入filter_conditions。3. 所有用户共享同一个向量数据库集合Collection。1. 确保每条记忆在创建时都带有正确的user_id和session_id。2. 所有检索操作必须附带用户/会话过滤条件。3. 考虑为每个用户或租户创建独立的集合实现物理隔离。向量数据库性能慢1. 嵌入模型过大编码耗时。2. 向量索引未优化。3. 记忆条目过多。1. 权衡精度与速度选择更轻量的嵌入模型。2. ChromaDB默认使用HNSW索引可调整参数hnsw:space,hnsw:M。对于生产环境可评估Qdrant、Pinecone等专业向量库。3. 实施记忆清理策略定期归档或删除不重要的旧记忆。重要记忆被“遗忘”1. 短期记忆滑动窗口机制简单FIFO。2. 长期记忆检索相关性不高未被召回。3. 缺乏记忆重要性评估与固化机制。1. 实现更复杂的短期记忆管理如基于重要性得分保留关键记忆。2. 优化检索策略结合关键词检索与向量检索混合搜索。3. 实现“反思”机制定期由LLM分析对话将关键信息主动存储为长期记忆。6. 生产级最佳实践与进阶优化将上述Demo升级到生产环境需要考虑更多工程和架构问题。6.1 记忆的抽象与压缩原始对话记录非常冗长。直接存储和检索效率低下。摘要Summarization定期或基于事件使用LLM对一段对话进行总结将总结而非原文存入长期记忆。分块Chunking对于长文档或长回复将其分割成有重叠的语义块分别嵌入和存储以提高检索粒度。结构化提取Structured Extraction使用LLM配合Pydantic模型从对话中提取结构化的实体如人物、地点、事件、偏好和关系存入图数据库或关系型数据库实现更精确的查询。6.2 混合检索策略单一向量检索可能受限于嵌入模型的质量。关键词向量混合检索先使用BM25等关键词算法进行初步筛选再对结果进行向量重排序Re-ranking兼顾召回率与精确度。元数据过滤优先先通过user_id,session_id,memory_type等强过滤条件缩小范围再进行向量相似度计算大幅提升效率。6.3 记忆的重要性与衰减并非所有记忆都同等重要。重要性评分Importance Scoring设计一个评分函数考虑记忆的访问频率、最近访问时间、来源是用户明确陈述的事实还是系统的推断等。记忆衰减Memory Decay模拟人类遗忘曲线对长期记忆中长时间未被访问的、低重要性的记忆进行降权或归档。记忆主动触发除了根据当前查询被动检索系统可以主动在对话中注入重要的长期记忆例如“我记得您上周提到对Python异步编程感兴趣这里有一篇新文章……”。6.4 安全与隐私记忆系统存储了大量用户数据安全至关重要。数据加密对存储在向量数据库和磁盘上的记忆内容进行加密。记忆隔离如前所述必须在数据层实现严格的租户/用户隔离。记忆遗忘权提供API让用户查看、编辑和删除其个人记忆。匿名化处理在存储前对敏感个人信息如电话号码、地址进行匿名化处理。6.5 与现有框架集成我们的自建模块可以无缝集成到LangChain等框架中。封装为LangChain Memory继承BaseMemory类实现load_memory_variables和save_context方法即可在LangChain Chain或Agent中使用。在LangGraph中作为状态的一部分在LangGraph的持久化状态中将我们的ShortTermMemory和LongTermMemory管理器作为状态属性在图的节点间传递和更新。# 示例将自定义记忆封装为LangChain Memory概览 from langchain.memory import BaseMemory from typing import Dict, List, Any class CustomAgentMemory(BaseMemory): 自定义记忆类适配LangChain def __init__(self, user_id: str): super().__init__() self.agent BookRecommendationAgent(user_iduser_id) property def memory_variables(self) - List[str]: return [chat_history, relevant_memories] def load_memory_variables(self, inputs: Dict[str, Any]) - Dict[str, Any]: # 根据inputs中的查询构建上下文 query inputs.get(input, ) context self.agent.memory_fusion.build_context(...) return { chat_history: self.agent.stm.get_context(), relevant_memories: context } def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) - None: # 保存用户输入和助手输出到记忆 user_input inputs.get(input, ) assistant_output outputs.get(output, ) self.agent.process_query(user_input) # 简化处理 # ... 保存助手输出构建一个高效、可靠的Agent记忆系统是开发现实世界智能体应用的关键一步。本文从原理出发通过代码逐层实现了短期记忆管理、长期记忆的向量化存储与检索、以及两者的智能融合。我们不仅解决了上下文长度限制和记忆混淆的核心痛点还提供了一个可运行、可扩展的工程案例。要深化理解建议从以下几个方向继续探索替换更强的LLM和嵌入模型将模拟LLM替换为GPT-4、Claude或本地模型使用text-embedding-ada-002或bge-large-zh等更优的嵌入模型。实现更复杂的记忆策略如基于LLM的自动反思与摘要、记忆重要性动态评分。接入真实数据源让智能体记忆来自数据库、知识库或API的外部信息。性能优化与监控对记忆的存储、检索延迟进行监控建立索引优化和缓存机制。记忆系统是智能体迈向“智能”的桥梁。希望这篇保姆级教程能为你打下坚实的基础助你构建出更强大、更个性化的AI智能体应用。
