☰
Agent记忆架构全链路实战:从会话窗口到遗忘机制
2026/10/6 17:57:04 网站建设 项目流程

前阵子有个做企业知识库系统的朋友跟我抱怨:他的Agent明明接了大模型API,但用户多聊几轮就"失忆"——用户刚在前文确认过的订单编号,下一轮就要重新问一遍;昨天刚聊完的技术方案,今天换个会话打开就完全不记得。这种"健忘症"不是个例,几乎所有商用Agent在没有专门做记忆架构之前都会栽在这上面。

问题的根源在于:大模型本身是无状态的,每次API调用都是一次独立的推理过程,模型根本不知道上一次对话发生了什么。你看到的「连续对话」感,其实完全靠上层应用把历史消息一遍遍塞进Prompt里。这就像一个人每次见你都被格式化记忆,全靠你递给他一张纸条回忆你们上次聊了什么。如果不系统性地设计会话记忆、短期记忆、长期记忆和遗忘机制,Agent在真实业务场景里的可用性会非常差。

这篇文章我直接用一段完整的Python代码实战,拆解商用Agent记忆架构的全链路实现——从最基础的会话窗口管理,到短期状态快照,再到长期向量/结构化记忆,最后是很多人最容易忽略的遗忘机制。读者对象是对Agent开发已经上手的工程师,或者准备把Agent推向生产环境、正在纠结记忆模块怎么设计的团队。

1. 健忘的根源:不是模型不行,是没有记忆层

1.1 大模型的"金鱼脑"到底是怎么造成的

先彻底说清楚一件事:你调GPT、Claude、文心这类模型的接口时,模型本身是完全无状态的。你发一个请求,它返回一个结果,然后就结束了,服务器不会偷偷把你的对话存下来给下次用。所谓"多轮对话能力",是平台SDK或你自己在应用层做了历史消息拼接,把前几轮对话按角色排好,当作Prompt的一部分发给模型。

这意味着什么?意味着"记忆"这件事,从架构层面就落在了你身上。如果你不维护消息列表,模型每次都是"睁眼瞎":

  • 用户上一句说的关键实体(订单号、客户名、需求点),模型无从知晓
  • 跨会话的偏好和长期目标,模型一概不知
  • 甚至同一个会话内,如果消息列表超过上下文窗口,早期信息被截断,模型就会"选择性忘记"

我见过很多团队第一版就是「把所有消息全部塞进messages数组,不做任何裁剪」,然后跑到第20轮对话时接口直接报错——上下文超限。这不算健忘,更准确说是"爆仓"。但用户体感上是一样的:Agent聊着聊着就不对劲了。

1.2 商用记忆架构的四个层次

要根治"健忘",需要在应用层补一套分层记忆系统。我用大白话类比一下:人的记忆也不是只靠一个脑子,而是分感官记忆、工作记忆、长时记忆,并且有遗忘机制在调节。Agent的记忆架构可以对应拆成四层:

记忆层对应人脑存储介质生命周期解决什么问题
会话记忆(Session Memory)感官记忆/短期工作记忆进程内list,附着在每次请求里当前对话持续期间让模型知道"本轮聊到哪了"
短期记忆(Working Memory)工作记忆结构化状态对象、摘要任务进行期间,跨请求存在让模型记住"当前任务做了一半的状态"
长期记忆(Long-term Memory)长时记忆数据库/向量库天、周、月级别让模型记住"这个用户是谁、上次聊过什么"
遗忘机制(Forgetting)遗忘定时任务/评分策略按策略触发防止记忆膨胀、污染和过时信息干扰

很多入门教程只讲了第一层,也就是"把消息塞回上下文",但商用系统真正拉开差距的是后三层。尤其是遗忘机制,绝大多数团队做都没做,导致记忆库无限膨胀、检索噪音越来越大,最后Agent反而被"不需要的记忆"拖累,回答质量越来越差。

后面我就按这四层逐一写代码,最后给一个打通全链路的MemoryManager整合类。

2. 会话记忆:先把"这一轮对话"管明白

2.1 为什么直接堆messages列表是"废柴"方案

