Agent 记忆机制:短期、长期与工作记忆工程实现
Agent 的记忆问题本质是上下文窗口的管理问题:对话轮次越多,messages 列表越长,成本和延迟线性增长,超过窗口上限后模型开始”遗忘”早期信息。工程上需要把不同时效的信息放到合适的存储层,而不是全部堆进 messages。
你大概率遇到过这种场景:客服 Agent 上线一周后,某个老用户连续聊了两百多轮,某天突然开始答非所问——不是模型变笨了,是 messages 列表已经堆到几万 token,早期的关键信息(用户到底买了什么型号、报过什么故障)被挤到窗口最前面,模型的注意力早就分散到后面无关的寒暄里去了。如果调用的是有硬上限的接口,还会直接收到类似”maximum context length is 128000 tokens, however you requested 132456 tokens”的报错,请求直接 400 掉。这不是靠换更贵的模型能解决的,是内存管理没做。
四种记忆类型
| 记忆类型 | 存储位置 | 生命周期 | 典型数据 | 实现复杂度 |
|---|---|---|---|---|
| 短期记忆 | messages 列表 | 当前会话 | 对话上下文、工具调用历史 | 低 |
| 工作记忆 | Scratch Pad(prompt 内变量) | 当前任务 | 任务中间状态、计算结果 | 低 |
| 情节记忆 | 向量数据库 | 跨会话 | 历史对话摘要、用户偏好 | 中 |
| 语义记忆 | 知识库(RAG) | 永久 | 领域知识、产品文档 | 中 |
大多数 Agent 只需要短期记忆 + 情节记忆两层;工作记忆和语义记忆按需引入。
判断该上哪一层,其实就问自己一句话:这条信息”过了这次对话还有没有用”。计算过程中的临时变量(比如某个订单号、某次工具调用返回的中间结果)过了这次任务就该扔,塞进工作记忆;用户说过”我对花生过敏”这种跨会话都要记住的偏好,才值得放进情节记忆去持久化。反过来,如果你把订单查询这种一次性数据也往向量库里存,库很快会被灌满噪音,检索命中率反而下降——见过团队把每一次工具调用返回值都存成”记忆”,三个月后向量库里九成是垃圾,召回排在前面的全是没用的旧数据。
短期记忆:滚动窗口控制
最常见的错误是无限追加 messages,不做任何截断。推荐两种策略:
策略一:Token 计数截断
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
def count_tokens(messages: list[dict]) -> int:
total = 0
for msg in messages:
content = msg.get("content") or ""
total += len(enc.encode(content)) + 4 # 每条消息固定开销
return total
def trim_messages(
messages: list[dict],
max_tokens: int = 6000,
keep_system: bool = True
) -> list[dict]:
"""
保留 system prompt,从最旧的非 system 消息开始丢弃,
直到 token 数降到 max_tokens 以内。
"""
if count_tokens(messages) <= max_tokens:
return messages
system_msgs = [m for m in messages if m["role"] == "system"]
non_system = [m for m in messages if m["role"] != "system"]
# 从头开始删,保留最近的对话
while non_system and count_tokens(system_msgs + non_system) > max_tokens:
non_system.pop(0)
return system_msgs + non_system
这段代码里有两个容易被忽略的细节。第一,len(enc.encode(content)) + 4 里的 +4 不是拍脑袋加的——OpenAI 的 chat 格式里每条消息都要包上 role、消息分隔符这些元数据的 token 开销,官方文档给出的经验值大概是每条消息 3~4 个 token,如果你只算 content 文本的 token 数,实际请求会比你算出来的数值偏大,长对话攒起来能差出几百 token,正好卡在窗口边界报错。第二,tiktoken.encoding_for_model 认的是模型名,如果你的模型不在 tiktoken 内置的映射表里(比如某些国产模型或者自定义部署的模型),会直接抛 KeyError,稳妥的做法是显式 tiktoken.get_encoding("cl100k_base"),虽然编码规则和目标模型不完全一致,但作为长度估算够用,比抛异常强。
还有一个更隐蔽的坑:trim_messages 从头开始整条丢弃非 system 消息,如果丢到的正好是一条带 tool_calls 的 assistant 消息,但对应的 tool 角色回复消息没有一起丢,下一次请求就会报类似”messages with role ‘tool’ must be a response to a preceding message with ‘tool_calls‘“的结构性错误。所以实际生产代码里不能简单 pop(0),得按”assistant+tool_calls 和它对应的 tool 消息”成对判断,要丢就一起丢,单独留一半会直接把请求打挂。
策略二:摘要压缩(超过 N 轮时)
def summarize_history(messages: list[dict], keep_recent: int = 4) -> list[dict]:
"""
对历史消息做摘要,保留最近 keep_recent 轮原始对话
"""
if len(messages) <= keep_recent + 1: # +1 for system
return messages
system = messages[0]
to_summarize = messages[1:-keep_recent]
recent = messages[-keep_recent:]
history_text = "\n".join(
f"{m['role']}: {m.get('content', '')}"
for m in to_summarize
if m.get("content")
)
summary_resp = client.chat.completions.create(
model="gpt-4o-mini", # 摘要用轻量模型节省成本
messages=[
{"role": "system", "content": "请用3-5句话总结以下对话的关键信息:"},
{"role": "user", "content": history_text}
]
)
summary = summary_resp.choices[0].message.content
return [
system,
{"role": "system", "content": f"[历史对话摘要]\n{summary}"},
*recent
]
摘要压缩本质是用一次额外的 LLM 调用换取上下文空间,这笔账要算清楚:每次触发摘要就多一次请求延迟(哪怕用 mini 模型也有几百毫秒),如果每一轮都判断”要不要摘要”,等于给主流程加了一道串行等待。更合理的做法是设阈值触发——比如消息数超过 20 轮或者 token 数超过窗口的 70% 才摘要一次,不要每轮都算。
摘要压缩还有个真实存在的风险:LLM 做摘要不是复制粘贴,是”重新表述”,如果原始对话里有精确数值(订单号、金额、日期),摘要模型有一定概率会记错或者简化掉小数点后的数字——这不是危言耸听,gpt-4o-mini 这类轻量模型在长文本摘要时对具体数字的保真度明显低于长模型。所以像订单号、金额这类不能出错的关键数据,不要指望摘要帮你记住,得在摘要之前用正则或者结构化解析单独抽出来,存成结构化字段(哪怕就是个 dict),摘要只负责保留”用户情绪、大致诉求”这种模糊信息即可。
工作记忆:Scratch Pad 模式
对于多步骤复杂任务,让模型维护一个”便签本”记录中间状态,避免在推理中丢失计算结果:
SYSTEM_PROMPT = """你是一个智能助理。
## 工作便签(Scratch Pad)
在执行复杂任务时,使用以下格式维护中间状态:
<scratch_pad>
任务目标:[当前要完成的任务]
已完成步骤:
- [步骤1] 结果:[...]
- [步骤2] 结果:[...]
待执行步骤:
- [步骤3]
关键变量:
- order_id: ORD-20240601001
- total_amount: 299.00
</scratch_pad>
最终回答时不需要展示便签内容,直接给出结论。"""
Scratch Pad 内容保留在当前对话的 system/assistant 消息中,任务完成后随对话结束自动清空,不需要持久化存储。
这个模式说白了是”骗模型自己给自己做笔记”,本质上不是真正的外部状态存储,而是靠 prompt 里的格式约束让模型每次推理都重新读一遍前面写下的内容——它能不能生效,完全取决于模型有没有老老实实按格式更新便签。实测中最容易翻车的情况是任务步骤一多,模型开始”偷懒”,只更新”已完成步骤”却忘了同步”关键变量”,下一步引用变量时就用了旧值。想让它更可靠,有两个办法:一是把便签结构做成强约束的 JSON 而不是自由格式的文本块,配合 function calling 强制模型每步都要”调用”一个 update_scratchpad 工具来写状态,工具调用的参数校验能挡住格式错误;二是任务步骤超过 5~6 步就别指望纯 prompt 里的便签兜得住,老老实实上真正的状态机(哪怕就是个 Python 字典在你自己的代码里维护),模型只负责”读”和”决策”,别指望它自己”记账”记得准。
情节记忆:跨会话向量存储
对需要记住用户偏好、历史行为的 Agent,需要把重要信息持久化到向量库:
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
import hashlib, time
memory_client = QdrantClient("localhost", port=6333)
def save_memory(user_id: str, content: str, memory_type: str = "preference"):
"""保存一条情节记忆"""
vector = embed(content) # 向量化
doc_id = int(hashlib.md5(f"{user_id}{time.time()}".encode()).hexdigest(), 16) % (2**63)
memory_client.upsert(
collection_name="agent_memories",
points=[PointStruct(
id=doc_id,
vector=vector,
payload={
"user_id": user_id,
"content": content,
"type": memory_type,
"created_at": time.time()
}
)]
)
def recall_memory(user_id: str, query: str, top_k: int = 3) -> list[str]:
"""根据当前上下文召回相关记忆"""
query_vec = embed(query)
results = memory_client.search(
collection_name="agent_memories",
query_vector=query_vec,
query_filter={"must": [{"key": "user_id", "match": {"value": user_id}}]},
limit=top_k
)
return [r.payload["content"] for r in results]
# 在 Agent 开始时注入记忆
def build_system_prompt(user_id: str, current_query: str) -> str:
memories = recall_memory(user_id, current_query)
memory_section = ""
if memories:
memory_section = "\n\n## 用户历史记忆\n" + "\n".join(f"- {m}" for m in memories)
return BASE_SYSTEM_PROMPT + memory_section
几个实操细节值得单独说一下。首先 query_filter 这段写法要小心 Qdrant 客户端版本,早期版本接受原始 dict(如上面代码),但较新版本的 qdrant-client 推荐用 models.Filter(must=[models.FieldCondition(key="user_id", match=models.MatchValue(value=user_id))]) 这种强类型写法,混用旧写法在新版客户端上可能直接抛 ValidationError 或者过滤条件被静默忽略——过滤失效是最阴的坑,因为代码不报错,只是查出来的记忆混进了别的用户数据,排查起来得对着返回结果一条条核对 user_id 才能发现。
其次是 embed() 这个函数:存记忆时用的 embedding 模型和后续检索时用的必须是同一个模型、同一个版本,换了模型意味着向量维度可能都不一样(比如从 1536 维换成 3072 维),Qdrant 会直接报 Wrong input: Vector dimension error: expected dim: 1536, got 3072,而且旧数据不会自动重新向量化,你要么整库重建索引,要么新开一个 collection 平滑迁移,别指望模型升级了向量库能自动兼容。
再有就是 user_id 的隔离策略——上面代码是用 filter 做逻辑隔离,所有用户的记忆存在同一个 collection 里,这种方式简单但当用户量上到百万级、单个 collection 索引巨大时,过滤检索的性能会下降;如果你的场景是强多租户(比如 SaaS 给不同企业客户),更稳的做法是按租户拆 collection 做物理隔离,虽然多了运维成本,但避免了”一次过滤条件写错就跨租户读到别人数据”的合规风险,这对企业客户来说是红线。
| 存储方案 | 适合规模 | 优势 | 劣势 |
|---|---|---|---|
| Qdrant / Milvus 等专用向量库 | 十万级以上记忆条目 | 检索快、过滤能力强、支持分片 | 多一套独立服务要运维 |
| pgvector(PostgreSQL 插件) | 万级到十万级 | 复用现有数据库,事务和过滤都用 SQL | 数据量大时索引和查询性能不如专用向量库 |
| Redis + 向量搜索模块 | 小规模、要求低延迟 | 部署简单,能和缓存共用一套基础设施 | 持久化和扩容能力弱于专用方案 |
如果你的记忆条目预计长期在几万条以内,别一上来就上 Qdrant 这类专用向量库——团队已经在用 PostgreSQL 的话,pgvector 插件够用,还能少维护一套服务;等规模真的涨上去了再迁移也不迟,工程上没必要为了”看起来专业”提前上重型方案。
记忆架构选型建议
业务场景
│
├── 单次任务型(查询/翻译/分析)
│ → 仅短期记忆(滚动窗口),无需持久化
│
├── 对话型(客服/助理)
│ → 短期记忆 + 情节记忆(向量库存储摘要)
│
└── 复杂任务型(自动化工作流)
→ 短期记忆 + 工作记忆(Scratch Pad)+ 语义记忆(RAG)
常见问题
记忆应该存原始对话还是摘要? 推荐存摘要。原始对话包含大量冗余信息,向量检索时噪声多;用 gpt-4o-mini 压缩成 3~5 句话的摘要,既节省存储,检索精度也更高。对关键数据点(用户偏好、重要决策)单独抽取存储效果更好。
记忆多了会越来越慢吗? 向量检索的延迟主要取决于索引规模,百万级别内都在 10ms 以内。影响更大的是”注入多少记忆到上下文”——建议每次最多注入 3~5 条,过多反而造成注意力稀释。
如何防止记忆包含错误信息? 定期对记忆做置信度标记:高置信记忆(用户明确确认的偏好)保留,低置信记忆(Agent 自行推断的)设过期时间自动清除。对关键业务数据,用结构化存储(数据库)代替向量记忆更可靠。
← 返回 应用模式总览:从 Prompt 到 Agent | 应用模式专题
相关阅读:ReAct 模式实现 · Agent 工具调用设计 · 向量数据库选型
Agent 上下文管理、多模型切换统一接入?力达云聚合 API 帮你屏蔽底层差异,专注业务逻辑。