【TypeSafe】出力を絞ればLLMも速くなる？ Gemini 3.8 Flash と Jev で同じ107ケースの速度・コストを比較してみた

#!/usr/bin/env python3
"""生成型 LLM(Gemini)で、JEV と全く同じ横突合ケースを測る。

確かめたいのは「JEV が速くて安いのは文章を生成しないからで、生成型 LLM でも
出力を極小に絞れば同じ速度・料金になるのでは」という仮説。

条件を揃えるために次のようにしている。
- ケースは JEV 側の cases.jsonl をそのまま読む(コピーしない)
- 質問文も JEV 側 run_eval.py の QUESTIONS を import して同じ文言を使う
- JEV 側と同じく逐次実行でレイテンシを測る。接続は既定で1件ずつ新規、--keepalive で使い回し
- 出力は structured output(JSON スキーマ強制)で2項目だけに絞る

API: POST https://generativelanguage.googleapis.com/v1beta/models/<model>:generateContent
     https://ai.google.dev/api/generate-content

標準ライブラリだけで動く。
"""

from __future__ import annotations

import argparse
import http.client
import json
import os
import statistics
import sys
import threading
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

# JEV 側の run_eval.py と cases.jsonl を探す。同じディレクトリに置いてあればそれを、
# 無ければ隣の jev-cross-match-validation/ を使う。
HERE = Path(__file__).resolve().parent
JEV_DIR = HERE if (HERE / "run_eval.py").exists() else HERE.parent / "jev-cross-match-validation"
sys.path.insert(0, str(JEV_DIR))
from run_eval import QUESTIONS, STATE_KEYS  # noqa: E402  JEV と同じ質問文を使う

HOST = "generativelanguage.googleapis.com"
ENDPOINT_PATH = "/v1beta/models/{model}:generateContent"
ENDPOINT = f"https://{HOST}{ENDPOINT_PATH}"
DEFAULT_MODEL = "gemini-3.8-flash"

# USD / 1M tokens(Standard、2026-09 時点の公開価格)。出力単価は thinking トークンにもかかる。
# https://ai.google.dev/gemini-api/docs/pricing
PRICES = {
    "gemini-3.8-flash": (0.75, 3.75),       # 2027-01-01 から 1.50 / 7.50
    "gemini-3.5-flash-lite": (0.30, 2.50),
}

# モデルごとに指定できる最小の thinking level。
# gemini-3.8-flash は minimal 非対応(400 エラー)で、thinking を完全には切れない。
MIN_THINKING = {
    "gemini-3.8-flash": "low",
    "gemini-3.5-flash-lite": "minimal",
    # Gemma 4 の thinking は on/off の2値で、API 上は minimal が off にあたる。
    "gemma-4-26b-a4b-it": "minimal",
    "gemma-4-31b-it": "minimal",
}

RETRY_STATUS = {429, 500, 503}

def build_prompt(case: dict, lang: str) -> tuple[str, str]:
    """JEV の questions / state と同じ情報を、system と user の2つの文字列にする。"""
    src = QUESTIONS[lang]
    key_a, key_b = STATE_KEYS[lang]
    lines = []
    for key in ("exact", "semantic"):
        q = src[key]
        lines.append(f"[{key}] {q['instructions']}")
        lines.append(f"  true: {q['criteria']['true']}")
        lines.append(f"  false: {q['criteria']['false']}")
    state = json.dumps({key_a: case["a"], key_b: case["b"]}, ensure_ascii=False)
    return "\n".join(lines), state

def build_schema(mode: str) -> dict:
    """bool は true/false だけ、prob は JEV の noul に合わせて 0〜1 の確率を返させる。"""
    if mode == "bool":
        prop = {"type": "BOOLEAN"}
    else:
        prop = {"type": "NUMBER", "minimum": 0, "maximum": 1}
    return {
        "type": "OBJECT",
        "properties": {"exact": prop, "semantic": prop},
        "required": ["exact", "semantic"],
        "propertyOrdering": ["exact", "semantic"],
    }

_local = threading.local()