会话记忆是最基础的一层,也是最容易出现低级错误的地方。早期Demo大家都这么写:

messages.append({"role": "user", "content": user_input}) response = llm.chat(messages) messages.append({"role": "assistant", "content": response})

这一段代码跑不了几轮就完蛋。原因有两点:

第一,上下文窗口是硬限制。以8K窗口的模型为例,去掉系统提示词和预留的输出空间,实际能塞的输入历史可能只有6000左右。一个长篇文档复制粘贴进来可能就占掉一半。不做预算控制,迟早爆掉。

第二,历史消息不是等价的。对话早期那些寒暄、确认、试错内容,对当前轮推理的参考价值很低,但还占着宝贵的Token预算。更合理的方式是:优先保留最近的N轮对话,早期内容做压缩摘要,而不是一股脑全留着。

2.2 滑动窗口裁剪实战

我建议的会话记忆模块至少要做到两件事:全局Token预算监控和滑动窗口裁剪。Token数估算在没有分词器的环境里可以用一个简单经验公式:英文按4字符约等于1 Token,中文按1到1.5字符约等于1 Token。当然更精确可以接模型自己的tokenizer,但工程上为了性能,先用估算函数做粗粒度控制完全够。

import time import json def estimate_tokens(text: str) -> int: """粗粒度Token估算:中文按1.5字符/token,英文按4字符/token""" if not text: return 0 total_chars = len(text) # 简单分词统计更是性能开销,直接用字符级估算即可 return int(total_chars / 1.5) class SessionMemory: def __init__(self, max_context_tokens: int = 6000, reserved_output_tokens: int = 2000): self.messages = [] self.max_context_tokens = max_context_tokens self.reserved_output_tokens = reserved_output_tokens # 系统级消息单独存,不允许被裁剪掉 self.system_messages = [] def add_system_message(self, content: str): self.system_messages.append(content) def add_message(self, role: str, content: str): self.messages.append({"role": role, "content": content, "ts": time.time()}) def _total_tokens(self) -> int: system_tokens = sum(estimate_tokens(m) for m in self.system_messages) history_tokens = sum(estimate_tokens(m["content"]) for m in self.messages) return system_tokens + history_tokens def build_prompt_messages(self) -> list: """组装最终发给模型的messages""" budget = self.max_context_tokens - self.reserved_output_tokens system_tokens = sum(estimate_tokens(m) for m in self.system_messages) remaining = budget - system_tokens selected = [] # 从最近的消息往前取,直到预算耗尽 for msg in reversed(self.messages): msg_tokens = estimate_tokens(msg["content"]) if remaining - msg_tokens < 0: break selected.append(msg) remaining -= msg_tokens selected.reverse() final_messages = [] for sys_content in self.system_messages: final_messages.append({"role": "system", "content": sys_content}) final_messages.extend(selected) return final_messages

这段代码的核心逻辑是从后往前取消息,优先保留最新的上下文,因为离当前轮次越近的消息对推理的影响越大。早期消息被挤出budget后,直接丢掉,等后续章节的"摘要滚动"机制接管去压缩它们。注意我把system_messages单独存了,因为系统指令一旦被裁掉,Agent的人设和能力设定就崩了,这是新手最容易踩的雷区。

2.3 裁剪时的"关键信息抢救"

滑动窗口简单粗暴但有个副作用:如果早期对话里藏着用户明确说过的硬性约束(比如"不要用短信验证码"、"价格必须含税"),一旦被裁掉,后面Agent就可能违反约定。所以商用级的会话记忆必须在裁剪前做一次关键信息抢救。

我常用的方案是写一个约束提取器,用轻量规则或正则去扫描早期消息,把"用户明确表达的偏好/禁止项"捞出来,附加到系统消息或单独的关键信息卡片里:

