cached_tokensが0｜OpenAI APIで比べて直す

# 未実行サンプル。既定で有料の生成リクエストを2回送ります。
# Python 3の標準ライブラリだけを使用。自動再試行はしません。
# 先にOPENAI_API_KEYを環境変数へ設定してください。
# fixed-prefix.txtに実際の共通指示を保存し、対象モデルで事前計数します。
# FIXED_PREFIX_TOKENSにはそのトークン数を設定します（文字数ではありません）。
import copy
import hashlib
import json
import os
from datetime import datetime, timezone
from decimal import Decimal
from pathlib import Path
from urllib.request import Request, urlopen

model = os.getenv("OPENAI_MODEL", "gpt-6-luna")
fixed = Path("fixed-prefix.txt").read_text(encoding="utf-8")
fixed_tokens = int(os.environ["FIXED_PREFIX_TOKENS"])
if not fixed or fixed_tokens < 1024:
    raise SystemExit("対象モデルで固定部を計数し、1,024トークン以上を用意してください。")

base = {
    "model": model,
    "service_tier": "default",
    "reasoning": {"effort": "none"},
    "text": {"verbosity": "low"},
    "max_output_tokens": 256,
    "store": False,
    "tools": [{"type": "function", "name": "lookup_note",
               "description": "Look up a note by key.",
               "parameters": {"type": "object", "properties": {
                   "key": {"type": "string"}}, "required": ["key"],
                   "additionalProperties": False}, "strict": True}],
    "tool_choice": "none",
    "prompt_cache_options": {"mode": "explicit", "ttl": "30m"},
    "input": [{"role": "developer", "content": [{
        "type": "input_text", "text": fixed,
        "prompt_cache_breakpoint": {"mode": "explicit"}}]},
        {"role": "user", "content": "Reply with exactly OK."}],
}

# 費用は手動で確認した3単価と適用条件を設定したときだけ計算します。
# INPUT_USD_PER_M / READ_USD_PER_M / WRITE_USD_PER_M: USD/100万tokens
# PRICE_CONDITION: モデル・実際のtier・長文/地域条件・確認日
price_names = ["INPUT_USD_PER_M", "READ_USD_PER_M", "WRITE_USD_PER_M"]
rates = [os.getenv(name) for name in price_names]
price_condition = os.getenv("PRICE_CONDITION")
prices = [Decimal(value) for value in rates] if all(rates) and price_condition else None
if prices and not all(value.is_finite() and value >= 0 for value in prices):
    raise ValueError("単価には有限の非負数を指定してください。")

def input_cost(n, r, w):
    if not prices or not all(value is not None for value in (n, r, w)):
        return None
    if min(n, r, w, n - r - w) < 0:
        return None
    amount = sum(Decimal(tokens) * price for tokens, price in
                 zip((n - r - w, r, w), prices)) / Decimal(1000000)
    return str(amount)

def run(label, body, tools_version, change):
    request = Request("https://api.openai.com/v1/responses",
                      data=json.dumps(body).encode("utf-8"),
                      headers={"Authorization": "Bearer " + os.environ["OPENAI_API_KEY"],
                               "Content-Type": "application/json"}, method="POST")
    sent_at = datetime.now(timezone.utc).isoformat()
    with urlopen(request, timeout=120) as response:
        result = json.load(response)
    usage = result.get("usage") or {}
    details = usage.get("input_tokens_details") or {}
    diagnostic = result.get("prompt_cache_diagnostics") or {}
    n, r, w = (usage.get("input_tokens"), details.get("cached_tokens"),
               details.get("cache_write_tokens"))
    record = {
        "case": label, "sent_at_utc": sent_at, "changed_item": change,
        "response_id": result.get("id"), "status": result.get("status"),
        "comparison_response_id": body["prompt_cache_options"].get("comparison_response_id"),
        "model_requested": body["model"], "model_returned": result.get("model"),
        "service_tier_requested": body["service_tier"],
        "service_tier_returned": result.get("service_tier"),
        "tools_version": tools_version,
        "tools_sha256": hashlib.sha256(json.dumps(body["tools"], sort_keys=True).encode()).hexdigest(),
        "reasoning_effort": body["reasoning"]["effort"],
        "verbosity": body["text"]["verbosity"], "text_format": body["text"].get("format"),
        "fixed_prefix_tokens": fixed_tokens,
        "count_method": os.getenv("FIXED_PREFIX_COUNT_METHOD", "事前計数・方法は診断票へ記入"),
        "fixed_prefix_sha256": hashlib.sha256(fixed.encode("utf-8")).hexdigest(),
        "variable_input_position": "固定部の境界より後",
        "cache_options": body["prompt_cache_options"],
        "cache_breakpoint": "developerの固定部末尾", "prompt_cache_key": body.get("prompt_cache_key"),
        "input_tokens": n, "cached_tokens": r, "cache_write_tokens": w,
        "diagnostics_type": diagnostic.get("type"), "diagnostics_reason": diagnostic.get("reason"),
        "prices_usd_per_m": dict(zip(price_names, rates)), "price_condition": price_condition,
        "total_input_cost_usd": input_cost(n, r, w),
    }
    print(json.dumps(record, ensure_ascii=False, indent=2))
    if result.get("status") != "completed" or not result.get("id"):
        raise SystemExit("完了したresponse IDを得られないため、次の比較を中止しました。")
    return result["id"]

baseline_id = run("A", copy.deepcopy(base), "tools-v1", "基準")
second = copy.deepcopy(base)
second["prompt_cache_options"]["comparison_response_id"] = baseline_id
run("B", second, "tools-v1", "キャッシュ条件の変更なし・診断用比較IDのみ追加")

# 任意の3回目。実行する場合だけRUN_TOOL_NAME_TEST=1を設定します。
if os.getenv("RUN_TOOL_NAME_TEST") == "1":
    third = copy.deepcopy(base)
    third["tools"][0]["name"] = "lookup_note_v2"
    third["prompt_cache_options"]["comparison_response_id"] = baseline_id
    run("C", third, "tools-v2-name-only", "tools[0].nameのみ変更")