Harness实战：上下文压缩——三层策略让Agent永不断档

第1轮: read_file → main.py              (输出: 200行, ~800 token)
第2轮: bash → pytest --cov              (输出: 80行,  ~400 token)
第3轮: read_file → test_main.py         (输出: 150行, ~600 token)
第4轮: edit_file → test_main.py         (输出: "Edited", ~10 token)
第5轮: bash → pytest                    (输出: 30行,  ~150 token)
第6轮: 告诉用户 "修好了"

第1轮结果在后续5轮中重复传输: 800 × 5 = 4,000 token
第2轮结果在后续4轮中重复传输: 400 × 4 = 1,600 token
第3轮结果在后续3轮中重复传输: 600 × 3 = 1,800 token
───────────────────────────────────
仅6轮对话，重复传输的废 token:    ~7,400 token

替换前: {"type": "tool_result", "content": "200行main.py的完整内容..."}
替换后: {"type": "tool_result", "content": "[Previous: used read_file]"}

┌───────────────────────────────────────────────────┐
│                   agent_loop                       │
│                                                    │
│  每轮开头 → Layer 1: micro_compact (静默替换)      │
│              将旧 tool_result 替换为占位符          │
│              成本: 零（纯字符串操作）               │
│              ↓                                     │
│  token > 阈值 → Layer 2: auto_compact (LLM摘要)   │
│              备份完整对话 → LLM 生成摘要            │
│              成本: 一次 LLM 调用                    │
│              ↓                                     │
│  模型主动调用 → Layer 3: compact 工具 (手动触发)   │
│              同 Layer 2 的摘要机制，按需执行         │
│                                                    │
│  激进度:  低 ──────────────────────────→ 高        │
└───────────────────────────────────────────────────┘

import time  # 新增 import，transcript 需要时间戳

# ── 上下文压缩配置 ──────────────────────────────────────
THRESHOLD = 50_000               # 超过此值自动触发 auto_compact
TRANSCRIPT_DIR = WORKDIR / ".transcripts"
KEEP_RECENT = 3                  # micro_compact 保留最近 N 个已消化的 tool_result
PRESERVE_RESULT_TOOLS = {"read_file"}  # 这些工具的结果不压缩（参考材料）
SUMMARY_SYSTEM = (               # auto_compact 摘要时的 system prompt
    "Summarize this conversation for continuity. Include: "
    "1) What was accomplished, 2) Current state, 3) Key decisions made. "
    "Be concise but preserve critical details."
)

SYSTEM = f"""...（原有内容不变）

When context feels long or cluttered, use the `compact` tool to compress the conversation."""

def estimate_tokens(messages: list) -> int:
    """粗略估算 token 数：每 4 个字符 ≈ 1 token。"""
    return len(str(messages)) // 4

def micro_compact(messages: list) -> list:
    # 找到最后一条 assistant 消息的位置——只替换模型已经"消化"过的结果
    last_assistant_idx = -1
    for i in range(len(messages) - 1, -1, -1):
        if messages[i]["role"] == "assistant":
            last_assistant_idx = i
            break

    # 只收集 last_assistant_idx 之前的 tool_result（模型已见过的）
    consumed_results = []
    for msg_idx, msg in enumerate(messages):
        if msg_idx >= last_assistant_idx:
            break
        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":
                    consumed_results.append((msg_idx, part_idx, part))
    if len(consumed_results) <= KEEP_RECENT:
        return messages

    # 从 assistant 消息中收集 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

    # 替换旧的已消化结果（保留最后 KEEP_RECENT 个）
    to_clear = consumed_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
        result["content"] = f"[Previous: used {tool_name}]"
    return messages

last_assistant_idx = ...  # 最后一条 assistant 消息的位置
for msg_idx, msg in enumerate(messages):
    if msg_idx >= last_assistant_idx:
        break  # 这之后的 tool_result 模型还没见过，不能动

consumed_results = [...]  # 只包含模型已见过的 tool_result
to_clear = consumed_results[:-KEEP_RECENT]

if not isinstance(result.get("content"), str) or len(result["content"]) <= 100:
    continue

if tool_name in PRESERVE_RESULT_TOOLS:
    continue

messages 状态（第5轮 LLM 调用前）:

  [0] user: "帮我看看测试覆盖率"
  [1] assistant: → tool_use(read_file, "main.py")
  [2] user: tool_result("200行代码...")
  [3] assistant: → tool_use(bash, "pytest --cov")
  [4] user: tool_result("覆盖率报告...")
  [5] assistant: → tool_use(read_file, "test_main.py")
  [6] user: tool_result("150行测试代码...")
  [7] assistant: → tool_use(edit_file, "test_main.py")
  [8] user: tool_result("Edited test_main.py")      ← ≤100字符，跳过

