基于 s06_context_compact.py 源码逐行分析，配合 s06-context-compact.md` 设计思路。

一、问题：上下文是有限的，Agent 却想一直干活

前面几章的 agent 有一个共同假设：上下文够用。但现实中不是。

读一个 1000 行的 Python 文件，大约吃掉 4000 tokens。一次探索任务读 20 个文件、跑 15 条 bash 命令，轻松突破 80k tokens。当前主流模型的上下文窗口在 100k-200k 之间——看起来很大，但一个稍微复杂的任务就能填满。

更根本的问题：即使上下文窗口够大，模型处理超长上下文的效率也会下降。窗口越满，注意力越分散，决策质量越低。s03 的 Nag Reminder 解决的是模型"忘记目标"的问题；s06 要解决的是"上下文物理满了"的问题。

二、解决方案：三层压缩，激进程度递增

s06 设计了一个三层的压缩管道：

每一轮循环之前：

[Layer 1: micro_compact]        静默执行，每轮都跑
  把超过 3 轮的非 read_file 工具结果
  替换为 "[Previous: used {tool_name}]"
        ↓
[检查: tokens > 50000?]
   |              |
   否             是
   |              ↓
  继续      [Layer 2: auto_compact]
              完整对话存到 .transcripts/
              让 LLM 自己做摘要
              用摘要替换全部 messages
                  ↓
          [Layer 3: compact 工具]
              模型主动调 compact → 立即摘要
              和 auto_compact 同一个底层函数

核心思想：“The agent can forget strategically and keep working forever.” — agent 可以有策略地遗忘，从而无限工作。

三、和 s05 相比，多了什么？

组件 s05 s06
上下文管理 无 三层压缩管道
新工具 load_skill compact（手动触发压缩）
持久化 无 对话记录存 .transcripts/
agent_loop 简单循环 循环前加 micro_compact + auto_compact 检查

四、一些基础概念

什么是 token？ LLM 不是逐字阅读的，而是把文本切成小块（token），一个 token 大约是 3-4 个英文字母或 1 个中文字。API 按 token 计费，上下文窗口也以 token 为单位。s06 的 estimate_tokens() 用了一个简单的近似：文本长度 / 4 ≈ token 数。

messages[:] = new_list 是什么意思？ messages[:] = ... 是 Python 的原地替换语法——清空列表内容然后填入新元素。和 messages = new_list 不同，后者只是改变局部变量指向，不会影响调用方看到的列表。因为 agent_loop(messages) 通过引用传递列表，必须用 [:] 才能让外层的 history 也看到变化。

json.dumps(messages, default=str) 的 default=str？ 当 JSON 序列化遇到无法处理的对象（如 Anthropic SDK 的 ContentBlock 对象）时，default=str 告诉 Python：对它调 str() 转成字符串。这是保证序列化不崩溃的兜底策略，和 skill 系统的容错设计同源。

五、Layer 1：micro_compact — 静默替换旧结果

5.1 核心逻辑

KEEP_RECENT = 3              # 保留最近 3 个工具结果
PRESERVE_RESULT_TOOLS = {"read_file"}  # read_file 的结果不压缩

def micro_compact(messages: list) -> list:
    # 第一步：收集所有 tool_result 的位置
    tool_results = []
    for msg_idx, msg in enumerate(messages):
        if msg["role"] == "user" and isinstance(msg.get("content"), list):
            for part_idx, part in enumerate(msg["content"]):
                if isinstance(part, dict) and part.get("type") == "tool_result":
                    tool_results.append((msg_idx, part_idx, part))

    if len(tool_results) <= KEEP_RECENT:
        return messages       # 不够 3 个，不动

    # 第二步：建立 tool_use_id → tool_name 的映射
    tool_name_map = {}
    for msg in messages:
        if msg["role"] == "assistant":
            content = msg.get("content", [])
            if isinstance(content, list):
                for block in content:
                    if hasattr(block, "type") and block.type == "tool_use":
                        tool_name_map[block.id] = block.name

    # 第三步：清理旧结果（保留最近 3 个）
    to_clear = tool_results[:-KEEP_RECENT]
    for _, _, result in to_clear:
        if not isinstance(result.get("content"), str) or len(result["content"]) <= 100:
            continue           # 已经很短了，跳过
        tool_id = result.get("tool_use_id", "")
        tool_name = tool_name_map.get(tool_id, "unknown")
        if tool_name in PRESERVE_RESULT_TOOLS:
            continue           # read_file 的结果保留原文
        result["content"] = f"[Previous: used {tool_name}]"

    return messages

5.2 逐步拆解

第一步：定位所有 tool_result。 遍历 messages 的每一条消息、每一个 content block，找到类型为 tool_result 的条目。记录它们在 messages 树中的精确位置（msg_idx, part_idx, part），后面要直接修改 part["content"]。

第二步：建立 ID 映射。 每个 tool_result 有一个 tool_use_id，对应之前某条 assistant 消息中的 tool_use block。通过遍历找到每个 id 对应的工具名（bash / read_file / write_file / edit_file），这样替换时能写出有意义的占位符：[Previous: used bash] 比 [Previous: used unknown] 有用得多。

第三步：替换旧结果。 tool_results[:-KEEP_RECENT] 取的是除了最后 3 个以外的所有结果。对每一个：

  • 如果已经是短内容（≤100 字符），不替换——没有压缩收益
  • 如果是 read_file 的结果，不替换——文件内容是需要反复参考的，压缩了模型会重复读文件
  • 其他情况（bash 输出、write_file 确认、edit_file 确认），替换为 [Previous: used {tool_name}]

5.3 为什么保留 read_file 的结果？

read_file 读取的文件内容是参考资料，不是操作日志。如果压缩掉，模型下次需要这个文件内容时会重新调用 read_file——额外消耗 tokens 且无意义。相比之下，bash 的 2000 行输出在命令执行后就过时了，replace 掉损失不大。

这也是一个有趣的权衡：保留 read_file 结果意味着 micro_compact 的压缩效率取决于"读文件和跑命令的比例"。如果 agent 大量读文件但要留原文，micro_compact 的效果就有限。此时就需要 Layer 2 来解决。

六、Layer 2：auto_compact — LLM 自己摘要自己的记忆

6.1 触发条件

THRESHOLD = 50000  # tokens 阈值

if estimate_tokens(messages) > THRESHOLD:
    messages[:] = auto_compact(messages)

每轮循环在调用 API 之前检查。estimate_tokens 用 len(str(messages)) // 4 粗略估算——不需要精确计数，只需要判断"是不是该压缩了"。

6.2 auto_compact 做了什么

def auto_compact(messages: list) -> list:
    # Step 1: 把完整对话保存到磁盘
    TRANSCRIPT_DIR.mkdir(exist_ok=True)
    transcript_path = TRANSCRIPT_DIR / f"transcript_{int(time.time())}.jsonl"
    with open(transcript_path, "w") as f:
        for msg in messages:
            f.write(json.dumps(msg, default=str) + "\n")

    # Step 2: 让 LLM 做摘要
    conversation_text = json.dumps(messages, default=str)[-80000:]
    response = client.messages.create(
        model=MODEL,
        messages=[{"role": "user", "content":
            "Summarize this conversation for continuity. Include: "
            "1) What was accomplished, 2) Current state, 3) Key decisions made. "
            "Be concise but preserve critical details.\n\n" + conversation_text}],
        max_tokens=2000,
    )
    summary = next((block.text for block in response.content
                    if hasattr(block, "text")), "")

    # Step 3: 用摘要替换全部 messages
    return [
        {"role": "user",
         "content": f"[Conversation compressed. Transcript: {transcript_path}]\n\n{summary}"},
    ]

三步走：

Step 1：存档。 完整对话原文写入 .transcripts/transcript_{timestamp}.jsonl。JSONL 格式每行一个 JSON 对象，方便后续用 grep 或脚本分析。default=str 处理 SDK 对象的序列化——一个常见的坑是 ContentBlock 不是原生 JSON 可序列化的。

Step 2：摘要。 取对话文本末尾 80000 字符（最新的内容最重要），发给同一个模型，要求它总结：完成了什么、当前状态是什么、关键决策是什么。max_tokens=2000 限制摘要长度——压缩后整个上下文只有 ~2000 tokens。

Step 3：替换。 返回一条消息，告诉模型"对话已被压缩，原始记录在某某文件"。之后 agent_loop 会把 messages[:] 设为这个单元素列表——整个对话历史被一个摘要替代，token 占用从 50000+ 骤降至 ~2000。

6.3 关键设计决策

  • [-80000:] 取末尾而非全量：早期的对话内容对摘要贡献低，取最近 80000 字符足够了。而且全量可能超过模型的 max_tokens 限制。
  • 摘要保留"关键决策"：不是简单的"做了什么"，而要求模型判断什么重要。这利用了 LLM 的理解能力——让它自己判断什么值得记住。
  • 只留一条消息：压缩后的 messages 只有 [{role: "user", content: summary}]。这不影响后续对话——agent 可以继续在上面累积新消息。

七、Layer 3：compact 工具 — 模型主动触发

"compact": lambda **kw: "Manual compression requested.",

compact 工具的 handler 不执行压缩——只返回一条确认信息。真正的压缩发生在 agent_loop 的工具执行之后：

manual_compact = False
for block in response.content:
    if block.type == "tool_use":
        if block.name == "compact":
            manual_compact = True
            output = "Compressing..."
        else:
            # ...正常执行其它工具...

messages.append({"role": "user", "content": results})

if manual_compact:
    messages[:] = auto_compact(messages)
    return    # 压缩后退出本次 agent_loop，让用户看到结果

为什么不在 handler 里直接调 auto_compact？因为压缩会修改 messages 列表（替换为摘要），而这个修改必须发生在工具结果追加之后、下一轮 API 调用之前。handler 只是工具执行的一环，不控制循环流程。将 trigger 放在循环层是架构干净的做法——handler 做标记，循环做决策。

八、agent_loop 的整合

def agent_loop(messages: list):
    while True:
        # Layer 1: 每次 API 调用前先静默压缩
        micro_compact(messages)

        # Layer 2: token 超阈值自动压缩
        if estimate_tokens(messages) > THRESHOLD:
            print("[auto_compact triggered]")
            messages[:] = auto_compact(messages)

        response = client.messages.create(...)
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return

        # ...执行工具...

        # Layer 3: 模型主动触发压缩
        if manual_compact:
            messages[:] = auto_compact(messages)
            return

三层在循环中的位置不同：Layer 1 和 2 在调用 API 之前（预防性压缩），Layer 3 在工具执行之后（响应式压缩）。它们的时间节奏也不同：Layer 1 每轮都跑，Layer 2 偶尔触发，Layer 3 由模型决定。

九、完整流程走读

假设一个 agent 正在重构项目，messages 里已经累积了 20 轮对话。

第 1-20 轮（正常循环）

每轮 API 调用前，micro_compact 静默运行：把超过 3 轮的老旧 bash 输出替换为 [Previous: used bash]。上下文在不断膨胀，但增速被抑制了——micro_compact 压掉的是"低信息密度"的历史工具输出。

第 21 轮（auto_compact 触发）

estimate_tokens(messages) 超过 50000。循环暂停，auto_compact 执行：

  1. 完整对话存入 .transcripts/transcript_1712345678.jsonl
  2. LLM 摘要：“用户要求重构 utils.py：已添加类型注解（第 1-5 轮），已提取公共函数（第 6-12 轮），正在进行测试迁移（第 13-20 轮）。关键决策：保留向后兼容的 shim 层，使用 pytest 替换 unittest。”
  3. messages 被替换为单条摘要消息

第 22 轮起

agent 从摘要恢复——它"遗忘"了具体每一步的操作，但记住了完成了什么、当前状态、关键决策。transcript 文件在磁盘上，需要时可以用 read_file 回溯。上下文回到 ~2000 tokens，agent 可以继续工作。

十、设计洞察

10.1 分层遗忘策略

不是所有信息都该被遗忘——read_file 的结果要保留，短输出没必要压缩。micro_compact 的保留/替换逻辑体现了"信息密度"的考量：高密度内容（文件原文）保留，低密度内容（过时的命令输出）替换为占位符。遗忘不是开关，是一个连续的取舍。

10.2 元认知：让模型总结自己

auto_compact 最精妙的地方是让 LLM 自己摘要自己的对话。没有预定义的"什么重要"的规则——模型自己判断哪些决定、哪些状态值得保留。这是用 LLM 的理解能力解决传统程序难以处理的问题（“什么是关键信息？”）。

10.3 磁盘是逃生舱

transcript 文件确保信息没有真正丢失——只是移出了活跃上下文。这为后续章节（s08 的后台任务、s09 的多 agent 协作）提供了基础：压缩后的 agent 可以重新加载历史来恢复状态。内存是快的，但它会满；磁盘是慢的，但它无限。

10.4 压缩后退出

Layer 3（manual compact）执行后直接 return，而不是继续循环。为什么？因为此时 messages 只剩一条摘要消息，stop_reason 已经是 end_turn 或下一个 task 需要用户的新输入。让控制权回到 REPL 层，用户可以看到压缩结果再决定下一步。

# windows # python
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

计算机毕业设计：Python医疗知识图谱可视化与智能问答一体化平台 Django框架 Bert模型 深度学习 知识图谱 大模型（建议收藏）✅

avatar AtomGit开源社区
cover

计算机毕业设计：Python基于知识图谱的医疗问答与数据挖掘分析系统 Django框架 Bert模型 深度学习 知识图谱 大模型（建议收藏）✅

avatar AtomGit开源社区
cover

一个高颜值、轻量级的精品网址导航工具

avatar AtomGit开源社区