代码讲解.md 13 KB
S08 Context Compact 代码讲解
这一节只讲相对 S07 新增的内容：
上下文会越来越长，所以 Agent 在调用模型之前，要先把旧消息和大工具结果变小。
1. 本节新增内容
CONTEXT_LIMIT：触发自动摘要压缩的上下文大小阈值。KEEP_RECENT：保留最近几个完整tool_result。PERSIST_THRESHOLD：单个工具输出超过多大时写入磁盘。snip_compact()：裁剪中间历史。micro_compact()：把旧工具结果换成占位符。tool_result_budget()：处理最后一轮过大的工具输出。compact_history()：调用 LLM 总结完整历史。reactive_compact()：API 报上下文过长时兜底压缩。compact工具：让模型主动请求压缩。
2. Messages 数据结构
messages 是：
messages: list[dict]
一段带工具调用的历史通常长这样：
messages = [
{"role": "user", "content": "请读取 README.md"},
{"role": "assistant", "content": [tool_use_block]},
{"role": "user", "content": [tool_result_block]},
]
tool_result_block 是字典：
{
"type": "tool_result",
"tool_use_id": "toolu_01xxx",
"content": "工具返回的大段文本"
}
3. 四层压缩顺序
在 agent_loop() 里，压缩发生在调用模型之前：
messages[:] = tool_result_budget(messages)
messages[:] = snip_compact(messages)
messages[:] = micro_compact(messages)
if estimate_size(messages) > CONTEXT_LIMIT:
messages[:] = compact_history(messages)
执行顺序是：
L3 大工具结果预算 -> L1 裁剪中间历史 -> L2 压缩旧工具结果 -> L4 LLM 摘要
核心原则：
先用便宜的办法压缩，最后才调用 LLM 做摘要。
4. tool_result_budget(messages)
目标：
如果最后一轮工具结果总大小超过 max_bytes，就优先把最大的工具输出写入磁盘。
关键代码：
last = messages[-1] if messages else None
等价于：
if messages:
last = messages[-1]
else:
last = None
这里的 last 预期是：
last: dict = {
"role": "user",
"content": [tool_result_block, tool_result_block]
}
筛选工具结果：
blocks = [
(i, b)
for i, b in enumerate(last["content"])
if isinstance(b, dict) and b.get("type") == "tool_result"
]
blocks 的类型可以理解成：
blocks: list[tuple[int, dict]]
例如：
[
(0, {"type": "tool_result", "tool_use_id": "toolu_01", "content": "..."}),
(1, {"type": "tool_result", "tool_use_id": "toolu_02", "content": "..."})
]
计算总大小：
total = sum(len(str(b.get("content", ""))) for _, b in blocks)
_ 表示这个值不用。这里不关心下标，只关心 b 这个工具结果字典。
按输出长度排序：
ranked = sorted(
blocks,
key=lambda p: len(str(p[1].get("content", ""))),
reverse=True
)
p 是 (i, b)，所以：
p[0] # 下标
p[1] # tool_result 字典
排序后，最大的工具输出排在前面，优先被写入磁盘。
5. persist_large_output(tool_use_id, output)
压缩前：
{
"type": "tool_result",
"tool_use_id": "toolu_01",
"content": "非常非常长的输出......"
}
压缩后：
{
"type": "tool_result",
"tool_use_id": "toolu_01",
"content": "<persisted-output>文件路径 + Preview</persisted-output>"
}
完整输出在磁盘：
.task_outputs/tool-results/toolu_01.txt
上下文里只保留路径和前 2000 字符预览。
6. snip_compact(messages)
作用：消息数量超过阈值时，裁掉中间历史，只保留：
开头几条消息 + 一个裁剪标记 + 最近几条消息
这样模型还能知道“最开始的任务是什么”，也能看到“最近正在做什么”，中间太长的历史就先省掉。
源码：
def snip_compact(messages, max_messages=50):
if len(messages) <= max_messages: return messages
keep_head, keep_tail = 3, max_messages - 3
head_end, tail_start = keep_head, len(messages) - keep_tail
if head_end > 0 and _message_has_tool_use(messages[head_end - 1]):
while head_end < len(messages) and _is_tool_result_message(messages[head_end]):
head_end += 1
if (tail_start > 0 and tail_start < len(messages)
and _is_tool_result_message(messages[tail_start])
and _message_has_tool_use(messages[tail_start - 1])):
tail_start -= 1
if head_end >= tail_start:
return messages
snipped = tail_start - head_end
return messages[:head_end] + [{"role": "user", "content": f"[snipped {snipped} messages]"}] + messages[tail_start:]
6.1 最简单的裁剪例子
为了方便理解，先把 max_messages 想成 8。
假设现在有 12 条消息：
messages = [
"msg0", "msg1", "msg2",
"msg3", "msg4", "msg5", "msg6",
"msg7", "msg8", "msg9", "msg10", "msg11"
]
代码第一步：
keep_head, keep_tail = 3, max_messages - 3
如果 max_messages = 8：
keep_head = 3
keep_tail = 8 - 3 = 5
意思是：
保留开头 3 条
保留结尾 5 条
再看这行：
head_end, tail_start = keep_head, len(messages) - keep_tail
代入数字：
head_end = 3
tail_start = 12 - 5 = 7
这两个变量可以这样理解：
head_end = 开头保留到哪里结束
tail_start = 结尾从哪里开始保留
对应到列表下标：
messages[:head_end]
就是：
messages[:3] = ["msg0", "msg1", "msg2"]
而：
messages[tail_start:]
就是：
messages[7:] = ["msg7", "msg8", "msg9", "msg10", "msg11"]
中间被裁掉的是：
messages[3:7] = ["msg3", "msg4", "msg5", "msg6"]
所以：
snipped = tail_start - head_end
就是：
snipped = 7 - 3 = 4
最终结果变成：
[
"msg0", "msg1", "msg2",
{"role": "user", "content": "[snipped 4 messages]"},
"msg7", "msg8", "msg9", "msg10", "msg11"
]
这就是 snip_compact() 的基本逻辑。
6.2 为什么变量叫 head_end 和 tail_start
这两个名字是从“切片边界”来的。
messages[:head_end]
表示保留头部。
messages[tail_start:]
表示保留尾部。
中间要裁掉的区间是：
messages[head_end:tail_start]
所以：
head_end
是头部保留区的结束位置。
tail_start
是尾部保留区的开始位置。
6.3 为什么不能直接裁剪
普通聊天消息可以直接裁。
但工具调用消息有配对关系：
assistant: tool_use
user: tool_result
例子：
messages = [
{"role": "user", "content": "读取文件"},
{"role": "assistant", "content": [tool_use_read_file]},
{"role": "user", "content": [tool_result_read_file]},
]
这里第 1 条和第 2 条是一组：
模型说：我要调用工具
程序说：这是工具结果
如果裁剪后只剩：
{"role": "assistant", "content": [tool_use_read_file]}
但对应的 tool_result 被裁掉了，消息结构就不完整。
所以 snip_compact() 里有一段逻辑，专门避免把这组消息从中间切断。
6.4 第一段边界保护：头部结尾不能只留下 tool_use
源码：
if head_end > 0 and _message_has_tool_use(messages[head_end - 1]):
while head_end < len(messages) and _is_tool_result_message(messages[head_end]):
head_end += 1
假设：
head_end = 3
原本保留：
messages[:3]
也就是保留下标：
0, 1, 2
现在检查：
messages[head_end - 1]
就是：
messages[2]
如果 messages[2] 是一条 assistant tool_use，说明头部最后一条消息是：
模型要求调用工具
那下一条很可能就是对应的工具结果：
messages[3]
所以代码会继续看：
while head_end < len(messages) and _is_tool_result_message(messages[head_end]):
head_end += 1
变量变化示例：
head_end = 3
messages[3] 是 tool_result
head_end += 1
head_end = 4
这样保留头部就从：
messages[:3]
变成：
messages[:4]
也就是把对应的 tool_result 也保留下来。
6.5 第二段边界保护：尾部开头不能只留下 tool_result
源码：
if (tail_start > 0 and tail_start < len(messages)
and _is_tool_result_message(messages[tail_start])
and _message_has_tool_use(messages[tail_start - 1])):
tail_start -= 1
假设：
tail_start = 7
原本保留尾部：
messages[7:]
也就是保留下标：
7, 8, 9, 10, 11
现在检查：
messages[tail_start]
就是：
messages[7]
如果 messages[7] 是 tool_result，说明尾部第一条就是工具结果。
再检查：
messages[tail_start - 1]
也就是：
messages[6]
如果 messages[6] 是对应的 tool_use，说明原本的切法会变成：
裁掉 tool_use
保留 tool_result
这也不完整。
所以代码做：
tail_start -= 1
变量变化：
tail_start = 7
tail_start -= 1
tail_start = 6
尾部保留范围从：
messages[7:]
变成：
messages[6:]
这样 tool_use 和 tool_result 就一起保留下来了。
6.6 head_end >= tail_start 是什么意思
源码：
if head_end >= tail_start:
return messages
正常情况下：
head_end < tail_start
中间才有东西可以裁。
例如：
head_end = 3
tail_start = 7
中间可裁：
messages[3:7]
但如果边界保护后变成：
head_end = 7
tail_start = 6
说明头部和尾部已经重叠了。
这时候再裁剪就没有意义，甚至可能裁错，所以直接：
return messages
6.7 完整执行示例
假设：
max_messages = 8
len(messages) = 12
先计算：
keep_head = 3
keep_tail = 5
head_end = 3
tail_start = 7
假设消息结构是：
0 user text
1 assistant text
2 assistant tool_use
3 user tool_result
4 user text
5 assistant text
6 assistant tool_use
7 user tool_result
8 user text
9 assistant text
10 user text
11 assistant text
第一段保护：
messages[head_end - 1] = messages[2]
它是 tool_use。
所以检查：
messages[head_end] = messages[3]
它是 tool_result。
于是：
head_end = 4
第二段保护：
messages[tail_start] = messages[7]
它是 tool_result。
并且：
messages[tail_start - 1] = messages[6]
它是 tool_use。
于是：
tail_start = 6
现在：
head_end = 4
tail_start = 6
要裁掉：
messages[4:6]
也就是下标 4 和 5。
最终保留：
messages[:4]
+ [{"role": "user", "content": "[snipped 2 messages]"}]
+ messages[6:]
结果是：
0 user text
1 assistant text
2 assistant tool_use
3 user tool_result
[snipped 2 messages]
6 assistant tool_use
7 user tool_result
8 user text
9 assistant text
10 user text
11 assistant text
你会发现，两个工具调用配对都没有被切断。
6.8 一句话总结
snip_compact() 做的是：
先算出头部保留边界和尾部保留边界
再检查边界有没有切断 tool_use / tool_result
如果切断了，就移动边界
最后把中间消息替换成一个 [snipped ... messages] 标记
它不是为了精确总结内容，而是一个便宜、快速、不会调用 LLM 的上下文裁剪方式。
7. micro_compact(messages)
作用：保留最近几个完整工具结果，把更早的大工具结果替换成短文本。
block["content"] = "[早前工具结果已压缩。如有需要请重新运行。]"
它不会调用 LLM，所以成本很低。
8. compact_history(messages)
作用：调用 LLM，把历史总结成一条新消息。
压缩前：
[msg1, msg2, msg3, ..., msg100]
压缩后：
[
{
"role": "user",
"content": "[Compacted]\n\n这里是历史摘要"
}
]
9. messages[:] = ...
messages[:] = ... 表示原地替换列表内容。
如果写：
messages = compact_history(messages)
只是函数内部变量换了一个新列表。
如果写：
messages[:] = compact_history(messages)
外面传进来的 history 也会看到压缩后的内容。
这就是为什么 Agent Loop 里使用 messages[:]。