def post_fresh(body: bytes, api_key: str, args) -> tuple[int, str]:
    """JEV 側と同じ測り方。1件ごとに新規接続するので TLS ハンドシェイクが毎回乗る。"""
    req = urllib.request.Request(
        ENDPOINT.format(model=args.model), data=body, method="POST",
        headers={"x-goog-api-key": api_key, "Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=args.timeout) as resp:
            return resp.status, resp.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read().decode("utf-8", errors="replace")

def post_keepalive(body: bytes, api_key: str, args) -> tuple[int, str]:
    """接続を使い回す。実運用のサーバーはこちらに近く、ハンドシェイク分だけ速い。"""
    for retry in (False, True):
        conn = getattr(_local, "conn", None)
        if conn is None:
            conn = _local.conn = http.client.HTTPSConnection(HOST, timeout=args.timeout)
        try:
            conn.request("POST", ENDPOINT_PATH.format(model=args.model), body=body,
                         headers={"x-goog-api-key": api_key,
                                  "Content-Type": "application/json"})
            resp = conn.getresponse()
            return resp.status, resp.read().decode("utf-8", errors="replace")
        except (http.client.HTTPException, ConnectionError):
            # サーバー側がアイドル接続を切っていた場合は1回だけ張り直す
            conn.close()
            _local.conn = None
            if retry:
                raise
    raise AssertionError("unreachable")

def call_gemini(case: dict, api_key: str, args) -> dict:
    """1ケースを投げ、応答とレイテンシを返す。レイテンシは成功した試行の分だけ。"""
    system, state = build_prompt(case, args.lang)
    if args.mode == "prob":
        system += ("\nAnswer each question with the probability (0 to 1) that it is true."
                   if args.lang == "en" else
                   "\n各質問について、true である確率を 0〜1 の数値で答える。")

    body = json.dumps({
        "systemInstruction": {"parts": [{"text": system}]},
        "contents": [{"role": "user", "parts": [{"text": state}]}],
        "generationConfig": {
            "responseMimeType": "application/json",
            "responseSchema": build_schema(args.mode),
            "thinkingConfig": {"thinkingLevel": args.thinking_level},
        },
    }, ensure_ascii=False).encode("utf-8")

    for attempt in range(4):
        started = time.perf_counter()
        try:
            status, raw = (post_keepalive if args.keepalive else post_fresh)(body, api_key, args)
        except Exception as exc:  # ネットワーク断・タイムアウト
            return {"error": f"{type(exc).__name__}: {exc}",
                    "latency_ms": (time.perf_counter() - started) * 1000}
        latency_ms = (time.perf_counter() - started) * 1000
        if status == 200:
            return {"payload": json.loads(raw), "retries": attempt,
                    "latency_ms": latency_ms}
        if status in RETRY_STATUS and attempt < 3:
            time.sleep(2 ** attempt)
            continue
        return {"error": f"HTTP {status}: {raw[:500]}", "latency_ms": latency_ms}
    raise AssertionError("unreachable")

def judge(case: dict, result: dict, args) -> dict:
    """応答を期待値と突き合わせる。行の形は JEV 側の results_*.jsonl に揃える。"""
    row = {
        "id": case["id"],
        "category": case["category"],
        "note": case["note"],
        "latency_ms": round(result["latency_ms"], 1),
    }
    if "error" in result:
        row["error"] = result["error"]
        return row

    payload = result["payload"]
    usage = payload.get("usageMetadata") or {}
    row["model"] = payload.get("modelVersion") or args.model
    row["retries"] = result["retries"]
    row["input_tokens"] = usage.get("promptTokenCount")
    row["output_tokens"] = usage.get("candidatesTokenCount")
    row["thinking_tokens"] = usage.get("thoughtsTokenCount") or 0

    try:
        parts = payload["candidates"][0]["content"]["parts"]
        text = "".join(p.get("text", "") for p in parts if not p.get("thought"))
        answers = json.loads(text)
    except (KeyError, IndexError, json.JSONDecodeError) as exc:
        row["error"] = f"応答を解釈できない: {type(exc).__name__}: {json.dumps(payload, ensure_ascii=False)[:300]}"
        return row

    thresholds = {"exact": args.threshold_exact, "semantic": args.threshold_semantic}
    for key in ("exact", "semantic"):
        value = answers.get(key)
        expected = case[f"expect_{key}"]
        if args.mode == "bool":
            prob, got = None, value
        else:
            prob = value
            got = None if value is None else value >= thresholds[key]
        row[f"{key}_prob"] = prob
        row[f"{key}_threshold"] = thresholds[key] if args.mode == "prob" else None
        row[f"{key}_got"] = got
        row[f"{key}_expected"] = expected
        row[f"{key}_ok"] = None if expected is None else (got == expected)
    return row

def summarize(rows: list[dict], args) -> None:
    errors = [r for r in rows if "error" in r]
    ok_rows = [r for r in rows if "error" not in r]

    print("\n=== 結果サマリ ===")
    print(f"モデル: {args.model} / thinking {args.thinking_level} / 出力 {args.mode}")
    print(f"ケース数: {len(rows)}  成功: {len(ok_rows)}  失敗: {len(errors)}")

    for key in ("exact", "semantic"):
        scored = [r for r in ok_rows if r.get(f"{key}_ok") is not None]
        hit = sum(1 for r in scored if r[f"{key}_ok"])
        rate = f"{hit / len(scored):.1%}" if scored else "-"
        print(f"{key:9s} 正答 {hit}/{len(scored)} ({rate})")

    print("\n--- カテゴリ別(semantic) ---")
    for cat in sorted({r["category"] for r in ok_rows}):
        scored = [r for r in ok_rows
                  if r["category"] == cat and r.get("semantic_ok") is not None]
        if scored:
            hit = sum(1 for r in scored if r["semantic_ok"])
            print(f"{cat:12s} {hit}/{len(scored)} ({hit / len(scored):.0%})")

    misses = [r for r in ok_rows if r.get("semantic_ok") is False]
    if misses:
        print("\n--- semantic を外したケース ---")
        for r in misses:
            print(f"{r['id']:5s} {r['note']}  期待={r['semantic_expected']} "
                  f"実際={r['semantic_got']} (p={r['semantic_prob']})")

    if ok_rows:
        lat = sorted(r["latency_ms"] for r in ok_rows)
        p = lambda q: lat[min(int(len(lat) * q), len(lat) - 1)]  # noqa: E731
        print("\n--- レイテンシ(ms) ---")
        print(f"平均 {statistics.mean(lat):.0f} / 中央値 {p(0.5):.0f} / "
              f"p95 {p(0.95):.0f} / 最小 {lat[0]:.0f} / 最大 {lat[-1]:.0f}")

        n = len(ok_rows)
        tin = sum(r.get("input_tokens") or 0 for r in ok_rows)
        tout = sum(r.get("output_tokens") or 0 for r in ok_rows)
        tthink = sum(r.get("thinking_tokens") or 0 for r in ok_rows)
        print("\n--- トークン ---")
        print(f"入力 合計 {tin} (平均 {tin / n:.0f}/件) / 出力 合計 {tout} (平均 {tout / n:.1f}/件) / "
              f"thinking 合計 {tthink} (平均 {tthink / n:.1f}/件)")
        price_in = args.price_in if args.price_in is not None else PRICES.get(args.model, (None, None))[0]
        price_out = args.price_out if args.price_out is not None else PRICES.get(args.model, (None, None))[1]
        if price_in is None or price_out is None:
            print("概算コスト: 単価が未登録。--price-in / --price-out で指定する")
        else:
            cost = tin / 1e6 * price_in + (tout + tthink) / 1e6 * price_out
            print(f"概算コスト: ${cost:.6f} (入力 ${price_in} / 出力+thinking ${price_out} per 1Mtok)")

    for r in errors:
        print(f"[ERROR] {r['id']}: {r['error']}", file=sys.stderr)

def main() -> int:
    ap = argparse.ArgumentParser(description="Gemini で JEV と同じ横突合ケースを測る")
    ap.add_argument("--cases", default=str(JEV_DIR / "cases.jsonl"))
    ap.add_argument("--out", help="既定は results_<model>_<mode>_<lang>.jsonl")
    ap.add_argument("--lang", choices=("ja", "en"), default="en")
    ap.add_argument("--model", default=DEFAULT_MODEL)
    ap.add_argument("--mode", choices=("bool", "prob"), default="bool",
                    help="bool は true/false だけ返させる。prob は 0〜1 の確率を返させて閾値で判定")
    ap.add_argument("--thinking-level", choices=("minimal", "low", "medium", "high"),
                    help="既定はモデルが受け付ける最小値")
    ap.add_argument("--threshold-exact", type=float, default=0.8)
    ap.add_argument("--threshold-semantic", type=float, default=0.8)
    ap.add_argument("--price-in", type=float, help="USD / 1M 入力トークン")
    ap.add_argument("--price-out", type=float, help="USD / 1M 出力トークン")
    ap.add_argument("--concurrency", type=int, default=1,
                    help="並列数。レイテンシを正しく測るなら1のまま")
    ap.add_argument("--keepalive", action="store_true",
                    help="HTTP 接続を使い回す。JEV 側の測り方(毎回新規接続)とは条件が変わる")
    ap.add_argument("--limit", type=int, help="先頭 N 件だけ実行(疎通確認用)")
    ap.add_argument("--timeout", type=float, default=60.0)
    args = ap.parse_args()
    if args.thinking_level is None:
        args.thinking_level = MIN_THINKING.get(args.model, "low")
    suffix = "_keepalive" if args.keepalive else ""
    out_path = args.out or f"results_{args.model}_{args.mode}_{args.lang}{suffix}.jsonl"

    with open(args.cases, encoding="utf-8") as f:
        cases = [json.loads(line) for line in f if line.strip()]
    if args.limit:
        cases = cases[:args.limit]

    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
        print("GEMINI_API_KEY が未設定です。export してください。", file=sys.stderr)
        return 1

    print(f"{len(cases)} ケースを {args.model} に投げます (質問文 {args.lang} / "
          f"thinking {args.thinking_level} / 出力 {args.mode} / 並列 {args.concurrency})...",
          file=sys.stderr)

    run = lambda c: judge(c, call_gemini(c, api_key, args), args)  # noqa: E731

    wall = time.perf_counter()
    if args.concurrency > 1:
        with ThreadPoolExecutor(max_workers=args.concurrency) as pool:
            rows = list(pool.map(run, cases))
    else:
        rows = [run(c) for c in cases]
    wall = time.perf_counter() - wall

    with open(out_path, "w", encoding="utf-8") as f:
        for row in rows:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")

    summarize(rows, args)
    print(f"\n全体所要: {wall:.1f}s ({wall / len(rows):.2f}s/件)")
    print(f"明細: {out_path}")
    return 0

if __name__ == "__main__":
    raise SystemExit(main())