preference_patterns = [ (r"(?:不要|别|禁止|avoid).{0,30}(?:使用|用|采用)", "prohibited"), (r"(?:记住|务必|一定|必须).{0,30}", "required"), (r"我(?:喜欢|偏好|希望).{0,30}", "preference"), ] def extract_key_facts(messages: list) -> list: facts = [] for msg in messages: if msg["role"] != "user": continue content = msg["content"] for pattern, fact_type in preference_patterns: matches = re.findall(pattern, content) for m in matches: facts.append({"type": fact_type, "text": m, "source_ts": msg.get("ts")}) return facts # 在build_prompt_messages里,把这些facts合并成一张"用户约束卡片"

注意这里不需要做得特别重,regex+固定句式就能覆盖大部分场景。真正的语义级信息抽取可以交给后面的短期/长期记忆层做,会话层只负责兜底。我的原则是:越底层越轻量,防止性能浪费;越上层越智能,体现业务价值。

3. 短期记忆:让Agent记得"刚才做了一半的事"

3.1 短期记忆的两种形态:结构化状态与摘要滚动

会话窗口再大也有边界。真实业务里用户经常聊着聊着岔开话题又绕回来,期间几十轮对话塞进窗口不现实。这时候需要短期记忆层,它有两大职能:状态快照和摘要滚动。

状态快照解决的是"当前任务做了一半"的问题。举例:用户在表单填写流程里已经填好了姓名和电话,下一步是要填地址。你不可能每次请求都让用户把所有信息重新说一遍,而是应该把一个结构化的表单状态存起来,每次请求时注入到Prompt里。

摘要滚动解决的是"早期对话太长"的问题。当历史消息超过某个阈值(比如窗口预算的60%),就把最早的一批消息交给LLM生成一段摘要,用摘要代替原始消息。模型看到的是"用户已确认了退款方案,等待商家处理"这样一句高度浓缩的话,而不是那段冗长的来回确认过程。

3.2 摘要滚动实战代码

我封装一个Summarizer类,触发条件是历史消息占用的Token数超过阈值。为了控制成本,摘要不是每一轮都做,而是设置了水位线,低水位不触发,高水位才触发,中间留一个缓冲带:

class ShortTermMemory: def __init__(self, llm_func, summarize_threshold_tokens: int = 2000): self.llm_func = llm_func # 传入一个调用LLM的函数 self.summarized_text = "" self.low_watermark = summarize_threshold_tokens * 0.6 self.high_watermark = summarize_threshold_tokens self.full_history = [] def compress_if_needed(self, history: list) -> str: """当历史超过阈值时,对最早的N条消息做摘要,保留压缩结果""" current_tokens = sum(estimate_tokens(m["content"]) for m in history) if current_tokens < self.high_watermark: return self.summarized_text # 只压缩最早的50%历史,保留最近的消息用于精确推理 to_summarize = history[: len(history) // 2] messages = [ {"role": "system", "content": "请用不超过150字总结这段对话的核心信息,包括用户的明确决定、关键数据和任务进度。保持客观,不要添加推断。"} ] for m in to_summarize: messages.append(m) summary = self.llm_func(messages) self.summarized_text = summary # 触发压缩后,这半个会话被摘要取代,最粗粒度的实现可以直接丢弃 return self.summarized_text

这里有个关键设计:只压缩最早的50%,而不是一次性全量压缩。原因是最近几轮的原始对话包含大量逻辑细节,比如用户在最后补充了"发票抬头改成公司名",这种信息一旦被摘要压缩,模型可能记不精确。而早期那些背景信息、寒暄、多次往返确认,压缩掉影响不大。

3.3 状态快照:比摘要更可靠的"结构化记忆"

摘要毕竟是自然语言,模型理解会打折扣。对于业务系统,我更推荐把关键状态结构化存储。比如客服场景的工单处理状态、多轮表单的填写进度、搜索场景的筛选条件,这些用JSON比用纯文本更可靠:

class StateSnapshot: """维护一个可注入Prompt的结构化状态卡片""" def __init__(self, initial_state: dict = None): self.state = initial_state or {} def update(self, key: str, value): self.state[key] = value def get_state_card(self) -> str: """把状态渲染成固定格式卡片,注入system prompt""" if not self.state: return "" card_lines = ["当前任务状态记录(这是系统维护的结构化数据,请优先参考):"] for k, v in self.state.items(): card_lines.append(f"- {k}: {v}") card_lines.append("如果用户提到以上字段,说明该值已被用户确认,无需再次询问。") return "\n".join(card_lines)

