s08：上下文压缩管道

s08：上下文压缩管道（Context Compaction Pipeline）

核心问题

---

上下文窗口是稀缺资源：LLM 能处理的文本长度是有限的，长时间运行的 Agent 对话历史会迅速膨胀并超出这个限制，导致 API 报错或成本剧增。上下文压缩是 Agent 系统维持长期运行的基础设施。

---

分层压缩策略（Layered Compaction）

---

管道中的每一层都代表一种成本不同的压缩方法，从几乎无成本的简单修剪，到需要一次 API 调用的 LLM 摘要，递进式地释放上下文空间。

主动压缩 vs 被动压缩

类型	触发时机	作用
主动压缩（Proactive）	在每次调用 LLM 之前，自动运行压缩管道（L1–L4）	预防性地腾出空间，避免超限
被动/紧急压缩（Reactive/Emergency）	当 LLM API 仍然返回 prompt_too_long 错误时触发	"兜底"机制，确保系统不会因上下文过长而崩溃

设计哲学：主动压缩是"体检"，被动压缩是"急救"。两者配合，保证系统在任何情况下都能继续运行。

持久化与引用（内存-磁盘交换）

L3 不是简单的丢弃信息，而是将大型工具结果持久化到磁盘，并在对话中留下一个文件路径作为引用。模型未来可以通过 read_file 工具按需重新读取——这是一种**“索引"或"交换到磁盘”**的思想（类比操作系统虚拟内存的 swap）。

转录存档（Transcript Archive）

在 L4 进行摘要前，会将完整的对话历史序列化保存到磁盘（.transcripts 目录）。这保证了即使进行了有损压缩，历史信息也不会永久丢失，可用于事后审计、调试或恢复。

---

压缩管道基础设施（常量与辅助判断函数）

---

管道运行前需要一些"基础设施"：配置阈值参数，以及识别消息类型的辅助函数。

CONTEXT_LIMIT = 50000        # 触发 L4 摘要的字符串长度阈值
KEEP_RECENT = 3              # L2 微压缩时，保留最近 N 个工具结果不被压缩
PERSIST_THRESHOLD = 30000    # L3 持久化时，单个工具结果超过此长度才被持久化

辅助判断函数：被 L1 和 紧急压缩 用于识别**"工具调用-工具结果"的配对关系**，从而进行更智能的修剪，避免将一次完整的工具交互拦腰截断。

def _message_has_tool_use(msg):
    # 判断一条 assistant 消息是否包含工具调用

def _is_tool_result_message(msg):
    # 判断一条 user 消息是否包含工具结果

---

四层压缩管道（L1–L4）解析

---

每一层都是一个纯函数，接收并返回 messages 列表。层级编号 L1→L4 对应"成本从低到高"。

L1：snip_compact — 基于消息数量的滑动窗口修剪

成本最低的压缩，模拟一个固定窗口滑过对话历史，只保留最新的 max_messages 条消息和头部几条（keep_head）消息。

亮点：边界保护逻辑——在剪切窗口边缘时，它会智能地向下包含完整的工具结果，或向上包含发起的工具调用，确保 LLM 看到的剩余对话是逻辑自洽的，不会出现"一个没有下文的工具调用"或"一个没有来源的工具结果"。

def snip_compact(messages, max_messages=50):
    # 1. 消息数量未超限，直接返回
    if len(messages) <= max_messages: return messages
    # 2. 计算保留的头尾部分
    keep_head, keep_tail = 3, max_messages - 3
    head_end, tail_start = keep_head, len(messages) - keep_tail
    # 3. 边界保护：向下调整头部的结束位置，防止切断一个"工具调用-结果"对
    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
    # 4. 边界保护：向上调整尾部的开始位置，防止从一个"工具结果"的中间开始
    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
    # 5. 如果调整后头尾重叠或交叉，放弃修剪
    if head_end >= tail_start:
        return messages
    # 6. 执行修剪，拼接头、占位符、尾
    snipped = tail_start - head_end
    return messages[:head_end] + [{"role": "user", "content": f"[snipped {snipped} messages]"}] + messages[tail_start:]

---

L2：micro_compact — 工具结果占位符替换

更精细的压缩。它不删除消息，而是原地替换内容。核心思想：对于很久以前的工具执行结果，其具体细节大概率已不重要，但保留其**"发生过"的事实**（即保留消息结构）对 LLM 理解对话流程仍有价值。

保留最近 KEEP_RECENT 个结果，是因为它们与当前任务最相关。

