第6课：无限对话——上下文压缩三层策略

读 5 个文件      → 20,000 token
跑 10 次命令     → 15,000 token
写 3 个文件      → 12,000 token
来回 8 轮对话    → 8,000 token
系统提示 + 工具定义 → 5,000 token
───────────────────────────
合计：             ~60,000 token

┌──────────────────────────────────────────────────────┐
│                    Agent Loop                         │
│                                                       │
│   每轮结束 ──→ Layer 1: micro_compact (静默裁剪)       │
│                  替换旧 tool_result 为占位符            │
│                  ↓                                    │
│   token > 阈值 → Layer 2: auto_compact (自动摘要)      │
│                  保存 transcript → LLM 做摘要           │
│                  ↓                                    │
│   模型主动调用 → Layer 3: compact 工具 (手动触发)       │
│                  同样的摘要机制，按需执行                 │
│                                                       │
│   激进度:  低 ──────────────────────────────→ 高       │
└──────────────────────────────────────────────────────┘

def micro_compact(messages: list, keep_recent: int = 3):
    """保留最近 keep_recent 轮完整内容，
    更早的 tool_result 替换为占位符。"""
    assistant_indices = [
        i for i, m in enumerate(messages)
        if m["role"] == "assistant"
    ]
    if len(assistant_indices) <= keep_recent:
        return

    cutoff_index = assistant_indices[-keep_recent]

    for i, msg in enumerate(messages):
        if i >= cutoff_index:
            break
        if msg["role"] != "user":
            continue
        content = msg.get("content", [])
        if not isinstance(content, list):
            continue
        for j, block in enumerate(content):
            if (isinstance(block, dict)
                    and block.get("type") == "tool_result"):
                tool_name = _find_tool_name(
                    messages, block["tool_use_id"]
                )
                content[j] = {
                    **block,
                    "content": f"[Previous: used {tool_name}]",
                }

def save_transcript(messages: list) -> Path:
    """保存完整对话到 .transcripts/，压缩前的快照。"""
    transcript_dir = Path(".transcripts")
    transcript_dir.mkdir(exist_ok=True)
    filepath = transcript_dir / f"conversation_{int(time.time())}.jsonl"
    with open(filepath, "w") as f:
        for msg in messages:
            f.write(json.dumps(msg, ensure_ascii=False) + "\n")
    return filepath


def summarize_conversation(client, messages: list) -> str:
    """让 LLM 把完整对话压缩为结构化摘要。"""
    response = client.messages.create(
        model=MODEL, max_tokens=4000,
        system=(
            "把对话压缩为结构化摘要。保留：1) 用户原始目标 "
            "2) 已完成步骤 3) 关键发现和决策 4) 当前待办。"
            "丢弃：具体文件内容、命令输出、中间调试过程。"
        ),
        messages=[{"role": "user",
                    "content": format_messages_for_summary(messages)}],
    )
    return response.content[0].text


def auto_compact(client, messages: list) -> list:
    """保存 transcript + 生成摘要 + 替换消息列表。"""
    filepath = save_transcript(messages)
    summary = summarize_conversation(client, messages)
    return [{
        "role": "user",
        "content": (
            f"[Context compacted. Full transcript: {filepath}]\n\n"
            f"## Conversation Summary\n\n{summary}\n\n"
            "Continue from where we left off."
        ),
    }]

COMPACT_TOOL = {
    "name": "compact",
    "description": (
        "Compress the conversation history into a summary. "
        "Use when the context is getting long."
    ),
    "input_schema": {"type": "object", "properties": {}},
}

def handle_compact(client, messages: list) -> tuple:
    """手动触发，复用 auto_compact 的摘要流程。"""
    new_messages = auto_compact(client, messages)
    return new_messages, "Context compacted successfully."

TOKEN_THRESHOLD = 100_000

def agent_loop(client, messages: list, tools: list):
    while True:
        micro_compact(messages)                    # ① Layer 1

        if estimate_tokens(messages) > TOKEN_THRESHOLD:
            messages = auto_compact(client, messages)  # ② Layer 2

        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            messages=messages, tools=tools, max_tokens=8000,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return

        results = []
        for block in response.content:
            if block.type != "tool_use":
                continue
            if block.name == "compact":            # ③ Layer 3
                messages, output = handle_compact(client, messages)
            else:
                output = dispatch_tool(block)
            results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": output,
            })
        messages.append({"role": "user", "content": results})

# 克隆仓库
git clone https://github.com/shareAI-lab/learn-claude-code
cd learn-claude-code

# 安装依赖
pip install -r requirements.txt

# 配置 API Key
cp .env.example .env
# 编辑 .env，填入你的 ANTHROPIC_API_KEY 和 MODEL_ID

# 启动第六课
python agents/s06_context_compact.py

s06 >> Read every Python file in the agents/ directory one by one
> bash: (列出 agents/ 目录下所有文件)
> read_file: __init__.py
> read_file: s01_agent_loop.py
> read_file: s02_tool_use.py
> read_file: Error: No such file: 'agents/s03_context.py'
> bash: (重新 ls 确认文件名)
> read_file: s03_todo_write.py
> read_file: s04_subagent.py
> read_file: s05_skill_loading.py
> read_file: Error: No such file: 'agents/s06_memory.py'
> bash: (再次 ls 确认)
> read_file: s06_context_compact.py
> read_file: s07_task_system.py
> read_file: s08_background_tasks.py
> read_file: Error: No such file: 'agents/s09_guardrails.py'
> bash: (第三次 ls 确认)
> read_file: s09_agent_teams.py
> read_file: s10_team_protocols.py
> read_file: s11_autonomous_agents.py
> read_file: s12_worktree_task_isolation.py
> read_file: s_full.py

[auto_compact triggered]
[transcript saved: .transcripts/transcript_1775572349.jsonl]

I have full context of the entire codebase.
13 Python files in agents/ — a progressive tutorial:
| s01 | Basic chat loop | s07 | File-based task board |
| s02 | Tool use | s08 | Background execution |
| s03 | TodoWrite | s09 | Multi-agent messaging |
| ... | ... | s12 | Git worktree isolation |

s06 >> Keep reading files until compression triggers automatically
> bash: (列出所有文件)
> read_file: hello.py
> read_file: LICENSE
> read_file: requirements.txt
> read_file: mypackage/utils.py
> read_file: mypackage/__init__.py
> read_file: mypackage/tests/test_utils.py
> read_file: README.md
> read_file: README-zh.md
> read_file: greet.py
> bash: ls -la agents/
> read_file: s01_agent_loop.py
> read_file: s02_tool_use.py
> read_file: s03_todo_write.py
> read_file: s04_subagent.py
> read_file: s05_skill_loading.py
> read_file: s06_context_compact.py
> read_file: s07_task_system.py
> read_file: s08_background_tasks.py
> read_file: s09_agent_teams.py
> read_file: s10_team_protocols.py
> read_file: s11_autonomous_agents.py
> read_file: s12_worktree_task_isolation.py
> bash: ls docs/

[auto_compact triggered]
[transcript saved: .transcripts/transcript_1775572551.jsonl]

s06 >> Use the compact tool to manually compress the conversation
> compact:
Compressing...
[manual compact]
[transcript saved: .transcripts/transcript_1775572604.jsonl]

# 预告：s07 的任务持久化
task = TaskManager.load("task_001")  # 从磁盘恢复
task.status = "in_progress"
# ... Agent 继续工作 ...
task.save()  # 随时保存进度