实际调用时,把这张"状态卡片"塞进system message里,模型就能准确知道"这个表单已经填到哪一步了"。文字描述可能引发歧义,但结构化键值对不会——"表单地址字段:UNFILLED"比 "用户还没填地址" 清楚得多。

这里分享一个我踩过的坑:状态快照一开始只记录了last_user_input,结果模型经常把上一轮的用户消息当作当前轮继续处理,逻辑全乱了。后来我把状态改成"当前意图+已完成字段+等待字段"三段式,模型的上下文理解才稳定下来。核心教训是:短期记忆的状态字段要面向"任务进度"设计,而不是面向"聊天内容"设计。

4. 长期记忆:跨天跨周的"往事回忆"怎么存怎么取

4.1 从向量库到轻量SQLite的选型考量

长期记忆是商用Agent区别于Demo的关键。用户今天问过的偏好、上周提交过的资料、三个月前购买过的产品型号,如果全压在会话记忆里,任何窗口都扛不住。所以长期记忆必须落地到外部存储,按需检索、按需注入。

存储选型上,国内团队落地时我见过两个主流方向:一是上向量数据库(Milvus/pgvector/Redis向量扩展)+ Embedding模型,适合大厂有GPU资源的场景;二是在中小项目里用SQLite/MySQL+结构化字段存记忆条目,用关键词或简单的Embedding检索。第二个方案成本极低,效果在多数垂直场景里足够。

这里我给出一个不依赖外部矢量库的简化方案,用SQLite存记忆条目,配合关键词检索做TopK召回。它的优点是零额外依赖,生产环境也能跑,缺点是语义检索能力弱。如果你的团队已经有了向量库,替换检索函数即可,上层MemoryManager的接口不用动。

4.2 长期记忆表结构与写入函数

我们设计的记忆表有这些字段:user_id区分用户,content存记忆正文,keywords存搜索关键词(从文本里抽出来的实体词),importance给遗忘机制用,last_access和access_count做热度统计。建表SQL和写入函数如下:

import sqlite3 import json SCHEMA = """ CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL, content TEXT NOT NULL, keywords TEXT DEFAULT '', importance REAL DEFAULT 0.5, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_access TIMESTAMP DEFAULT CURRENT_TIMESTAMP, access_count INTEGER DEFAULT 0 ); CREATE INDEX idx_memories_user ON memories(user_id); CREATE INDEX idx_memories_user_keywords ON memories(user_id, keywords); """ def init_db(db_path="agent_memory.db"): conn = sqlite3.connect(db_path) conn.executescript(SCHEMA) conn.commit() return conn def write_memory(conn, user_id: str, content: str, keywords: list, importance: float = 0.5): cursor = conn.execute( "INSERT INTO memories(user_id, content, keywords, importance) VALUES(?,?,?,?)", (user_id, content, json.dumps(keywords, ensure_ascii=False), importance) ) conn.commit() return cursor.lastrowid

写入时机很讲究。不是用户每轮说话都往长期记忆里写,那样记忆库会爆炸。我常用的写入策略是:在任务结束、会话结束或检测到"重要决策"时才写入。检测重要决策可以看用户是否用了"定了/就这样/确认/决定"这类词,也可以用LLM做一个轻量判断。写少了无所谓,写多了才是灾难。

4.3 关键词检索与TopK召回

召回是长期记忆的灵魂。检索时先把主检索条件(当前用户消息中的核心名词)和keywords字段做SQLite的LIKE匹配,取出候选集,再按相关性+重要度+新鲜度综合排序,最后取TopK注入上下文:

