Agent记忆系统架构：Session Memory与Long-term Memory实战

1. 为什么 Agent 的“记忆”是个绕不开的坎

做过对话类 Agent 的人都有一个共同体会：模型本身很聪明，但它像个失忆症患者。用户上一轮说了“我住在杭州，平时喜欢喝美式”，下一轮问“明天适合穿什么”，它完全不知道你在哪个城市、什么季节、甚至不记得你刚说过喜欢喝什么。这不是模型能力问题，而是架构问题——大模型本身是无状态的，每次调用都是一次全新的开始。

记忆系统要解决的核心问题就一个：让 Agent 在多轮交互中保持连贯性，并且能跨会话记住用户的偏好、习惯和历史信息。这里面其实分两层：一层是Session Memory（会话记忆），管的是当前这次对话的上下文；另一层是Long-term Memory（长期记忆），管的是跨会话、跨时间段的持久化信息。两层解决的是完全不同的问题，用的技术手段也不一样，混在一起做很容易翻车。

这篇文章适合谁看？如果你正在做客服机器人、个人助理、教育辅导类 Agent，或者任何需要“记住用户”的产品，这套思路可以直接参考。如果你只是调 API 做个一次性问答，那暂时用不上，但了解一下架构思路也没坏处。我踩过的坑主要集中在：上下文窗口爆掉、长期记忆检索不准、记忆写入时机不对导致存了一堆垃圾。下面把这些东西拆开讲。

2. 记忆系统的整体架构设计

2.1 两层记忆的分工与边界

先把概念理清楚。Session Memory 和 Long-term Memory 不是简单的“短期 vs 长期”的关系，它们的数据结构、存储介质、读写时机都不一样。

Session Memory 本质上是当前对话的消息列表，加上一些从对话中实时提取的结构化信息（比如用户当前提到的地点、时间、任务状态）。它的生命周期就是这次会话，会话结束就可以丢弃。存储上通常放在内存或者 Redis 里，读写频率极高，要求毫秒级响应。

Long-term Memory 则是跨会话的持久化存储。它记录的是用户的稳定特征：偏好、身份信息、历史行为模式、重要事件。比如“这个用户是素食主义者”“上个月咨询过退款流程”“习惯用简洁风格回复”。这些信息不会因为一次会话结束就消失，需要写进数据库或者向量库，下次会话开始时再检索出来注入上下文。

两者的边界在哪里？我的经验是：当前会话内产生的、只对本次对话有意义的信息放 Session Memory；跨会话仍然有价值的信息才写 Long-term Memory。举个例子，用户说“帮我查一下明天杭州的天气”，这个“杭州”和“明天”是 Session Memory，会话结束就没用了。但如果用户说“我以后都在杭州”，那“用户常驻杭州”就该进 Long-term Memory。

2.2 为什么不能只做一层

有人会想，我直接把所有历史消息都塞进上下文不就行了？理论上可以，但实际做不到。原因有三个：

第一，上下文窗口有限。就算现在模型支持 128K 甚至更长的上下文，但每轮都塞几千条历史消息，token 成本会爆炸，响应速度也会明显下降。而且研究表明，上下文过长时模型对中间部分的注意力会衰减，关键信息反而被淹没。

第二，跨会话无法延续。上下文窗口再长，也是单次请求内的。用户今天聊完关掉页面，明天再来，上下文是空的。没有持久化层，Agent 永远只能记住“这一次”。

第三，信息密度问题。原始对话里大量内容是寒暄、确认、重复，真正有价值的用户信息可能只占 5%。全部保留既浪费资源又干扰检索。长期记忆需要的是提炼后的结构化信息，不是原始对话的堆砌。

所以两层架构不是可选项，是必选项。Session Memory 保证当前对话流畅，Long-term Memory 保证跨会话的个性化体验。

2.3 整体数据流设计

我采用的架构大致是这样的数据流：

