#!/usr/bin/env python3
"""cache_breakpoint_planner.py — prompt caching 断点设计与成本对比（无需 API key，本地可跑）

演示：
1. 自动缓存（顶层 cache_control）vs 显式断点（块级 cache_control）
2. 前缀一致性规则：tools -> system -> messages 顺序；断点后内容可变
3. 5m/1h TTL 计价（写 1.25x / 2x，读 0.1x）与命中率收益测算
4. 反模式检测：易变内容出现在断点之前

官方依据（截至 2026-08-30）:
https://platform.claude.com/docs/en/build-with-claude/prompt-caching（访问于 2026-08-30）

运行: python3 cache_breakpoint_planner.py
"""

import copy
import json

PRICING = {"claude-sonnet-5": {"in": 2.0, "out": 10.0}}   # USD/MTok，截至 2026-08-30


def make_request(mode: str) -> dict:
    """构造带缓存策略的请求。mode: none | auto | explicit"""
    req = {
        "model": "claude-sonnet-5",
        "max_tokens": 1024,
        "system": [
            {"type": "text", "text": "你是企业知识库助手。以下是产品规范文档……" * 50},
        ],
        "tools": [
            {"name": "search_docs", "description": "检索文档", "input_schema": {"type": "object", "properties": {}}},
        ],
        "messages": [{"role": "user", "content": "帮我查 Q2 销售数据"}],
    }
    if mode == "auto":
        req["cache_control"] = {"type": "ephemeral"}      # 顶层自动缓存：断点自动打到最后可缓存块
    elif mode == "explicit":
        req["tools"][0]["cache_control"] = {"type": "ephemeral", "ttl": "1h"}
        req["system"][0]["cache_control"] = {"type": "ephemeral", "ttl": "1h"}
    return req


def cost_estimate(model: str, cacheable_tokens: int, mutable_tokens: int,
                  out_tokens: int, hit_rate: float, ttl: str) -> float:
    """估算单请求成本。cacheable 部分按命中率分读/写；mutable 部分全价。"""
    p = PRICING[model]
    write_mult = 1.25 if ttl == "5m" else 2.0
    c = cacheable_tokens / 1e6 * (
        p["in"] * write_mult * (1 - hit_rate) + p["in"] * 0.1 * hit_rate
    )
    c += mutable_tokens / 1e6 * p["in"]
    c += out_tokens / 1e6 * p["out"]
    return c


def anti_pattern_check(req: dict) -> list[str]:
    """检测常见缓存反模式。"""
    problems = []
    sys_text = json.dumps(req.get("system", ""), ensure_ascii=False)
    for volatile in ("2026-08-30 20:15", "user_id", "session_id"):
        if volatile in sys_text[:200]:
            problems.append(f"易变内容 {volatile!r} 出现在 system 前部 -> 前缀永远失配")
    return problems


if __name__ == "__main__":
    print("== 三种模式请求体对比（节选）==")
    for mode in ("none", "auto", "explicit"):
        req = make_request(mode)
        marker = "顶层 cache_control" if mode == "auto" else (
            "块级 cache_control(ttl=1h)" if mode == "explicit" else "无缓存")
        print(f"  {mode:8s}: {marker}")

    # 反模式演示
    bad = make_request("auto")
    bad["system"][0]["text"] = "当前时间 2026-08-30 20:15:33。user_id=8891。" + bad["system"][0]["text"]
    problems = anti_pattern_check(bad)
    assert problems, "应检测到易变前缀"
    print("\n反模式检测:", problems[0])

    # 成本对比：20K 稳定前缀 + 200 可变输入 + 500 输出，100 轮对话后命中率 80%
    print("\n== 单请求成本（20K 稳定前缀, 200 可变输入, 500 输出, 命中率 80%）==")
    for label, hit, ttl in [("无缓存", 0.0, "5m"), ("5m TTL", 0.8, "5m"), ("1h TTL", 0.8, "1h")]:
        c = cost_estimate("claude-sonnet-5", 20_000, 200, 500, hit, ttl)
        print(f"  {label:8s}: ${c:.5f}")
    none_c = cost_estimate("claude-sonnet-5", 20_000, 200, 500, 0.0, "5m")
    best_c = cost_estimate("claude-sonnet-5", 20_000, 200, 500, 0.8, "5m")
    print(f"  缓存节省: {100*(1-best_c/none_c):.0f}%（未计 ITPM 限速收益——cache reads 不占 ITPM）")

    # 显式断点结构合法性与顺序说明
    explicit = make_request("explicit")
    assert explicit["tools"][0].get("cache_control") and explicit["system"][0].get("cache_control")
    print("\n显式断点结构: tools[0] + system[0] 各一个断点（前缀顺序 tools -> system -> messages）")
    print("规则提醒: 断点之前的内容必须逐字节稳定；易变内容放在断点之后的消息里。")