AI Agent 的记忆系统——短期/长期/情景记忆的工程实现
2026/4/30大约 8 分钟
AI Agent 的记忆系统——短期/长期/情景记忆的工程实现
适读人群：做 Agent 系统的工程师 | 阅读时长：约18分钟 | 核心价值：三层记忆架构的完整工程实现，Redis + 向量数据库方案
去年我接了一个做"智能客服 Agent"的项目，甲方的要求是：用户说过的事情，下次来不用重复，Agent 要记得。
这听起来是个小需求，结果我发现这背后藏着一整套工程问题：记什么？存哪里？用什么格式？怎么检索？什么时候更新？多少条合适？
我把这套问题拆解了三个月，踩了很多坑，最终搭出来一套可用的记忆架构。今天把这套方案完整写出来，包括代码。
为什么 LLM 自身的上下文不够用
很多人刚开始做 Agent 的时候，觉得 context window 越来越大，记忆问题自然就解决了。128K token 够了吧？这个想法有几个问题。
成本问题。 128K token 每次请求都带着，按 claude-sonnet-4-5 的价格，输入端每次多带 10 万 token 就要多花 $0.30，一天 1000 次请求就是 $300——这条路走不通。
召回质量问题。 研究表明 LLM 对长 context 的处理存在"中间迷失"现象，早期和末尾的信息召回效果好，中间的信息容易被忽略。把所有历史塞进去，未必比精选的关键信息效果好。
时效性问题。 用户三个月前说的偏好，和昨天说的偏好，权重应该不同。原始对话历史无法体现这个衰减。
所以需要一套真正的记忆管理系统，而不是把上下文塞满。
三层记忆架构
我最终用的是一个类比人类记忆的三层架构：
+------------------+
| 工作记忆 | 当前对话的上下文，存在内存/Redis，短暂
| Working Memory | 容量：最近 20 轮对话
+------------------+
|
v（对话结束时压缩摘要）
+------------------+
| 情景记忆 | 历史对话的压缩摘要，按时间索引
| Episodic Memory | 容量：最近 100 次对话摘要
+------------------+
|
v（持续提取更新）
+------------------+
| 语义记忆 | 用户的稳定偏好和知识，向量索引
| Semantic Memory | 容量：无限（向量数据库）
+------------------+三层各司其职：工作记忆保证当前对话连贯，情景记忆保证跨会话上下文，语义记忆保证长期用户画像。
工作记忆：Redis 实现
工作记忆就是当前会话的对话历史，最简单，但有几个细节要处理。
import redis
import json
from datetime import datetime
from typing import List, Dict
class WorkingMemory:
def __init__(self, redis_client: redis.Redis, max_turns: int = 20):
self.redis = redis_client
self.max_turns = max_turns
self.ttl = 3600 # 1小时不活跃自动过期
def _key(self, session_id: str) -> str:
return f"working_memory:{session_id}"
def add_turn(self, session_id: str, role: str, content: str):
key = self._key(session_id)
turn = {
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
}
# 用 Redis List，RPUSH 追加，LTRIM 保持最大长度
self.redis.rpush(key, json.dumps(turn))
self.redis.ltrim(key, -self.max_turns * 2, -1) # 保留最近 max_turns 轮（每轮2条）
self.redis.expire(key, self.ttl)
def get_history(self, session_id: str) -> List[Dict]:
key = self._key(session_id)
raw_turns = self.redis.lrange(key, 0, -1)
return [json.loads(t) for t in raw_turns]
def clear(self, session_id: str):
self.redis.delete(self._key(session_id))
def get_token_estimate(self, session_id: str) -> int:
"""粗略估算当前工作记忆占用的 token 数"""
history = self.get_history(session_id)
total_chars = sum(len(t["content"]) for t in history)
return total_chars // 4 # 粗略：4个字符约1个token这里有个细节：用 max_turns * 2 而不是 max_turns，是因为每轮对话有用户和助手两条消息。
情景记忆：会话摘要的生成和存储
情景记忆的核心是：对话结束时，用 LLM 生成这次对话的结构化摘要，压缩存储。
import anthropic
from dataclasses import dataclass
from typing import Optional
@dataclass
class EpisodeRecord:
session_id: str
user_id: str
start_time: str
end_time: str
summary: str # 自然语言摘要
key_facts: List[str] # 提取的关键事实
user_requests: List[str] # 用户在这次对话里请求了什么
resolved: bool # 需求是否被满足
class EpisodicMemory:
def __init__(self, redis_client: redis.Redis, max_episodes: int = 100):
self.redis = redis_client
self.max_episodes = max_episodes
self.client = anthropic.Anthropic()
def _episodes_key(self, user_id: str) -> str:
return f"episodic_memory:{user_id}"
async def compress_session(
self,
user_id: str,
session_id: str,
conversation_history: List[Dict]
) -> EpisodeRecord:
"""对话结束后，把整个对话历史压缩成结构化摘要"""
# 把对话历史格式化成文本
dialogue_text = "\n".join([
f"{t['role'].upper()}: {t['content']}"
for t in conversation_history
])
prompt = f"""请分析以下对话，提取关键信息。
对话内容：
{dialogue_text}
请按以下 JSON 格式返回：
{{
"summary": "一两句话概括这次对话的核心内容",
"key_facts": ["关于用户的重要事实1", "重要事实2"],
"user_requests": ["用户的请求1", "用户的请求2"],
"resolved": true/false
}}
注意：key_facts 应该是对未来有参考价值的用户信息，比如偏好、背景、已知情况等。"""
response = self.client.messages.create(
model="claude-sonnet-4-5",
max_tokens=500,
messages=[{"role": "user", "content": prompt}]
)
import re
json_match = re.search(r'\{.*\}', response.content[0].text, re.DOTALL)
parsed = json.loads(json_match.group())
episode = EpisodeRecord(
session_id=session_id,
user_id=user_id,
start_time=conversation_history[0]["timestamp"],
end_time=conversation_history[-1]["timestamp"],
summary=parsed["summary"],
key_facts=parsed["key_facts"],
user_requests=parsed["user_requests"],
resolved=parsed["resolved"]
)
# 存入 Redis（按用户 ID 的 List）
key = self._episodes_key(user_id)
self.redis.rpush(key, json.dumps(episode.__dict__))
self.redis.ltrim(key, -self.max_episodes, -1)
return episode
def get_recent_episodes(self, user_id: str, n: int = 5) -> List[EpisodeRecord]:
key = self._episodes_key(user_id)
raw = self.redis.lrange(key, -n, -1)
records = []
for r in raw:
d = json.loads(r)
records.append(EpisodeRecord(**d))
return records
def format_for_context(self, episodes: List[EpisodeRecord]) -> str:
"""把最近的情景记忆格式化成 system prompt 的一部分"""
if not episodes:
return ""
lines = ["【历史对话摘要】"]
for ep in episodes:
lines.append(f"- {ep.start_time[:10]}：{ep.summary}")
if ep.key_facts:
for fact in ep.key_facts:
lines.append(f" * {fact}")
return "\n".join(lines)语义记忆：向量数据库存用户偏好
语义记忆是最复杂的部分。它存的是用户的稳定知识——偏好、习惯、背景信息——用向量索引，通过语义搜索召回相关片段。
我用的是 Qdrant（开源向量数据库，可以自部署），配合 Anthropic 的 Embedding（或者用 OpenAI 的 text-embedding-3-small，更便宜）。
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid
import openai # 用 OpenAI embedding，便宜
class SemanticMemory:
def __init__(self, qdrant_client: QdrantClient, collection_name: str = "user_memory"):
self.qdrant = qdrant_client
self.collection = collection_name
self.openai = openai.OpenAI()
self._ensure_collection()
def _ensure_collection(self):
existing = [c.name for c in self.qdrant.get_collections().collections]
if self.collection not in existing:
self.qdrant.create_collection(
collection_name=self.collection,
vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)
def _embed(self, text: str) -> List[float]:
response = self.openai.embeddings.create(
input=text,
model="text-embedding-3-small"
)
return response.data[0].embedding
def upsert_fact(self, user_id: str, fact: str, category: str = "general"):
"""存入或更新一条用户事实"""
vector = self._embed(fact)
point_id = str(uuid.uuid4())
self.qdrant.upsert(
collection_name=self.collection,
points=[
PointStruct(
id=point_id,
vector=vector,
payload={
"user_id": user_id,
"fact": fact,
"category": category,
"created_at": datetime.now().isoformat(),
"access_count": 0
}
)
]
)
return point_id
def search_relevant(self, user_id: str, query: str, top_k: int = 5) -> List[Dict]:
"""根据当前查询，检索最相关的用户记忆"""
query_vector = self._embed(query)
results = self.qdrant.search(
collection_name=self.collection,
query_vector=query_vector,
query_filter={
"must": [{"key": "user_id", "match": {"value": user_id}}]
},
limit=top_k,
with_payload=True
)
return [
{
"fact": r.payload["fact"],
"category": r.payload["category"],
"score": r.score,
"created_at": r.payload["created_at"]
}
for r in results
]
def extract_and_store_from_episode(
self,
user_id: str,
episode: EpisodeRecord,
llm_client: anthropic.Anthropic
):
"""从情景记忆中提取值得长期存储的语义知识"""
# 用 LLM 判断哪些事实值得长期记忆
prompt = f"""以下是一次对话中提取的关键事实，请判断哪些值得长期记忆（用户的稳定偏好、背景信息、重要状态），哪些只是一次性的：
事实列表：
{json.dumps(episode.key_facts, ensure_ascii=False)}
请返回 JSON：
{{
"persistent_facts": ["值得长期记忆的事实"],
"categories": ["对应的分类，如preference/background/status"]
}}"""
response = llm_client.messages.create(
model="claude-haiku-20240307", # 用便宜的模型做分类
max_tokens=300,
messages=[{"role": "user", "content": prompt}]
)
import re
json_match = re.search(r'\{.*\}', response.content[0].text, re.DOTALL)
parsed = json.loads(json_match.group())
for fact, category in zip(
parsed.get("persistent_facts", []),
parsed.get("categories", [])
):
self.upsert_fact(user_id, fact, category)三层整合：Agent 调用记忆的完整流程
三层记忆搭好了，关键是在 Agent 里怎么用。
class MemoryAwareAgent:
def __init__(
self,
redis_client: redis.Redis,
qdrant_client: QdrantClient
):
self.working_memory = WorkingMemory(redis_client)
self.episodic_memory = EpisodicMemory(redis_client)
self.semantic_memory = SemanticMemory(qdrant_client)
self.client = anthropic.Anthropic()
def build_system_prompt(
self,
user_id: str,
current_query: str
) -> str:
"""根据当前查询，动态构建包含记忆的 system prompt"""
# 1. 检索语义记忆（与当前查询最相关的用户事实）
relevant_facts = self.semantic_memory.search_relevant(
user_id, current_query, top_k=5
)
# 2. 获取最近的情景记忆
recent_episodes = self.episodic_memory.get_recent_episodes(
user_id, n=3
)
# 组装 system prompt
memory_section = []
if relevant_facts:
memory_section.append("【关于用户的已知信息】")
for fact in relevant_facts:
if fact["score"] > 0.7: # 相关度阈值
memory_section.append(f"- {fact['fact']}")
episode_context = self.episodic_memory.format_for_context(recent_episodes)
if episode_context:
memory_section.append(episode_context)
base_prompt = "你是一个智能助手。"
if memory_section:
base_prompt += "\n\n" + "\n".join(memory_section)
return base_prompt
async def chat(self, user_id: str, session_id: str, user_message: str) -> str:
# 存入工作记忆
self.working_memory.add_turn(session_id, "user", user_message)
# 构建包含记忆的 system prompt
system_prompt = self.build_system_prompt(user_id, user_message)
# 获取当前会话历史
history = self.working_memory.get_history(session_id)
messages = [
{"role": t["role"], "content": t["content"]}
for t in history
]
# 调用 Claude
response = self.client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1000,
system=system_prompt,
messages=messages
)
assistant_message = response.content[0].text
# 存入工作记忆
self.working_memory.add_turn(session_id, "assistant", assistant_message)
return assistant_message
async def end_session(self, user_id: str, session_id: str):
"""会话结束时，压缩存入情景记忆，提取更新语义记忆"""
history = self.working_memory.get_history(session_id)
if len(history) < 2: # 太短的对话不值得记忆
return
# 压缩成情景记忆
episode = await self.episodic_memory.compress_session(
user_id, session_id, history
)
# 从情景记忆中提取语义知识
self.semantic_memory.extract_and_store_from_episode(
user_id, episode, self.client
)
# 清理工作记忆
self.working_memory.clear(session_id)几个踩过的坑
记忆污染问题。 如果 Agent 在某次对话里理解错了用户意图，把错误的事实存进语义记忆，后续会持续影响表现。要加一个"置信度"字段，低置信度的事实过段时间自动过期，不要无限期保留。
隐私合规问题。 语义记忆存的是用户信息，GDPR 和国内《个人信息保护法》都要求支持"删除我的数据"。向量数据库删除要同时删掉向量和 payload，确保彻底。Qdrant 支持按 payload 过滤删除，这点没问题。
记忆检索的延迟。 每次对话都要先检索向量数据库，会增加首响应延迟。我的做法是把检索做成异步预取——用户发消息的同时就开始检索，等 LLM 处理完第一个 token 的时候检索通常也结束了，延迟基本被掩盖。
工作记忆 token 预算。 要动态控制三层记忆加起来占的 token 不超过 context window 的 40%，剩下留给当前对话。写一个简单的 token 预算分配器，当总量超限时先砍情景记忆，再砍语义记忆，工作记忆的最近几轮始终保留。
这套架构跑起来之后，用户满意度分明显高了，最直接的感受就是：用户不用重复介绍自己了。对于做客服 Agent 的团队，这个体验差距相当明显。