How to Build an AI Agent
How to Build an AI Agent
s06

圧縮

メモリ管理

Three-Layer Compression

205 LOC5 ツールmicro-compact + auto-compact + archival
Context will fill up; three-layer compression strategy enables infinite sessions

s01 > s02 > s03 > s04 > s05 > [ s06 ] | s07 > s08 > s09 > s10 > s11 > s12

"コンテキストはいつか溢れる、空ける手段が要る" -- 3層圧縮で無限セッションを実現。

問題

コンテキストウィンドウは有限だ。1000行のファイルに対するread_file1回で約4000トークンを消費する。30ファイルを読み20回のbashコマンドを実行すると、100,000トークン超。圧縮なしでは、エージェントは大規模コードベースで作業できない。

解決策

積極性を段階的に上げる3層構成:

Every turn:
+------------------+
| Tool call result |
+------------------+
        |
        v
[Layer 1: micro_compact]        (silent, every turn)
  Replace tool_result > 3 turns old
  with "[Previous: used {tool_name}]"
        |
        v
[Check: tokens > 50000?]
   |               |
   no              yes
   |               |
   v               v
continue    [Layer 2: auto_compact]
              Save transcript to .transcripts/
              LLM summarizes conversation.
              Replace all messages with [summary].
                    |
                    v
            [Layer 3: compact tool]
              Model calls compact explicitly.
              Same summarization as auto_compact.

仕組み

  1. 第1層 -- micro_compact: 各LLM呼び出しの前に、古いツール結果をプレースホルダーに置換する。
def micro_compact(messages: list) -> list:
    tool_results = []
    for i, msg in enumerate(messages):
        if msg["role"] == "user" and isinstance(msg.get("content"), list):
            for j, part in enumerate(msg["content"]):
                if isinstance(part, dict) and part.get("type") == "tool_result":
                    tool_results.append((i, j, part))
    if len(tool_results) <= KEEP_RECENT:
        return messages
    for _, _, part in tool_results[:-KEEP_RECENT]:
        if len(part.get("content", "")) > 100:
            part["content"] = f"[Previous: used {tool_name}]"
    return messages
  1. 第2層 -- auto_compact: トークンが閾値を超えたら、完全なトランスクリプトをディスクに保存し、LLMに要約を依頼する。
def auto_compact(messages: list) -> list:
    # Save transcript for recovery
    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")
    # LLM summarizes
    response = client.messages.create(
        model=MODEL,
        messages=[{"role": "user", "content":
            "Summarize this conversation for continuity..."
            + json.dumps(messages, default=str)[:80000]}],
        max_tokens=2000,
    )
    return [
        {"role": "user", "content": f"[Compressed]\n\n{response.content[0].text}"},
        {"role": "assistant", "content": "Understood. Continuing."},
    ]
  1. 第3層 -- manual compact: compactツールが同じ要約処理をオンデマンドでトリガーする。

  2. ループが3層すべてを統合する:

def agent_loop(messages: list):
    while True:
        micro_compact(messages)                        # Layer 1
        if estimate_tokens(messages) > THRESHOLD:
            messages[:] = auto_compact(messages)       # Layer 2
        response = client.messages.create(...)
        # ... tool execution ...
        if manual_compact:
            messages[:] = auto_compact(messages)       # Layer 3

トランスクリプトがディスク上に完全な履歴を保持する。何も真に失われず、アクティブなコンテキストの外に移動されるだけ。

s05からの変更点

ComponentBefore (s05)After (s06)
Tools55 (base + compact)
Context mgmtNoneThree-layer compression
Micro-compactNoneOld results -> placeholders
Auto-compactNoneToken threshold trigger
TranscriptsNoneSaved to .transcripts/

試してみる

python agents/s06_context_compact.py
  1. Read every Python file in the agents/ directory one by one (micro-compactが古い結果を置換するのを観察する)
  2. Keep reading files until compression triggers automatically
  3. Use the compact tool to manually compress the conversation