def recall_memories(conn, user_id: str, query_keywords: list, top_k: int = 5): """基于关键词召回TopK长期记忆""" if not query_keywords: return [] like_conditions = " OR ".join(["keywords LIKE ?"] * len(query_keywords)) params = [f"%{kw}%" for kw in query_keywords] + [user_id] sql = f""" SELECT id, content, importance, last_access, access_count FROM memories WHERE user_id = ? AND ({like_conditions}) ORDER BY importance DESC, last_access DESC LIMIT ? """ cursor = conn.execute(sql, params[::-1]) # 注意参数顺序 rows = cursor.fetchall() results = [] now = time.time() for row in rows: mem_id, content, importance, last_access, access_count = row # 时间衰减因子:最近访问的略微加权 hours_since = (now - datetime.strptime(last_access, "%Y-%m-%d %H:%M:%S").timestamp()) / 3600.0 recency_boost = 1.0 / (1.0 + hours_since / 24.0) score = 0.6 * importance + 0.2 * recency_boost + 0.2 * min(access_count / 10.0, 1.0) results.append({"content": content, "score": score, "id": mem_id}) # 更新访问信息 conn.execute("UPDATE memories SET last_access = CURRENT_TIMESTAMP, access_count = access_count + 1 WHERE id = ?", (mem_id,)) conn.commit() results.sort(key=lambda x: x["score"], reverse=True) return results[:top_k]

这里我加了分数排序,分数由三部分构成:重要度权重最高(0.6),访问新鲜度次之(0.2),访问频次兜底(0.2)。这样既保留了用户明确标记的重要信息,又能让频繁问过的热点内容排前面,同时给最近刚写过的新记忆一定曝光机会。单纯按时间倒序或者按重要度倒序都会有偏科问题,经验是综合评分更稳。

召回结果的注入点放在系统提示词里最合适。我会拼一个"用户长期记忆参考"区块,告诉模型:这些是用户历史档案,如果当前问题无直接相关可以忽略,如果相关则优先参考。注意用词——"可以忽略"这四个字很重要,否则模型容易强行把不相关的历史记忆编进回答里。

5. 遗忘机制:会忘的Agent才是好Agent

5.1 "记住一切"的危害,你可能还没意识到

大多数团队做记忆系统时只做加法不做减法,结果记忆库越来越大。危害有三:

第一,Token膨胀成本激增。你每次把TopK记忆注入上下文,如果TopK设得太大,光记忆就吃掉上千Token,成本直线上升。

第二,检索噪音污染推理。记忆库里存了大量过期信息,比如用户半年前说"我住北京",现在搬去了上海。要是Agent把半年前的旧地址当作当前事实用,那比没有记忆还糟。

第三,隐私合规风险。用户明确说"不要再提我信用卡的事",结果长期记忆里还留着,你拿用户全量历史去喂模型,法律上也有风险。

所以遗忘机制不是锦上添花,是记忆系统的必修边界。商用架构里要说有什么功能最能体现成熟度,遗忘机制绝对排第一。

5.2 TTL过期与访问频次整合

最简单的遗忘策略是TTL(生存时间)。每条记忆写入时设一个有效期,比如30天,到期后后台任务自动清理或降权。但这太粗暴——有些记忆虽然老但是关键决定(比如"公司主体是X科技"),有效期过了忘了就麻烦了。

我建议用**TTL+访问频次+LRI(Least Recently Important)**三者结合。核心思想是:记忆的重要性会随着时间衰减,但被频繁访问的记忆会"续命"。实现上不删数据,而是给每条记忆维护一个新鲜度分数,每次召回时更新,遗忘清扫按分数清理:

def forget_score(importance, last_access_days, access_count, base_ttl_days=45): """遗忘评分:分数越低越优先被清理""" time_penalty = min(last_access_days / base_ttl_days, 1.0) access_bonus = min(access_count / 20.0, 0.5) score = importance * (1 - time_penalty) + access_bonus return score def cleanup_memories(conn, user_id: str, max_keep: int = 200, min_score: float = 0.15): """记忆清理:保留条数上限+最低分数双阈值""" rows = conn.execute( "SELECT id, content, importance, last_access, access_count FROM memories WHERE user_id = ?", (user_id,) ).fetchall() scored = [] for row in rows: mem_id, content, importance, last_access, access_count = row # 解析last_access并计算天数 last_dt = datetime.strptime(last_access, "%Y-%m-%d %H:%M:%S") days = (datetime.now() - last_dt).days score = forget_score(importance, days, access_count) scored.append((score, mem_id, content)) scored.sort(key=lambda x: x[0]) # 低于最低分数的先删除 to_delete = [s[1] for s in scored if s[0] < min_score] # 如果仍然超过max_keep,从低分开始继续删 if len(scored) - len(to_delete) > max_keep: excess = len(scored) - len(to_delete) - max_keep to_delete.extend([s[1] for s in scored if s[1] not in to_delete][:excess]) for mem_id in to_delete: conn.execute("DELETE FROM memories WHERE id = ?", (mem_id,)) conn.commit() return len(to_delete)

