長時間エージェントのトークン肥大を context editing と memory tool で抑える

context_management={
    "edits": [
        {
            "type": "clear_tool_uses_20250919",
            # この入力トークン数を超えたら消去を始める
            "trigger": {"type": "input_tokens", "value": 30000},
            # 直近 3 件のツール結果は残す
            "keep": {"type": "tool_uses", "value": 3},
            # 一度の消去で最低これだけのトークンを削る
            "clear_at_least": {"type": "input_tokens", "value": 5000},
            # この道具の結果は消さない
            "exclude_tools": ["web_search"],
        }
    ]
}

# Before（危険）: 受け取った path をそのまま結合している
from pathlib import Path
 
ROOT = Path("./memories")
 
def view(path: str) -> str:
    target = ROOT / path.lstrip("/")          # "../../etc/passwd" を防げない
    return target.read_text(encoding="utf-8")

# After（保護あり）: canonical 解決後に /memories 配下か検証する
from pathlib import Path
 
ROOT = Path("./memories").resolve()
 
def _safe(path: str) -> Path:
    # 先頭の /memories を実ディレクトリへ写像し、正規化して配下チェック
    rel = path.replace("/memories", "", 1).lstrip("/")
    target = (ROOT / rel).resolve()
    try:
        target.relative_to(ROOT)             # 配下でなければ ValueError
    except ValueError:
        raise PermissionError(f"パスが /memories の外を指しています: {path}")
    return target
 
def handle_memory(cmd: dict) -> str:
    c = cmd["command"]
    if c == "view":
        p = _safe(cmd["path"])
        if p.is_dir():
            items = "\n".join(f"{f.stat().st_size}\t/memories/{f.relative_to(ROOT)}"
                              for f in sorted(p.rglob("*")) if f.is_file())
            return f"Files under /memories:\n{items}"
        lines = p.read_text(encoding="utf-8").splitlines()
        return "\n".join(f"{i+1:>6}\t{ln}" for i, ln in enumerate(lines))
    if c == "create":
        p = _safe(cmd["path"])
        if p.exists():
            return f"Error: File {cmd['path']} already exists"
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(cmd["file_text"], encoding="utf-8")
        return f"File created successfully at: {cmd['path']}"
    if c == "str_replace":
        p = _safe(cmd["path"])
        body = p.read_text(encoding="utf-8")
        if body.count(cmd["old_str"]) != 1:
            return "No replacement was performed; old_str must appear exactly once."
        p.write_text(body.replace(cmd["old_str"], cmd["new_str"]), encoding="utf-8")
        return "The memory file has been edited."
    if c == "delete":
        p = _safe(cmd["path"])
        p.unlink(missing_ok=True)
        return f"Successfully deleted {cmd['path']}"
    return f"Error: unsupported command {c}"

def run_turn(messages):
    response = client.beta.messages.create(
        model="claude-opus-4-8",
        max_tokens=4096,
        messages=messages,
        tools=[
            {"type": "memory_20250818", "name": "memory"},
            {"type": "web_search_20250305", "name": "web_search"},
        ],
        betas=["context-management-2025-06-27"],
        context_management={
            "edits": [{
                "type": "clear_tool_uses_20250919",
                "trigger": {"type": "input_tokens", "value": 30000},
                "keep": {"type": "tool_uses", "value": 3},
            }]
        },
    )
    return response
 
# memory コマンドが来たらバックエンドで処理して tool_result を返す
def dispatch_tools(response, messages):
    results = []
    for block in response.content:
        if block.type == "tool_use" and block.name == "memory":
            out = handle_memory(block.input)
            results.append({"type": "tool_result",
                            "tool_use_id": block.id, "content": out})
    if results:
        messages.append({"role": "user", "content": results})
    return messages

counted = client.beta.messages.count_tokens(
    model="claude-opus-4-8",
    messages=long_messages,                       # 肥大化した実際の履歴
    betas=["context-management-2025-06-27"],
    context_management={
        "edits": [{
            "type": "clear_tool_uses_20250919",
            "trigger": {"type": "input_tokens", "value": 30000},
            "keep": {"type": "tool_uses", "value": 5},
        }]
    },
)
 
original = counted.context_management.original_input_tokens
after = counted.input_tokens
print(f"消去前: {original} / 消去後: {after} / 削減: {original - after}")