Agent 干得越久,历史越肥:一次读文件就是几千 token 进历史,而循环的每一轮都要全量重发。第 6 篇装上的读文件工具正是头号推手——单次调用几千 token,读三个文件历史就破万。本文是「从零手写 Agent」系列第 7 篇,治理 Agent 的两大绝症——失忆(进程一死记忆清零)与破产(历史无限膨胀),实现摘要压缩与磁盘持久化。
本文按"膨胀的账单 → 压缩的艺术 → 持久化 → 实测与总结"组织。
1、概述:账单膨胀在 Agent 下的恶化
系列第 2 篇算过聊天场景的账:十轮对话九成输入在为重复历史付费。Agent 场景下恶化一个量级——工具结果是实打实的大块文本,几次代码探索后历史轻松过万 token,而 Agent 循环每转一轮,这些历史就重发一遍。不加治理,长任务的结局只有两种:撑爆上下文窗口报错,或者账单爆炸。
先把一个概念摆正:上下文窗口是工作台,不是仓库。摊在工作台上的资料每一轮都重新计费,把"需要长期记住的东西"堆在窗口里,等于按轮数给同一份资料付复利——治理的本质是把该搬走的搬走。
治理思路对应两招:历史太长就压缩——把旧对话变成摘要;进程要死就持久化——把历史写盘,下次读回来。
1.1、token 统计实现
度量先于治理:没有 token 计数,压缩何时触发、效果如何都无从谈起。估算函数的实现如下:
import json, re
def estimate_tokens(messages) -> int:
"""粗估口径:中文约 1 字 1 token,其他字符约 4 个折 1 token。"""
text = json.dumps(messages, ensure_ascii=False)
cjk = len(re.findall(r"[一-鿿]", text))
return int((cjk + (len(text) - cjk) / 4) * 1.3) # 1.3:结构开销系数
解读:口径按本项目实测校准,误差在两成以内——触发压缩阈值够用,拿它结算账单不行。精确口径有两个:每个响应自带的 usage 字段是真实消耗的事后实报;官方的 count_tokens 接口可以在发送前精确预算,代价是多一次网络往返,接口说明见 token 计数文档。工程折中是"估算触发、实报记账":压缩阈值用估算函数判,账单按 usage 累加。估算还有一层好处:纯本地计算,不花钱不花网络往返,可以每轮都跑;精确接口反而只适合低频抽查。
1.2、三招的分工
压缩、持久化、分级记忆各管一段,互不替代。对比如下:
| 手段 | 解决什么 | 代价 |
|---|---|---|
| 压缩 | 单会话内历史过长 | 有损,细节会丢 |
| 持久化 | 进程重启即失忆 | 无损,但僵尸历史照涨 |
| 分级记忆 | 摘要也保不住的关键事实 | 要多维护一份存储 |
解读:三者是递进关系——压缩管会话内,持久化管会话间,分级记忆把最不该丢的事实彻底搬出对话流。省钱还有第四条路:prompt caching,缓存命中的输入按远低于基础价的折扣计费;但缓存的是前缀、省的是钱不是窗口,与压缩正交,配置说明见 prompt caching 文档。第 4 节会看到它与压缩的一次正面冲突。
2、压缩的艺术
压缩的通用策略是"摘要加尾巴":把历史切成旧消息与最近几条,旧的部分让模型自己写摘要,新的部分原样保留。新历史的结构如下:
def compress_history(client, model, history, keep_tail=6):
cut = _safe_cut_index(history, len(history) - keep_tail)
old_part, recent = history[:cut], history[cut:]
resp = client.messages.create(
model=model, max_tokens=1024,
messages=[{"role": "user", "content": "请把以下对话历史压缩成摘要,"
"必须保留用户偏好、任务结论、关键数字与待办。\n\n"
+ json.dumps(old_part, ensure_ascii=False)[:30000]}],
)
summary = "".join(b.text for b in resp.content if b.type == "text")
return [
{"role": "user", "content": f"[此前对话的摘要]\n{summary}"},
{"role": "assistant", "content": "(已了解背景,继续任务)"},
*recent,
]
解读:摘要由模型自己写,摘要提示词点名要保留什么——用户是谁、做完了什么、关键数字、待办事项。摘要做成一问一答两条消息,是为了满足角色交替的协议要求。压缩是有损交易:主干保住、细节丢失,摘要提示词就是你在"记忆与成本"上的落点。
摘要提示词值得反复打磨:保留清单每多一项,活下来的记忆就多一类。实战里值得点名保留的还有三样——已确认的文件路径、失败过的尝试与原因、用户的否定反馈。尤其"失败记录":不点名的话,摘要倾向于只写进展,踩过的坑被压掉,同一个弯路再拐一次,是长任务里最冤枉的重复开销。
keep_tail 的取值同样是设计点而非玄学:尾部太短,正在进行的工具链会被切进摘要区——刚发出的调用还没等到结果就被"总结"掉,下一步决策失去依据。经验值是保留最近三组工具往返加一条用户消息,任务节奏快的场景再往上加。尾部消息永远原样在场,这是"压缩只动过去、不伤当下"的底线。
2.1、切口避开工具配对
压缩有一个隐蔽的协议大坑:切口不能拆散工具配对。历史里助手消息(含调用块)与紧随的用户消息(含结果块)是一对,切口落在这对中间,接口直接报 400。_safe_cut_index 的职责就是发现切口落在结果消息上时往前退一位,让这对搭档整体留在保留区。实现如下:
def _safe_cut_index(history, cut: int) -> int:
"""切口落在工具结果消息上时,回退到其配对的助手调用之前。"""
if cut <= 0 or history[cut].get("role") != "user":
return cut # 切口不在结果消息上,安全
prev = history[cut - 1]
if prev.get("role") == "assistant" and any(
b.get("type") == "tool_use" for b in prev["content"]):
return cut - 1 # 调用与结果同进保留区
return cut
解读:判断只看两条——切口处是不是 user 消息、它前面是不是带 tool_use 块的助手消息,命中即回退一位,这对搭档整体留在保留区。接口报 400 时的错误信息通常直指"tool_result 缺少对应的 tool_use",看到它先查切口。这个函数不到十行,却是压缩功能上线前必须过的协议门。
2.2、接进 Agent 循环
压缩要有触发时机,工程上挂在循环每轮请求模型之前。自动触发的实现如下:
COMPRESS_THRESHOLD = 6000 # 估算 token 越线即压缩
KEEP_TAIL = 6 # 保留最近 6 条消息
def maybe_compress(agent):
if estimate_tokens(agent.history) <= COMPRESS_THRESHOLD:
return # 未越线,不动历史
before = estimate_tokens(agent.history)
agent.history = compress_history(
agent.client, agent.model, agent.history, keep_tail=KEEP_TAIL)
print(f"[压缩] {before} → {estimate_tokens(agent.history)} token")
解读:三行打印让每次压缩的账面变化可见——省了多少,一眼可查。阈值要给"尾部保留加本轮新增"留足余量,否则刚压完又越线,来回抖动;实测把阈值放在窗口八成以下是安全区(本项目口径)。至此三件套合体:循环每轮先 maybe_compress 再请求模型,任务结束落盘。
3、持久化与分级记忆
持久化朴素得出人意料:所谓跨会话记忆,就是把 messages 列表 JSON 写盘,下次启动读回来。聊天产品的历史记录、编码工具的会话恢复,数据结构层面都是这一个动作。
def save_history(history, path):
Path(path).write_text(json.dumps(history, ensure_ascii=False, indent=1),
encoding="utf-8")
def load_history(path):
return json.loads(Path(path).read_text(encoding="utf-8"))
解读:两个函数就是完整的"存档读档"。值得进阶的是分级记忆:把"用户叫什么、偏好什么"这类长期事实从对话历史里抽出来,单独存一个事实列表,每次请求拼进系统提示词。工作记忆(随压缩衰减)与长期记忆(永不遗忘、不占历史)分离,是各家记忆功能的共同骨架。
3.1、长期事实独立存取
把"值得永生"的事实从对话流里搬出来单独存,实现如下:
import json
from pathlib import Path
FACTS_PATH = Path("memory/facts.json")
def load_facts() -> list[str]:
if FACTS_PATH.exists():
return json.loads(FACTS_PATH.read_text(encoding="utf-8"))
return []
def add_fact(fact: str):
facts = load_facts()
if fact not in facts: # 去重,防同一事实反复堆积
facts.append(fact)
FACTS_PATH.write_text(json.dumps(
facts, ensure_ascii=False, indent=1), encoding="utf-8")
def build_system_prompt(base: str) -> str:
facts = "\n".join(f"- {f}" for f in load_facts())
return f"{base}\n\n已知长期事实:\n{facts}" if facts else base
解读:长期事实每次请求拼进系统提示词——不占对话历史,也就永远不会被压缩吃掉。add_fact 的写入口有两种做法:做成一个显式的"记事"工具让模型随时调用,或在会话结束时跑一次模型抽取。第 4 节实测里"自发写资料文件"那个意外收获,其实就是模型自己发明了分级记忆的雏形。生产系统里吹得天花乱坠的记忆目录,骨架就是这一个 JSON 文件。
事实列表也有自己的膨胀问题:会话越攒越多,事实迟早长成第二份历史。两个廉价的治理手段:一是合并——定期让模型把事实列表本身压一遍,"喜欢深色主题"与"偏好暗色"并成一条;二是淘汰——每条事实记一个最后使用时间,长期没被引用的降权或清除。长期记忆不设治理,就会从"记忆"变成"包袱"。
4、实测记录
一次完整剧情的实测数据如下:
自我介绍(埋记忆点:名字与偏好)
读三个源码文件喂肥历史 → 估算 6925 token,触发压缩 → 3770
再读一个文件 → 5480,再次压缩 → 2276
压缩后提问"我叫什么、喜欢什么颜色" → 正确答出
提问"第一个文件里 run 方法在第几行" → 诚实回答不记得细节
新进程读档后提问 → 认识用户
解读:三次压缩把历史稳定控制在四千估算 token 以内;名字与偏好活过了压缩(摘要提示词点名保留),行号细节丢了(摘要不点名就丢)——记得什么、忘了什么,由摘要提示词决定,这是压缩器的可编程接口。一个意外收获:模型在自我介绍环节自发调用写文件工具,把用户信息写进了磁盘上的资料文件——没有被要求的持久化行为自然涌现,工具与记忆的组合会催生设计之外的能力。
再从账单角度复盘:首次压缩把估算 token 从 6925 降到 3770,单轮省下三千余;此后每轮循环都按这个差值复利——一个还要再转十轮的任务,累计能省三万 token 量级(估算口径)。压缩的收益随剩余轮数放大,这正是它必须挂进循环、而不是任务结束时做一次的原因。
有一个要留心的冲突:压缩会改写历史前缀,而 prompt caching 恰恰依赖前缀稳定——每压一次,缓存就失效重建一次,缓存写入的加价反而成了新开销。两招打架的折中是拉长压缩间隔、把阈值放宽松些,让"省窗口"与"省钱"的调度错开;缓存档位与失效规则以官方文档为准。
最后是一个容易被忽略的工程习惯:压缩与落盘都要写日志。哪个任务、第几轮、压掉多少、还剩多少——这些数字平时没人看,出问题时是唯一的现场。上面那张"剧情表",就是照着压缩日志整理出来的;没有日志的压缩,等于黑箱里做有损操作。
5、总结
上下文管理的两招都是朴素工程:压缩是"摘要加尾巴",用有损换省钱,切口必须避开工具配对;持久化是"列表落盘再读回",分级记忆把长期事实从对话流里分离。对于准备上生产的读者,建议先加 token 估算与自动压缩,再考虑 prompt caching 与分级记忆的组合。下一篇让 Agent 学会分身:子 Agent 派工与上下文隔离。上下文管理设计可参考 Anthropic 文档。

