AI Agent 记忆系统实战：从短期上下文到长期向量记忆的分层架构设计
当你的 Agent 在第三轮对话就"忘了"用户是谁，你才会真正理解记忆系统为何是 Agent 的灵魂。
引言：为什么 AI Agent 需要记忆系统
大语言模型（LLM）本身是无状态的。每一次推理调用，模型都像一张白纸，只能看到你在 prompt 里塞进去的内容。这种"金鱼记忆"在简单问答场景下还能勉强应付，但当你试图构建一个能陪伴用户、能学习、能自主规划的多轮 Agent 时，记忆缺失就成了致命短板。
想象这样一个场景：用户周一告诉 Agent "我对花生过敏"，周三又说"帮我订一份餐厅推荐"。一个真正有用的 Agent 应该自动过滤掉所有含花生菜品的餐厅。这背后依赖的不是模型的聪明，而是一套设计良好的记忆系统——它能在正确的时刻，把正确的上下文"想"起来。
记忆系统要解决三个核心问题：
- 上下文窗口有限：即便 128K 的上下文，也无法装下一年的对话历史，更不用说海量知识。
- 信息密度不均：一句"我对花生过敏"价值极高，而十句寒暄几乎没有价值，记忆系统必须做信息筛选。
- 检索时机难把握：记住是一回事，在合适的时刻想起来是另一回事，这涉及检索与重排序策略。
本文将从记忆的分类讲起，逐步构建一个分层记忆架构，并用 Python 给出完整可运行的 MemoryManager 实现。
AI Agent 记忆的五大类型
借鉴认知科学的分类，并结合工程实践，AI Agent 的记忆通常分为以下五类：
1. 短期记忆（Short-term Memory）
即当前会话的工作记忆，对应模型的上下文窗口。它存储最近几轮对话，容量小、读取快、随会话结束而消失。本质上是 prompt 的一部分。
2. 长期记忆（Long-term Memory）
跨会话持久化的记忆，通常以向量形式存入向量数据库。它突破了上下文窗口限制，让 Agent 能"回忆"起几天甚至几个月前的事。检索靠语义相似度而非精确匹配。
3. 语义记忆（Semantic Memory）
对事实、概念、知识的记忆，类似"用户对花生过敏""公司有 30 名员工"这类客观陈述。它通常是结构化或半结构化的，更新频率低但权威性高。
4. 情节记忆（Episodic Memory）
对具体事件、经历的回忆，带有时空标签，比如"上周三下午用户让我订餐厅，最终订了日料"。情节记忆保留事件的因果链，是 Agent 进行复盘与经验学习的基础。
5. 程序性记忆（Procedural Memory）
关于"怎么做"的记忆，对应工具使用流程、操作 SOP、已掌握的技能。它通常以代码、提示词模板或技能库的形式存在，是 Agent 能力复用的关键。
工程要点：这五类记忆并非彼此孤立，而是协同工作。短期记忆负责"此刻"，长期/语义记忆负责"我知道什么"，情节记忆负责"我经历过什么"，程序性记忆负责"我会怎么做"。
分层记忆架构设计
一个生产可用的 Agent 记忆系统，通常采用分层架构。下面用文字描述这一架构（你可以把它想象成一张从上到下的数据流图）：
┌─────────────────────────────────────────────────┐
│ 用户输入 / Agent 输出 │
└───────────────────────────┬─────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ 第 0 层：记忆路由层（Memory Router） │
│ 判断本轮是否需要检索长期记忆、是否触发摘要压缩 │
└───────────────────────────┬─────────────────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌──────────────────┐ ┌──────────────┐ ┌──────────────┐
│ 第 1 层：短期记忆 │ │ 第 2 层：摘要 │ │ 第 3 层：长期 │
│ (Redis / 内存) │ │ 记忆(压缩层) │ │ 记忆(向量库) │
│ 最近 N 轮原文 │ │ 历史对话摘要 │ │ 语义+情节检索 │
└────────┬─────────┘ └──────┬───────┘ └──────┬───────┘
│ │ │
└──────────────────┴────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ 第 4 层：记忆重排序与组装（Rerank & Assemble） │
│ 对召回的记忆打分、去重、截断，拼装进最终 prompt │
└───────────────────────────┬─────────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ LLM 推理 / 工具调用 │
└─────────────────────────────────────────────────┘
各层职责说明：
- 短期记忆层：热数据，低延迟，存放最近对话原文，保证上下文连贯性。
- 摘要记忆层：当短期记忆超过阈值时触发，将较早的对话压缩成摘要，既保留信息又腾出窗口。
- 长期记忆层：冷数据 + 语义检索，存储跨会话的重要事实与情节，按相似度召回。
- 重排序层：多路召回后，记忆可能冗余或顺序混乱，需要重新打分并按相关性排序，再注入 prompt。
这种分层设计的核心思想是用空间换时间、用压缩换容量、用检索换精度。
实战：用 Python 实现一个完整的 Agent 记忆系统
下面给出一个完整可运行的实现。我们先安装依赖。
环境准备
pip install redis openai faiss-cpu numpy tiktoken python-dotenv
此外需要本地或远程运行一个 Redis 服务（
docker run -d -p 6379:6379 redis即可快速启动）。如果你没有 OpenAI API Key，可以把 embedding 与 summarize 方法替换为本地模型（如 sentence-transformers），本文为简洁起见使用 OpenAI。
在项目根目录创建 .env 文件：
OPENAI_API_KEY=sk-你的key
REDIS_URL=redis://localhost:6379/0
短期记忆：基于 Redis 的会话上下文管理
短期记忆用 Redis 的 List 结构存储，每个 session 一个 key，按时间顺序追加消息，超过阈值后触发摘要压缩。
长期记忆：基于 FAISS 的语义检索
长期记忆把每条"值得记住"的信息向量化后存入 FAISS 索引，同时把原文与元数据存到一份并行列表里，检索时根据向量距离召回 Top-K。
完整 MemoryManager 实现
import os
import json
import time
import hashlib
from typing import List, Dict, Any, Optional
import numpy as np
import redis
import faiss
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
# ---------------------------------------------------------------------------
# 配置
# ---------------------------------------------------------------------------
EMBEDDING_MODEL = "text-embedding-3-small" # 1536 维
EMBEDDING_DIM = 1536
LLM_MODEL = "gpt-4o-mini"
SHORT_TERM_LIMIT = 8 # 短期记忆保留最近 8 条消息
SUMMARIZE_TRIGGER = 6 # 当短期记忆达到 6 条时触发压缩
TOP_K = 5 # 长期记忆召回数量
# ---------------------------------------------------------------------------
# MemoryManager：分层记忆系统的核心
# ---------------------------------------------------------------------------
class MemoryManager:
"""一个集成了短期、摘要、长期三层记忆的 Agent 记忆管理器。"""
def __init__(self, redis_url: str = "redis://localhost:6379/0"):
# 1. Redis 客户端：负责短期记忆与会话隔离
self.redis = redis.from_url(redis_url, decode_responses=True)
# 2. OpenAI 客户端：负责 embedding 与摘要生成
self.client = OpenAI() # 自动读取 OPENAI_API_KEY
# 3. FAISS 索引：长期记忆的向量存储
self.index = faiss.IndexFlatIP(EMBEDDING_DIM) # 内积=余弦相似度(向量归一化后)
self.long_term_store: List[Dict[str, Any]] = [] # 与索引平行的元数据
# ---------------------- 向量化工具 ----------------------
def _embed(self, text: str) -> np.ndarray:
"""把文本转成归一化的 embedding 向量。"""
resp = self.client.embeddings.create(
model=EMBEDDING_MODEL, input=text
)
vec = np.array(resp.data[0].embedding, dtype="float32")
vec /= np.linalg.norm(vec) # 归一化，使内积=余弦相似度
return vec
# ---------------------- 短期记忆 ----------------------
def _session_key(self, session_id: str) -> str:
return f"memory:short:{session_id}"
def add_message(self, session_id: str, role: str, content: str) -> None:
"""向短期记忆追加一条消息。"""
msg = {"role": role, "content": content, "ts": time.time()}
self.redis.rpush(self._session_key(session_id), json.dumps(msg))
# 触发摘要压缩
if self.redis.llen(self._session_key(session_id)) >= SHORT_TERM_LIMIT:
self._summarize_session(session_id)
def get_short_term(self, session_id: str, limit: int = SHORT_TERM_LIMIT) -> List[Dict]:
"""读取最近 N 条短期记忆。"""
raw = self.redis.lrange(self._session_key(session_id), -limit, -1)
return [json.loads(r) for r in raw]
# ---------------------- 摘要记忆 ----------------------
def _summarize_session(self, session_id: str) -> None:
"""把较早的对话压缩成摘要，写回短期记忆头部，并归档到长期记忆。"""
msgs = self.get_short_term(session_id, limit=SUMMARIZE_TRIGGER)
if len(msgs) < SUMMARIZE_TRIGGER:
return
transcript = "\n".join(
f"{m['role']}: {m['content']}" for m in msgs
)
prompt = (
"请把以下对话压缩成一段不超过 150 字的摘要，"
"重点保留用户的关键偏好、事实和待办事项，忽略寒暄：\n\n"
f"{transcript}"
)
resp = self.client.chat.completions.create(
model=LLM_MODEL,
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
)
summary = resp.choices[0].message.content
# 用摘要替换被压缩的消息
self.redis.ltrim(self._session_key(session_id), SUMMARIZE_TRIGGER, -1)
summary_msg = {
"role": "system",
"content": f"[历史摘要] {summary}",
"ts": time.time(),
}
self.redis.lpush(self._session_key(session_id), json.dumps(summary_msg))
# 同时把摘要沉淀到长期记忆
self.remember(summary, meta={"type": "summary", "session_id": session_id})
# ---------------------- 长期记忆 ----------------------
def remember(self, text: str, meta: Optional[Dict] = None) -> None:
"""把一条信息写入长期记忆（向量化 + 元数据存储）。"""
vec = self._embed(text)
self.index.add(np.array([vec]))
record = {
"id": hashlib.md5(text.encode()).hexdigest()[:12],
"text": text,
"meta": meta or {},
"ts": time.time(),
}
self.long_term_store.append(record)
def recall(self, query: str, top_k: int = TOP_K) -> List[Dict]:
"""根据 query 从长期记忆中召回 Top-K 相关条目。"""
if self.index.ntotal == 0:
return []
qvec = self._embed(query).reshape(1, -1)
scores, ids = self.index.search(qvec, min(top_k, self.index.ntotal))
results = []
for score, idx in zip(scores[0], ids[0]):
if idx == -1:
continue
item = dict(self.long_term_store[idx])
item["score"] = float(score)
results.append(item)
return results
# ---------------------- 检索 + 重排序 ----------------------
def retrieve_and_rerank(
self, session_id: str, query: str, top_k: int = TOP_K
) -> Dict[str, Any]:
"""多路召回 + 重排序，组装最终喂给 LLM 的上下文。"""
# 召回路 1：长期语义记忆
candidates = self.recall(query, top_k=top_k * 2)
# 重排序：结合相似度分、时间新鲜度、类型权重
now = time.time()
for c in candidates:
freshness = 1.0 / (1.0 + (now - c["ts"]) / 86400) # 越新分越高
type_weight = 2.0 if c["meta"].get("type") == "fact" else 1.0
c["rerank_score"] = (
0.6 * c["score"] + 0.25 * freshness + 0.15 * (type_weight / 2.0)
)
candidates.sort(key=lambda x: x["rerank_score"], reverse=True)
# 去重 + 截断
seen, reranked = set(), []
for c in candidates:
if c["text"] in seen:
continue
seen.add(c["text"])
reranked.append(c)
if len(reranked) >= top_k:
break
return {
"short_term": self.get_short_term(session_id),
"long_term": reranked,
}
# ---------------------- 持久化（可选） ----------------------
def save(self, path: str = "long_term_memory.json") -> None:
"""把长期记忆元数据落盘（FAISS 索引也可单独 save）。"""
with open(path, "w", encoding="utf-8") as f:
json.dump(self.long_term_store, f, ensure_ascii=False, indent=2)
faiss.write_index(self.index, path + ".faiss")
def load(self, path: str = "long_term_memory.json") -> None:
if os.path.exists(path):
with open(path, "r", encoding="utf-8") as f:
self.long_term_store = json.load(f)
if os.path.exists(path + ".faiss"):
self.index = faiss.read_index(path + ".faiss")
# ---------------------------------------------------------------------------
# 端到端示例：如何在一个对话回合里使用 MemoryManager
# ---------------------------------------------------------------------------
if __name__ == "__main__":
mm = MemoryManager()
session = "user_001"
# 1. 先把一些关键事实沉淀到长期记忆
mm.remember("用户对花生过敏，严重程度为致命级。",
meta={"type": "fact"})
mm.remember("用户偏好日料和轻食，不喜欢太辣的食物。",
meta={"type": "preference"})
mm.remember("上周三为用户预订了东京日料餐厅，反馈很好。",
meta={"type": "episodic", "session_id": session})
# 2. 模拟几轮短期对话
mm.add_message(session, "user", "你好，帮我规划今晚的晚餐。")
mm.add_message(session, "assistant", "好的，请问有什么口味偏好？")
mm.add_message(session, "user", "想吃点清淡的日料。")
# 3. 在新一轮对话前，做记忆检索与重排序
query = "今晚晚餐推荐，要安全、清淡"
context = mm.retrieve_and_rerank(session, query, top_k=3)
print("=== 短期记忆 ===")
for m in context["short_term"]:
print(f" [{m['role']}] {m['content']}")
print("\n=== 召回的长期记忆（已重排序）===")
for m in context["long_term"]:
print(f" score={m['rerank_score']:.3f} | {m['text']}")
# 4. 组装 prompt 喂给 LLM（这里仅演示拼装）
prompt_messages = [{"role": "system", "content":
"你是一个贴心的餐厅推荐助手。请结合用户记忆回答。"}]
# 注入长期记忆
facts = "\n".join(f"- {m['text']}" for m in context["long_term"])
prompt_messages.append({"role": "system", "content": f"用户记忆：\n{facts}"})
# 注入短期记忆
prompt_messages.extend(
{"role": m["role"], "content": m["content"]}
for m in context["short_term"]
)
prompt_messages.append({"role": "user", "content": query})
print("\n=== 最终 prompt 消息数 ===", len(prompt_messages))
运行后，你会看到短期记忆、重排序后的长期记忆被分别打印出来，最后组装成可喂给 LLM 的消息列表。这个 MemoryManager 虽然精简，但已经涵盖了分层架构的四个关键能力：短期存储、摘要压缩、向量召回、重排序组装。
实操步骤小结
- 启动 Redis：
docker run -d -p 6379:6379 redis。 - 配置
.env，填入OPENAI_API_KEY。 - 安装依赖：
pip install redis openai faiss-cpu numpy tiktoken python-dotenv。 - 把上面的代码保存为
memory_manager.py，运行python memory_manager.py。 - 观察：长期记忆是否被正确召回、短期记忆是否被注入 prompt。
- 进阶：把
retrieve_and_rerank的输出接到真实的 LLM 调用上，完成一个端到端对话 Agent。
生产部署最佳实践
把上面的原型搬上生产，还需要注意以下几点：
1. 短期记忆要设 TTL 与会话隔离
给 Redis key 设置过期时间（如 24 小时），避免内存无限膨胀；用 session_id 严格隔离不同用户的上下文，防止串号。
2. 摘要触发要异步化
摘要生成是一次 LLM 调用，耗时几百毫秒到数秒。生产中应把 _summarize_session 放进异步队列（如 Celery），不要阻塞主对话链路。
3. 长期记忆要区分重要度
不是所有消息都值得长期存储。应设置一个"记忆写入判定器"——可以是一条 LLM 规则，判断当前消息是否包含事实/偏好/待办，只有高价值信息才调用 remember()，避免向量库被噪声淹没。
4. 向量索引要可扩展
IndexFlatIP 是暴力检索，数据量上万后延迟会上升。生产环境推荐使用 IndexIVFFlat 或 HNSW，或者直接用托管的向量数据库（Milvus、Qdrant、Pinecone）。
5. 记忆要有遗忘机制 人会遗忘，Agent 也应该。对长期记忆设置衰减策略：超过 N 天未被召回、或召回后从未被采纳的条目，可降级或删除，保持记忆库的高信噪比。
6. 做好可观测性 记录每次召回的 query、命中条目、重排序分数、最终是否被 LLM 引用。这些指标是评估记忆系统质量、调优权重参数的关键依据。
常见问题 FAQ
Q1：短期记忆和长期记忆的边界到底怎么划分？ A：以"会话"为界。当前会话内、放在 prompt 上下文窗口里的就是短期记忆；跨会话、需要检索才能进入 prompt 的就是长期记忆。一个实用经验是：短期记忆放最近 5-10 轮原文，超出的部分要么摘要、要么沉淀为长期记忆。
Q2：FAISS 和 Milvus / Qdrant 该怎么选？ A：FAISS 是库，适合单机、数据量在百万以内的场景，部署简单；Milvus / Qdrant 是服务，支持分布式、水平扩展、持久化和过滤，适合多实例部署和数据量大的生产场景。原型阶段用 FAISS，规模化后迁移到向量数据库。
Q3：摘要压缩会不会丢失关键信息？ A：会，这是压缩的本质代价。缓解方法有三：一是只压缩较早的消息，保留最近几轮原文；二是摘要 prompt 中明确要求保留事实和待办；三是把被压缩的原文同时存入长期记忆，需要时仍可向量召回原文。
Q4：记忆系统会不会让 Agent 产生"幻觉"或引用错误记忆？ A：会。向量检索本质是相似度匹配，可能召回语义相近但事实错误的条目。缓解方法：对写入记忆做事实校验；在重排序中加入"可信度"权重；在 prompt 中明确告知 LLM "以下记忆仅供参考，若与用户最新说法冲突以用户为准"。
Q5：用户隐私数据存进向量库安全吗？ A：需要特别处理。建议：对 PII（个人身份信息）做脱敏后再向量化；按用户/租户做向量索引隔离；提供"遗忘接口"，支持用户删除自己的全部记忆；遵守 GDPR 等数据合规要求。
Q6：记忆检索的 Top-K 设多少合适？ A：取决于上下文窗口和单条记忆长度。一般 Top-K 在 3-10 之间。太少会漏掉相关信息，太多会挤占窗口并引入噪声。建议配合重排序，先召回 Top-K×2，再重排序截断到 Top-K，效果更好。
Q7：程序性记忆（技能）怎么和这套系统集成？ A：程序性记忆通常不走向量检索，而是走"技能路由"。可以把技能注册成一个技能表（name→prompt 模板/代码），由 LLM 通过 function calling 决定调用哪个技能；技能内部再使用上述记忆系统管理自己的上下文。
总结
AI Agent 的记忆系统不是单一组件，而是一套分层协作的架构。短期记忆保证当下连贯，摘要记忆在有限窗口内压榨更多信息，长期记忆突破时间与容量限制，重排序层则确保在正确的时刻想起正确的事。
本文给出的 MemoryManager 实现虽然精简，但已经覆盖了分层架构的核心链路：Redis 管短期上下文、FAISS 管长期向量、LLM 负责摘要、加权打分负责重排序。你可以在此基础上替换组件（如把 FAISS 换成 Milvus、把 OpenAI 换成本地模型），但架构思想是通用的。
记住一句话：Agent 的上限由模型决定，但 Agent 的连续性与可信度，由记忆系统决定。 把记忆系统当作 Agent 的"第二大脑"来认真设计，你的 Agent 才真正具备陪伴用户、持续成长的能力。
相关文章推荐：
本文发布于 1630.top，转载请注明出处。