智能体记忆系统
记忆是智能体的基石。没有记忆，每次交互都从零开始。本技能涵盖智能体记忆的架构：短期记忆（上下文窗口）、长期记忆（向量存储），以及组织它们的认知架构。
关键洞察：记忆不仅仅是存储——更是检索。一百万个存储的事实，如果找不到正确的那一个，也毫无意义。分块、嵌入和检索策略决定了你的智能体是记住还是遗忘。
该领域术语不一且零散。我们使用 CoALA 认知架构框架：语义记忆（事实）、情景记忆（经历）和程序记忆（操作知识）。
原则
- 记忆质量 = 检索质量，而非存储数量
- 为检索分块，而非为存储分块
- 上下文隔离是记忆的敌人
- 为正确的信息选择正确的记忆类型
- 遗忘旧记忆——并非所有东西都应永久保存
- 在生产前测试检索准确性
- 后台记忆形成优于实时记忆
能力
- agent-memory
- long-term-memory
- short-term-memory
- working-memory
- episodic-memory
- semantic-memory
- procedural-memory
- memory-retrieval
- memory-formation
- memory-decay
范围
- vector-database-operations → data-engineer
- rag-pipeline-architecture → llm-architect
- embedding-model-selection → ml-engineer
- knowledge-graph-design → knowledge-engineer
工具
记忆框架
- LangMem (LangChain) - 适用场景：需要持久记忆的 LangGraph 智能体 注意：支持语义、情景、程序记忆类型
- MemGPT / Letta - 适用场景：虚拟上下文管理，类似操作系统的记忆 注意：分层记忆层级，自动换页
- Mem0 - 适用场景：用于个性化的用户记忆层 注意：专为用户偏好和历史设计
向量存储
- Pinecone - 适用场景：托管式、企业级规模（数十亿向量） 注意：最佳查询性能，成本最高
- Qdrant - 适用场景：复杂元数据过滤，开源 注意：基于 Rust，过滤能力出色
- Weaviate - 适用场景：混合搜索，知识图谱功能 注意：GraphQL 接口，适合关系型数据
- ChromaDB - 适用场景：原型开发，中小型应用 注意：开发者友好，在 10 万向量下 p50 约 20 毫秒
- pgvector - 适用场景：已使用 PostgreSQL，设置更简单 注意：适合小于 100 万向量，工具熟悉
嵌入模型
- OpenAI text-embedding-3-large - 适用场景：最佳质量，3072 维度 注意：$0.13/100 万 token
- OpenAI text-embedding-3-small - 适用场景：良好平衡，1536 维度 注意：$0.02/100 万 token，便宜 5 倍
- nomic-embed-text-v1.5 - 适用场景：开源，本地部署 注意：768 维度，质量不错
- all-MiniLM-L6-v2 - 适用场景：轻量级，快速本地嵌入 注意：384 维度，最低延迟
模式
记忆类型架构
为不同信息选择合适的记忆类型
适用场景：设计智能体记忆系统
记忆类型架构 (CoALA 框架)：
""" 三种记忆类型用于不同目的：
-
语义记忆：事实和知识
- 你对世界的了解
- 用户偏好、领域知识
- 存储在配置文件（结构化）或集合（非结构化）中
-
情景记忆：经历和事件
- 发生了什么（带时间戳的事件）
- 过去的对话、任务结果
- 用于从经验中学习
-
程序记忆：如何做事
- 规则、技能、工作流程
- 通常作为少样本示例实现
- “我以前是怎么解决这个的？” """
LangMem 实现
""" from langmem import MemoryStore from langgraph.graph import StateGraph
初始化记忆存储
memory = MemoryStore( connection_string=os.environ["POSTGRES_URL"] )
语义记忆：用户 profile
await memory.semantic.upsert( namespace="user_profile", key=user_id, content={ "name": "Alice", "preferences": ["dark mode", "concise responses"], "expertise_level": "developer", } )
情景记忆：过去交互
await memory.episodic.add( namespace="conversations", content={ "timestamp": datetime.now(), "summary": "Helped debug authentication issue", "outcome": "resolved", "key_insights": ["Token expiry was root cause"], }, metadata={"user_id": user_id, "topic": "debugging"} )
程序记忆：学到的模式
await memory.procedural.add( namespace="skills", content={ "task_type": "debug_auth", "steps": ["Check token expiry", "Verify refresh flow"], "example_interaction": few_shot_example, } ) """
运行时的记忆检索
""" async def prepare_context(user_id, query): # 获取用户 profile（语义） profile = await memory.semantic.get( namespace="user_profile", key=user_id )
# 查找相关过去经验（情景）
similar_experiences = await memory.episodic.search(
namespace="conversations",
query=query,
filter={"user_id": user_id},
limit=3
)
# 查找相关技能（程序）
relevant_skills = await memory.procedural.search(
namespace="skills",
query=query,
limit=2
)
return {
"profile": profile,
"past_experiences": similar_experiences,
"relevant_skills": relevant_skills,
}
"""
向量存储选择模式
为你的用例选择合适的向量数据库
适用场景：设置持久化记忆存储
向量存储选择：
""" 决策矩阵：
| Pinecone | Qdrant | Weaviate | ChromaDB | pgvector | |
|---|---|---|---|---|---|
| 规模 | 数十亿 | 1亿+ | 1亿+ | 100万 | 100万 |
| 托管 | 是 | 均可 | 均可 | 自托管 | 自托管 |
| 过滤 | 基础 | 最佳 | 良好 | 基础 | SQL |
| 混合搜索 | 否 | 是 | 最佳 | 否 | 是 |
| 成本 | 高 | 中 | 中 | 免费 | 免费 |
| 延迟 | 5ms | 7ms | 10ms | 20ms | 15ms |
| """ |
Pinecone（企业级规模）
""" from pinecone import Pinecone
pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("agent-memory")
使用元数据进行 Upsert
index.upsert( vectors=[ { "id": f"memory-{uuid4()}", "values": embedding, "metadata": { "user_id": user_id, "timestamp": datetime.now().isoformat(), "type": "episodic", "content": memory_text, } } ], namespace=namespace )
带过滤器的查询
results = index.query( vector=query_embedding, filter={"user_id": user_id, "type": "episodic"}, top_k=5, include_metadata=True ) """
Qdrant（复杂过滤）
""" from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, Filter, FieldCondition
client = QdrantClient(url="http://localhost:6333")
使用 Qdrant 进行复杂过滤
results = client.search( collection_name="agent_memory", query_vector=query_embedding, query_filter=Filter( must=[ FieldCondition(key="user_id", match={"value": user_id}), FieldCondition(key="type", match={"value": "semantic"}), ], should=[ FieldCondition(key="topic", match={"any": ["auth", "security"]}), ] ), limit=5 ) """
ChromaDB（原型开发）
""" import chromadb
client = chromadb.PersistentClient(path="./memory_db") collection = client.get_or_create_collection("agent_memory")
简单快速，适用于原型
collection.add( ids=[str(uuid4())], embeddings=[embedding], documents=[memory_text], metadatas=[{"user_id": user_id, "type": "episodic"}] )
results = collection.query( query_embeddings=[query_embedding], n_results=5, where={"user_id": user_id} ) """
分块策略模式
将文档分解为可检索的块
适用场景：处理文档以进行记忆存储
分块策略：
""" 分块困境：
- 太大：向量失去特异性
- 太小：丢失上下文
最佳块大小取决于：
- 文档类型（代码 vs 散文 vs 数据）
- 查询模式（事实性 vs 探索性）
- 嵌入模型（每个模型有其最佳点）
一般指导：大多数用例为 256-512 个 token """
固定大小分块（基线）
""" from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 字符 chunk_overlap=50, # 重叠可防止切断句子 separators=["\n\n", "\n", ". ", " ", ""] # 优先级顺序 )
chunks = splitter.split_text(document) """
语义分块（质量更好）
""" from langchain_experimental.text_splitter import SemanticChunker from langchain_openai import OpenAIEmbeddings
基于语义相似度分割
splitter = SemanticChunker( embeddings=OpenAIEmbeddings(), breakpoint_threshold_type="percentile", breakpoint_threshold_amount=95 )
chunks = splitter.split_text(document) """
结构感知分块（具有层级结构的文档）
""" from langchain.text_splitter import MarkdownHeaderTextSplitter
尊重文档结构
splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] )
chunks = splitter.split_text(markdown_doc)
每个块都有标题元数据用于上下文
"""
上下文分块（Anthropic 的方法）
"""
在嵌入前为每个块添加上下文
将检索失败率降低 35%
def add_context_to_chunk(chunk, document_summary): context_prompt = f''' 文档摘要：{document_summary}
以下是本文档的一个块：
{chunk}
'''
return context_prompt
嵌入上下文化后的块，而不是原始块
for chunk in chunks: contextualized = add_context_to_chunk(chunk, summary) embedding = embed(contextualized) store(chunk, embedding) # 存储原始块，嵌入上下文化的块 """
代码专用分块
""" from langchain.text_splitter import Language, RecursiveCharacterTextSplitter
语言感知分割
python_splitter = RecursiveCharacterTextSplitter.from_language( language=Language.PYTHON, chunk_size=1000, chunk_overlap=200 )
尊重函数/类边界
chunks = python_splitter.split_text(python_code) """
后台记忆形成
异步处理记忆以获得更好的质量
适用场景：希望在不减慢交互的情况下获得更高的召回率
后台记忆形成：
""" 实时记忆提取会减慢对话，并为智能体工具调用增加复杂性。对话结束后进行后台处理能产生更高质量的记忆。
模式：潜意识记忆形成 """
LangGraph 后台处理
""" from langgraph.graph import StateGraph from langgraph.checkpoint.postgres import PostgresSaver
async def background_memory_processor(thread_id: str): # 在对话结束或空闲时运行 conversation = await load_conversation(thread_id)
# 在没有时间压力的情况下提取见解
insights = await llm.invoke('''
分析此对话并提取：
1. 关于用户了解到的关键事实
2. 用户表露的偏好
3. 已完成或待完成的任务
4. 用户行为中的模式
请详尽——这将在后台运行。
对话：
{conversation}
''')
# 存储到长期记忆
for insight in insights:
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(insight),
content=insight,
metadata={"source_thread": thread_id}
)
在对话结束或空闲超时时触发
@on_conversation_idle(timeout_minutes=5) async def process_conversation(thread_id): await background_memory_processor(thread_id) """
记忆整合（类似睡眠）
"""
定期整合和去重记忆
async def consolidate_memories(user_id: str): # 获取用户的所有记忆 memories = await memory.semantic.list( namespace="user_insights", filter={"user_id": user_id} )
# 查找相似记忆（潜在重复）
clusters = cluster_by_similarity(memories, threshold=0.9)
# 合并相似记忆
for cluster in clusters:
if len(cluster) > 1:
merged = await llm.invoke(f'''
将这些相关记忆整合为一个：
{cluster}
保留所有重要信息。
''')
await memory.semantic.upsert(
namespace="user_insights",
key=generate_key(merged),
content=merged
)
# 删除原始记忆
for old in cluster:
await memory.semantic.delete(old.id)
"""
记忆衰减模式
遗忘旧的、不相关的记忆
适用场景：记忆增长变大，检索速度变慢
记忆衰减：
""" 并非所有记忆都应永久存在：
- 旧的偏好可能已过时
- 任务细节失去相关性
- 冲突的记忆会混淆检索
实现基于以下因素的智能衰减：
- 时效性（记忆何时创建/访问？）
- 频率（它被检索的频率？）
- 重要性（是核心事实还是细节？） """
基于时间的衰减
""" from datetime import datetime, timedelta
async def decay_old_memories(namespace: str, max_age_days: int): cutoff = datetime.now() - timedelta(days=max_age_days)
old_memories = await memory.episodic.list(
namespace=namespace,
filter={"last_accessed": {"$lt": cutoff.isoformat()}}
)
for mem in old_memories:
# 软删除（标记为已归档）
await memory.episodic.update(
id=mem.id,
metadata={"archived": True, "archived_at": datetime.now().isoformat()}
)
每周运行
@schedule(cron="0 0 * * 0") async def weekly_cleanup(): await decay_old_memories("conversations", max_age_days=90) await decay_old_memories("user_insights", max_age_days=365) """
基于频率的衰减
"""
低频记忆先被遗忘
def calculate_importance_score(memory): age_days = (datetime.now() - memory.created_at).days access_count = memory.metadata.get("access_count", 0) frequency = access_count / max(age_days, 1) return frequency
async def prune_by_importance(namespace: str, keep_ratio: float = 0.8): memories = await memory.semantic.list(namespace=namespace)
# 按重要性评分排序
memories.sort(key=calculate_importance_score)
# 删除最低重要性的 20%
delete_count = int(len(memories) * (1 - keep_ratio))
for mem in memories[:delete_count]:
await memory.semantic.delete(mem.id)
"""
常见陷阱
陷阱 1：上下文窗口作为记忆
上下文窗口不是记忆：
"""
问题：将会话历史作为唯一的记忆机制
后果：
- 上下文窗口大小受限（即使有 200K 也是如此）
- 每次对话都从零开始
- 无法跨会话共享学习
解决方案：使用长期记忆存储
- 提取关键信息
- 存储到向量数据库
- 在下次对话时检索 """
陷阱 2：忽视元数据过滤
元数据过滤至关重要：
"""
问题：仅使用向量相似度搜索，没有元数据过滤
后果：
- 来自不同用户或会话的混淆记忆
- 需要信任阈值
- 没有细粒度的检索控制
始终添加元数据：
- user_id（用户隔离）
- session_id（会话边界）
- memory_type（语义/情景/程序）
- created_at（时效性过滤）
- importance（优先级） """
陷阱 3：生产中使用错误的分块
为你的用例选择正确的分块：
"""
问题：对所有文档类型使用相同的分块
后果：代码文档中函数被截断，散文语境不完整
根据内容类型匹配分块：
- 代码：结构感知分割（按函数/类）
- 规则/格式：语义分割
- 问答：分辨率为每个问答对 """
陷阱 4：在用户交互期间提取记忆
不要在关键路径上更新记忆：
"""
问题：在对话中添加“保存记忆”工具调用
后果：
- 用户体验减慢
- 提取逻辑增加 token 成本
- 为智能体添加了不必要的工具调用
解决方案：后台处理
- 处理空闲会话
- 批量记忆更新
- 如果延迟，稍后仍会捕获 """
故障排除
检索正在返回不相关的结果
可能的原因：
"""
- 分块太大 → 减小块大小
- 嵌入模型错误 → 切换到 text-embedding-3-large
- 缺少元数据过滤 → 根据 user_id 或会话添加 where 子句
- 相似度阈值太低 → 增加到 0.7-0.8 """
记忆存储增长失控
可能的原因：
"""
- 没有衰减策略 → 实施基于时间或频率的衰减
- 去重效率低 → 运行记忆整合，如 LangGraph 示例
- 无意义数据收集 → 审核你的记忆形成提示以过滤噪音 """
跨会话记忆丢失
可能的原因：
"""
- 命名空间不一致 → 标准化命名空间（如 user_id + 会话类型）
- 刷新导致会话 ID 更改 → 存储并检索全局 user_id
- 向量存储索引删除 → 在部署前检查保留策略 """