用户发消息 → 加载 Session Memory（当前对话历史 + 会话级结构化状态）→ 检索 Long-term Memory（根据当前消息语义，从向量库/数据库中拉取相关用户记忆）→ 组装成完整 Prompt → 调用模型 → 得到回复 → 异步更新 Session Memory（追加消息、更新状态）→ 异步判断是否需要写入 Long-term Memory → 需要则提炼后写入。

这里有两个关键设计决策。一是Long-term Memory 的检索是“按需拉取”而不是“全量注入”。用户可能有几百条长期记忆，不可能全部塞进 Prompt，必须根据当前对话语义做相关性检索。二是记忆写入是异步的，不能阻塞主流程。用户等回复的时候，后台默默做记忆提炼和存储，不增加响应延迟。

3. Session Memory 的核心实现细节

3.1 消息列表的管理策略

Session Memory 最基础的部分就是消息列表。看起来简单，但有几个细节容易出问题。

首先是消息的裁剪策略。当对话轮次多了，消息列表会越来越长。我的做法是保留最近 N 轮完整对话（N 通常取 10-20），加上一个“历史摘要”。摘要由模型在对话过程中定期生成，把更早的对话压缩成一段简短描述。这样既保留了近期细节，又不丢失远期脉络。

具体实现上，我设置了一个阈值：当消息列表的 token 数超过 3000 时，触发摘要生成。把最早的一半消息交给模型，让它输出一段 200 字以内的摘要，替换掉原始消息。这个过程可以递归进行，摘要也可以再被摘要。

其次是消息角色的处理。system、user、assistant 三种角色的消息在裁剪时要区别对待。system prompt 永远保留，不能被裁掉。user 和 assistant 的消息按轮次成对裁剪，避免出现“只有用户问没有助手答”的断裂情况。

注意：裁剪时不要简单按 token 数截断，一定要按“对话轮次”为单位。我早期图省事按字符数截，结果经常把一轮对话从中间切断，模型看到半截问题直接懵了。

3.2 会话级结构化状态的维护

光有消息列表还不够。很多信息散落在对话里，每次都要模型重新理解一遍，效率低还容易出错。我的做法是维护一个会话级的状态字典，实时从对话中抽取关键信息。

比如一个订餐 Agent 的会话状态可能是这样的：

{ "current_task": "订晚餐", "location": "杭州", "cuisine_preference": "川菜", "budget": "100-150元", "dietary_restriction": "不吃香菜", "confirmed_items": ["水煮鱼", "麻婆豆腐"], "pending_question": "需要确认送达时间" }

这个状态字典怎么更新？两种方式结合。一是规则抽取，对于地点、时间、数字这类格式固定的信息，用正则或轻量 NER 模型直接抽。二是模型抽取，每轮对话结束后，把最新一轮消息和当前状态一起给模型，让它输出更新后的状态 JSON。规则抽取快且稳，模型抽取灵活但偶尔会抽风，两者互补。

状态字典的好处是，组装 Prompt 时可以直接把结构化状态放在 system 消息里，比让模型从一堆对话历史里自己找要可靠得多。而且状态字典很小，token 消耗可以忽略。

3.3 上下文组装的最佳实践

把 Session Memory 组装成最终 Prompt 时，顺序很重要。我实测下来比较稳的结构是：

1. System Prompt：角色定义、能力边界、输出格式要求
2. 用户画像摘要：从 Long-term Memory 检索到的相关信息，用自然语言描述
3. 会话状态：当前的结构化状态字典，JSON 格式
4. 历史摘要：早期对话的压缩摘要
5. 近期对话：最近 N 轮的完整消息
6. 当前用户输入

这个顺序的逻辑是：从稳定到动态，从概括到具体。模型先看到“你是谁、你面对的是谁”，再看到“当前在聊什么”，最后看到“用户刚说了什么”。实测这个顺序比把用户画像放在最后效果更好，因为模型在生成回复时已经充分“进入角色”了。

