長時間走るエージェントの文脈が静かに劣化する — コンテキスト予算と圧縮の設計

from dataclasses import dataclass, field
 
@dataclass
class ContextBudget:
    total: int = 200_000          # モデルのウィンドウ
    reserve_output: int = 16_000  # 出力用に確保する分
    # 入力に使える上限から各層へ配分
    def allocation(self) -> dict[str, int]:
        usable = self.total - self.reserve_output
        return {
            "system":  int(usable * 0.08),
            "work":    int(usable * 0.22),
            "knowledge": int(usable * 0.30),
            "history": int(usable * 0.40),
        }
 
def count_tokens(client, model, blocks) -> int:
    # 概算ではなく公式のトークンカウントAPIで実測する
    return client.messages.count_tokens(model=model, messages=blocks).input_tokens

def context_health(usage: dict, alloc: dict) -> dict:
    history_ratio = usage["history"] / alloc["history"]
    system_distance = usage["history"] + usage["knowledge"] + usage["work"]
    return {
        "history_ratio": round(history_ratio, 2),
        "system_distance": system_distance,
        "compact_now": history_ratio > 0.8 or system_distance > 150_000,
    }

SUMMARIZE_SYSTEM = (
    "以下はエージェントの過去ログです。後続の判断に必要な"
    "決定事項・制約・未解決の課題だけを、箇条書きで簡潔に残してください。"
    "完了済みの冗長なやりとりは省いて構いません。"
)
 
def compact_history(client, model, old_turns, keep_recent=6):
    archived = old_turns[:-keep_recent]      # 古い部分だけ要約対象
    recent = old_turns[-keep_recent:]        # 直近は手をつけない
    if not archived:
        return old_turns
    summary = client.messages.create(
        model=model,
        max_tokens=1024,
        system=SUMMARIZE_SYSTEM,
        messages=[{"role": "user",
                   "content": _render(archived)}],
    ).content[0].text
    memo = {"role": "user",
            "content": f"[これまでの要約]\n{summary}"}
    return [memo] + recent

resp = client.beta.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=4096,
    messages=conversation,
    tools=tools,
    betas=["context-management-2025-06-27"],
    context_management={
        "edits": [{
            "type": "clear_tool_uses",
            # 直近のツール結果は残し、古いものから消す
            "keep": {"type": "tool_uses", "value": 8},
            "clear_at_least": {"type": "tokens", "value": 20_000},
        }]
    },
)

def retrieve(query, store, k_search=40, k_final=5):
    candidates = store.search(query, top_k=k_search)   # まず広く
    reranked = rerank(query, candidates)               # 関連度で並べ直す
    picked = reranked[:k_final]                         # 数件に絞る
    # 知識層の予算内に収める。溢れたら末尾から落とす
    return trim_to_budget(picked, budget="knowledge")

system = [
    {"type": "text", "text": ROLE_AND_RULES,
     "cache_control": {"type": "ephemeral"}},   # 不変層をキャッシュ
]
# 知識層・履歴層はキャッシュの後ろ。入れ替わってもキャッシュは無効化されない