micro_compact:
  last_assistant_idx = 7  (messages[7] 是最后一条 assistant)
  只收集 index < 7 的已消化 tool_result:
    [2] "200行代码..."    → 保留（read_file 在 PRESERVE_RESULT_TOOLS 中）
    [4] "覆盖率报告..."    → "[Previous: used bash]"   (400→10 token)
    [6] "150行测试代码..." → 保留（read_file 在 PRESERVE_RESULT_TOOLS 中）
  [8] 的 tool_result 在 last_assistant_idx 之后 → 模型还没见过，不动!

  consumed_results=3, KEEP_RECENT=3 → 不触发替换?
  ❌ 错了，这里 consumed 刚好等于 3，不触发。
  等下一轮 [7] assistant 后面再追加新的 tool_result 时才会开始替换。

def auto_compact(messages: list) -> list:
    """保存完整对话到磁盘，用 LLM 生成摘要，替换所有消息。"""
    # ① 保存 transcript
    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")
    print(f"[transcript saved: {transcript_path}]")

    # ② 用 LLM 生成摘要（截取最后 80000 字符防止超长）
    conversation_text = json.dumps(messages, default=str)[-80000:]
    response = client.messages.create(
        model=MODEL,
        system=SUMMARY_SYSTEM,
        messages=[{"role": "user", "content": conversation_text}],
        max_tokens=2048,
    )
    summary = next(
        (block.text for block in response.content if hasattr(block, "text")),
        "No summary generated.",
    )

    # ③ 用压缩后的单条消息替换全部历史
    return [
        {
            "role": "user",
            "content": (
                f"[Conversation compressed. Transcript: {transcript_path}]\n\n"
                f"{summary}"
            ),
        },
    ]

COMPACT_TOOL = {
    "name": "compact",
    "description": "Trigger manual conversation compression.",
    "input_schema": {
        "type": "object",
        "properties": {
            "focus": {"type": "string", "description": "What to preserve in the summary"}
        },
    },
}

results = []
manual_compact = False
for block in response.content:
    if block.type == "tool_use":
        if block.name == "compact":
            manual_compact = True
            output = "Compressing..."
        else:
            handler = TOOL_HANDLERS.get(block.name)
            try:
                output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
            except Exception as e:
                output = f"Error: {e}"
        results.append({"type": "tool_result", "tool_use_id": block.id,
                        "content": str(output)})
messages.append({"role": "user", "content": results})

# Layer 3: compact 触发后，立即摘要并返回
if manual_compact:
    print("[manual compact]")
    messages[:] = auto_compact(messages)
    return

def agent_loop(messages: list):
    while True:
        # ★ Layer 1: micro_compact — 每轮静默替换旧 tool_result
        micro_compact(messages)

        # ★ Layer 2: auto_compact — token 超阈值时，LLM 摘要压缩
        if estimate_tokens(messages) > THRESHOLD:
            print("[auto_compact triggered]")
            messages[:] = auto_compact(messages)

        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 = []
        manual_compact = False
        for block in response.content:
            if block.type == "tool_use":
                if block.name == "compact":
                    manual_compact = True
                    output = "Compressing..."
                else:
                    handler = TOOL_HANDLERS.get(block.name)
                    try:
                        output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
                    except Exception as e:
                        output = f"Error: {e}"
                print(f"> {block.name}:")
                print(str(output)[:200])
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": str(output)})
        messages.append({"role": "user", "content": results})

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

elif query.strip() == "/compact":
    if not history:
        print("  没有可压缩的对话")
        continue
    print("[manual compact]")
    history[:] = auto_compact(history)
    print(f"  ✓ 当前上下文: ~{estimate_tokens(history)} tokens\n")
    continue

❯ Read every Python file in the harness/ directory

> read_file: harness.py
> read_file: multi-turn.py
> read_file: tool-function-calling.py
> read_file: plan-demo.py
> read_file: harness-subagent.py
> read_file: harness-parallel.py
> read_file: harness-background.py
> read_file: agentic-demo.py

  [micro_compact] 替换了 5 个旧 tool_result

I've read all 8 Python files. Here's a summary...

❯ Now analyze the architecture and write a report

  [micro_compact] 替换了 2 个旧 tool_result

> bash: wc -l harness/*.py
> write_file: ANALYSIS_REPORT.md

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

Report written to ANALYSIS_REPORT.md.

❯ /compact

[manual compact]
[transcript saved: .transcripts/transcript_1712736100.jsonl]
  ✓ 当前上下文: ~900 tokens

# 错误：只改了局部变量，调用方的 history 不变
messages = auto_compact(messages)

# 正确：替换列表内容，调用方的 history 同步更新
messages[:] = auto_compact(messages)