另外一个小技巧：在用户画像摘要部分，用第二人称描述比第三人称效果好。比如写“你是一位常驻杭州的用户，偏好川菜，不吃香菜”，比写“用户常驻杭州，偏好川菜”更能让模型产生代入感，回复的个性化程度明显提升。

4. Long-term Memory 的存储与检索

4.1 记忆的提炼与写入时机

Long-term Memory 最大的坑是写太多垃圾。如果每轮对话都往长期记忆里塞东西，很快库就脏了，检索出来的全是无关信息。我的策略是按事件触发写入，而不是按轮次。

触发写入的时机有这么几个：

• 会话结束时：对整个会话做一次总结，提炼出值得长期保留的信息
• 检测到明确的用户偏好表达时：比如“我喜欢”“我习惯”“以后都”这类句式
• 检测到重要事实时：比如用户告知了姓名、职业、地址等身份信息
• 检测到任务完成时：比如一次订餐、一次咨询结束，把结果和用户反馈存下来

写入前必须经过提炼这一步。不能把原始对话直接存进去，要让模型输出结构化的记忆条目。我用的格式是这样的：

{ "memory_type": "preference", "content": "用户偏好川菜，不吃香菜", "confidence": 0.9, "source_session": "sess_20240115_001", "created_at": "2024-01-15T19:30:00Z", "last_accessed": "2024-01-15T19:30:00Z", "access_count": 0 }

memory_type我分了几个大类：preference（偏好）、fact（事实）、event（事件）、relationship（关系）。不同类型在检索时的权重和时效性处理不一样。比如 preference 通常长期有效，event 可能有过期时间。

实操心得：写入时加一个confidence字段很有用。模型提炼的记忆不一定百分百准确，比如用户说“今天不太想吃辣”，这到底是一时兴起还是长期偏好？让模型给个置信度，低于阈值的先存着但检索时降权，等多次出现类似表述再提升置信度。这个机制能有效过滤噪声。

4.2 向量检索与关键词检索的混合方案

Long-term Memory 的检索，纯向量检索和纯关键词检索都有问题。向量检索擅长语义匹配，但容易召回“意思相近但实际无关”的记忆；关键词检索精确，但用户换个说法就匹配不上了。

我的方案是混合检索 + 重排序。具体流程：

第一步，向量检索。把用户当前输入 embedding 后，在向量库里做相似度搜索，取 Top 20。向量库我用的 FAISS 本地索引，数据量大了可以换 Milvus 或 Qdrant。

第二步，关键词检索。对当前输入做分词，提取关键词，在记忆库的 content 字段做全文检索，同样取 Top 20。这一步用 Elasticsearch 或者简单的 SQL LIKE 都能做。

第三步，合并去重。两路结果合并，按记忆 ID 去重。

第四步，重排序。用一个轻量交叉编码器（cross-encoder）对合并后的候选做精排，输出最终 Top 5。交叉编码器比向量相似度准得多，但计算量大，所以只用在精排阶段。

第五步，时效性加权。根据last_accessed和access_count做微调。最近被访问过的记忆适当提权，很久没被访问的降权。但注意，偏好类记忆不应该因为长期没访问就降权，用户不说≠偏好变了。

这套流程实测下来，检索准确率比单路方案提升明显。代价是延迟增加了大概 50-80ms，对于非实时场景完全可以接受。

4.3 记忆的更新与遗忘机制

记忆不是只增不减的。用户偏好会变，事实会过时，无效记忆会越积越多。需要一套更新和遗忘机制。

更新方面，当新记忆和旧记忆冲突时，不能简单覆盖。我的做法是保留历史版本，但把旧版本标记为superseded，检索时默认不返回，除非用户明确问“我以前是不是说过……”。这样既保证了当前信息的准确性，又保留了追溯能力。

遗忘方面，我设了几个规则：

• 事件类记忆：超过 90 天且未被访问的，归档到冷存储，不参与常规检索
• 偏好类记忆：不主动遗忘，但如果检测到用户明确改变偏好（“我现在不吃辣了”），旧偏好标记为失效
• 低置信度记忆：如果一条记忆创建后 30 天内从未被检索命中，且置信度低于 0.6，直接删除

