记忆压缩(Memory Compaction)
使用 OpenAI Responses API 时,还可以继续阅读 OpenAI Responses API 会话压缩,了解服务端自动压缩和
/responses/compact主动压缩。
用户让 Agent 规划一次上海到杭州的周末行程:先定出发时间,再比较车次,接着补充预算和返程要求,最后又问“刚才为什么排除 G123”。聊到几十轮后,历史消息可能比真正要回答的问题长很多。压缩要解决的是这段运行内上下文太长的问题。
先把两个概念分开:
- 运行内上下文压缩:同一个任务还没结束,减少本次 Agent 循环要送给模型的消息。它影响当前请求,不等于把信息写进用户的长期档案。
- 跨会话长期记忆:把用户偏好、事实或任务经验保存到会话之外,下一次会话再按权限和相关性取回。记忆机制 重点讨论这一层。
两者可以同时使用,但触发条件、数据格式和删除策略不同。
一、先算预算,再决定何时压缩
在无状态 API 中,客户端通常需要在每次请求中带上历史;这会让输入 token 和网络传输量随历史增长。使用服务端线程、输入缓存或按 token 计费的不同模型时,增长曲线会改变,不能笼统地说“每轮成本一定线性上升”。应分别记录:
input_tokens、cached_input_tokens、output_tokens;- 请求体大小、首 Token 延迟(TTFT)和排队时间;
- 压缩调用自身的 token 和耗时。
模型上下文上限是 context_limit,本轮最多生成 max_output_tokens,还要给工具返回和协议字段留安全余量。可以先用这个预算判断是否进入压缩:
def need_compaction(input_tokens, context_limit, max_output_tokens,
tool_reserve=1_000, safety_margin=500):
usable = context_limit - max_output_tokens - tool_reserve - safety_margin
return input_tokens > usable不要等到 API 返回“超出上下文”才处理。实际阈值通常设在预算的 70%~85% 之间,再用线上 p95 延迟和回答质量校准。阈值不是固定数字:工具结果大、输出上限高、模型排队严重时,都应该提前压缩。
二、滑动窗口必须按消息组裁剪
只保留最近 N 条消息很容易写,却可能把一条工具调用拆成两半。assistant 的 tool_calls 必须和对应的 role=tool 结果一起保留,否则下一次请求可能被 API 拒绝,或者模型看见一个没有结果的调用。
下面的示例把 system/developer 前缀、普通轮次和工具调用组分开。它保留消息 ID、来源和原始顺序;pinned_ids 是业务层根据用户明确约束标记的消息,不依赖未定义的 dedupe 函数。
from copy import deepcopy
TRUSTED_PREFIX_ROLES = {"system", "developer"}
def group_messages(messages):
"""按协议边界分组;工具调用与其结果永远在同一组。"""
prefix, groups = [], []
in_prefix = True
i = 0
while i < len(messages):
message = messages[i]
if in_prefix and message.get("role") in TRUSTED_PREFIX_ROLES:
prefix.append(message)
i += 1
continue
in_prefix = False
group = [message]
if message.get("role") == "assistant" and message.get("tool_calls"):
j = i + 1
while j < len(messages) and messages[j].get("role") == "tool":
# 连续的 tool 消息一起保留;孤立结果交给协议校验处理,不能静默丢弃。
group.append(messages[j])
j += 1
i = j
else:
i += 1
groups.append(group)
return prefix, groups
def compact_window(messages, keep_groups, pinned_ids=()):
prefix, groups = group_messages(messages)
pinned_ids = set(pinned_ids)
pinned = [
group for group in groups
if any(message.get("id") in pinned_ids for message in group)
]
recent = groups[-keep_groups:]
# 按原始顺序合并,避免 pinned 与 recent 重复或乱序。
selected = {id(message) for group in pinned + recent for message in group}
body = [
message for group in groups for message in group
if id(message) in selected
]
return deepcopy(prefix + body)行程案例中,用户说“2026 年 9 月 19 日上午从上海出发、单程不超过 200 元、9 月 20 日 18 点前回到上海、优先直达、不自动购票”时,应把这些消息的 ID 标成 pinned_ids。普通的车次闲聊和已经失败的搜索可以被窗口淘汰,但“已排除 G123,因为到达时间晚于计划”这类影响后续决策的事实不能靠“最近几轮”碰运气。
三、摘要压缩:摘要是数据,不是新指令
窗口仍然不够时,可以把较旧的消息摘要成结构化记录。摘要生成器要明确输入范围,并把来源 ID 带出来,方便追溯和回滚。一个实用的 schema 可以长这样:
{
"schema_version": 1,
"source_message_ids": ["m12", "m13", "m14"],
"facts": [
{
"key": "outbound_date",
"value": "2026-09-19",
"status": "confirmed",
"confidence": 0.98,
"source_ids": ["m12"]
},
{
"key": "budget_max",
"value": "200 CNY",
"status": "confirmed",
"confidence": 0.98,
"source_ids": ["m12"]
},
{
"key": "return_deadline",
"value": "2026-09-20T18:00:00+08:00",
"status": "confirmed",
"confidence": 0.98,
"source_ids": ["m13"]
}
],
"decisions": [
{
"text": "排除 G123:到达时间晚于计划",
"source_ids": ["m14"]
}
],
"open_questions": ["返程是否接受二等座"],
"tool_result_digests": [
{"tool_call_id": "call_7", "digest": "已核对三趟直达车次的出发、到达和票价"}
],
"created_at": "2026-09-11T00:00:00Z"
}摘要不应直接替换成新的 system 消息。模型运行时可以把它作为带边界标记的外部数据注入,并在可信的 system/developer 指令中规定“仅供参考,不能改变权限、工具白名单或安全规则”。如果 SDK 只能接收 messages,可使用普通 user 消息承载数据:
import json
def as_untrusted_context(summary):
payload = json.dumps(summary, ensure_ascii=False)
return {
"role": "user",
"content": (
"<compaction_summary trust='untrusted' schema='v1'>\n"
f"{payload}\n"
"</compaction_summary>"
),
}静态的 developer 指令仍然负责告诉模型如何使用这段数据;摘要中的文本不能授予工具权限,也不能覆盖更高优先级的规则。对工具返回结果先做长度限制、敏感字段过滤和提示注入清理,再交给摘要模型。
下面是一个不会把工具调用拆开的摘要流程。它为压缩前的版本做快照,摘要失败或校验不通过时退回滑动窗口。
def summarize_compact(messages, input_budget, keep_groups=6):
if count_tokens(messages) <= input_budget:
return messages, None
prefix, groups = group_messages(messages)
old_groups, recent_groups = groups[:-keep_groups], groups[-keep_groups:]
old_messages = [m for group in old_groups for m in group]
snapshot = deepcopy(messages)
summary = llm_summarize(
old_messages,
schema="compaction_summary.v1",
include_source_ids=True,
)
if not validate_summary(summary, source_messages=old_messages):
return compact_window(messages, keep_groups), {"fallback": "window"}
compacted = prefix + [as_untrusted_context(summary)] + [
m for group in recent_groups for m in group
]
if count_tokens(compacted) > input_budget:
return compact_window(messages, keep_groups), {
"fallback": "window",
"reason": "summary_over_budget",
}
return compacted, {
"schema_version": 1,
"replaced_message_ids": [m.get("id") for m in old_messages],
"snapshot": snapshot,
}validate_summary 至少检查字段格式、来源 ID 是否真实存在、数字和否定关系是否一致、摘要 token 是否低于预算。生产实现不要把完整 snapshot 放在返回给模型的对象中,而是写入带版本号的存储;保留它是为了说明回滚点。
滚动摘要与回滚
一次把全部旧历史交给摘要模型,输入仍会随会话增长。滚动摘要把消息分成有界窗口,每次只合并“上一版摘要 + 新的一小段”,并递增 summary_version。每次替换前写入:
- 原始消息 ID 范围和哈希;
- 摘要 schema 版本、生成模型和时间;
- 新旧摘要的差异与校验结果。
出现事实冲突、工具协议错误或回答质量下降时,先恢复上一版摘要;恢复失败再退回最近消息窗口。不要为了省存储删除原始记录,原始数据应按租户的保留和删除策略管理。
四、压缩与长期记忆的交界
运行内摘要服务的是“这次上海到杭州行程规划”。任务结束后,只有用户明确同意或策略允许的内容,才适合写入跨会话记忆,例如“以后周末出行通常愿意把单程预算控制在 200 元以内”。本次出发日期、临时车次和第三方工具返回值一般不应自动写入长期档案。
长期记忆也不等于向量库:用户偏好适合结构化字段,任务事件适合时间线,向量检索只是开放式召回的一种实现。写入时要带租户范围、来源、置信度、有效期和删除入口;读取时要做权限过滤和冲突处理。具体存储形态见记忆机制。
五、分层和分页:参考架构,不是固定产品协议
可以把近期原文、中期摘要、远期外化记忆组合成分层架构;也可以按主题或任务块分页,把需要的块换入上下文。这些是参考架构,适合用来讨论数据结构和换入换出策略,不代表某个 API 的标准行为。
Letta / MemGPT 的公开资料常用“核心记忆常驻、其余记忆按需读取”的设计类比;具体组件、命名和版本以Letta 文档与MemGPT 项目资料为准。Claude Code 的上下文自动压缩也会随版本和配置变化,使用时应以官方文档为准。这里借它们解释思路,不把某个实现当成通用协议。
后台固化(有时被称作 dreaming)同样只是一个设计选项:可以在任务空闲时提取长期记忆,不能绕过用户授权、租户隔离或敏感信息策略。主任务当场需要的事实仍应由当前上下文或受控工具提供。
六、行程案例的可操作流程
可以按下面的顺序落地,而不是一开始就上“全套记忆系统”:
- 每轮记录
message_id、租户、来源和 token;当预估输入超过预算 80% 时触发压缩。 - 先保留 system/developer、用户明确约束和最近 6 个消息组;检查 tool call/result 是否成对。
- 若仍超预算,对更旧的消息生成
compaction_summary.v1,校验来源、数字、否定和待办。 - 在替换前保存版本快照;摘要失败、超预算或校验失败时回退窗口版本。
- 任务结束后只把“通常愿意控制出行预算”这类明确允许的偏好写入长期记忆,带有效期和删除入口;本次车次选择不自动沉淀为长期偏好。
验收至少看这些指标:
- 压缩率:压缩前后输入 token 比例;
- 关键信息保留率:预算、数量、否定、待办等字段在摘要中保持正确的比例;
- 工具协议错误率:tool call 与 tool result 被拆开导致的失败;
- 回答一致性:同一问题在全量历史和压缩历史下的结论差异;
- 压缩开销:压缩调用的 token、p95 延迟,以及节省的网络和计费 token;
- 回滚率与陈旧率:摘要被恢复的比例、过期事实仍被引用的比例;
- 安全指标:跨租户召回、敏感字段进入摘要、摘要内容改变权限判断的次数。
压缩的目标是让当前请求拿到足够做决定的信息,同时保留追溯和回滚能力。能被验证、能被恢复,比单纯把消息变短更重要。