Gemini API の usageMetadata で本番アプリのコストを記録する — リクエスト単位で請求書と突き合わせる実装パターン

# pricing.py
# 1M トークンあたりの USD（2026 年 4 月時点・必ず公式の最新表で確認）
PRICING = {
    "gemini-2.5-pro": {
        "input": 1.25, "cached": 0.31, "output": 10.00, "thoughts": 10.00,
    },
    "gemini-2.5-flash": {
        "input": 0.30, "cached": 0.075, "output": 2.50, "thoughts": 2.50,
    },
    "gemini-2.5-flash-lite": {
        "input": 0.10, "cached": 0.025, "output": 0.40, "thoughts": 0.40,
    },
}
 
def calc_usd(record: dict) -> float:
    p = PRICING.get(record["model"])
    if p is None:
        # 未知のモデルは安全側に倒して 0 にせず Pro 単価で見積もる
        p = PRICING["gemini-2.5-pro"]
    input_billable = max(0, record["prompt"] - record.get("cached", 0))
    return (
        input_billable / 1_000_000 * p["input"]
        + record.get("cached", 0) / 1_000_000 * p["cached"]
        + record.get("candidates", 0) / 1_000_000 * p["output"]
        + record.get("thoughts", 0) / 1_000_000 * p["thoughts"]
    )

import json
from collections import defaultdict
from pricing import calc_usd
 
usd_by_endpoint = defaultdict(float)
with open("logs/gemini_usage.jsonl") as f:
    for line in f:
        rec = json.loads(line)
        usd_by_endpoint[rec["endpoint"]] += calc_usd(rec)
 
USD_TO_JPY = 156  # 月初の TTM レートで固定するのが運用上ラク
for ep, usd in sorted(usd_by_endpoint.items(), key=lambda x: -x[1]):
    print(f"{ep:40s}  ${usd:>8.4f}  ¥{usd * USD_TO_JPY:>9.2f}")

import json
from datetime import datetime, timezone, timedelta
from pricing import calc_usd
 
DAILY_BUDGET_USD = 3.0  # 月 90 ドルを想定
 
def today_total_usd() -> float:
    today = datetime.now(timezone(timedelta(hours=9))).date()
    total = 0.0
    with open("logs/gemini_usage.jsonl") as f:
        for line in f:
            rec = json.loads(line)
            d = datetime.fromtimestamp(rec["ts"], tz=timezone(timedelta(hours=9))).date()
            if d == today:
                total += calc_usd(rec)
    return total
 
# 各リクエストの直後に呼ぶ
if today_total_usd() > DAILY_BUDGET_USD:
    # 例: Slack Webhook へ通知、または当日の API 呼び出しを停止
    raise RuntimeError("Daily Gemini budget exceeded")

sqlite3 usage.db <<'SQL'
.mode json
CREATE TABLE IF NOT EXISTS usage AS SELECT * FROM read_json('logs/gemini_usage.jsonl');
SQL

-- 直近 7 日のエンドポイント別コスト Top 10（gemini-2.5-flash）
SELECT endpoint,
       SUM(prompt - cached) * 0.30 / 1e6 AS input_usd,
       SUM(candidates + thoughts) * 2.50 / 1e6 AS output_usd
FROM usage
WHERE ts > strftime('%s','now') - 7*86400
  AND model = 'gemini-2.5-flash'
GROUP BY endpoint
ORDER BY input_usd + output_usd DESC
LIMIT 10;

import { GoogleGenAI } from "@google/genai";
 
const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY });
const resp = await ai.models.generateContent({
  model: "gemini-2.5-flash",
  contents: "Hello",
});
 
const um = resp.usageMetadata;
console.log({
  prompt: um.promptTokenCount,
  cached: um.cachedContentTokenCount ?? 0,
  candidates: um.candidatesTokenCount ?? 0,
  thoughts: um.thoughtsTokenCount ?? 0,
});

def call_gemini_logged(model: str, prompt: str, *, user_id: str, endpoint: str):
    started = time.time()
    resp = client.models.generate_content(model=model, contents=prompt)
    um = resp.usage_metadata
 
    record = {
        "ts": time.time(),
        "elapsed_ms": int((time.time() - started) * 1000),
        "user_id": user_id,
        "endpoint": endpoint,
        "requested_model": model,                 # こちらが要求した名前（エイリアス含む）
        "served_model": resp.model_version,       # 実際に応答したモデル ← 単価引き当てはこちら
        "prompt": um.prompt_token_count,
        "cached": um.cached_content_token_count or 0,
        "candidates": um.candidates_token_count or 0,
        "thoughts": um.thoughts_token_count or 0,
        "tool_use": um.tool_use_prompt_token_count or 0,
        "total": um.total_token_count,
    }
    with LOG_PATH.open("a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")
    return resp

def resolve_price_key(served_model: str) -> str | None:
    """model_version を PRICING のキーへ正規化する。未知なら None を返す"""
    if not served_model:
        return None
    m = served_model.lower()
    for key in sorted(PRICING, key=len, reverse=True):   # 長い順に部分一致を取る
        if m.startswith(key) or key in m:
            return key
    return None
 
def calc_usd(record: dict) -> float:
    key = resolve_price_key(record.get("served_model") or record.get("requested_model"))
    if key is None:
        # 黙って Pro 単価に倒さず、気づける形で止める
        raise KeyError(f"unknown model: {record.get('served_model')}")
    p = PRICING[key]
    input_billable = max(0, record["prompt"] - record.get("cached", 0))
    return (
        input_billable / 1_000_000 * p["input"]
        + record.get("cached", 0) / 1_000_000 * p["cached"]
        + record.get("candidates", 0) / 1_000_000 * p["output"]
        + record.get("thoughts", 0) / 1_000_000 * p["thoughts"]
    )

import json
from collections import Counter
from datetime import datetime, timezone
 
def audit_recent(days: int = 1) -> dict:
    since = datetime.now(timezone.utc).timestamp() - days * 86400
    drift = Counter()          # (requested, served) の食い違い
    unknown = Counter()        # PRICING に無い served_model
    total = 0
    with open("logs/gemini_usage.jsonl") as f:
        for line in f:
            rec = json.loads(line)
            if rec["ts"] < since:
                continue
            total += 1
            served = rec.get("served_model")
            if resolve_price_key(served) is None:
                unknown[served] += 1
            if served and rec.get("requested_model") and \
               resolve_price_key(served) != resolve_price_key(rec["requested_model"]):
                drift[(rec["requested_model"], served)] += 1
    return {"total": total, "drift": drift, "unknown": unknown}
 
report = audit_recent(days=1)
if report["unknown"]:
    print("単価表に無いモデルが応答しています:", dict(report["unknown"]))
if report["drift"]:
    ratio = sum(report["drift"].values()) / max(report["total"], 1)
    print(f"要求と異なるモデルでの応答が {ratio:.1%}:", dict(report["drift"]))