返回列表构建AI Agent的记忆系统：从短时记忆到长期记忆的架构实践
本文系统探讨AI Agent的记忆系统设计，从短时记忆的上下文管理，到中时的向量检索，再到长期记忆的持久化存储，详解三层架构的实现思路与代码示例，并讨论记忆的触发写入、检索激活与衰减策略。
构建AI Agent的记忆系统：从短时记忆到长期记忆的架构实践
当我们谈论AI Agent的能力时，往往关注它的规划、工具调用和多步推理能力，却忽略了一个最基础却至关重要的维度——记忆。
没有记忆的Agent，就像一个每次见面都重新认识你的朋友。它能帮你完成今天的任务，却无法记住你昨天的偏好、上周的决策、或上个月讨论的方向。在一次性的对话中，这或许可以接受；但在真实的业务场景里，这种"金鱼记忆"几乎毫无用处。
本文将从架构层面系统探讨AI Agent的记忆系统设计，涵盖短时记忆、长期记忆、记忆检索与衰减等核心组件，并结合代码示例说明实现思路。
一、为什么AI Agent需要记忆系统
大语言模型本身是无状态的——每一次API调用都是独立的，模型不会自动保留上下文之外的信息。即便在单个会话中，通过将对话历史作为输入上下文传递，也只能利用有限的上下文窗口（通常是128K到1M token不等）。
当Agent需要：
- 跨会话持久化：记住用户偏好、长期目标、项目背景
- 长期任务延续：跨越数天甚至数周执行多阶段项目
- 经验积累：从历史行为中学习，避免重复错误
- 个性化：基于用户特点调整交互策略
这时，就必须构建专门的记忆系统。
二、记忆系统的分层架构
一个完善的Agent记忆系统通常分为三层：
| 层次 | 容量 | 访问速度 | 典型存储 | 典型保留时间 |
|---|
| 短时记忆 | ~128K tokens | 极快 | 对话上下文 | 当前会话 |
| 中时记忆 | 约数十条重要记录 | 快 | Vector DB / 缓存 | 数天～数周 |
| 长期记忆 | 无限制 | 较慢 | 持久化存储 | 永久 |
2.1 短时记忆：会话上下文
短时记忆即当前会话的上下文窗口，是最直接可用的记忆形式。其核心职责是：
- 保留当前任务的背景信息
- 支持多轮对话的连贯性
- 为Agent提供"工作台"式的临时存储
实现上，通常在每次调用LLM时，将相关的历史消息、工具调用结果、用户反馈一并注入上下文。关键在于上下文管理策略——如何决定哪些信息值得保留、哪些需要压缩或丢弃。
常见策略：
#Agent#Agent Memory#AI工程#上下文工程
全部保留
摘要压缩：定期将对话历史压缩为摘要，保留关键信息重要性筛选：基于启发式规则或小模型判断，保留高价值消息class ShortTermMemory:
def __init__(self, max_tokens: int = 128000):
self.max_tokens = max_tokens
self.messages = []
self.summary = None
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
def get_context(self) -> list[dict]:
if self._token_count() > self.max_tokens * 0.8:
self._compress()
return self.messages
def _compress(self):
self.summary = compress_messages(self.messages)
self.messages = [{"role": "system", "content": f"会话摘要: {self.summary}"}]
2.2 中时记忆：跨会话重要信息
中时记忆用于存放跨会话仍然重要的信息——用户偏好、系统设计决策、项目背景等。其访问频率低于短时记忆，但需要更高的持久性。
典型实现方式是将关键信息结构化存储在向量数据库中，通过语义检索快速召回。
class MediumTermMemory:
def __init__(self, vector_store):
self.vector_store = vector_store
def store(self, user_id: str, content: str, metadata: dict):
embedding = get_embedding(content)
self.vector_store.insert({
"user_id": user_id,
"content": content,
"embedding": embedding,
"metadata": metadata,
"created_at": datetime.now()
})
def retrieve(self, user_id: str, query: str, top_k: int = 5):
query_embedding = get_embedding(query)
results = self.vector_store.search(
vector=query_embedding,
filter={"user_id": user_id},
top_k=top_k
)
return results
中时记忆的检索质量直接决定了Agent能否有效利用历史信息。检索策略的设计要点包括：
- 元数据过滤：支持按时间范围、信息类型等条件过滤
- 混合检索：结合关键词精确匹配和向量相似度检索
- 重排序：对初步检索结果进行更精准的排序
2.3 长期记忆：持久化知识与经验
长期记忆存储需要永久保留或跨极长时间使用的信息，包括：
- Agent的学习经验（从历史任务中学到的模式）
- 公共知识（行业知识、产品文档）
- 用户持续积累的个人知识库
实现上通常结合结构化数据库（如PostgreSQL）和向量数据库，分别支持精确查询和语义检索。
class LongTermMemory:
def __init__(self, db, vector_store):
self.db = db
self.vector_store = vector_store
def store_episode(self, user_id: str, task: str, outcome: str, lessons: str):
self.db.execute("""
INSERT INTO agent_episodes (user_id, task, outcome, lessons, timestamp)
VALUES (%s, %s, %s, %s, NOW())
""", (user_id, task, outcome, lessons))
self.vector_store.insert({
"user_id": user_id,
"content": f"任务: {task}\n结果: {outcome}\n经验: {lessons}",
"type": "episode"
})
def get_lessons(self, user_id: str, task_type: str):
return self.db.execute("""
SELECT lessons FROM agent_episodes
WHERE user_id = %s AND task LIKE %s
ORDER BY timestamp DESC LIMIT 10
""", (user_id, f"%{task_type}%"))
三、记忆的写入策略：何时存储、存储什么
不是所有信息都值得记忆。记忆写入过于激进会导致存储成本飙升、检索质量下降；过于保守则会让Agent错失重要上下文。
3.1 基于事件的触发写入
- 任务完成时：存储任务目标、执行步骤、结果和经验教训
- 用户明确反馈时：存储偏好修正、纠正信息
- 检测到重要决策时：存储决策背景和依据
- 定期摘要：对话进行到一定阶段时自动生成摘要
class MemoryWriter:
def should_store(self, event_type: str, content: str) -> bool:
rules = {
"task_completion": True,
"user_correction": True,
"important_decision": True,
"preference_update": True,
"casual_chat": False,
"routine_query": False
}
return rules.get(event_type, False)
3.2 内容的自动摘要与结构化
原始对话往往包含大量冗余信息，直接存储效率很低。更优的做法是在写入前进行结构化处理：
def process_for_storage(raw_content: str, event_type: str) -> dict:
prompt = f"""从以下内容中提取关键信息，结构化为JSON格式：
类型: {event_type}
内容: {raw_content}
输出字段: summary(摘要), facts(关键事实列表), tags(标签列表)"""
result = llm.invoke(prompt)
return json.loads(result)
四、记忆的检索与激活
有记忆是一回事，能否在正确的时机召回正确的记忆是另一回事。
4.1 上下文感知的记忆检索
Agent在执行任务时，应能根据当前上下文自动检索相关记忆。这通常通过以下方式实现：
def retrieve_relevant_memory(agent_context: dict, user_id: str) -> list[dict]:
query_parts = [
agent_context.get("current_task", ""),
agent_context.get("recent_activities", []),
agent_context.get("stated_goals", [])
]
query = " ".join(query_parts)
short = short_term.get_context()
medium = medium_term.retrieve(user_id, query, top_k=5)
long = long_term.get_episodes(user_id, query, limit=3)
merged = deduplicate_and_rank(short, medium, long)
return merged
4.2 记忆的衰减与遗忘
记忆并非永恒有效。用户的偏好可能改变，项目背景可能过时，过时的记忆不仅无益，甚至可能造成误导。
def memory_relevance_score(memory: dict, current_time: datetime) -> float:
age = (current_time - memory["created_at"]).days
base_score = memory.get("importance", 1.0)
decay_rate = memory.get("decay_rate", 0.05)
return base_score * math.exp(-decay_rate * age)
对于中时记忆和长期记忆，可以设置自动清理机制——超过一定时间且未被再次访问的记忆标记为"归档"，进一步则可删除。
五、一个完整的记忆系统实现
以下是一个简化但完整的记忆系统实现，展示了各组件如何协同工作：
class AgentMemorySystem:
def __init__(self):
self.short_term = ShortTermMemory(max_tokens=128000)
self.medium_term = MediumTermMemory(vector_store)
self.long_term = LongTermMemory(db, vector_store)
self.writer = MemoryWriter()
self.retriever = MemoryRetriever()
def store(self, event_type: str, content: str, user_id: str):
if not self.writer.should_store(event_type, content):
return
processed = process_for_storage(content, event_type)
if event_type in ["task_completion", "important_decision"]:
self.long_term.store_episode(
user_id, processed["summary"],
processed["facts"], processed.get("lessons", "")
)
elif event_type in ["preference_update", "user_feedback"]:
self.medium_term.store(user_id, processed["summary"],
{"type": event_type, "tags": processed["tags"]})
def recall(self, agent_context: dict, user_id: str) -> dict:
return {
"short_term": self.short_term.get_context(),
"medium_term": self.medium_term.retrieve(
user_id, agent_context["query"], top_k=5
),
"long_term": self.long_term.get_episodes(
user_id, agent_context["task_type"], limit=3
)
}
六、实践中的挑战与权衡
6.1 存储成本与检索质量的平衡
向量检索虽然语义匹配效果好，但计算成本不低。在高并发场景下，需要考虑：
- 检索结果的缓存策略
- 分层索引（先快速筛选再精细排序）
- 离线批量预处理embedding
6.2 隐私与安全
- 存储内容加密，尤其是个人偏好和私人信息
- 检索时严格基于用户ID隔离，防止跨用户信息泄露
- 提供记忆删除机制，满足数据合规要求
6.3 记忆的真实性与一致性
LLM存在"幻觉"问题，当记忆被注入上下文后，Agent可能将其作为事实继续传播。需要在记忆存储和检索环节引入验证机制，或在注入时标记信息来源和置信度。
七、未来方向
当前AI Agent的记忆系统仍处于早期阶段，以下方向值得关注：
- 主动记忆：Agent不只是被动存储，而是主动分析用户行为，预测可能需要记住的信息
- 记忆组合：将多个相关记忆片段组合成更高层的知识表示，而非碎片化存储
- 元认知：Agent能够感知自身记忆的局限性，主动询问用户以填补关键空白
- 跨Agent记忆共享：在多Agent协作场景中，部分记忆可以安全地共享，提升团队协作效率
结语
记忆是智能的基础——无论是人类还是AI。在Agent从"工具"走向"助手"的过程中，记忆系统的完善程度将在很大程度上决定其价值的上限。
构建好的记忆系统不在于存储更多信息，而在于：在正确的时机，召回正确的记忆，并让它服务于当前的决策。
这看似简单，实则是整个AI Agent架构中最具挑战性的问题之一。