def micro_compact(messages):
    tool_results = collect_tool_results(messages)  # 收集所有工具结果块
    if len(tool_results) <= KEEP_RECENT: return messages # 结果数量不多则跳过
    # 遍历除了最近 KEEP_RECENT 个之外的所有旧工具结果
    # 每个 tool_results 中的元素是一个元组，_,_ 忽略前两个不需要的值
    for _, _, block in tool_results[:-KEEP_RECENT]:
        # 如果结果内容较长，替换为占位符，短结果保留原样
        if len(block.get("content", "")) > 120:
            block["content"] = "[Earlier tool result compacted. Re-run if needed.]"
    return messages

---

L3：tool_result_budget — 大型结果的"内存-磁盘"交换

L3 处理单个工具调用可能产生海量输出的问题（例如 cat 一个巨大日志文件）。它引入了一个 max_bytes 预算，针对每一批工具结果进行管控。其核心是用磁盘空间换取 LLM 宝贵的上下文窗口。

persist_large_output 创建的 <persisted-output> 标签，是一种约定，告诉模型"完整数据在文件里，你可以用 read_file 去读"。

def persist_large_output(tool_use_id, output):
    # 检查是否超过单条消息的持久化阈值
    if len(output) <= PERSIST_THRESHOLD: return output
    # 将完整输出写入磁盘文件
    TOOL_RESULTS_DIR.mkdir(parents=True, exist_ok=True)
    path = TOOL_RESULTS_DIR / f"{tool_use_id}.txt"
    if not path.exists(): path.write_text(output)
    # 返回一个包含文件路径和预览内容的占位符
    return f"\nFull output: {path}\nPreview:\n{output[:2000]}\n"

def tool_result_budget(messages, max_bytes=200_000):
    # 1. 只处理最新一条 user 消息（即刚执行完的那批工具结果）
    last = messages[-1] if messages else None
    if not last or last.get("role") != "user" ...: return messages
    # 2. 计算当前批次所有工具结果的总大小
    blocks = [(i, b) for i, b in enumerate(last["content"]) if ...]
    total = sum(len(str(b.get("content", ""))) for _, b in blocks)
    if total <= max_bytes: return messages # 未超预算，直接返回
    # 3. 超预算：将结果按大小降序排列
    ranked = sorted(blocks, key=lambda p: len(str(p[1].get("content", ""))), reverse=True)
    # 4. 从最大的结果开始，尝试持久化，直到总大小回到预算内
    for _, block in ranked:
        if total <= max_bytes: break
        # ... 持久化逻辑 ...
        block["content"] = persist_large_output(tid, content)
        total = sum(...) # 重新计算总大小
    return messages

---

L4：compact_history — LLM 完整摘要（最昂贵的一层）

不再修剪或替换，而是用一次额外的 LLM 调用，将整个对话历史压缩成一段摘要文本，然后彻底丢弃原有的 messages，用一个只包含摘要的新列表替换。

# 确保了信息的可恢复性
def write_transcript(messages):
    # 将当前完整的 messages 序列化并写入 .jsonl 文件，作为存档点
    ...

# 提示词精确地列出了需要保留的关键信息（目标、决策、文件等），指导 LLM 进行高质量的有损压缩。
def summarize_history(messages):
    # 截断 messages 以控制摘要的输入大小，然后调用 LLM 进行摘要
    conversation = json.dumps(messages, default=str)[:80000]
    prompt = ("Summarize this coding-agent conversation so work can continue.\n"
              "Preserve: 1. current goal, 2. key findings/decisions, ...")
    response = client.messages.create(model=MODEL, messages=[{"role": "user", "content": prompt}], ...)
    return extract_text(response) # 返回摘要文本

def compact_history(messages):
    # 1. 存档完整对话历史
    transcript_path = write_transcript(messages)
    # 2. 调用 LLM 生成摘要
    summary = summarize_history(messages)
    # 3. 【关键】返回一个全新的、仅包含摘要的 messages 列表
    return [{"role": "user", "content": f"[Compacted]\n\n{summary}"}]

---

紧急压缩：reactive_compact

---

reactive_compact 是 L4 的变种，用于处理**“预算没算准"或"摘要后仍然超限”**的紧急情况。

与 L4 丢弃全部历史不同，reactive_compact 更加保守——它只摘要较旧的历史，而保留最近的一小部分消息（尾部）。原因是：LLM 很可能就是被这最后几条消息中过大的结果撑爆的。它试图在"摘要释放空间"和"保留近期工作记忆"之间取得平衡。

