DeepSeek V4 FlashとGPT-4oを同条件で比較する方法 — 速度・料金・品質をPythonで測る

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade openai

export COMETAPI_KEY="your_api_key_here"
export DEEPSEEK_MODEL_ID="deepseek-v4-flash"
export GPT4O_MODEL_ID="gpt-4o"

import json
import math
import os
import random
import statistics
import time
from pathlib import Path

from openai import OpenAI


def required(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"環境変数{name}を設定してください。")
    return value


def percentile(values: list[float], fraction: float) -> float:
    ordered = sorted(values)
    index = max(0, math.ceil(fraction * len(ordered)) - 1)
    return ordered[index]


runs = int(os.environ.get("BENCHMARK_RUNS", "5"))
if runs < 3:
    raise RuntimeError("BENCHMARK_RUNSは3以上にしてください。")

# 2026-07-23時点のCometAPIモデル一覧。実行前に更新してください。
models = [
    {
        "label": "DeepSeek V4 Flash",
        "id": os.environ.get("DEEPSEEK_MODEL_ID", "deepseek-v4-flash"),
        "input_usd_per_m": 0.15,
        "output_usd_per_m": 0.30,
    },
    {
        "label": "GPT-4o",
        "id": os.environ.get("GPT4O_MODEL_ID", "gpt-4o"),
        "input_usd_per_m": 2.50,
        "output_usd_per_m": 10.00,
    },
]

tasks = [
    {
        "id": "json_extract",
        "prompt": (
            "JSONオブジェクトだけを返してください。必須キーはcustomer、plan、"
            "renewal_date、riskです。入力: Northwind LabsはProプラン。"
            "更新日は2026-09-30。2回のエクスポート失敗が更新リスクです。"
        ),
        "check": "JSONとして読み込め、4項目が入力と一致すること",
    },
    {
        "id": "code_fix",
        "prompt": (
            "次のPython関数のmutable default引数を修正し、修正版と2文の説明を"
            "返してください。\n\ndef add_item(item, bucket=[]):\n"
            "    bucket.append(item)\n    return bucket"
        ),
        "check": "デフォルトをNoneにし、関数内でリストを作成すること",
    },
    {
        "id": "reasoning",
        "prompt": (
            "バッチ処理は毎分120件です。600件ごとに30秒停止します。"
            "開始時は停止せず、1,800件を完了する時間を分単位で計算してください。"
            "最後の600件後の停止を含めるか、前提も明記してください。"
        ),
        "check": "最終停止の扱いを明記し、その前提と計算が一致すること",
    },
]

client = OpenAI(
    api_key=required("COMETAPI_KEY"),
    base_url="https://api.cometapi.com/v1",
    timeout=120.0,
    max_retries=0,
)

records = []

for run in range(1, runs + 1):
    work = [(model, task) for task in tasks for model in models]
    random.shuffle(work)

    for model, task in work:
        started = time.perf_counter()
        record = {
            "run": run,
            "model": model["label"],
            "model_id": model["id"],
            "task": task["id"],
            "quality_check": task["check"],
        }

        try:
            response = client.chat.completions.create(
                model=model["id"],
                messages=[{"role": "user", "content": task["prompt"]}],
                temperature=0,
                max_completion_tokens=800,
            )
            elapsed = time.perf_counter() - started
            usage = response.usage
            input_tokens = usage.prompt_tokens if usage else 0
            output_tokens = usage.completion_tokens if usage else 0
            estimated_cost = (
                input_tokens * model["input_usd_per_m"] / 1_000_000
                + output_tokens * model["output_usd_per_m"] / 1_000_000
            )

            record.update(
                {
                    "ok": True,
                    "elapsed_seconds": elapsed,
                    "input_tokens": input_tokens,
                    "output_tokens": output_tokens,
                    "estimated_cost_usd": estimated_cost,
                    "output": response.choices[0].message.content or "",
                }
            )
        except Exception as exc:
            record.update(
                {
                    "ok": False,
                    "elapsed_seconds": time.perf_counter() - started,
                    "error_type": type(exc).__name__,
                }
            )

        records.append(record)
        print(
            f"run={run} model={model['label']} task={task['id']} "
            f"ok={record['ok']} seconds={record['elapsed_seconds']:.2f}"
        )

Path("comparison_results.json").write_text(
    json.dumps(records, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

print("\n| Model | Success | Median latency | P95 latency | Total cost |")
print("|---|---:|---:|---:|---:|")

for model in models:
    rows = [row for row in records if row["model"] == model["label"]]
    successful = [row for row in rows if row["ok"]]

    if not successful:
        print(f"| {model['label']} | 0/{len(rows)} | n/a | n/a | n/a |")
        continue

    latencies = [row["elapsed_seconds"] for row in successful]
    total_cost = sum(row["estimated_cost_usd"] for row in successful)
    print(
        f"| {model['label']} | {len(successful)}/{len(rows)} "
        f"| {statistics.median(latencies):.2f}s "
        f"| {percentile(latencies, 0.95):.2f}s "
        f"| ${total_cost:.6f} |"
    )

print("\n回答全文はcomparison_results.jsonへ保存しました。")

python compare_models.py

BENCHMARK_RUNS=20 python compare_models.py

品質合格率 = 合格した回答数 / 成功したAPIリクエスト数

合格1件あたりのコスト = 全リクエストの料金 / 合格した回答数