长任务Agent上下文丢失怎么办?prime-agent的监控、压缩与恢复方案 如果你用 Agent 跑过那种“一口气完成几十步”的任务一定见过这样的场景任务刚开始的时候Agent 思路清晰每一步都记得住之前确认过的结论。跑到二十多步它开始反复问同一个问题——这个问题明明在第一步就已经说清楚了。再往后它甚至会对早已经否决掉的方案重新给出建议仿佛换了一个新会话。这不是个别模型的 bug。长任务跑着跑着就丢上下文是当前 Agent 工程里最普遍的痛点之一。更准确地说这不一定是因为“模型变笨了”而是因为上下文没有被当作工程资源来管理。它本来应该像数据库事务一样可控但大多数 Agent 应用只是简单地把历史消息越堆越多直到溢出、截断、遗忘。prime-agent 这类项目出现的原因正是要把上下文从“一次性输入”变成“可监控、可压缩、可恢复”的资源。这篇文章会先讲清楚长任务丢上下文的底层机制再拆解上下文治理的几种主流路线最后用一个可运行的最小原型演示监控、压缩、检查点这三个关键动作是怎么落地的。无论你是正在搭 Agent 应用的工程师还是被长任务折磨过的使用者都能按这篇文章的思路把问题定位到具体环节。核心判断先放在这里解决长任务上下文问题不一定需要换更大上下文窗口的模型更靠谱的方向是把上下文管理拆成可观测、可压缩、可恢复三个动作。谁先把这三个动作做扎实谁的长任务稳定性就会有本质提升。1. 长任务丢上下文的真实场景1.1 三个典型的“失忆”现场场景一批量代码重构。你让 Agent 把一个老项目的工具类从 Java 迁移到 Kotlin涉及几十个文件。前五个文件它记得住统一的命名规范和异常处理风格到第十五个文件它开始使用另一种风格和前面完全不一致。你回看对话发现规范讨论早就在阶段切换时被截断了。场景二多阶段数据迁移。Agent 第一步读取了源表结构第二步确认了主键和去重逻辑。跑到第五步生成数据校验脚本时它把主键记错了因为它之后又读过其他几张表早期的表结构信息被挤出了有效范围。场景三自动化测试修复。Agent 在修一个测试用例先查了日志又改了代码再跑测试。当测试再次失败时它没有结合之前的失败原因而是重新开始猜甚至把已经在第三步验证过的错误方案又试了一遍。这三个场景的共同点不是模型没有能力而是任务需要的信息总量超过了模型当前能看到和有效使用的范围或者关键信息在调度过程中被静默丢弃了。上下文对回答的影响在长任务里会被放大得极其明显。1.2 为什么长任务特别容易触发Agent 每一轮都在消耗上下文。用户指令、模型回复、工具调用结果、代码片段、报错日志全部追加到对话里。Token 数量随着轮数近似线性增长而任务复杂度稍微提高几轮工具调用就会消耗成千上万个 token。更麻烦的是工具结果。一次 grep、一次读文件、一次编译输出往往就是几百上千 token。如果 Agent 习惯把完整结果原样塞进对话二十轮之后上下文窗口就会非常紧张。这也是“workbuddy 上下文用量满了怎么办”“LM Studio 如何设置上下文”这类问题频繁出现的原因——很多人先遇到的不是模型能力问题而是上下文被无意识消耗完了。1.3 丢上下文的底层机制这里要区分三种情况显式截断开发者在代码里用history[-n:]只保留最近几轮最前面的决策被直接删除。隐式丢失模型在长上下文中对中间部分注意力下降学术界常称为 “Lost in the Middle”。摘要失真用摘要代替原始历史后摘要没有保留关键约束模型以为自己知道其实知道的是删减版。三种情况经常叠加。长任务丢上下文几乎不会只有一个原因。所以在动手解决之前先搞清楚当前系统究竟是哪一种丢失比盲目换模型更重要。2. 上下文机制基础Agent 是如何“记住”事情的2.1 一段上下文的构成在 Agent 应用中发往模型的内容可以拆成四块System Prompt系统指令包括角色、任务边界、输出格式。用户的原始指令任务目标本身。历史对话用户与模型的交互记录。工具调用记录工具参数、返回值、报错信息。模型没有独立的“记忆”它只能基于本次请求拼接出来的上下文做推理。所以讨论上下文管理本质上是在讨论拼接策略哪些内容放进去、哪些内容压缩后放进去、哪些内容彻底不放进去。这个视角很重要。很多人以为“上下文”是模型自带的记忆能力实际上它是每次请求时我们亲手组装的输入。App 负责决定模型“记得”什么责任在工程侧不在模型侧。2.2 上下文窗口既是容量也是成本上下文窗口是模型一次能接收的最大 token 数。窗口越大一次能处理的信息越多但成本和延迟也越高。长任务里真正的问题是无用信息占满窗口关键信息反而放不进去。更重要的是对上下文的理解要换一下——它不只是容量而是任务的“工作记忆”。工作记忆不仅要能存还要能在需要时被有效访问。窗口大到 200k也不代表模型能有效利用中间那 100k token 的信息。这就带来一个工程判断把历史做结构化、分层、摘要化比单纯扩大窗口更可控。2.3 三个容易混淆的概念概念含义作用范围上下文Context本次请求实际发送给模型的文本当前请求记忆Memory跨请求保留和恢复信息的机制多个请求知识Knowledge模型参数中固化或外部检索到的信息全局Agent 真正需要的是“记忆”但实现记忆最直接的方式还是操作上下文。prime-agent 这类工具的切入点就在这里在上下文层面实现记忆效果让长任务在多次请求之间保持稳定的信息状态。3. prime-agent 的核心设计思路管理上下文而不是延长上下文3.1 三种主流治理路线目前行业里应对长任务上下文问题主要有三条路线路线核心做法优点代价扩大窗口换用更长上下文的模型短期直接解决容量问题成本高中段信息仍可能丢失上下文压缩把早期历史摘要化保留决策控制成本摘要可能丢细节检索增强历史写入外部存储按需召回扩展性好可支撑超长任务工程复杂度高召回质量影响大三条路线并不互斥。成熟系统通常是“压缩 检索”配合使用窗口大小只是兜底上限。3.2 prime-agent 的核心能力定位从项目定位看prime-agent 主要解决的是“长任务持续运行时的上下文可靠性”问题核心动作可以归纳为三件事用量可观测实时知道当前上下文用了多少距离上限还有多少。阶段性压缩在关键节点把历史对话压缩成结构化摘要保住决策、待办、关键参数。检查点恢复任务中断或超限后能从最近一个一致状态继续而不是从头再来。这三件事正好对应长任务丢上下文的三个根源无感知、无压缩、无恢复。值得强调的是这三件事并不依赖某个特定大模型属于通用工程能力。具体的压缩触发时机、摘要格式、检查点保存策略每个项目可以根据自身任务类型调整。3.3 与直接使用大上下文窗口的区别如果你只是把模型从 32k 换成 200k短期看确实能装下更多历史。但窗口变大并不意味着信息被有效利用长上下文中间部分的信息召回率会下降。窗口只是容量不是检索能力。更重要的问题是成本每轮请求都携带全部历史token 消耗会在长任务里快速膨胀。在 Agent 场景里正确的目标不是“所有历史都在窗口里”而是“每个阶段需要的信息都能以低成本拿到”。这也是 prime-agent 把压缩与检查点作为核心能力的原因——先用摘要留住高价值信息再用检查点保证可恢复最后才考虑窗口容量。4. 环境准备与最小原型设计4.1 环境要求下面要演示的最小原型只需 Python 3.10 以上依赖两个库tiktoken用于 token 计数与窗口用量估算。openai用于调用摘要压缩模型。安装命令python -m pip install tiktoken openai如果你连接的模型服务兼容 OpenAI 接口openai库同样可用通过base_url和api_key指向本地或私有部署服务即可。例如from openai import OpenAI client OpenAI( api_keysk-xxxx, base_urlhttp://localhost:8000/v1, )需要特别说明下面代码是通用工程原型用来演示上下文治理的设计思路不等同于 prime-agent 项目的官方实现。模型名称和窗口大小请以你实际使用的模型为准本文重点在机制本身。4.2 原型目录结构context-agent-demo/ ├── context_monitor.py # 上下文用量监控 ├── history_compressor.py # 历史压缩 ├── checkpoint_manager.py # 检查点持久化 ├── main.py # 组装构建发送给模型的 messages └── test_run.py # 模拟长任务验证5. 核心流程拆解从监控到恢复5.1 第一步上线前先能“看到”上下文如果不测量就不知道上下文在哪一步超限。监控要做两件事统计当前 messages 的 token 总量。计算占窗口上限的比例并设置告警阈值。关键点统计时要包括 system prompt、历史、工具返回。很多人只统计业务对话漏掉工具结果导致实际用量远高于预期。上下文用量满了才发现问题往往已经晚了应该在任务运行过程中实时暴露。5.2 第二步设计压缩触发策略压缩不是越频繁越好。压缩本身是一次模型调用有成本和延迟而且过度压缩会让模型丢失细节。常见策略有三种阈值触发用量超过窗口 70% 时触发。阶段触发任务每完成一个明确阶段时触发。事件触发出现大段工具返回、报错日志时对那段内容即时裁剪。实践中更推荐“阈值触发 阶段触发”结合。阈值触发负责兜底阶段触发负责在任务自然边界做更高质量的整理。5.3 第三步把历史压成“决策/待办/关键信息”摘要压缩不是简单“总结一下”。有效的历史压缩要回答三个问题哪些结论已经确定后续不能改变哪些事情还没做下一步做什么哪些参数、路径、限定条件必须原样保留把这三个问题写进压缩提示词压缩结果才有工程价值而不是一段泛泛的剧情简介。如果只让模型“总结对话”它会倾向于把过程描述出来而不是把决策、待办、约束拆出来这样的摘要对后续步骤没有约束力。5.4 第四步检查点保存与恢复检查点本质上是“任务状态快照”。设计要点如下在每步成功完成之后保存不要在失败状态保存。保存内容包括结构化摘要、最近 N 轮对话、当前任务 ID、更新时间。加载检查点后要把摘要作为 system 消息注入而不是简单追加历史。这样即使进程崩溃、API 超时、本地重启任务也能从中断处继续。这也回应了“如何保存当前上下文下次打开时可以继续”这类需求——检查点就是答案。6. 完整示例代码一个可运行的上下文治理原型6.1 上下文用量监控# context_monitor.py import tiktoken class ContextMonitor: 统计 messages 的 token 总量并计算窗口占用比例。 def __init__(self, model: str gpt-4o): self.encoder tiktoken.encoding_for_model(model) # 常见模型的窗口大小不同服务商可能有差异请以官方文档为准 self._limits { gpt-4o: 128000, gpt-4o-mini: 128000, gpt-4: 8192, gpt-3.5-turbo: 16385, } self.limit self._limits.get(model, 128000) def count_text(self, text: str) - int: return len(self.encoder.encode(text)) def count_messages(self, messages: list[dict]) - int: total 0 for message in messages: content message.get(content) if content: total self.count_text(str(content)) return total def usage_ratio(self, messages: list[dict]) - float: return self.count_messages(messages) / self.limit关键逻辑count_messages遍历所有消息的 content 字段统计的是“实际要发给模型的文本量”。这里没有计算角色字段和元信息量级影响不大但在真实系统里可以把 overhead 也加进去。6.2 历史摘要压缩# history_compressor.py from openai import OpenAI client OpenAI() COMPRESS_PROMPT 请把下面的对话历史压缩成结构化摘要。 必须保留 1. 已经确认的决策、结论、约束条件 2. 尚未完成的任务和下一步计划 3. 关键文件路径、字段名、命令、参数 4. 被明确否决过的方案防止模型再次采用。 不要保留 - 寒暄、重复解释、中间试探过程。 输出格式 决策... 待办... 关键信息... 否决过的方案... 对话历史 {history} def compress_history(history: list[dict], max_chars: int 20000) - str: text \n.join( f{m[role]}: {m[content]} for m in history if m.get(content) ) # 防止历史太长超过摘要模型自己的窗口先截断 text text[-max_chars:] resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是一个对话历史压缩引擎只做信息保真压缩。}, {role: user, content: COMPRESS_PROMPT.format(historytext)}, ], temperature0.2, ) return resp.choices[0].message.content这里把“决策 / 待办 / 关键信息 / 否决过的方案”做成固定格式。固定格式不是形式主义后续模型读取摘要时能够快速定位“决策”和“否决过的方案”比读一大段自然语言摘要可靠得多。“否决过的方案”尤其重要因为 Agent 丢上下文后最容易犯的错误就是重新采用已经被否定的方案。6.3 检查点持久化# checkpoint_manager.py import json import time from pathlib import Path class CheckpointManager: def __init__(self, workdir: str .agent_state): self.dir Path(workdir) self.dir.mkdir(exist_okTrue) def save(self, task_id: str, state: dict) - str: payload { updated_at: time.time(), task_id: task_id, state: state, } path self.dir / f{task_id}.json path.write_text( json.dumps(payload, ensure_asciiFalse, indent2), encodingutf-8, ) return str(path) def load(self, task_id: str) - dict | None: path self.dir / f{task_id}.json if not path.exists(): return None payload json.loads(path.read_text(encodingutf-8)) return payload.get(state)检查点文件里存的是可 JSON 序列化的状态包括摘要和最近轮次。实际项目中如果状态里有非序列化对象要先转换成字典再保存。6.4 组装主流程# main.py from context_monitor import ContextMonitor from history_compressor import compress_history from checkpoint_manager import CheckpointManager monitor ContextMonitor(modelgpt-4o) checkpoints CheckpointManager(workdir.agent_state) # 保留最近多少轮原始消息 KEEP_RECENT_TURNS 6 # 窗口占用超过该比例时触发压缩 MAX_RATIO 0.7 def build_messages(task_id: str, new_user_content: str, history: list[dict]) - list[dict]: checkpoint checkpoints.load(task_id) if checkpoint: # 恢复用摘要压住早期历史只保留最近几轮 compact_history [ {role: system, content: f此前任务阶段摘要\n{checkpoint[summary]}} ] history[-KEEP_RECENT_TURNS:] else: compact_history history messages compact_history [{role: user, content: new_user_content}] if monitor.usage_ratio(messages) MAX_RATIO: summary compress_history(compact_history) checkpoints.save(task_id, { summary: summary, recent_history: history[-KEEP_RECENT_TURNS:], progress: 已触发一次阶段压缩, }) messages [ {role: system, content: f此前任务阶段摘要\n{summary}}, {role: user, content: new_user_content}, ] print(f[context] current usage: {monitor.usage_ratio(messages):.1%}) return messages这个流程把监控、压缩、检查点串起来了先看有没有检查点有就恢复超过阈值就压缩并保存检查点否则正常返回 messages。注意压缩后最近几轮历史和摘要之间可能有信息重叠这在原型阶段是可接受的生产系统里可以做去重或更精细的截断。6.5 模拟长任务验证脚本# test_run.py from main import build_messages # 模拟一个已经执行了很多轮的历史 fake_history [] for step in range(1, 31): fake_history.append({ role: user, content: f第 {step} 步处理订单数据约束是只读取测试库不使用 DELETE 语句。, }) fake_history.append({ role: assistant, content: f第 {step} 步完成字段 order_id/user_id/amount/status 已确认。, }) # 第 31 步继续任务 messages build_messages(task_order_001, 继续第 31 步生成对账脚本, fake_history) print(发送给模型的 messages 数量:, len(messages)) for m in messages[:2]: print(m[role], m[content][:120])第一次运行时会因为历史太长触发压缩