注意：遗忘机制一定要有，但不能太激进。我早期设了个“30天未访问就删除”，结果把一个用户三个月前说的“我对花生过敏”给删了，后来推荐餐厅时差点出事。涉及安全、健康、身份类的记忆，我后来加了白名单，永不自动删除。

5. 记忆系统的完整实操流程

5.1 环境准备与依赖选型

先说一下我用的技术栈，不是唯一方案，但经过实际验证比较稳：

• Session Memory 存储：Redis，用 Hash 结构存会话状态，List 存消息队列
• Long-term Memory 存储：PostgreSQL 存结构化记忆元数据 + FAISS 存向量索引
• Embedding 模型：BGE-M3，中文效果好，维度 1024，本地部署无 API 成本
• 重排序模型：BGE-Reranker-Base，轻量，CPU 上单次推理约 30ms
• 记忆提炼模型：用主模型即可，通过 Prompt 控制输出格式

安装依赖：

pip install redis psycopg2-binary faiss-cpu sentence-transformers pip install FlagEmbedding # BGE 系列模型

Redis 和 PostgreSQL 用 Docker 起就行，不赘述。FAISS 索引我建议用IndexFlatIP做基础版本，数据量超过 10 万条再换 IVF 索引。

5.2 Session Memory 的代码实现

先定义会话状态的数据结构：

import json import redis from datetime import datetime class SessionMemory: def __init__(self, redis_client, session_id, max_recent_turns=15): self.redis = redis_client self.session_id = session_id self.max_recent_turns = max_recent_turns self.msg_key = f"session:{session_id}:messages" self.state_key = f"session:{session_id}:state" self.summary_key = f"session:{session_id}:summary" def add_message(self, role, content): msg = json.dumps({ "role": role, "content": content, "ts": datetime.utcnow().isoformat() }) self.redis.rpush(self.msg_key, msg) self.redis.expire(self.msg_key, 86400) # 24小时过期 self._maybe_summarize() def get_recent_messages(self): raw = self.redis.lrange(self.msg_key, -self.max_recent_turns*2, -1) return [json.loads(m) for m in raw] def get_state(self): state = self.redis.hgetall(self.state_key) return {k.decode(): json.loads(v) for k, v in state.items()} def update_state(self, key, value): self.redis.hset(self.state_key, key, json.dumps(value)) self.redis.expire(self.state_key, 86400)

_maybe_summarize方法负责在消息过多时触发摘要：

def _maybe_summarize(self): msg_count = self.redis.llen(self.msg_key) if msg_count <= self.max_recent_turns * 2: return # 取出最早的一半消息做摘要 old_msgs = self.redis.lrange(self.msg_key, 0, msg_count // 2 - 1) old_text = "\n".join( f"{json.loads(m)['role']}: {json.loads(m)['content']}" for m in old_msgs ) # 调用模型生成摘要（这里省略模型调用细节） new_summary = self._call_summarize_model(old_text) # 更新摘要，删除旧消息 self.redis.set(self.summary_key, new_summary) self.redis.ltrim(self.msg_key, msg_count // 2, -1)

这里有个细节：摘要不是覆盖式的，而是累积式的。新的摘要应该包含旧摘要的内容加上新压缩的部分。我在_call_summarize_model里会把旧摘要一起传进去，让模型输出合并后的摘要。

5.3 Long-term Memory 的写入流程

写入流程分三步：触发判断、信息提炼、存储。

触发判断用一个轻量分类器或者规则引擎。我用的规则 + 模型混合：