def reactive_compact(messages):
    # 1. 存档：把**完整消息落盘**到 `.transcripts/*.jsonl`
    transcript = write_transcript(messages)
    # 2. 找到尾部几条消息，保留最后一次完整的"工具交互"
    tail_start = max(0, len(messages) - 5)
    # ... 边界保护，与 L1 类似 ...
    # 3. 仅对 tail_start 之前的历史进行摘要
    summary = summarize_history(messages[:tail_start])
    # 4. 将摘要作为头部，拼接上未摘要的尾部消息，返回新列表
    return [{"role": "user", "content": f"[Reactive compact]\n\n{summary}"}, *messages[tail_start:]]

---

核心循环改造：agent_loop 的集成

---

管道前置：廉价优先

调用 LLM 之前，管道按照**“廉价→昂贵”**的顺序运行。L1–L3 几乎无成本，它们尽力为 L4 这个昂贵的操作"减负"，减少不必要的 LLM 摘要调用。

注意执行顺序：实际代码中 messages[:] = tool_result_budget(messages)（L3）→ snip_compact（L1）→ micro_compact（L2），顺序是 L3 → L1 → L2，而非按编号 L1 → L2 → L3。这是因为 L3 需要先控制单批结果的体量，再做结构性裁剪。

def agent_loop(messages: list):
    reactive_retries = 0
    while True:
        # === s08 核心插入：主动压缩管道 ===
        # 执行顺序：L3(预算) -> L1(修剪) -> L2(微压缩)。注意 messages[:] = ... 的原地修改。
        messages[:] = tool_result_budget(messages)
        messages[:] = snip_compact(messages)
        messages[:] = micro_compact(messages)
        # L4 触发点：如果压缩后上下文大小仍超限，执行 LLM 摘要
        if estimate_size(messages) > CONTEXT_LIMIT:
            print("[auto compact]")
            messages[:] = compact_history(messages)
        # **`try-except` 作为架构组件**：紧急压缩 `reactive_compact` 被巧妙地嵌入到 `try-except` 块中。
        # 它不是普通的 Python 错误处理，而是一个**架构级的反馈控制回路**。
        # 当主动压缩没能满足 API 的限制时，API 的报错信息本身成为一个信号，触发更高级别的压缩策略。
        try:
            # 调用 LLM
            response = client.messages.create(...)
            reactive_retries = 0 # 成功则重置计数器
        except Exception as e:
            # === s08 核心插入：紧急压缩兜底 ===
            # API 仍返回 `prompt_too_long` 时触发
            # `estimate_size()` 只是 `len(str(messages))` 的**粗略代理**，不是真实 tokenizer 计数，
            # 所以"主动压缩"可能估算不准——必须有一道"被动兜底"。
            if ("prompt_too_long" in str(e).lower() ...) and reactive_retries < MAX_REACTIVE_RETRIES:
                print("[reactive compact]")
                messages[:] = reactive_compact(messages) # 更激进的摘要
                reactive_retries += 1
                continue # 用压缩后的 messages 重新开始本轮循环（再次调用 LLM）
            raise # 不是长度错误或重试次数用尽，则抛出异常
        # ... (后续的 assistant 消息追加与工具执行逻辑) ...

estimate_size() 的说明：它只是 len(str(messages)) 的粗略代理，并非真实的 tokenizer 计数。因此主动压缩的估算可能不准，必须依赖被动兜底（reactive_compact）作为双保险。

---

新工具 compact（把控制权交给 Agent）

---

• 架构意义：压缩不再只是 Harness 自动触发，Agent 自己"感觉上下文紧了"可以主动请求压缩，形成自动 + 手动双通道。

if block.name == "compact":
    messages[:] = compact_history(messages)
    results.append({...tool_result...})
    messages.append({"role":"user","content":results})
    break   # 结束本轮，用压缩后的上下文重启

---

本模块知识点

---

• 上下文窗口是稀缺资源，分层压缩维持 Agent 长期运行。
• 主动压缩（每次调 LLM 前跑 L1–L3）+ 被动兜底（reactive_compact 处理 prompt_too_long）。
• 持久化与引用：L3 把大结果落盘 .transcripts，对话中留路径引用（内存-磁盘交换）。
• 转录存档：L4 摘要前先把完整历史序列化落盘，保证有损压缩也可恢复。

层	函数	压缩方式	成本
L1	snip_compact	滑动窗口修剪旧消息（含边界保护）	极低
L2	micro_compact	旧工具结果替换为占位符	极低
L3	tool_result_budget	大结果持久化落盘，对话留引用	低
L4	compact_history	LLM 完整摘要，丢弃原历史	高（一次 API 调用）
紧急	reactive_compact	只摘要旧历史，保留近期尾部	高（保守摘要）