这段代码的关键点是:importance是遗忘评分的基础分,"一段时间没被访问"会拉低分数,"经常被访问"会把分数救回来。这样即使一条记忆看起来很重要,但如果半年都没被用到过一次,也该让位给新知识和当前活跃内容。

5.3 主动遗忘与用户删除接口

被动清理之外,还有一个产品层面的诉求:用户主动要求遗忘。你不能让Agent记住了用户想删除的东西,然后还理直气壮地回"我没有删除功能"。商用系统至少提供两个接口:

  • delete_memory_by_id(user_id, memory_id):删除单条记忆
  • clear_user_memories(user_id):清空某个用户全部记忆

一个更进阶的用法是让Agent在对话里识别用户的"遗忘指令",比如用户说"忘了刚才说的价格吧"或者"把上个月聊的方案删掉",Agent通过工具调用来执行删除。实现时把"delete_user_memory"注册成Agent的一个tool即可,这里不多展开。

我还想强调一个容易被坑的点:遗忘不能只在长期记忆层做,会话记忆层也要做。比如用户明确说"换个话题,别再提我孩子的事了",你可以在会话层把这个话题相关的早期消息标记为disabled,后续build_prompt_messages组装时直接跳过。否则模型看到的上下文中还有相关内容,就可能又绕回去。

6. 全链路整合:MemoryManager让Agent不再健忘

6.1 整合的数据流设计

到这里,四层能力都齐了,但真正要让Agent"不易健忘",还得把它们捏成一个对外统一的门面类,让业务流程只关心一件事:组装Prompt之前,先把记忆查好拼好;每一轮对话结束后,把新的信息按规则写回各层。

完整数据流如下:

  1. 用户消息进来,先更新会话记忆(追加到messages列表)
  2. 从长期记忆层召回TopK相关历史记忆
  3. 从短期记忆层取出当前任务的状态卡片
  4. 检查短期记忆是否需要触发摘要压缩
  5. 把系统提示词(含状态卡片)、长期记忆参考区块、会话摘要、最近对话消息,按顺序拼成最终的messages
  6. 调用LLM获取回复
  7. 回复追加进会话记忆,同时执行记忆写入规则(重要信息写长记、状态变更更新快照、刷新访问热度)

6.2 MemoryManager核心代码

我把上面三层的类全部装进一个门面类,对外接口就两个:process_user_message和build_prompt。这样业务侧调用非常干净:

class MemoryManager: def __init__(self, user_id: str, llm_func, db_path="agent_memory.db"): self.user_id = user_id self.llm_func = llm_func self.session = SessionMemory() self.short_term = ShortTermMemory(llm_func=llm_func) self.state = StateSnapshot() self.conn = init_db(db_path) self.endpoint_facts = [] def update_state(self, key: str, value): self.state.update(key, value) def build_prompt(self, user_input: str) -> list: # 1. 追加当前用户消息到会话 self.session.add_message("user", user_input) # 2. 从长期记忆召回 query_keywords = extract_keywords(user_input) # 简单关键词抽取 recall = recall_memories(self.conn, self.user_id, query_keywords, top_k=5) memory_block = "" if recall: lines = ["以下是该用户的部分历史记录,若与当前话题无关请忽略:"] for item in recall: lines.append(f"- {item['content']}") memory_block = "\n".join(lines) # 3. 短期摘要 summary = self.short_term.compress_if_needed(self.session.messages) # 4. 组装messages final_messages = [] system_lines = [ self.state.get_state_card(), memory_block, ] if summary: system_lines.append(f"早期对话摘要:{summary}") if self.endpoint_facts: system_lines.append("用户明确约束:" + ";".join(self.endpoint_facts)) sys_content = "\n".join([line for line in system_lines if line.strip()]) final_messages.append({"role": "system", "content": sys_content}) final_messages.extend(self.session.build_prompt_messages()) return final_messages def process_new_user_input(self, user_input: str, api_responder): """完整的一轮处理流程:build prompt -> call llm -> post process""" prompt_messages = self.build_prompt(user_input) reply = api_responder(prompt_messages) self.session.add_message("assistant", reply) # 会话结束后写入长期记忆的简单策略:如果回复中检测到重要实体 important_entities = detect_important_entities(user_input, reply) if important_entities: write_memory( self.conn, user_id=self.user_id, content=f"用户需求/决策: {user_input}", keywords=important_entities, importance=0.6 ) # 定期触发遗忘清理 if random.random() < 0.05: # 5%概率触发清理,避免每次请求都扫库 cleanup_memories(self.conn, self.user_id) return reply