def should_write_longterm(user_input, assistant_reply, session_state): # 规则1：明确的偏好表达 preference_patterns = ["我喜欢", "我习惯", "我偏好", "以后都", "我一直"] if any(p in user_input for p in preference_patterns): return True # 规则2：身份信息 identity_patterns = ["我是", "我叫", "我的职业", "我住在"] if any(p in user_input for p in identity_patterns): return True # 规则3：会话结束信号 if session_state.get("session_ending"): return True # 规则4：模型判断（兜底） return _model_judge(user_input, assistant_reply)

信息提炼的 Prompt 大致长这样：

你是一个记忆提炼助手。请从以下对话中提取值得长期记住的用户信息。 只提取关于用户的稳定特征、偏好、事实，不要提取临时性信息。 输出 JSON 数组，每个元素包含 type、content、confidence 三个字段。 type 可选值：preference、fact、event、relationship。 confidence 为 0-1 的浮点数。 对话内容： 用户：{user_input} 助手：{assistant_reply} 输出：

存储时，除了写 PostgreSQL，还要同步更新 FAISS 索引：

def store_memory(memory_item, embedding_model, faiss_index, pg_conn): # 写 PostgreSQL with pg_conn.cursor() as cur: cur.execute(""" INSERT INTO memories (type, content, confidence, created_at, last_accessed, access_count) VALUES (%s, %s, %s, NOW(), NOW(), 0) RETURNING id """, (memory_item["type"], memory_item["content"], memory_item["confidence"])) memory_id = cur.fetchone()[0] # 写 FAISS vec = embedding_model.encode(memory_item["content"]) faiss_index.add_with_ids(vec.reshape(1, -1), np.array([memory_id])) return memory_id

5.4 检索与注入的完整链路

检索链路我在 4.2 节讲了思路，这里给核心代码：

def retrieve_memories(query, embedding_model, faiss_index, pg_conn, top_k=5): # 向量检索 query_vec = embedding_model.encode(query) vec_scores, vec_ids = faiss_index.search(query_vec.reshape(1, -1), 20) # 关键词检索 keywords = extract_keywords(query) kw_ids = keyword_search(pg_conn, keywords, limit=20) # 合并去重 candidate_ids = list(set(vec_ids[0].tolist() + kw_ids)) # 拉取记忆内容 candidates = fetch_memories_by_ids(pg_conn, candidate_ids) # 重排序 reranked = rerank_model.rerank(query, [c["content"] for c in candidates]) top_memories = [candidates[i] for i in reranked[:top_k]] # 更新访问记录 update_access_stats(pg_conn, [m["id"] for m in top_memories]) return top_memories

注入 Prompt 时，把检索到的记忆格式化成自然语言：

def format_memories_for_prompt(memories): if not memories: return "" lines = ["关于当前用户，你已知以下信息："] for m in memories: lines.append(f"- {m['content']}") return "\n".join(lines)

这段文本放在 system prompt 之后、会话状态之前，效果最好。

6. 常见问题与排查技巧实录

6.1 记忆检索不准的排查思路

检索不准是最常见的问题，表现是“明明存了相关信息，但就是没检索出来”。排查按这个顺序来：

第一步，确认记忆是否真的写进去了。直接查 PostgreSQL，看 content 字段有没有目标信息。我遇到过因为提炼 Prompt 写得太严格，模型把用户偏好当成“临时信息”过滤掉了，根本没入库。

第二步，检查 embedding 质量。把查询和记忆内容分别 encode，算余弦相似度。如果相似度低于 0.5，说明 embedding 模型对这类文本区分度不够。中文场景下 BGE-M3 一般够用，但如果是专业领域术语，可能需要微调。

第三步，看关键词检索有没有命中。如果向量检索没中但关键词检索中了，说明是语义匹配的问题；如果两路都没中，说明查询和记忆的表述差异太大，需要在提炼阶段做同义词扩展。

第四步，检查重排序是否把正确结果排下去了。把重排序前后的顺序打出来对比，如果正确记忆在向量检索里排第 3，重排序后掉到第 10，说明重排序模型和你的场景不匹配，考虑换模型或者调整权重。

6.2 上下文超长的应急处理

