AI Agent 记忆系统全解：短期、长期与外部记忆的工程实践
为什么 AI Agent 需要记忆系统？
大语言模型（LLM）本质上是无状态的——每次推理都基于当前输入，不保留任何历史信息。这对于简单的问答场景足够，但对于需要多轮交互、个性化服务、持续学习的 AI Agent 而言，缺乏记忆就意味着每次对话都是"失忆"状态，无法积累经验、理解用户偏好，更无法执行跨会话的复杂任务。
现代 AI Agent 记忆系统借鉴了人类认知心理学中的记忆分类理论，通常分为以下几个层次：
感知记忆（Sensory Memory）：当前输入的原始信息，处理时间极短
短期/工作记忆（Short-term / Working Memory）：上下文窗口内的活跃信息
长期记忆（Long-term Memory）：持久化存储的知识与经验
情节记忆（Episodic Memory）：具体事件和交互历史
语义记忆（Semantic Memory）：抽象的知识和概念
理解这些层次，是设计高质量 Agent 记忆系统的第一步。
短期记忆：上下文窗口的工程管理
短期记忆对应 LLM 的上下文窗口（Context Window）。GPT-4 支持 128K tokens，Claude 3.5 支持 200K tokens，但更大的窗口并不意味着可以无限堆砌内容——超长上下文会导致"中间内容遗忘"（Lost in the Middle）问题，且推理成本呈线性增长。
工程上，管理短期记忆的核心策略有以下几种：
滑动窗口（Sliding Window）：保留最近 N 条消息，丢弃过早的历史
摘要压缩（Summarization）：定期将历史对话压缩为摘要，节省 token
Token Budget 控制：动态计算当前上下文 token 数，超阈值时触发压缩
以下是一个基于 LangChain 的滑动窗口 + 摘要混合记忆实现：
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 超过 2000 tokens 时自动压缩为摘要
memory = ConversationSummaryBufferMemory(
llm=llm,
max_token_limit=2000,
return_messages=True,
memory_key="chat_history"
)
# 添加对话历史
memory.save_context(
{"input": "我想学习 Python，从哪里开始？"},
{"output": "建议从官方文档和 Python Crash Course 入手，先掌握基础语法。"}
)
# 获取当前记忆（自动决定是否压缩）
history = memory.load_memory_variables({})
print(history["chat_history"])长期记忆：向量数据库驱动的语义检索
长期记忆的核心是将信息持久化并支持语义检索。与传统的关键词搜索不同，基于 Embedding 的向量检索能够理解语义相似性，让 Agent 找到"相关"而非仅仅"相同"的记忆片段。
主流的向量数据库方案对比：
Chroma：本地部署友好，适合原型开发，Python 原生支持
Pinecone：全托管云服务，生产级可用，支持元数据过滤
Weaviate：支持混合搜索（向量 + 关键词），开源可自托管
Qdrant：高性能 Rust 实现，支持复杂过滤条件
pgvector：PostgreSQL 扩展，适合已有 PG 基础设施的团队
以下是使用 Chroma + OpenAI Embeddings 构建 Agent 长期记忆的完整示例：
import chromadb
from openai import OpenAI
client = OpenAI()
chroma = chromadb.PersistentClient(path="./agent_memory")
collection = chroma.get_or_create_collection(
name="agent_memories",
metadata={"hnsw:space": "cosine"}
)
def store_memory(memory_id: str, content: str, metadata: dict = None):
"""存储记忆片段到向量数据库"""
response = client.embeddings.create(
model="text-embedding-3-small",
input=content
)
embedding = response.data[0].embedding
collection.upsert(
ids=[memory_id],
embeddings=[embedding],
documents=[content],
metadatas=[metadata or {}]
)
def recall_memory(query: str, n_results: int = 5, filter_meta: dict = None):
"""语义检索相关记忆"""
response = client.embeddings.create(
model="text-embedding-3-small",
input=query
)
query_embedding = response.data[0].embedding
results = collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
where=filter_meta
)
return results["documents"][0]
# 存储记忆
store_memory(
"mem_001",
"用户偏好使用 Python 3.11，不喜欢过度封装的框架",
{"user_id": "user_123", "type": "preference"}
)
# 检索记忆
memories = recall_memory(
"用户的技术栈偏好",
filter_meta={"user_id": "user_123"}
)
print(memories)情节记忆：结构化存储交互历史
情节记忆（Episodic Memory）记录 Agent 与用户的具体交互事件，包括时间戳、上下文、操作结果等结构化信息。它与语义记忆的区别在于：情节记忆关注"发生了什么"，语义记忆关注"知道什么"。
一个完整的情节记忆条目通常包含：
时间戳：事件发生时间，用于时序推理（"上次做了什么"）
会话 ID：关联同一对话的所有事件
动作类型：用户请求、Agent 响应、工具调用等
实体提取：从内容中提取的关键实体（人名、项目名、URL 等）
情感标记：用户满意度、交互质量评分
import json
from datetime import datetime
from dataclasses import dataclass, asdict
@dataclass
class Episode:
episode_id: str
session_id: str
timestamp: str
action_type: str # "user_input" | "agent_response" | "tool_call"
content: str
entities: list[str]
metadata: dict
class EpisodicMemoryStore:
def __init__(self, db_path: str):
import sqlite3
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS episodes (
episode_id TEXT PRIMARY KEY,
session_id TEXT,
timestamp TEXT,
action_type TEXT,
content TEXT,
entities TEXT,
metadata TEXT
)
""")
self.conn.execute(
"CREATE INDEX IF NOT EXISTS idx_session ON episodes(session_id)"
)
self.conn.commit()
def store(self, episode: Episode):
self.conn.execute(
"INSERT OR REPLACE INTO episodes VALUES (?,?,?,?,?,?,?)",
(
episode.episode_id,
episode.session_id,
episode.timestamp,
episode.action_type,
episode.content,
json.dumps(episode.entities, ensure_ascii=False),
json.dumps(episode.metadata, ensure_ascii=False)
)
)
self.conn.commit()
def get_session_history(self, session_id: str) -> list[Episode]:
cursor = self.conn.execute(
"SELECT * FROM episodes WHERE session_id=? ORDER BY timestamp",
(session_id,)
)
rows = cursor.fetchall()
return [Episode(
r[0], r[1], r[2], r[3], r[4],
json.loads(r[5]), json.loads(r[6])
) for r in rows]记忆融合：构建统一的 Agent 记忆接口
在实际 Agent 系统中，需要将短期、长期、情节记忆统一封装，提供一致的 API 给 Agent 推理层使用。记忆融合的核心设计原则是：
透明性：Agent 无需关心记忆存储的底层实现
相关性排序：综合时间衰减、语义相关度、重要性评分来排序召回结果
写入策略：哪些信息值得记忆？（重要性过滤）
class UnifiedMemorySystem:
def __init__(self, user_id: str):
self.user_id = user_id
self.working_memory = [] # 短期上下文
self.long_term = VectorMemoryStore() # 向量长期记忆
self.episodic = EpisodicMemoryStore("episodes.db")
def add_to_working(self, role: str, content: str):
"""添加到工作记忆（短期）"""
self.working_memory.append({"role": role, "content": content})
# 超过阈值时触发压缩
if self._count_tokens() > 3000:
self._compress_working_memory()
def recall(self, query: str, top_k: int = 3) -> str:
"""综合召回：从长期记忆检索相关内容"""
memories = self.long_term.search(query, user_id=self.user_id, k=top_k)
if not memories:
return ""
return "相关记忆：\n" + "\n".join(f"- {m}" for m in memories)
def consolidate(self, importance_threshold: float = 0.7):
"""记忆巩固：将重要的工作记忆写入长期存储"""
for msg in self.working_memory:
score = self._rate_importance(msg["content"])
if score >= importance_threshold:
self.long_term.store(
content=msg["content"],
metadata={"user_id": self.user_id, "score": score}
)
def get_context_for_llm(self, current_query: str) -> list[dict]:
"""构建发送给 LLM 的完整上下文"""
recalled = self.recall(current_query)
messages = []
if recalled:
messages.append({
"role": "system",
"content": f"以下是你对该用户的历史记忆：\n{recalled}"
})
messages.extend(self.working_memory[-10:]) # 最近10条
return messages生产级记忆系统的最佳实践
将记忆系统从原型推向生产，需要关注以下几个工程质量维度：
隐私与安全：用户记忆必须严格隔离，避免跨用户数据泄露；敏感信息需脱敏存储
记忆遗忘机制：实现 GDPR 合规的"被遗忘权"，支持用户级别的完整删除
记忆冲突解决：用户偏好更新时，新信息应覆盖旧信息（upsert 而非 insert）
召回质量评估：通过 A/B 测试持续优化 Embedding 模型和检索参数
冷启动问题：新用户无历史记忆时，使用群体偏好作为初始化数据
推荐的技术栈组合（2025年）：
Embedding 模型：
text-embedding-3-small（成本低）或text-embedding-3-large（精度高）向量数据库：小规模用 Chroma，中大规模用 Qdrant 或 Weaviate
关系型存储：PostgreSQL + pgvector（统一向量和结构化数据）
记忆框架：mem0（专为 AI Agent 设计的记忆层）
AI Agent 的记忆系统仍然是快速发展的领域，随着 LLM 上下文窗口持续扩大和 Embedding 技术的进步，架构选型也在不断演进。但核心设计思想——合理分层、按需召回、持续巩固——将在相当长的时间内保持有效。
发布评论
热门评论区：