1. 为什么 Agent 的“记忆”是个绕不开的坎
做过对话式 Agent 的人都有一个共同体会:用户第一句话告诉你“我叫老张,做跨境电商的”,聊到第十轮你问“您贵姓”,它一脸茫然地回“您好,请问怎么称呼”。这不是模型笨,是它压根没记住。大模型的上下文窗口再大,也架不住会话结束就归零、跨会话更是彻底失忆。Session Memory(会话记忆)和Long-term Memory(长期记忆)这两个词,就是用来解决这个尴尬的。
简单说,Session Memory 管的是“这一次聊天里发生过什么”,Long-term Memory 管的是“这个用户过去所有聊天里值得留下来的东西”。前者像你手边的草稿纸,后者像你的客户档案柜。两者配合,Agent 才能从“每次都是陌生人”变成“越聊越懂你”。
这套东西适合谁参考?如果你正在做客服机器人、个人助理、教育陪练、角色扮演类应用,或者任何需要“记住用户”的对话产品,那这套记忆系统的设计思路你大概率用得上。哪怕你只是用现成框架搭了个 Demo,只要涉及多轮对话,记忆就是迟早要补的一课。下面我按自己实际搭过的一套方案,把设计思路、核心细节、落地步骤和踩过的坑完整拆一遍。
2. 记忆系统的整体设计与选型思路
2.1 两层记忆到底怎么分工
很多人一上来就想搞个“万能记忆库”,把所有对话都塞进去,结果检索又慢又乱。我的做法是明确分层,各司其职。
Session Memory 的定位是短期、高频、强时效。它保存当前会话的完整消息序列,包括用户说了什么、Agent 回了什么、调用了哪些工具、返回了什么结果。它的生命周期就是这次会话,会话结束可以选择落盘归档,也可以直接丢弃。它的读取方式是“全量拼接进上下文”,因为一次会话的消息量通常可控。
Long-term Memory 的定位是长期、低频、高价值。它不是把对话原文存下来,而是从对话里抽取结构化的事实和偏好,比如“用户叫老张”“用户偏好简洁回答”“用户在做跨境电商”“用户对价格敏感”。这些信息跨会话持久化,下次用户再来,Agent 先把它捞出来注入上下文,就能接着上次的认知继续聊。
提示:不要把 Long-term Memory 做成“对话历史全文检索”。全文检索适合找“某句话在哪”,但 Agent 需要的是“关于这个用户我知道什么”,这是两个不同的检索目标。
2.2 为什么不用“超长上下文”一把梭
有人会问:现在模型上下文都上百万 token 了,直接把所有历史都塞进去不就行了?我实测下来,这条路有三个坑。
第一是成本和延迟。上下文越长,推理越慢、越贵,而且大部分历史信息跟当前问题无关,纯属浪费。第二是注意力稀释。上下文里塞太多无关内容,模型反而抓不住重点,该记住的没记住。第三是跨会话根本做不到。上下文窗口再大,也是单次请求内的,用户关掉页面再回来,你还是得从存储里重新加载。
所以正确的思路是:Session Memory 控制单次会话的上下文规模,Long-term Memory 用抽取+检索的方式,只把“当前问题真正需要的那几条记忆”注入进去。这样既省成本,又提准确率。
2.3 存储选型的取舍
存储这块我对比过几种方案,最后的选择是关系库 + 向量库组合。
| 存储方案 | 适合存什么 | 优点 | 缺点 |
|---|---|---|---|
| 关系型数据库 | 结构化事实、用户画像、会话元数据 | 查询精确、事务可靠、易维护 | 语义检索弱 |
| 向量数据库 | 记忆片段的语义检索 | 模糊匹配强、支持相似度召回 | 精确过滤弱、需嵌入模型 |
| 纯内存 | 当前会话消息 | 快、简单 | 不持久、重启即丢 |
| 文件/对象存储 | 会话归档、原始日志 | 便宜、可追溯 | 检索能力差 |
我的组合是:当前会话消息放内存 + 定期落关系库;长期记忆的事实条目存关系库,同时把每条记忆做嵌入存向量库。检索时先用关系库做硬过滤(比如按用户 ID、按记忆类型),再用向量库做语义排序,两者结合召回质量明显比单用其一好。
3. 核心细节解析与实操要点
3.1 Session Memory 的数据结构设计
Session Memory 看起来简单,但结构设计不好,后面扩展会很痛苦。我用的结构大概是这样:
session = { "session_id": "sess_20240101_abc", "user_id": "user_10086", "created_at": "2024-01-01T10:00:00Z", "messages": [ {"role": "user", "content": "...", "ts": "..."}, {"role": "assistant", "content": "...", "ts": "..."}, {"role": "tool", "name": "search", "content": "...", "ts": "..."} ], "summary": "本次会话的滚动摘要", "token_count": 3200 }几个关键点值得说。messages 要保留 role 和工具调用记录,因为 Agent 的推理链依赖这些,丢了工具结果它可能重复调用。summary 字段是滚动摘要,当消息累积到一定长度,就把早期消息压缩成摘要,只保留最近 N 轮原文,这样上下文不会无限膨胀。token_count 用于触发压缩阈值,我一般设在模型窗口的 60% 左右,留出余量给系统提示和输出。
注意:滚动摘要不要用“把旧消息删掉”的方式,而是“把旧消息压缩成一段话再删原文”。删了就是真丢了,压缩了至少语义还在。
3.2 Long-term Memory 的抽取时机与策略
长期记忆的核心动作是“抽取”,也就是从对话里识别出值得长期保留的信息。抽取时机有两种:实时抽取和会话结束时批量抽取。
实时抽取的好处是信息新鲜、上下文完整,坏处是每轮都调一次抽取模型,成本和延迟都上去了。批量抽取在会话结束时跑一次,成本低,但如果会话中途断了就丢了。我的折中方案是:每 N 轮(比如 5 轮)触发一次增量抽取,会话结束时再补一次全量抽取。
抽取的内容我分成几类,用不同的 prompt 模板:
- 身份事实:姓名、职业、所在地、联系方式等
- 偏好设置:回答风格、语言、格式偏好、禁忌话题
- 任务上下文:正在做的项目、目标、截止时间
- 历史结论:之前达成的一致、做过的决定
抽取出来的每条记忆都带上user_id、type、content、confidence、source_session、created_at。confidence很重要,模型抽取有不确定性,低置信度的记忆在检索时可以降权或过滤。
3.3 记忆检索的排序逻辑
检索不是简单地把所有记忆捞出来,而是要按相关性排序,只取 top-k。我的排序公式大致是:
score = w1 * 语义相似度 + w2 * 时间衰减 + w3 * 置信度 + w4 * 类型权重语义相似度用向量余弦相似度,时间衰减用指数衰减(越新的记忆权重越高),置信度就是抽取时打的分,类型权重按场景调(比如身份事实权重高于一般偏好)。这几个权重我一般设成 0.5 / 0.2 / 0.2 / 0.1,具体按业务调。
提示:时间衰减的系数别设太狠。有些长期偏好(比如“用户是素食者”)不该因为时间久就失效,所以身份类记忆可以关掉时间衰减,只对任务类记忆做衰减。
3.4 记忆注入上下文的方式
检索出来的记忆怎么塞进 prompt,也有讲究。我试过两种方式:系统提示注入和对话前缀注入。
系统提示注入是把记忆放在 system message 里,好处是模型会当成“背景知识”,坏处是如果记忆太多会挤占系统指令的空间。对话前缀注入是把记忆伪装成一条“系统提醒”放在对话开头,好处是更贴近对话流,坏处是可能被模型当成普通消息忽略。
我最后用的是混合方式:高优先级的身份和偏好放 system message,任务相关的记忆放对话前缀。并且给每条记忆加上明确标签,比如[用户偏好] 喜欢简洁回答,让模型一眼知道这是什么。
4. 实操过程与核心环节实现
4.1 环境与依赖准备
这套系统我用 Python 实现,核心依赖不多:
pip install openai psycopg2-binary pgvector redis tiktokenopenai:调模型做抽取和对话psycopg2-binary+pgvector:关系库和向量检索redis:存当前会话的临时状态tiktoken:算 token 数,控制上下文长度
数据库我用 PostgreSQL 加 pgvector 扩展,一张表存记忆条目,一张表存会话归档。Redis 用来存活跃会话,设置 TTL 比如 2 小时,超时自动清理。
4.2 会话记忆的读写流程
会话开始时,先根据user_id和session_id从 Redis 加载会话;如果没有,就新建一个。每轮对话结束后,把新消息 append 进去,更新 token_count,如果超过阈值就触发压缩。
def append_message(session_id, message): session = redis.get(session_id) session["messages"].append(message) session["token_count"] += count_tokens(message["content"]) if session["token_count"] > COMPRESS_THRESHOLD: session = compress_session(session) redis.set(session_id, session, ex=SESSION_TTL) return session压缩逻辑是把最早的若干轮消息交给模型,让它生成一段摘要,替换掉原文:
def compress_session(session): old_messages = session["messages"][:KEEP_RECENT] recent = session["messages"][KEEP_RECENT:] summary = llm_summarize(old_messages, existing=session.get("summary")) session["messages"] = recent session["summary"] = summary session["token_count"] = count_tokens(summary) + sum( count_tokens(m["content"]) for m in recent ) return session这里KEEP_RECENT我一般设成最近 6 到 10 轮,保证当前对话的连贯性。
4.3 长期记忆的抽取实现
抽取用一个独立的 prompt,把最近几轮对话喂进去,让模型输出 JSON 格式的记忆条目:
EXTRACT_PROMPT = """ 从以下对话中抽取值得长期记住的用户信息。 只抽取明确的、稳定的信息,不要抽取临时性的内容。 输出 JSON 数组,每条包含 type、content、confidence。 type 可选:identity / preference / task / conclusion。 confidence 为 0 到 1 的小数。 对话: {dialogue} """拿到结果后做去重和合并。去重我用向量相似度,如果新记忆和已有记忆相似度超过 0.9,就更新旧记忆而不是新增。合并的逻辑是:同类型、同主题的记忆,保留最新的一条,旧的标记为 superseded。
def upsert_memory(user_id, new_mem): similar = vector_search(user_id, new_mem["content"], top_k=1) if similar and similar[0]["score"] > 0.9: update_memory(similar[0]["id"], new_mem) else: insert_memory(user_id, new_mem)4.4 记忆检索与注入的完整链路
用户发来新消息时,完整链路是这样的:
- 加载 Session Memory,拿到当前会话上下文
- 用用户消息去 Long-term Memory 做检索,取 top-k
- 把检索结果按优先级注入 system message 和对话前缀
- 拼接完整 prompt 调模型
- 模型返回后,append 到 Session Memory
- 每 N 轮触发一次长期记忆抽取
检索的代码大概长这样:
def retrieve_memories(user_id, query, top_k=5): query_vec = embed(query) candidates = db.query(""" SELECT id, type, content, confidence, created_at, 1 - (embedding <=> %s) AS sim FROM memories WHERE user_id = %s AND status = 'active' ORDER BY embedding <=> %s LIMIT 50 """, (query_vec, user_id, query_vec)) scored = [] for c in candidates: decay = time_decay(c["created_at"], c["type"]) score = 0.5 * c["sim"] + 0.2 * decay + 0.2 * c["confidence"] + 0.1 * type_weight(c["type"]) scored.append((score, c)) scored.sort(reverse=True) return [c for _, c in scored[:top_k]]注入时按类型分组,身份和偏好放 system,任务和结论放前缀:
def build_prompt(session, memories): identity = [m for m in memories if m["type"] == "identity"] prefs = [m for m in memories if m["type"] == "preference"] tasks = [m for m in memories if m["type"] in ("task", "conclusion")] system = BASE_SYSTEM + "\n\n已知用户信息:\n" for m in identity + prefs: system += f"- [{m['type']}] {m['content']}\n" prefix = "" if tasks: prefix = "相关背景:\n" + "\n".join(f"- {m['content']}" for m in tasks) messages = [{"role": "system", "content": system}] if prefix: messages.append({"role": "system", "content": prefix}) messages.extend(session["messages"]) return messages4.5 参数选择的计算过程
几个关键参数我都是算出来的,不是拍脑袋。
压缩阈值:假设模型窗口 8k token,系统提示占 500,输出预留 1000,那留给对话历史的就是 6500。设 60% 触发压缩,就是 3900 token 左右触发。这个值保证压缩后还有足够空间继续聊。
检索 top-k:k 太大上下文被记忆挤满,k 太小可能漏掉关键信息。我按经验设 5,然后看实际效果调。如果发现模型经常忽略某类记忆,就单独给那类提高 k。
时间衰减系数:用半衰期算。假设任务类记忆半衰期 7 天,那衰减公式是0.5 ** (days / 7)。身份类不衰减,直接给 1。
5. 常见问题与排查技巧实录
5.1 记忆抽取的典型问题速查
| 问题现象 | 可能原因 | 排查与解决 |
|---|---|---|
| 抽取出一堆废话 | prompt 没约束“稳定性” | 在 prompt 里明确排除临时信息,加 few-shot 示例 |
| 同一事实重复存 | 去重阈值太低 | 提高相似度阈值到 0.9,或加类型+主题联合去重 |
| 该记的没记住 | 抽取频率太低 | 缩短抽取间隔,或对关键类型实时抽取 |
| 记忆互相矛盾 | 没做 supersede | 同主题新记忆覆盖旧记忆,旧标记失效 |
| 检索召回不准 | 只用了向量检索 | 加关系库硬过滤,加类型权重和时间衰减 |
5.2 上下文膨胀的排查
有次上线后发现响应越来越慢,查下来是 Session Memory 没触发压缩。原因是 token_count 只在 append 时更新,但工具返回的长文本没算进去。修复方法是所有进入 messages 的内容都要过一遍 token 计数,包括工具结果和系统注入。
提示:工具返回的内容往往很长,是上下文膨胀的重灾区。我后来对工具结果也做了截断和摘要,超过 500 token 的结果先压缩再入会话。
5.3 记忆污染的坑
记忆污染是指错误的信息被当成事实长期保留。我遇到过一次,用户开玩笑说“我是外星人”,抽取模型当真了,之后每次对话都带着“用户是外星人”这条记忆。解决办法有两个:一是抽取时加置信度门槛,低于 0.7 的不入库;二是给用户提供记忆管理入口,让用户能查看和删除自己的记忆。
5.4 跨会话身份识别的坑
Long-term Memory 依赖user_id,但用户可能换设备、清缓存,导致user_id变了,记忆就接不上。我的做法是支持多种身份绑定:登录用户用账号 ID,游客用设备指纹,同时允许用户手动绑定。这样即使换了设备,只要登录同一个账号,记忆就能恢复。
5.5 隐私与合规的注意事项
记忆系统天然涉及用户数据,这块必须谨慎。我的原则是:只存必要信息,敏感信息脱敏,提供删除能力。具体做法包括:抽取时过滤掉密码、证件号等敏感字段;存储时加密;提供“清除我的记忆”接口;记忆保留设置有效期,过期自动清理。这些不是可选项,是底线。
6. 记忆系统的扩展方向
这套基础版跑通后,我陆续加了几个扩展。记忆重要性分级,让模型给每条记忆打重要性分,检索时高重要性优先。记忆关联图,把相关记忆连起来,检索一条能带出一串。主动记忆,Agent 在对话中主动询问“要不要记住这个”,让用户参与记忆决策。记忆遗忘曲线,模拟人类遗忘,长期不用的记忆逐渐降权直到归档。
这些扩展不是必须的,但如果你想让 Agent 从“能记住”进化到“会记忆”,这些方向值得试试。我个人的体会是,记忆系统的价值不在于技术多复杂,而在于是否真正贴合用户的使用场景。一个能记住用户名字和偏好的简单系统,比一个存了海量数据却检索不准的复杂系统有用得多。先把 Session Memory 和 Long-term Memory 这两层跑通,再谈花活。