有时候用户一次输入特别长，或者检索回来的记忆特别多，导致 Prompt 超出模型限制。应急处理方案：

实操心得：我建议在组装 Prompt 前先算一下 token 数，超过模型限制的 80% 就主动触发压缩，不要等到报错了再处理。用 tiktoken 算 token 很快，开销可以忽略。

6.3 记忆冲突与错误记忆的修正

用户说“我搬到上海了”，但长期记忆里存着“用户常驻杭州”。这种冲突怎么处理？

我的方案是新记忆写入时做冲突检测。在写入前，先用新记忆的内容去检索现有记忆，如果发现同类型、高相似度的旧记忆，就把新旧一起给模型，让它判断是“更新”“补充”还是“无关”。

def resolve_conflict(new_memory, existing_memories): if not existing_memories: return "insert" prompt = f""" 新记忆：{new_memory['content']} 已有记忆：{[m['content'] for m in existing_memories]} 请判断新记忆与已有记忆的关系，输出以下之一： - update：新记忆取代旧记忆 - supplement：新记忆是对旧记忆的补充 - unrelated：两者无关 只输出判断结果。 """ result = call_model(prompt).strip() return result

如果是update，把旧记忆标记为superseded，新记忆正常写入。如果是supplement，两条都保留。如果是unrelated，正常写入。

错误记忆的修正则靠用户反馈。如果用户说“你记错了，我不吃辣”，触发一个修正流程：检索相关记忆，标记为错误，写入正确信息。这个流程要做得轻量，不要让用户填表单，直接从对话里提炼就行。

6.4 性能优化的几个关键点

记忆系统做不好，延迟会很难看。我实测下来，几个优化点效果最明显：

Embedding 缓存。相同文本的 embedding 结果缓存起来，避免重复计算。用户查询往往有重复，缓存命中率能到 30% 以上。

FAISS 索引预热。服务启动时把索引加载到内存，不要每次查询都从磁盘读。FAISS 的IndexFlatIP加载 10 万条 1024 维向量大约占 400MB 内存，可以接受。

异步写入。Long-term Memory 的写入全部走消息队列异步处理，不阻塞主流程。我用 Redis 的 List 做简单队列，写入请求 push 进去，后台 worker 消费。

批量检索。如果一次请求需要检索多个查询（比如同时检索偏好和事实），合并成一次 FAISS 搜索，减少 IO 次数。

重排序模型量化。BGE-Reranker 用 ONNX 量化后，CPU 推理速度能提升 2-3 倍，精度损失很小。

7. 一些踩坑后的个人体会

这套记忆系统我从零搭到现在稳定运行，前后迭代了四五版。最大的体会是：记忆系统的难点不在技术，在于“判断什么值得记”。向量库、检索算法这些都是成熟方案，拿来用就行。但“用户说的这句话到底是不是长期偏好”“这个信息三个月后还有没有用”，这种判断需要结合具体业务场景反复调。

我早期犯的错是“宁可多记不可漏记”，结果记忆库膨胀到几万条，检索出来的全是噪声，反而干扰了模型判断。后来改成“宁缺毋滥”，只记高置信度的稳定信息，效果反而好了很多。用户其实不需要 Agent 记住所有事，只需要记住那些真正影响体验的关键信息。

另一个体会是记忆的透明度很重要。用户应该能知道 Agent 记住了什么，也应该能删除或修改。我在产品里加了一个“记忆管理”页面，用户可以看到所有被记住的信息，一键删除。这个功能上线后，用户对 Agent 的信任度明显提升。毕竟，谁也不想跟一个偷偷记小本本的助手打交道。

最后分享一个实用技巧：新会话开始时，不要一次性把所有长期记忆都注入。我的做法是先注入一个“用户画像摘要”（由系统定期生成，概括用户的核心特征），然后根据对话进展按需检索具体记忆。这样既保证了开场就有个性化，又避免了信息过载。摘要的更新频率我设的是每周一次，或者用户主动修改记忆时触发。