这只是一个骨架级实现,每个环节都可以往深做,但你把它跑起来之后,Agent的"健忘症"会有本质改善。测试方法很简单:故意在会话里说一个"重要决策"(比如"所有报表用企业微信接收"),新建一个会话问"我上次说过报告怎么接收?",看Agent能不能从长期记忆里召回并正确引用。

6.3 实测中的典型问题与调优方向

我自己跑这套架构时遇到过几个典型问题,提前给你排雷。

第一个是关键词抽取质量差导致召回为空。中文场景里用户说话太随意,"报表用企微接收"这种句子抽关键词可能只抽出"报表",导致记忆召回不全。优化方向:一是维护业务领域同义词表(报表/报告/数据汇总),二是直接用Embedding做相似度召回,三是把关键词抽取也交给LLM做,让它输出结构化实体,成本可控而且准确率大幅提升。

第二个是摘要压缩丢细节。摘要滚动压缩早期对话时,如果摘要prompt没写清楚"保留用户明确的约束和数字信息",模型可能把"手续费率是0.6%"这种关键数字给概括成"有手续费"。把约束写进摘要prompt,同时保留一份用户明确约束的独立字段,双保险。

第三个是重要度评分过于主观。importance=0.6这种赋值频率高了,所有记忆分都差不多,排序就退化成按时间排。优化方法是引入规则:包含数字、金额、明确偏好词的记忆自动提高重要度,寒暄、情绪类内容压低数值调低。只要规则清晰,这个评分体系就能保持区分度。

写在最后的经验之谈

这套记忆架构在真实商用项目里最大的价值不是"技术先进",而是让Agent的行为可预期。用户不会在乎你底层用了什么向量库、什么遗忘算法,他们在乎的是:昨天说了退款,今天打开对话还能看到进度;例行周报这件事,不用每次都重新解释一遍规则。

我在多个项目里踩过不少坑,最想提醒后来者的一句话是:记忆系统一定要从第一天就设计分层,不要等项目跑出问题再重构。会话层、短期层、长期层、遗忘层的边界一旦想清楚,后面加需求就是往里填代码;要是直接写一个"万能记忆函数"把什么都往里塞,三个月后你自己都看不懂那几百行逻辑在干什么。

另外,记忆的写入策略比检索策略更值得花时间。我见过很多团队把注意力全放在"怎么搜得准"上,却忽略了"什么该写、什么不该写"。你每写一条无效记忆,都在给未来的检索制造噪音。要学会克制——只有确定对后续对话有价值的用户决策、偏好、事实,才值得进入长期记忆。

最后留一个可以继续深挖的方向:多Agent场景下的记忆共享。上面整套代码都是围绕单用户单Agent设计的,如果团队在做多Agent协作(比如一个Agent负责客服,一个Agent负责工单处理),那么记忆层需要考虑跨Agent的数据共享与一致性。你可以把长期记忆的user_id扩展到conversation_id或者team_id维度,让多个Agent共享同一套用户档案,这会是一个很有意思的演进方向。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询