首页 / 资讯中心 / 文章详情

Agent上下文记忆管理:用LLM做消息相关性过滤,实现低成本长对话

Agent上下文记忆管理:用LLM做消息相关性过滤,实现低成本长对话 ★ FEATURED ARTICLE
如果你最近也在动手做一个类似 Manus 的通用任务型 Agent大概率已经撞上了同一堵墙任务还没跑完对话上下文先爆了。用户的需求、工具返回的结果、中间环节的推理过程每一轮都会往上下文里塞进大量 token窗口被占满之后要么强行截断导致后面的工具调用失去前文支撑要么全部保留然后成本直接起飞。我在这套手写 Agent 系列的第 06 个迭代里最终把“上下文记忆管理”从固定的滑动窗口换成了由 LLM 自己做决策的消息相关性过滤效果比自己预想的好不少。这篇文章就把这个方案的完整思路、Prompt 设计、落地代码和踩坑记录整理出来给同样被上下文问题折磨的人一个可以“抄作业”的参考。这套方案适合正在手撸 Agent / Multi-Agent 系统、被上下文长度和 token 成本双重夹击的开发者也适合那些已经接了某个大模型 API、但总觉得“对话一长就变蠢”的调参选手。核心其实就一句话别再傻傻地保留所有历史消息也别一刀切地截断末尾而是把“哪些消息值得保留”这件事交给 LLM 去判断。1. 为什么上下文记忆会成为 Agent 的瓶颈1.1 先看痛点消息是怎么把上下文窗口吃掉的一个完整任务的生命周期里Agent 往往要经历多轮“思考-调用工具-拿到结果-继续思考”的循环。比如用户丢过来一句“帮我做一份竞品分析报告”Agent 先搜索资料再打开几个链接再调用某个数据库查询每一个工具返回的都是几百甚至上千 token 的文本。这些结果不仅要喂给下一轮 LLM 做推理还有可能在后续步骤里被反复引用。我刚开始做项目的时候用的就是最简单粗暴的方案把历史消息全部拼进去窗口快满了就删最前面的。结果发现两个问题特别要命。第一个是成本一个复杂任务跑到一半上下文就能冲到几万 token每轮都要重发一次费用非常难看。第二个是“前半段失忆”很多早期的关键信息——比如用户最初交代的约束条件、某次工具调用返回的唯一一份数据——被当作最旧的消息直接丢掉了后续 Agent 的行为就会开始“跑偏”甚至会做出和初始目标完全相反的操作。后来我又试过固定长度的滑动窗口也就是只保留最近 N 轮对话。这个问题好一点但依然没有解决“重要但久远的信息”这个场景。比如用户在 20 轮之前说过一句“不要用外部 API所有请求走本地服务”这条信息如果落在窗口之外Agent 就会毫无心理负担地调用外部接口。1.2 三种常见记忆方案对比截断、摘要、相关性过滤处理 Agent 记忆社区里常见的做法大致有三条路。第一条是硬截断。不管是头部丢弃还是滑动窗口本质都是“按时间淘汰”。优点是实现简单缺点是没有任何选择性它不知道哪些信息对当前目标重要淘汰逻辑完全是盲目的。第二条是历史摘要。把所有旧消息定期喂给 LLM生成一段摘要然后只保留摘要。这个方案的优点是压缩比高但缺点同样明显摘要会失去细节尤其是工具调用返回的具体数值、代码片段、数据结构这类东西一摘要就没法用了。Agent 后续如果需要用到某个精确的 ID 或某个返回值摘要派不上用场。第三条就是我最终采用的消息相关性过滤。思路是把一条条消息当作独立的记忆单元每次要构建上下文之前先让 LLM 对历史消息做一轮相关性打分把分数高的挑出来拼进上下文分数低的丢弃或合并。它和摘要的区别在于它不是把所有消息揉成一段话而是保留原始消息本身只是有选择地保留。这样既不丢细节又能控制 token 总量。而“由 LLM 来过滤”这一步是最关键的只有模型才能理解一条历史信息和当前任务目标之间到底有没有语义关联。靠关键词匹配或规则过滤基本上杯水车薪。2. 消息相关性过滤的整体设计思路2.1 记忆分域短期工作记忆做保底长期线索做补充我一开始想把所有历史消息都交给 LLM 打分后来发现两个问题一是每次打分都要把全量历史重新发送一遍成本不低二是有些消息根本不需要打分比如最近一两轮里的系统输出和工具结果这些是当前推理的“直接前件”无论如何都要保留。所以我把记忆分成了两个域短期工作记忆最近 K 轮消息固定保留不参与相关性过滤。长期线索记忆更早的历史消息全部交给 LLM 打分只保留高相关部分。这个设计其实很像人的记忆机制。你不会记得昨天中午吃了什么长期记忆不相关但你一定记得刚才手里拿着杯子短期记忆直接相关。对 Agent 来说最近几轮的工具调用结果和推理过程就是手里的“杯子”必须放在触手可及的地方。而更早的任务约束、用户偏好、行业背景只要和当前目标相关也应该能从“记忆仓库”里被捞回来。实际实现时K 我取的是 2 到 3 轮太大会导致工作记忆本身就很占地方太小又容易丢掉推理链条的前半段。2.2 消息结构与元数据是过滤的地基要做消息级过滤前提是消息本身得是结构化对象而不是一串拼好的字符串。我在代码里定义了消息类包含 role、content、timestamp、msg_type、token_count、parent_id 这些字段。role 是消息角色user、assistant、tool、system。msg_type 标记消息的类型比如是一次工具调用、一次工具返回结果、一段普通推理、一条用户指令。timestamp 记录消息产生的时间后续可以做时间衰减。parent_id 标记消息之间的父子关系。比如某次工具调用产生了结果那“调用”和“结果”就共享同一个流程 ID过滤时要么一起保留要么一起丢弃。token_count 用来做预算控制。为什么元数据这么重要因为相关性不只看内容还要看类型。工具返回结果里的一个数字可能比一段长篇大论的用户闲聊重要得多。我在后续过滤时对不同消息类型设置了差异化的基础权重tool 类型消息默认权重最高assistant 推理次之user 指令也是高权重而那些寒暄类的用户消息则很容易被分数压下去。2.3 “打分-取舍-压缩”三层过滤策略完整的过滤流程我拆成了三步。第一步打分由 LLM 对每条历史消息输出一个 1 到 5 的相关性分数第二步取舍基于分数和 token 预算决定哪些消息进入上下文第三步压缩如果预算还不够就对低分段消息做一次摘要合并而不是直接丢弃。这三层是配合工作的。打分保证语义准确性取舍保证预算可控压缩则给了一个兜底方案。我最初只做了前两步后来发现复杂任务里高分段消息的数量也可能超预算这时候就不得不砍掉一些原本想保留的内容。加了摘要压缩之后至少能把那些“有点相关但没那么关键”的消息以极低成本保留下来。一句话总结这个策略先让 LLM 判断“哪条消息重要”再用工程手段保证“重要消息放得下”。3. 用 LLM 做相关性判断的落地实现3.1 打分 Prompt 的完整设计很多人在这一步翻车原因是觉得“让 LLM 打分很简单直接把任务目标和历史消息丢进去就行”。但实际上如果没有把评分标准定义清楚模型输出的分数会很飘。我给打分任务设计了一套固定格式的 Prompt核心结构包含四部分角色设定、任务目标、当前子任务、待打分消息列表。评分标准一共五档每一档都有明确行为描述5 分直接决定任务成败的关键事实比如用户的核心约束、唯一的 ID、关键数据。4 分影响后续关键步骤的信息比如正在执行的分析逻辑、中间结论。3 分提供有用背景可留可丢。2 分与当前任务关联度低优先丢弃。1 分与任务目标完全无关直接丢弃。Prompt 里我会明确要求模型输出 JSON 数组每个元素包含 id、score、reason 三个字段。reason 必须简短控制输出 token 成本。同时给一个硬性规则最近 2 轮的消息默认给 5 分不需要重新评估——这个规则避免了模型因为“上下文偏移”而误伤短期工作记忆。这是打分 Prompt 的模板骨架我做了简化SCORE_PROMPT 你是该 Agent 的记忆管理员。请判断以下历史消息对完成当前任务目标的价值。 任务目标 {task_goal} 当前正在执行的子任务 {current_subtask} 历史消息每条消息以 [id] 开头 {history} 请输出 JSON 数组数组长度必须等于消息条数。每条输出结构如下 [ {id: msg_001, score: 4, reason: 该消息包含用户对数据格式的硬性要求后续步骤需要遵守} ] 评分标准 5 直接决定任务成败必须保留 4 影响后续关键步骤建议保留 3 提供有用背景可保留也可丢弃 2 与当前任务关联度低优先丢弃 1 与任务目标无关直接丢弃 额外要求 1. 最近两轮消息 msg_098 和 msg_099 直接给 5 分不用解释。 2. score 必须是 1 到 5 的整数。 3. 不要遗漏任何一条消息。要注意的是不同模型对 JSON 格式的遵循程度不一样。我在使用时要确保拿到稳定的 JSON。如果你的模型支持 JSON mode 或结构化输出比在 Prompt 里强制要求要可靠得多。3.2 阈值、召回数量与 token 预算怎么定LLM 给出分数之后剩下的取舍逻辑就归工程侧管了。我默认采用“双阈值确认法”分数 ≥ 4 的消息直接进入上下文。分数 3 的消息进入候选池按照时间从近到远排序。分数 ≤ 2 的消息进入压缩区等待摘要合并。如果进入上下文的 token 总量还没有超过预算就从候选池里逐个补充消息直到预算接近上限。token 预算的设定要和目标模型的最大上下文窗口解耦。比如某个模型支持 128K 上下文但你不能真把 128K 全用掉因为模型还要留出空间做本轮推理输出。我通常的做法是给当前模型设置一个安全阈值比如最大窗口的 70%其中输入上下文占 60%给输出预留 10%。如果任务本身输出比较长比如要生成代码输出预留可以提高到 15% 到 20%。这里有一个经验值可以参考对大多数任务最终进入上下文的消息条数占全部历史消息的 20% 到 35% 是比较理想的。低于 20%说明过滤太激进很容易出现上下文割裂高于 35%说明相关性判断太保守几乎等于没过滤。我在项目里实测下来一个 10 万 token 的历史库经过打分过滤后通常能压到 2 万到 3 万 token 左右同时任务完成率比“永远只保留最近 N 轮”的方案高不少。3.3 落地代码一个可运行的消息过滤核心模块这部分给一段可运行的 Python 简化实现核心逻辑是消息存储、打分、组装上下文三个环节。实际项目里我会把它封装成一个独立的 MemoryManager 类。from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from datetime import datetime import json dataclass class Message: role: str # user / assistant / tool / system content: str timestamp: float msg_type: str normal # normal / tool_call / tool_result / user_instruction parent_id: Optional[str] None token_count: int 0 meta: Dict[str, Any] field(default_factorydict) property def msg_id(self) - str: return self.meta.get(msg_id, fmsg_{id(self)}) class RelevanceMemoryManager: def __init__(self, llm_client, max_input_tokens: int 24000): self.llm llm_client self.max_input_tokens max_input_tokens self.all_messages: List[Message] [] self.task_goal: str self.current_subtask: str def set_task_goal(self, goal: str) - None: self.task_goal goal def set_subtask(self, subtask: str) - None: self.current_subtask subtask def add(self, msg: Message) - None: self.all_messages.append(msg) def _estimate_tokens(self, text: str) - int: # 简化估算中文字符按 1 token 计英文按 0.25 token 计 # 实际项目中可以替换成对应模型的分词器 return int(len(text) * 0.75) def _build_score_prompt(self, history: List[Message]) - str: lines [] for m in history: content_preview m.content[:200] lines.append(f[{m.msg_id}] ({m.role}/{m.msg_type}) {content_preview}) history_text \n.join(lines) return SCORE_PROMPT.format( task_goalself.task_goal, current_subtaskself.current_subtask, historyhistory_text, ) def _score_messages(self, history: List[Message]) - Dict[str, int]: prompt self._build_score_prompt(history) response self.llm.chat( promptprompt, temperature0, response_formatjson # 你的 llm_client 可能需要调整 ) # 解析 JSON容错处理 markdown fence text response.strip() if text.startswith(): text text.strip() if json in text: text text.split(json)[1].split()[0].strip() result json.loads(text) return {item[id]: int(item[score]) for item in result} def build_context(self) - List[Message]: # Step 1: 固定保留最近三轮消息短期工作记忆 recent self.all_messages[-12:] # 三轮大约对应 12 条消息按实际调整 recent_ids {m.msg_id for m in recent} # Step 2: 对更早的历史消息做相关性打分 older [m for m in self.all_messages if m.msg_id not in recent_ids] if older: score_map self._score_messages(older) else: score_map {} # Step 3: 按分数与预算取舍 keep list(recent) candidates [] for m in older: score score_map.get(m.msg_id, 2) if score 4: keep.append(m) elif score 3: candidates.append(m) # score 2 暂时丢弃后续可进摘要压缩区 # Step 4: 候选池按时间从近到远补充直到预算上限 candidates.sort(keylambda x: x.timestamp, reverseTrue) current_tokens sum(self._estimate_tokens(m.content) for m in keep) for m in candidates: est self._estimate_tokens(m.content) if current_tokens est self.max_input_tokens: break keep.append(m) current_tokens est # Step 5: 按时间升序返回保证对话连贯性 keep.sort(keylambda x: x.timestamp) return keep def history_to_llm_messages(self) - List[Dict[str, str]]: context self.build_context() return [{role: m.role, content: m.content} for m in context]这段代码里_score_messages是对所有旧消息做一次批量打分然后根据分数和预算做取舍。阈值和预算都放在初始化参数里方便后续调优。我在真正集成这个模块时还做了两层加强。第一层最高优先级保留的不只是最近三轮而是所有parent_id处于“活动工具调用链”上的消息避免链路中断。第二层如果一次性要打分的历史消息太多我会先把它们按时间切片分多次打分再合并结果防止单次打分 Prompt 超长。4. 上下文组装与应用链路4.1 嵌入主循环每次调用 LLM 之前做一次过滤消息过滤不是一个一次性的动作而是 Agent 主循环里每一次调用 LLM 之前都要执行的步骤。我的完整流程是接收用户输入把用户消息写入 memory。如果用户输入改变了任务目标更新task_goal。在每次调用 LLM 之前执行build_context()构建本轮上下文。LLM 输出规划或工具调用结果。工具执行后的结果作为新的 tool 消息写入 memory。进入下一轮循环重新构建上下文。这里有一个特别容易忽略的点过滤是动态的。同样一条历史消息在任务前期可能是核心线索到后期可能就变成噪音了。所以每次调用 LLM 之前都要重新打分不能只打一次然后缓存结果。不过可以考虑加一层缓存比如同一子任务阶段内如果任务目标没变打分结果可以直接复用这样能省下不少 LLM 调用成本。我实际测过对一个 30 步左右的复杂任务如果每一步都重新打分额外会引入十几到几十次小请求。加上缓存之后同一个子任务阶段内只需要打一次分步骤切换时才重新打分成本能降一半以上。4.2 上下文组装顺序系统提示、任务目标、历史、最近过滤之后消息的排列顺序并不简单是按时间升序。上下文内部的顺序会影响模型的注意力分布。我最终固定了这样的组装顺序系统提示词固定不变描述 Agent 的整体行为和能力边界。任务目标task_goal每次组装时都会放在很靠前的位置。当前子任务描述current_subtask紧跟在任务目标之后。历史相关消息分数较高的早期消息按时间升序排列。最近几轮完整消息短期工作记忆放在最后保证模型对“当前正在发生什么”有最鲜活的感知。为什么任务目标要反复出现在上下文开头而不是靠模型记住因为相关性过滤本质上是在压缩和丢弃信息如果目标信息本身也被压缩了Agent 就会“忘本”。把 task_goal 从消息列表里单独抽出来每次组装时强制拼在最前面相当于给 Agent 一个锚点。这样哪怕是极端压缩的场景Agent 也始终记得自己最初要完成什么。我在调参过程中发现一个现象如果任务目标被放在很后面或者被埋在大量历史消息中间模型在长对话后半段经常会出现“目标漂移”——它开始专注于当前子任务忘记最初约束。把这行提到开头之后这个问题基本消失了。4.3 与工具调用链的配合防止“结果的爸爸”被滤掉工具调用链断裂是我在实现过程中踩到的最大的坑值得单独拿出来说。场景是这样的Agent 第一步调用搜索工具拿到一批结果第二步基于这些结果做了分析第三步又需要回到第一步的某条搜索结果做引用。如果相关性过滤在第二步到第三步之间把第一步的搜索结果判定为“低相关”并丢弃那第三步就彻底断了线模型会假装自己知道结果或者干脆重新搜索一遍。解决办法是引入parent_id和“调用链”。我在设计消息结构时就给每一条 tool 消息加了一个trace_id一次工具调用的完整链路包括assistant 的调用指令、tool 的原始请求参数、tool 的结果返回。这三者共享同一个trace_id。在过滤阶段规则如下如果一条 tool 消息被保留那么它所在 trace 链路里的所有消息全部保留。如果一条 tool 消息被丢弃但它对应的 assistant 调用指令仍然因为其他原因被保留则这个链路仍然完整保留。只要一个链路中任何一条消息分数达到 4 分整条链路升级为保留。实现起来就是在build_context()里加一个扩展步骤先按分数选定初始保留集合再遍历集合中所有带trace_id的消息把同链路消息全部拉进保留集合。这本质上是“关系完整性优先于分数优先”。5. 常见问题与排查技巧实录5.1 最常见的翻车现场先决条件被滤掉我最早跑这套过滤方案的时候最典型的问题是Agent 跑到一半突然忘了用户最初的格式要求。比如用户一开场说“报告用表格形式输出”20 步之后 Agent 开始输出大段文字完全没有表格。查 log 之后发现那条用户消息在过滤时被打了一个低分。原因是打分 Prompt 里的“当前子任务”是“查询数据库数据”而用户当时的格式要求从相关性上讲只是“背景信息”模型就给了 3 分以下。这个问题不是模型笨而是因为“任务目标”写得太窄。如果我只有一句“查询数据库”当任务目标那“输出格式”确实和它没关系。但如果把任务目标完整写成“查询数据库并以表格形式输出一份竞品对比报告”那条格式要求就会变成 4 分甚至 5 分。所以过滤效果很大程度上取决于任务目标描述得是否完整。我的经验是把任务目标拆成两层最终目标和当前子目标。最终目标写清楚用户所有硬性要求子目标负责描述眼下这一步要做什么。打分时同时给这两层信息LLM 才能区分“当前任务相关信息”和“全局任务相关信息”。5.2 问题速查表与修复方案我在项目里整理了一张速查表遇到问题可以直接对着排查症状根本原因修复方案Agent 在中途忘了初始约束task_goal 描述过窄过滤时把约束信息丢了把 task_goal 拆成最终目标子目标同时提供给打分 Prompt工具结果被滤掉导致后续推理断链没有处理 tool 消息调用链关系引入 trace_id链路内消息要么全留要么全弃LLM 打分不稳定同一条消息时高时低温度过高或 Prompt 评分标准不清晰temperature 设 0开启 JSON mode评分标准逐档细化过滤后上下文仍然超限阈值设置过低预算设置过高降 max_input_tokens提高 4 分阈值增加摘要压缩兜底上下文里重复出现同一份工具结果多个子任务引用同一结果过滤阶段重复带入对 tool 消息做哈希去重引用映射到占位符打分报告说“所有消息都重要”Prompt 缺少“建议丢弃”的强引导在评分标准里增加负面示例要求模型至少给出 10% 的 1 分消息早期关键数据被时间衰减误伤时间权重设置过强时间衰减只作用于边缘分数2-3 分4 分以上不做衰减5.3 让 LLM 打分更稳定的几个工程技巧先解决一个实际问题很多模型在要求输出 JSON 数组时偶尔会给出{scores: [...]}这种包装结构或者在开头加一段解释文字。单纯把响应转成json.loads()很容易崩溃。我在解析时做了容错先剥离 markdown fence再尝试提取最外层中括号内的 JSON 数组。如果解析失败就整批重试把上一次输出带上作为负面示例。还有一个稳定性的关键是让模型“先判断后打分”。我试过两种 Prompt 风格一种是让模型直接打分另一种是让模型先对每条消息写一句判断理由再打分。直接打分的结果比较随机而先写理由再打分的结果稳定性高很多。代价是输出 token 变多、响应变慢。我在生产环境里用了一个折中方案只对 3 分上下难以判断的消息要求写理由其他分数直接出结果。最后温度设置非常关键。打分任务本身就是提取模型内部判断不需要任何创造性temperature 必须设置为 0。我见过有人在打分 Prompt 里保留了默认的 0.7 温度结果同一个任务跑了两次过滤出来的消息集合完全不一样后续 Agent 行为也随之漂移。这种随机性在 Agent 系统里是不可接受的。6. 扩展与优化空间6.1 向量召回 LLM 精排过滤性能再上一个台阶纯 LLM 打分的方案在准确度上是够用的但有一个性能隐患如果 Agent 已经运行了非常久积累了上百万 token 的历史消息每次打分前把这些消息全部发给模型成本会非常高。我后来的优化方向是引入向量召回做初筛把所有历史消息先向量化存起来每次要过滤时用当前任务目标和最近几条消息拼接成一个查询向量先在向量库里用余弦相似度召回 Top 50 候选消息再把候选消息交给 LLM 做精排打分。这个方案相当于把“全量 LLM 打分”降级为“Vector 粗筛 LLM 精排”。粗筛保证相关消息不会漏精排保证召回的候选消息里哪些真正进入上下文是由 LLM 决定的。我在一个内部测试数据集上跑过采用“召回 Top 50 LLM 精排”之后过滤准确率和全量打分基本持平但打分阶段消耗的 token 下降了约 80%。6.2 时间衰减与依赖加权让记忆管理再精细一点纯语义相关性有一个短板它不区分“刚刚发生的事”和“很久以前的事”。早期的一条消息即使语义相关也可能已经是过时信息。我在打分结果上叠加了一个时间衰减因子分数越靠近 3 分衰减越明显分数 4 分以上不衰减。这样设计是因为“直接决定成败的消息”即使过了很久也仍然重要而“背景补充类消息”会随任务进展逐渐失去价值。另一个可以考虑的维度是依赖加权。一条消息如果被后续多条消息引用过说明它是任务中的关键节点。我在消息结构里增加了一个ref_count字段构建上下文时如果某条消息被其他保留消息反复引用则它的权重自动提高一档。这个逻辑模仿的是人类记忆里的“反复被调用的事件记得更牢”。6.3 摘要金字塔低分消息的利用思路我在压缩环节做的是“低分消息直接丢弃”但这浪费了一个信息源。更好的做法是摘要金字塔不相关的低分消息不是扔掉而是分两层压缩——所有低于 3 分的消息合并成一份“阶段摘要”保留在上下文末尾如果预算仍然紧张阶段摘要再压缩成“全局摘要”只保留与任务目标直接相关的少量关键词。这个分层的好处是Agent 永远有一条最粗粒度的“记忆备份”。当某个新步骤需要回溯早期信息时至少有一份摘要兜底。虽然摘要丢失了精确值但能提示模型“这一步需要回到工具 X 查原始数据”而不是完全失忆。我目前在生产环境里启用了两层摘要实测下来对任务完成率没有负面影响却能把长期运行 Agent 的记忆成本压在一个很低的水平。我自己在这套手写 Agent 项目里最深的一个体会是上下文记忆管理和算法模型同等重要甚至更影响最终体验。模型本身再强喂进去的上下文是残缺的、混乱的输出能力也会被严重拉低。把“记忆管理”变成 Agent 的一项显式能力由 LLM 自己决定哪些信息值得记住是我在这轮迭代里做的最值当的工程决策。最后再分享一个实操小技巧给消息打分时别只把消息正文发给模型把消息类型和来源也拼接进去。比如一条 tool 消息前面标注“这是工具调用结果来自 SQL 查询”LLM 对它的相关性判断会准很多。这算是我踩了几次坑之后总结出的低成本高收益改进推荐你也试试。
阅读完成 · 觉得有帮助?
咨询建站