★ FEATURED ARTICLE 最近在动手手写一个Manus风格的Agent系统做到第六节终于撞上最挠头的墙上下文记忆。多轮会话跑起来之后历史消息越堆越厚上下文窗口肉眼可见地膨胀。更麻烦的是Agent开始被一些边角料的旧消息带偏真正关键的信息反而被淹没。这一篇就是把我做的“消息相关性过滤”方案完整拆开讲核心是用LLM给Agent的上下文做动态整理属于用LLM管理Agent记忆的一条实用路线。适合正在手写Agent、被多轮上下文折磨得睡不着觉的朋友参考。我这个项目里已经跑通了过滤再回复的完整流程前后也踩了不少坑。如果你现在还在用“全量历史一股脑塞进模型”的写法这篇值得好好看一遍。1. 把上下文记忆交给LLM来管到底要解决什么问题1.1 上下文失控的三个典型症状先说我实际观察到的三个症状你在自己的Agent项目里大概率也遇到过。第一个是跑偏。历史对话里用户随口说过一句“我不太喜欢太正式的回复风格”Agent记住之后在后面所有回复里都刻意避开正式表达。问题是用户后来已经在聊严肃的技术方案了这条旧消息不仅没有帮助还一直在干扰当前判断。这种低相关性消息就像房间里一个一直在响的旧闹钟你知道它存在但每次听到都会分神。第二个是膨胀。每轮对话的思考过程、工具调用结果、中间输出全都被塞进历史列表几十轮跑下来光历史就有大几千甚至上万token。带来的直接后果是首token延迟变高、单次调用费用上升更隐蔽的问题是模型在超长上下文里对中段内容的注意力会明显衰减这已经是公认的模型行为规律不是你代码的问题。第三个是丢线索。用户前三轮说过“后面给方案的时候记得带上对比数据”结果聊到第十轮Agent完全忘了这回事。不是因为模型能力不行而是这条消息被埋在几十条工具调用日志和搜索摘要下面模型根本没把它当成重要约束。这三个症状指向同一个根因你没有一个机制来判断哪些历史消息对“当前这个问题”仍然重要哪些已经过期无效。1.2 常见的记忆管理方案横向对比在动手写过滤之前我把市面上常见的方案都过了一遍各有各的适用场景。方案核心思路优点缺点适合场景滑动窗口只保留最近N条消息实现简单零额外开销早期关键信息直接丢失短对话、强实时交互摘要压缩把旧消息总结成一段摘要token占用大幅降低压缩有损细节丢失后无法恢复每轮都做不太现实适合定时向量检索用embedding召回相似历史可扩展到很长的历史只做相似度匹配不理解对话目标外部知识库、长期记忆存储LLM相关性过滤让模型判断每条消息与当前问题的相关度语义理解到位可分级处理多一次模型调用有额外延迟和成本多轮Agent对话、工具调用密集场景滑动窗口最大的问题是你不知道“最近”到底应该有多近。有的信息聊了三轮之后还会被再次引用有的信息过两轮就没用了。固定窗口长度本质上是在赌赌错了就是丢线索。摘要压缩的问题是“一压就丢”。你把十轮工具调用压成两句话等后面哪一步需要具体参数值时你拿不回来了。它适合做归档不适合做唯一手段。向量检索本质上是“匹配”不是“理解”。它能找到和当前问题文本相似的旧消息但如果用户的当前问题是一句“还是按上次说的那个方案来”检索系统根本不知道“上次说的方案”是哪个方案因为这句话和方案正文的文本相似度可能很低。LLM相关性过滤是我最终选的主路线原因后面展开说。1.3 为什么“过滤”比“检索”更贴近对话本质我把检索和过滤做一个类比。向量检索像一个图书管理员你问他要什么主题的书他去书架上帮你找文本上相似的书。问题在于对话里的引用往往是“指代式”的——用户说的是“刚才那个问题”“上次的方案”“你之前答应我的事”这些表达只有在理解整段对话的语义流之后才知道指向哪里。LLM过滤则是让一个对话秘书从头到尾看一遍会议记录然后拿着今天的议题问哪些记录片段还需要放到桌面上这个秘书能理解语义能识别指代能判断一条消息是“过去式”还是“仍然有效”。所以我的结论很直接对Agent内部的多轮对话记忆来说过滤比检索更接近问题的本质。检索更适合对接外部知识库过滤更适合做内联记忆管理。两者不冲突但如果你只能先上一个先做过滤。2. 消息相关性过滤的整体设计与实现思路2.1 先给Agent定义清楚消息模型过滤的前提是你对消息本身有足够的信息可以判断。很多项目把历史记录只存成一个纯文本列表这是不够的。我在项目里定义了一个消息结构每条消息至少包含这些字段from dataclasses import dataclass, field from typing import Any, Optional dataclass class Message: msg_id: str # 全局唯一ID role: str # user / assistant / tool / system content: str # 文本内容 timestamp: int # 毫秒级时间戳 metadata: dict[str, Any] field(default_factorydict) # metadata 里我通常放 # tool_name: 本次工具名 # tool_call_id: 工具调用关联ID # ref_msg_ids: 这条消息引用/回复了哪些旧消息 # intent: 用户意图标签(可选) source: str session # session / memory / archive为什么msg_id和metadata这么重要因为在过滤之后你还要能追溯。如果一条被压缩的消息后来又被证实是问题里提到的“上次那个方案”你需要能定位到原始消息而不是面对一段没有来源的摘要。ref_msg_ids这个字段尤其有用它让我们可以构建消息间的引用链路后面做级联处理时省了很多事。2.2 过滤模块在Agent主循环中的位置过滤不能是独立跑一次的工具它必须嵌入Agent主循环。我实现的流程是这样的收到用户当前请求current_query。从记忆池中取出候选历史消息。把候选历史 当前请求一起送入过滤模块。过滤模块返回每条消息的动作保留/压缩/删除。按动作更新记忆池和归档池。用“保留消息 归档摘要”组装最终prompt。关键点在第二步和第三步的顺序过滤必须发生在组装prompt之前而且过滤的输入要包含当前请求。没有当前请求做参照相关性判断就是无源之水。有些方案用“定时压缩历史”而不是“按需过滤”每次都是无差别处理效果就差很多。我建议把这个过滤模块做成Agent组件里一个可插拔的独立对象不要和主循环耦合。这样后续你想换成别的记忆策略只需要替换filter_messages的传入类型不用改主循环。2.3 相关性判定的四个核心维度我让LLM在打分时重点考虑四个维度写进prompt里作为判定依据。第一个是主题相关性。这条消息的话题和当前问题是否在同一主题线上。用户当前在问部署方案历史里的部署参数就相关历史里聊的周末安排就不相关。第二个是实体引用。当前问题有没有提到历史中的特定实体比如文件名、API名、某个方案代号、某个数字。一旦出现这种引用被引用的那部分历史必须保留。第三个是意图延续。用户当前问题可能是对之前某个问题的一次追问、转折或反悔。比如用户之前要求“方案A”现在说“我觉得方案A不太行”如果不保留方案A的内容这个转折就会失去参照。第四个是时间衰减。同等相关度下更近的消息优先级更高。但注意时间衰减只能作为辅助权重不能主导判别。因为对话里经常有跨很多轮才收回的线索。这四个维度我会直接写进过滤prompt的判定标准里而不是让模型自己悟。实测下来给出明确维度说明之后打分稳定性明显高于只给一句“判断是否相关”。2.4 保留、压缩、删除的三级策略过滤结果如果只有“保留”和“删除”两个动作风险很大。因为LLM也不是每次判断都准一旦误删关键消息整场对话的上下文就断了。我改成了三级策略保留这条消息和当前问题直接相关原样进入上下文。压缩这条消息可能相关留着一时用不上删了又不放心。把它压成一句话摘要放入归档池。删除可以确认与当前问题无关从记忆池移除。还有一个兜底规则最近N条消息默认不参与过滤。我这里设的是最近20条。为什么因为对话的连续性很强最近的几轮不管看起来和当前问题多不相关都有可能在下一轮突然被引用。保护带是防止误删的第一道保险。压缩消息进入归档池之后组装prompt时会把归档摘要放在上下文末尾作为补充背景而不占用主体位置。这样既控制了token又保留了线索。3. 核心实现打分prompt、过滤主流程与阈值调优3.1 过滤prompt这样写模型输出才稳过滤模块本质上也是一次LLM调用。我最初用的prompt很随意就让模型“给下面每条历史消息打相关度分”结果输出格式五花八门有的给文字描述有的给星星有的直接不回JSON。后来我把prompt改成了结构化指令稳定性才上来。核心模板长这样FILTER_PROMPT 你是一个对话上下文过滤器。你的任务判断给定历史消息与当前用户问题是否相关。 评分标准重要 1. 主题相关消息讨论的话题是否与当前问题一致或直接关联 2. 实体引用当前问题是否引用了消息中出现的文件名、API、方案名、数字等实体 3. 意图延续消息是否是当前问题之前的关键决策、条件约束或用户偏好 4. 时间衰减同等相关度下近期的消息分值更高 输出要求 对每条消息严格输出一个JSON对象格式如下 {message_id: m12, score: 8, action: keep, reason: 该消息包含当前问题引用的部署参数} {message_id: m13, score: 4, action: compress, reason: 与当前部署问题弱相关压缩为一句摘要} {message_id: m14, score: 1, action: drop, reason: 闲聊内容与当前任务无关} 注意 - score必须是0到10的整数 - action只能取 keep / compress / drop 三者之一 - 不要输出任何JSON之外的文字 - 原文不要遗漏必须逐条输出 开始 当前用户问题 {current_query} 历史消息 {messages} 我特别加了一句话“原文不要遗漏必须逐条输出”。这句话很容易被忽略但没有它模型偶尔会偷偷省略几条它认为不重要的消息导致返回的JSON条目数和输入消息数对不上。还有一个小经验打分最好让模型用整数不要用小数。实测整数打分方差更大更容易拉开区分度而小数打分经常挤在6到8之间阈值很难切。3.2 过滤主流程的参考代码下面这段是我项目里过滤模块的核心实现注释写得很细。你把它理解完替换掉ask_llm这个函数名就能接入自己的模型客户端。import json from typing import List, Dict class MemoryFilter: def __init__( self, max_batch_size: int 12, keep_threshold: int 8, compress_threshold: int 6, protect_recent: int 20, ): self.max_batch_size max_batch_size self.keep_threshold keep_threshold # score 8 保留 self.compress_threshold compress_threshold # score 6 压缩 self.protect_recent protect_recent # 最近N条不参与过滤 def filter_messages( self, all_messages: List[Message], current_query: str, ask_llm, ) - Dict[str, str]: # 1. 切割保护带最近N条直接保留 if len(all_messages) self.protect_recent: return {m.msg_id: keep for m in all_messages} protect_msgs all_messages[-self.protect_recent:] candidate_msgs all_messages[:-self.protect_recent] actions {m.msg_id: keep for m in protect_msgs} archived_summaries [] # 2. 分批送入过滤模型 for batch_start in range(0, len(candidate_msgs), self.max_batch_size): batch candidate_msgs[batch_start:batch_start self.max_batch_size] batch_prompt FILTER_PROMPT.format( current_querycurrent_query, messagesself._format_messages(batch), ) response_text ask_llm(batch_prompt, temperature0) parsed self._safe_parse(response_text) if parsed is None: # 解析失败这批全部保留避免误删 for m in batch: actions[m.msg_id] keep continue for item in parsed: score item.get(score, 0) msg_id item.get(message_id, ) # 3. 按阈值分配动作 if score self.keep_threshold: actions[msg_id] keep elif score self.compress_threshold: actions[msg_id] compress archived_summaries.append( f[{msg_id}] {item.get(reason, )}: {self._find_msg(all_messages, msg_id).content[:50]} ) else: actions[msg_id] drop # 4. 把压缩摘要写回归档池 self.archive_pool.extend(archived_summaries) return actions def _safe_parse(self, text: str): text text.strip() # 有些模型会在JSON外包裹json需要剥离 if text.startswith(): text text.strip() if text.startswith(json\n): text text[5:] try: # 支持单条JSON和多条JSON拼接 if text.startswith([): return json.loads(text) lines [l for l in text.splitlines() if l.strip().startswith({)] return [json.loads(l) for l in lines] except Exception: return None def _format_messages(self, messages: List[Message]) - str: lines [] for m in messages: meta , .join([f{k}{v} for k, v in m.metadata.items()]) lines.append(f[{m.msg_id}] {m.role} (metadata: {meta}): {m.content[:200]}) return \n.join(lines) def _find_msg(self, messages, msg_id): for m in messages: if m.msg_id msg_id: return m return Message(msg_idmsg_id, roleunknown, contentnot found, timestamp0)注意几个细节。其一temperature0是必须的。过滤任务是确定性任务不是创意任务。temperature调高一点打分就在浮动阈值就成了摆设。其二ask_llm我传的是函数而不是具体模型对象。实践里的原因是我经常先用一个便宜的小模型跑粗筛再用一个大模型跑精筛这样接口可以在不同模型间切换。其三_safe_parse里有一层兜底解析失败就整批保留。很多实现一旦解析失败就报错重试但我选择保守处理。过滤任务里宁可多留不可误删。3.3 阈值怎么调先宽松再收紧阈值是过滤效果最敏感的参数没有之一。我推荐的调法是“先宽松后收紧”。第一次运行时把keep_threshold设成7compress_threshold设成5跑完看两件事一是被drop的消息里有没有让你觉得“哎呀这个其实后面可能要用”的二是最终prompt的token数有没有降下来。如果drop误删率高把keep_threshold往上调比如8compress_threshold往上调到6或7。如果token数降得不够明显说明保留太多了可以适当下调。我最终停在 keep8、compress6。这个组合在我的场景里误删率大概在5%左右token缩减率在40%到60%之间是一个比较健康的平衡点。永远不要追求“删得越多越好”。你的目标不是最大化删除而是最小化误删的同时把token压到合理区间。记住这一点阈值就好调了。3.4 别让过滤本身吃掉太多开销过滤功能多了一次LLM调用如果你每次回复都跑全量过滤token开销可能比省下来的还多。我踩过这个坑所以后来加了三道闸门。第一道闸门是触发条件。只有历史消息数超过保护带20条、或者历史总token超过某个阈值比如3000时才跑过滤。短对话直接跳过。第二道闸门是批大小。单条消息一条一条调用LLM肯定不行一次调用处理12条左右效果和性能最均衡。批次太小调用次数太多批次太大输出容易截断。按输出token估算每批12条历史消息输出大约600到900 token大多数模型都能一次稳定返回。第三道闸门是缓存。同一个会话如果当前问题和最近一轮问题高度相似那上一轮的过滤结果可以直接复用只对新产生的消息做增量过滤。我在会话对象里保存了上一轮的query向量相似度超过0.95就复用结果。4. 实操踩坑输出、误删、开销与评测方法4.1 JSON输出不稳定怎么兜底这是过滤模块最容易遇到的问题。模型输出不规范具体表现有三种JSON外面包了markdown代码块多个JSON对象之间出现逗号或说明文字score返回字符串而不是数字。我现在的处理分三层。第一层prompt约束。输出要求里明确写了“不要输出任何JSON之外的文字”“逐条输出”。这能解决80%的问题。第二层解析兜底。_safe_parse里先剥代码块围栏再尝试整体解析JSON失败就逐行解析。这里有个小技巧很多模型输出的是多个独立JSON对象拼接而不是一个JSON数组整体解析会失败按行解析就正常了。第三层解析失败就整批保留。宁可这次不省token也不冒误删风险。还要注意一个隐蔽问题模型偶尔会把两条消息合并成一条输出即message_id字段缺失或输出一个不存在的ID。这时候我会做一一对应校验ID对不上就触发整批重跑一次。4.2 误删关键消息症状、原因与对策误删是过滤方案最怕的事。我的经验是误删的典型案例不是那种直接相关的消息而是“目前暂不相关但后续会被引用”的消息。举个我实际遇到的例子。用户在第5轮提过一句“后端接口地址先记为localhost:8080后面部署时再改”第28轮突然问“部署环境的接口地址打算怎么处理”。如果第5轮的消息在第10轮就被drop了第28轮这个问题就成了无根之木Agent完全不知道“后面部署时再改”这个约定。对策有三个。第一个是保护带足够长。我把最近20条设为保护带就是给“后续引用”留出空间。第二个是用压缩代替删除。即使判断为弱相关也先压缩成一句话摘要放到归档池不直接物理删除。摘要虽然丢失了细节但至少保留了“曾经存在过这么一件事”的线索。第三个是引用链追溯。在metadata里记录ref_msg_ids当新消息引用旧消息时即使旧消息已经被压缩也可以根据引用ID找到归档池里的摘要必要时恢复原始内容。4.3 每次过滤都跑全量token反而更多这个坑尤其隐蔽。你不加触发条件过滤本身调用的token可能比过滤掉的token还多。特别是历史消息只有三五十条时过滤一次可能花1000多token删掉的只有几百token。我加了两个经验值历史超过80条或总token超过4000才触发过滤。在这个阈值以下直接全量保留因为模型处理起来也不吃力。还有一个小技巧过滤用的模型可以和最终回复用的模型不同。我用一个便宜的小模型跑过滤回复用能力更强的模型。过滤任务相对简单小模型完全够用成本能降一个量级。4.4 单条消息单独判断容易失真这是过滤模块里最难调的问题。单独看一条消息它可能完全无意义但放到上下文里它是某个关键方案的必要组成部分。我解决这个问题的方法是“按分钟窗口切片”。把连续几分钟内、同一话题下的消息打包成一个组让LLM对这个组整体打分而不是对单条消息打分。组内消息要么一起保留要么一起压缩。这样做有两个好处一是上下文连续性被保留二是减少了LLM的判断条目数变相降低了token消耗。实现上很简单_format_messages之前先把消息按时间戳和主题标签聚合成小组每个小组一个组ID打分结果映射到组内所有消息上。4.5 一个可复现的评测方法重建测试做过滤功能最怕“感觉还行但说不清好不好”。我搭了一套简单的离线评测方法叫“重建测试”。做法是准备几段多轮对话里面故意混入干扰话题。先把对话完整跑一遍作为baseline记录每个问题下Agent的回答质量。然后启用过滤再跑一遍同样的对话。最后随机抽取一些被drop或compress的消息人工构造与这些消息相关的问题问Agent看能否从归档池和摘要中恢复信息。这个测试能直接暴露两类问题一是过滤后回答质量下降二是被归档的消息是否真的还能找到。我跑下来发现好的过滤配置应该做到正常回答质量不降、干扰消息的引用率下降、关键信息恢复率在90%以上。如果你不想搞这么重可以做一个简化版在过滤后的上下文里插入一个“你记得我们之前聊过XX吗”的问题看Agent能否答对。多测几组比纯手感靠谱得多。5. 效果调优与记忆分层扩展5.1 测试集里怎么构造干扰话题评测用的测试集别用真实聊天记录隐私和维护成本都划不来。我用脚本生成模拟对话思路很简单每个测试场景包含一条关键信息线加两条干扰线。举一个例子主角线是“用户要求回复里加上技术方案的对比表格”干扰线一是“用户抱怨今天的天气”干扰线二是“用户问了一个和方案无关的历史问题”。过滤后的Agent应该保留主角线拿掉干扰线。我构造了三十个类似场景覆盖不同的关键信息类型数字参数、方案代号、用户偏好、历史约定。跑一轮评测大约消耗几千token但得到的结论比拍脑袋调参数可靠太多了。5.2 分层记忆过滤、摘要、检索的组合过滤不是记忆管理的终点。我的最终结构是三层记忆。第一层是工作记忆即过滤后保留的最近消息直接进入上下文。第二层是归档记忆即被压缩的弱相关消息以摘要形式存在归档池定期被查询。第三层是长期记忆把归档摘要定期做向量化支持跨会话的语义检索。这三层各有职责过滤负责完成第一层向第二层的转移。过滤做得越准确归档池的质量越高后续的摘要和检索才能站在好的基础上。实际实现中我每隔10轮会触发一次归档池整理把多条小摘要合并成一条大的主题摘要继续压缩token占用。这个可以做得很重但第一版先把过滤做扎实就够用了。5.3 过滤参数速查表最后把我常用的参数和推荐值整理成表方便你对照调整参数推荐值调整方向keep_threshold8误删多就上调到9保留太多就下调到7compress_threshold6想让归档更厚就下调到5protect_recent20对话引用跨度大就开到30max_batch_size12模型输出不稳定就降到8触发历史条数80除保护带外的候选消息数触发历史token4000超出这个值才跑过滤调参不要一次动多个值。每次只改一个跑一轮评测记录误删率和token缩减率。这样才能知道是哪个参数引起的效果变化。我在实际项目中跑下来最直观的体感是上下文从平均6000到7000 token降到2000到3000Agent回复跑偏的次数明显减少用户两次提到同一个约定的恢复能力也稳了不少。过滤本身多了一次小模型调用但省下的token和重试成本完全能覆盖掉。如果你也在手写Agent建议把过滤模块单独做成一个可插拔组件接口只留filter_messages和回传动作表。后面就算你想换摘要策略或者接向量检索也不会动到主循环。下一步我准备把归档摘要接进长期记忆索引让更久远的关键约定也能被主动想起来。
阅读完成 · 觉得有帮助？
