一覧へ

LLMベンダー比較ベンチマーク — 5つのベンダーに対して臨床的なIEタスクを実行し、3つの軸(精度/コスト/レイテンシー)で測定

複数のベンダーのLLM(Claude Sonnet · GPT-4o · Gemini 2.5 · Meditron)を使用して、トピック01の臨床ノートからの情報抽出タスクを実行します。精度(F1)· コスト(1000リクエストあたりのUSD)· レイテンシー(ms p50 / p95 / p99)の3軸ベンチマーク。プロンプトの公平性 · ベンダーごとの最適化 · JSON解析の成功率 · レート制限の処理を網羅した完全なパイプラインです。

中級
|
35
|
検証済み (2026-07)
進捗0/15 (0%)

LLMベンダー比較ベンチマーク — 臨床IEタスクを用いて、5つのベンダーに対して、精度/コスト/レイテンシーの3つの軸で測定する

Topic 01では、単一のベンダー(Claude)を使用して、臨床ノートからの情報抽出パイプラインを構築しました。しかし、現実の環境では、「このタスクに最適なベンダーはどれか」という疑問は必然的に生じます。最も精度の高いモデルが、必ずしもコスト/レイテンシーの観点から最適であるとは限らず、各ベンダーの強みと弱みは、タスクとプロンプトによって異なります。この記事では、Topic 01の臨床IEタスクを複数のベンダー(Claude Sonnet、Claude Opus、GPT-4o、GPT-4o-mini、o1、Gemini 2.5 Pro、Med-Gemini、Meditron)で実行し、精度、コスト、レイテンシーという3つの軸でそれらを比較する、実践的なベンチマークパイプラインを構築します。

📚 前提条件(強く推奨)

これは、AI×Bioの深掘りされたトピックです。始める前に、以下のDryBenchトピックをまず学習することを強くお勧めします。

これらの前提条件がないと、この記事では、ベンダーごとのプロンプトの特性、応答の検証、またはClaude Codeを使用してベンチマークパイプラインを自動化する原則を再説明することなく、実際のコードから直接進むため、理解が難しくなる可能性があります。


DryBenchで学んだこと

DryBench ai-native #7では、プロンプトがLLMの出力分布を制御するためのツールであり、最適なプロンプトのスタイルはベンダーによって異なることを学びました。#11では、すべてのLLMにハルシネーションが存在し、ベンダーやモデルによってそのばらつきが大きいことを学びました。#14では、Claude Codeを使用して反復的なパイプラインを自動化できることを確認しました。

これらの原則をベンダー比較ベンチマークに適用すると、いくつかの微妙な問題が生じます。何が「公平な」プロンプトなのでしょうか?各ベンダーが最適化しているシステムプロンプトを使用するのが正しいのか、それとも同一のプロンプトを強制するのが公平なのでしょうか?応答形式がベンダー間でわずかに異なる場合、解析精度の違いはベンダーのパフォーマンスなのか、それともパーサーの問題なのでしょうか?私たちは、これらのベンチマーク設計の落とし穴に正面から取り組み、徹底的に対処します。

ハードコアな問題定義

現実のシナリオ:Topic 01の臨床IEベンチマークを拡張する

Topic 01で定義された臨床ノートIEタスク(4つのフィールド:症状、投薬、検査、診断を抽出)を、以下の5つ以上のベンダーでベンチマークします。

  • Claude Opus 4.5: Anthropicの最上位モデル、精度を最優先。
  • Claude Sonnet 4.5: Anthropicの標準モデル、コストと速度のバランスが取れている。
  • GPT-4o: OpenAIの標準モデル。
  • GPT-4o-mini: OpenAIの軽量モデル、低コスト。
  • o1: OpenAIの推論モデル、最高の精度を持つが、レイテンシーとコストが高い。
  • Gemini 2.5 Pro: Googleの最上位モデル。
  • Gemini 2.5 Flash: Googleの軽量モデル。
  • Meditron 7Bまたは70B: EPFLのオープンソースモデル、ローカルで実行可能。
  • Llama-3-Med(コミュニティによるファインチューニング):オープンソースの代替モデル。

ベンチマークの現実世界の要件

  • 精度: フィールドごとのF1、完全一致、BLEU、タスクに適切な指標。
  • コスト: 1000件のリクエストを処理するのにかかるUSD(APIの価格表から計算)。
  • レイテンシー: p50、p95、p99のレイテンシー(ミリ秒)。
  • 安定性: 失敗率、タイムアウト、レート制限のヒット数、リトライ成功率。
  • 応答形式のコンプライアンス: 構造化された出力が要求された場合のJSON解析の成功率。
  • ベンダーごとの強み/弱みのマトリックス: どのベンダーがどのフィールドやタスクタイプで優れているか。

ベンダー比較の落とし穴(現実世界のベンチマーク設計で考慮する必要がある)

  • プロンプト最適化の偏り: ベンダーA用に最適化されたプロンプトでベンダーBを評価すると、ベンダーBのパフォーマンスが過小評価される可能性があります。
  • モデルバージョンのドリフト: ベンダーがデプロイするモデルは頻繁に更新されるため、ベンチマークの再現性が損なわれます。gpt-4o-2024-08-06のようなスナップショットピンが不可欠です。
  • コンテキストサイズの差: 最大コンテキストサイズはベンダーによって異なります(Claude 200k、GPT-4o 128k、Gemini 2M、Meditron 4〜8k)。
  • コスト構造の差: 入力/出力トークンの料金が異なり、キャッシュ割引のベンダーポリシーも異なります。
  • リージョナルレイテンシー: APIエンドポイントの地理的な違い(米国、EU、アジア)。
  • 構造化出力機能の違い: OpenAIのJSONモード、Claudeのtool_use、Geminiのresponse_schema。これらを使用すると、精度と公平性のトレードオフが生じます。

この記事のターゲット指標

  • 8〜10のベンダーそれぞれで、臨床IEタスクを実行します(100〜500のサンプル)。
  • 各ベンダーに対して、定量的な3軸レポート(F1、コスト、レイテンシー)を自動生成します。
  • ベンダーごとの既知の強み/弱みのマトリックス(例:Claudeは構造化出力に強く、Geminiはレイテンシーが低い)。
  • 再現可能なベンチマーク:プロンプト、データセット、モデルスナップショット、実行時間をピン留めします。
  • パレートフロンティアの視覚化(精度とコスト、精度とレイテンシー)。

ツールスタックとインフラストラクチャ要件

ツール役割ライセンス
Anthropic Python SDKClaude APIクライアントMIT
OpenAI Python SDKGPT-4o、o1、o3クライアントApache 2.0
google-generativeaiGemini APIクライアントApache 2.0
HuggingFace transformers + vLLM(オプション)Meditronローカル推論Apache 2.0
pandas、matplotlib、seaborn結果テーブル、パレート視覚化BSD
pytest(オプション)ベンチマーク再現性の検証MIT
asyncio、aiohttp並列API呼び出しPSF、MIT
structlog実行ログApache 2.0

インフラストラクチャ要件

  • GPUは不要(API中心)。オープンソースモデルのローカル実行の場合、小〜中規模/大規模GPU(Meditron 7Bは小規模GPU、70Bは24GB以上のVRAMを持つデータセンターGPU)。
  • 16GB以上のRAM。
  • ネットワーク:各ベンダーのAPIエンドポイントへのアクセス(地域によっては制限があることに注意)。

学習者の再現にかかる推定コスト: 8つのベンダー×500サンプル≈30〜150USD(各ベンダーの価格表から計算[1][2][3])。Meditronローカルは無料(GPU時間は除く)。

実世界のパイプライン実装

全体の流れ:

mermaid

ステップ1. ベンダーに依存しないプロンプトの設計

公平性を確保するために、ベンダー固有の構文(Claude XML、Gemini system_instruction、GPT function callingなど)ではなく、一般的な自然言語プロンプトをデフォルトのベンチマークとして使用します。

python
VENDOR_NEUTRAL_PROMPT = """あなたは臨床自然言語処理の専門家です。
与えられた臨床記録から4つのフィールドを構造化されたJSON形式で抽出してください。
JSON形式で1つのJSONのみを出力し、説明やコメントは含めないでください。
スキーマ:
{
"symptoms": ["文字列のリスト"],
"medications": [{"name": "薬の名前", "dose": "投与量", "frequency": "頻度"}],
"labs": [{"test": "検査名", "value": "数値", "unit": "単位"}],
"diagnoses": ["診断の文字列のリスト"]
}
原則:
1. 記録にない情報を推測したり、でっち上げたりしないでください(ハルシネーション)。
2. フィールドが存在しない場合は、空の配列 [] を使用します。
3. JSONのパースに失敗した場合、ベンチマークは無効になります。JSON形式のみを使用してください。
臨床記録:
{note}
JSON出力:
"""
# ベンダーごとの最適化されたプロンプト(オプションのベンチマーク)
VENDOR_OPTIMIZED_PROMPTS = {
"anthropic": VENDOR_NEUTRAL_PROMPT, # Claudeは自然言語プロンプトで強力
"openai": VENDOR_NEUTRAL_PROMPT + "\n\n(response_format=json_object)", # JSONモードのヒント
"google": VENDOR_NEUTRAL_PROMPT, # Geminiはresponse_schemaを別途使用
"opensource": VENDOR_NEUTRAL_PROMPT + "\n\nAnswer:", # Meditronなどは継続のためのキューが必要
}

ステップ2. ベンダーアダプター(共通インターフェース)

python
import asyncio
import time
import json
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
import anthropic
from openai import OpenAI
import google.generativeai as genai
@dataclass
class VendorResponse:
vendor: str
model: str
model_snapshot: str # スナップショットID(バージョン固定)
parsed_output: dict | None
raw_text: str
input_tokens: int
output_tokens: int
latency_ms: float
error: str | None
retry_count: int = 0
class VendorAdapter(ABC):
def __init__(self, model: str, max_retries: int = 3):
self.model = model
self.max_retries = max_retries
@abstractmethod
def query(self, prompt: str) -> VendorResponse:
...
async def query_async(self, prompt: str) -> VendorResponse:
"""asyncioのサポート。本番環境ではaiohttpやベンダーの非同期SDKを使用します。"""
return await asyncio.get_event_loop().run_in_executor(None, self.query, prompt)
class ClaudeAdapter(VendorAdapter):
def __init__(self, model: str = "claude-sonnet-4-5-20241022"):
super().__init__(model)
self.client = anthropic.Anthropic()
self.snapshot = model
def query(self, prompt: str) -> VendorResponse:
start = time.perf_counter()
retry = 0
for attempt in range(self.max_retries):
try:
resp = self.client.messages.create(
model=self.model,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}],
)
elapsed = (time.perf_counter() - start) * 1000
raw = resp.content[0].text
parsed = try_parse_json(raw)
return VendorResponse(
vendor="anthropic", model=self.model, model_snapshot=self.snapshot,
parsed_output=parsed, raw_text=raw,
input_tokens=resp.usage.input_tokens,
output_tokens=resp.usage.output_tokens,
latency_ms=elapsed, error=None, retry_count=retry,
)
except anthropic.RateLimitError:
retry += 1
time.sleep(2 ** attempt)
except Exception as e:
return VendorResponse(
vendor="anthropic", model=self.model, model_snapshot=self.snapshot,
parsed_output=None, raw_text="",
input_tokens=0, output_tokens=0,
latency_ms=(time.perf_counter() - start) * 1000,
error=str(e), retry_count=retry,
)
return VendorResponse(
vendor="anthropic", model=self.model, model_snapshot=self.snapshot,
parsed_output=None, raw_text="",
input_tokens=0, output_tokens=0,
latency_ms=(time.perf_counter() - start) * 1000,
error="max_retries_exceeded", retry_count=retry,
)
class OpenAIAdapter(VendorAdapter):
def __init__(self, model: str = "gpt-4o-2024-08-06", use_json_mode: bool = True):
super().__init__(model)
self.client = OpenAI()
self.snapshot = model
self.use_json_mode = use_json_mode
def query(self, prompt: str) -> VendorResponse:
start = time.perf_counter()
retry = 0
for attempt in range(self.max_retries):
try:
kwargs = {
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 2048,
}
if self.use_json_mode and not self.model.startswith("o"):
kwargs["response_format"] = {"type": "json_object"}
resp = self.client.chat.completions.create(**kwargs)
elapsed = (time.perf_counter() - start) * 1000
raw = resp.choices[0].message.content or ""
parsed = try_parse_json(raw)
return VendorResponse(
vendor="openai", model=self.model, model_snapshot=self.snapshot,
parsed_output=parsed, raw_text=raw,
input_tokens=resp.usage.prompt_tokens,
output_tokens=resp.usage.completion_tokens,
latency_ms=elapsed, error=None, retry_count=retry,
)
except Exception as e:
if "rate_limit" in str(e).lower():
retry += 1
time.sleep(2 ** attempt)
continue
return VendorResponse(
vendor="openai", model=self.model, model_snapshot=self.snapshot,
parsed_output=None, raw_text="",
input_tokens=0, output_tokens=0,
latency_ms=(time.perf_counter() - start) * 1000,
error=str(e), retry_count=retry,
)
return VendorResponse(
vendor="openai", model=self.model, model_snapshot=self.snapshot,
parsed_output=None, raw_text="",
input_tokens=0, output_tokens=0,
latency_ms=(time.perf_counter() - start) * 1000,
error="max_retries_exceeded", retry_count=retry,
)
class GeminiAdapter(VendorAdapter):
def __init__(self, model: str = "gemini-2.5-pro"):
super().__init__(model)
genai.configure()
self.snapshot = model
self.model_obj = genai.GenerativeModel(model)
def query(self, prompt: str) -> VendorResponse:
start = time.perf_counter()
try:
resp = self.model_obj.generate_content(prompt)
elapsed = (time.perf_counter() - start) * 1000
raw = resp.text if hasattr(resp, "text") else ""
parsed = try_parse_json(raw)
usage = getattr(resp, "usage_metadata", None)
input_toks = getattr(usage, "prompt_token_count", 0) if usage else 0
output_toks = getattr(usage, "candidates_token_count", 0) if usage else 0
return VendorResponse(
vendor="google", model=self.model, model_snapshot=self.snapshot,
parsed_output=parsed, raw_text=raw,
input_tokens=input_toks, output_tokens=output_toks,
latency_ms=elapsed, error=None,
)
except Exception as e:
return VendorResponse(
vendor="google", model=self.model, model_snapshot=self.snapshot,
parsed_output=None, raw_text="",
input_tokens=0, output_tokens=0,
latency_ms=(time.perf_counter() - start) * 1000,
error=str(e),
)
class MeditronAdapter(VendorAdapter):
"""ローカルのオープンモデル(Meditron 7B/70B)。vLLM、TGI、HFパイプラインを使用します。"""
def __init__(self, model_id: str = "epfl-llm/meditron-7b", device: str = "cuda"):
super().__init__(model_id)
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
self.tokenizer = AutoTokenizer.from_pretrained(model_id)
self.model_obj = AutoModelForCausalLM.from_pretrained(
model_id, torch_dtype=torch.float16, device_map=device,
)
self.device = device
self.snapshot = model_id
self.torch = torch
def query(self, prompt: str) -> VendorResponse:
start = time.perf_counter()
try:
inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device)
with self.torch.no_grad():
outputs = self.model_obj.generate(
**inputs, max_new_tokens=2048, do_sample=False, temperature=0.1,
)
elapsed = (time.perf_counter() - start) * 1000
raw = self.tokenizer.decode(
outputs[0][inputs.input_ids.size(1):], skip_special_tokens=True,
)
parsed = try_parse_json(raw)
return VendorResponse(
vendor="opensource", model=self.model, model_snapshot=self.snapshot,
parsed_output=parsed, raw_text=raw,
input_tokens=int(inputs.input_ids.size(1)),
output_tokens=int(outputs.size(1) - inputs.input_ids.size(1)),
latency_ms=elapsed, error=None,
)
except Exception as e:
return VendorResponse(
vendor="opensource", model=self.model, model_snapshot=self.snapshot,
parsed_output=None, raw_text="",
input_tokens=0, output_tokens=0,
latency_ms=(time.perf_counter() - start) * 1000,
error=str(e),
)
def try_parse_json(text: str) -> dict | None:
"""寛容なJSONパース。Markdownのフェンスと周囲のテキストを削除します。"""
if not text:
return None
match = re.search(r"\{.*\}", text, re.DOTALL)
if not match:
return None
try:
return json.loads(match.group(0))
except json.JSONDecodeError:
# もう一度試す:JSONブロックを囲むフェンス
fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
if fence:
try:
return json.loads(fence.group(1))
except json.JSONDecodeError:
pass
return None

ステップ3. コスト計算(固定されたAPI価格)

python
# 実行時のベンダー公式価格(参照。記事公開時に再確認する必要があります)
PRICING_TABLE_USD_PER_M_TOKENS = {
"claude-opus-4-5-20250219": {"input": 15.0, "output": 75.0},
"claude-sonnet-4-5-20241022": {"input": 3.0, "output": 15.0},
"claude-haiku-4-5-20251001": {"input": 1.0, "output": 5.0},
"gpt-4o-2024-08-06": {"input": 2.5, "output": 10.0},
"gpt-4o-mini-2024-07-18": {"input": 0.15, "output": 0.60},
"o1-2024-12-17": {"input": 15.0, "output": 60.0},
"o3-2025-04-16": {"input": 10.0, "output": 40.0},
"gemini-2.5-pro": {"input": 1.25, "output": 5.0},
"gemini-2.5-flash": {"input": 0.15, "output": 0.6},
"epfl-llm/meditron-7b": {"input": 0.0, "output": 0.0}, # ローカル
"epfl-llm/meditron-70b": {"input": 0.0, "output": 0.0},
}
def calculate_cost_usd(model_snapshot: str, input_tokens: int, output_tokens: int) -> float:
"""トークン数 × 料金 = USDコスト。"""
pricing = PRICING_TABLE_USD_PER_M_TOKENS.get(model_snapshot, {"input": 0.0, "output": 0.0})
return (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1_000_000

ステップ4. ベンチマーク実行オーケストレーター(非同期)

python
from dataclasses import dataclass
@dataclass
class BenchResult:
vendor_response: VendorResponse
cost_usd: float
f1_scores: dict[str, float]
gold_label: dict
async def benchmark_vendor(
adapter: VendorAdapter,
dataset: list[dict], # [{note, gold_label}]
prompt_template: str = VENDOR_NEUTRAL_PROMPT,
max_concurrent: int = 5,
) -> list[BenchResult]:
"""すべてのデータセットに対して1つのベンダーを実行します。"""
semaphore = asyncio.Semaphore(max_concurrent)
async def process_one(sample: dict) -> BenchResult:
async with semaphore:
prompt = prompt_template.replace("{note}", sample["note"])
resp = await adapter.query_async(prompt)
cost = calculate_cost_usd(resp.model_snapshot, resp.input_tokens, resp.output_tokens)
if resp.parsed_output:
f1 = evaluate_extraction(resp.parsed_output, sample["gold_label"])
else:
f1 = {"symptoms": 0.0, "medications": 0.0, "labs": 0.0, "diagnoses": 0.0}
return BenchResult(
vendor_response=resp, cost_usd=cost,
f1_scores=f1, gold_label=sample["gold_label"],
)
return await asyncio.gather(*[process_one(s) for s in dataset])
def evaluate_extraction(pred: dict, gold: dict) -> dict[str, float]:
"""ステップ01のF1スコア計算関数を再利用します。"""
scores = {}
for field in ["symptoms", "diagnoses"]:
p = {s.lower().strip() for s in pred.get(field, []) if isinstance(s, str)}
g = {s.lower().strip() for s in gold.get(field, []) if isinstance(s, str)}
scores[field] = f1_score(p, g)
for field in ["medications", "labs"]:
p = {json.dumps(x, sort_keys=True) for x in pred.get(field, []) if isinstance(x, dict)}
g = {json.dumps(x, sort_keys=True) for x in gold.get(field, []) if isinstance(x, dict)}
scores[field] = f1_score(p, g)
return scores
def f1_score(pred: set, gold: set) -> float:
if not pred and not gold:
return 1.0
tp = len(pred & gold)
if tp == 0:
return 0.0
precision = tp / len(pred)
recall = tp / len(gold)
return 2 * precision * recall / (precision + recall)

ステップ5. 3軸レポート + パレートフロンティア

python
import numpy as np
import pandas as pd
def aggregate_report(all_results: dict[str, list[BenchResult]]) -> pd.DataFrame:
"""ベンダーごとの3軸要約テーブル。"""
rows = []
for vendor_key, results in all_results.items():
valid = [r for r in results if r.vendor_response.error is None and r.vendor_response.parsed_output]
n_valid = len(valid)
n_total = len(results)
if n_valid == 0:
rows.append({
"vendor": vendor_key, "n_valid": 0, "n_total": n_total,
"macro_f1": 0.0, "cost_per_1k_usd": 0.0,
"latency_p50_ms": 0.0, "latency_p95_ms": 0.0, "latency_p99_ms": 0.0,
"failure_rate": 1.0,
})
continue
avg_f1 = {
field: float(np.mean([r.f1_scores[field] for r in valid]))
for field in ["symptoms", "medications", "labs", "diagnoses"]
}
macro_f1 = float(np.mean(list(avg_f1.values())))
total_cost = sum(r.cost_usd for r in valid)
cost_per_1k = (total_cost / n_valid) * 1000
latencies = [r.vendor_response.latency_ms for r in valid]
p50 = float(np.percentile(latencies, 50))
p95 = float(np.percentile(latencies, 95))
p99 = float(np.percentile(latencies, 99))
failure_rate = 1 - (n_valid / n_total)
rows.append({
"vendor": vendor_key,
"model": valid[0].vendor_response.model,
"snapshot": valid[0].vendor_response.model_snapshot,
"n_valid": n_valid, "n_total": n_total,
"macro_f1": round(macro_f1, 4),
**{f"f1_{k}": round(v, 4) for k, v in avg_f1.items()},
"cost_per_1k_usd": round(cost_per_1k, 3),
"latency_p50_ms": round(p50, 1),
"latency_p95_ms": round(p95, 1),
"latency_p99_ms": round(p99, 1),
"failure_rate": round(failure_rate, 4),
})
return pd.DataFrame(rows).sort_values("macro_f1", ascending=False)
def plot_3axis_pareto(report_df: pd.DataFrame, output_path: str = "bench_pareto.png") -> None:
"""3軸散布図:精度 vs コスト vs レイテンシー + パレートフロンティア。"""
import matplotlib.pyplot as plt
fig, axes = plt.subplots(1, 2, figsize=(16, 6))
# (a) 精度 vs コスト
ax = axes[0]
scatter = ax.scatter(
report_df["cost_per_1k_usd"], report_df["macro_f1"],
c=report_df["latency_p50_ms"], s=200, cmap="viridis",
)
for _, row in report_df.iterrows():
ax.annotate(row["model"], (row["cost_per_1k_usd"], row["macro_f1"]),
xytext=(5, 5), textcoords="offset points", fontsize=8)
ax.set_xlabel("コスト(1,000リクエストあたり、USD)")
ax.set_ylabel("マクロF1")
ax.set_xscale("log")
ax.set_title("精度 vs コスト(対数スケール)")
plt.colorbar(scatter, ax=ax, label="レイテンシーp50(ms)")
# (b) 精度 vs レイテンシー
ax2 = axes[1]
ax2.scatter(
report_df["latency_p50_ms"], report_df["macro_f1"],
s=200, c="steelblue",
)
for _, row in report_df.iterrows():
ax2.annotate(row["model"], (row["latency_p50_ms"], row["macro_f1"]),
xytext=(5, 5), textcoords="offset points", fontsize=8)
ax2.set_xlabel("レイテンシーp50(ms)")
ax2.set_ylabel("マクロF1")
ax2.set_title("精度 vs レイテンシー")
plt.tight_layout()
plt.savefig(output_path, dpi=150, bbox_inches="tight")
def identify_pareto_frontier(df: pd.DataFrame, higher_better: list[str], lower_better: list[str]) -> pd.DataFrame:
"""パレートフロンティアの候補のみを抽出します。"""
pareto = df.copy()
is_pareto = np.ones(len(df), dtype=bool)
for i in range(len(df)):
for j in range(len(df)):
if i == j:
continue
better_or_equal_all = True
strictly_better_any = False
for col in higher_better:
if df.iloc[j][col] < df.iloc[i][col]:
better_or_equal_all = False
break
if df.iloc[j][col] > df.iloc[i][col]:
strictly_better_any = True
if not better_or_equal_all:
continue
for col in lower_better:
if df.iloc[j][col] > df.iloc[i][col]:
better_or_equal_all = False
break
if df.iloc[j][col] < df.iloc[i][col]:
strictly_better_any = True
if better_or_equal_all and strictly_better_any:
is_pareto[i] = False
break
pareto = pareto[is_pareto]
return pareto

ステップ6. 統合パイプライン

python
from pathlib import Path
async def full_benchmark(
dataset: list[dict],
vendors: list[VendorAdapter],
output_dir: Path,
) -> pd.DataFrame:
"""完全なベンチマーク実行とレポート。"""
output_dir.mkdir(parents=True, exist_ok=True)
all_results = {}
for adapter in vendors:
key = f"{adapter.__class__.__name__}_{adapter.model.split('/')[-1]}"
print(f"ベンチマークを実行中:{key}")
results = await benchmark_vendor(adapter, dataset, VENDOR_NEUTRAL_PROMPT)
all_results[key] = results
# 個々の生の結果を保存
with open(output_dir / f"raw_{key}.jsonl", "w") as f:
for r in results:
f.write(json.dumps({
"model": r.vendor_response.model,
"parsed": r.vendor_response.parsed_output,
"f1": r.f1_scores,
"latency_ms": r.vendor_response.latency_ms,
"cost_usd": r.cost_usd,
"error": r.vendor_response.error,
}) + "\n")
report = aggregate_report(all_results)
report.to_csv(output_dir / "bench_report.csv", index=False)
plot_3axis_pareto(report, str(output_dir / "bench_pareto.png"))
# パレートフロンティアを抽出
pareto = identify_pareto_frontier(
report, higher_better=["macro_f1"], lower_better=["cost_per_1k_usd", "latency_p50_ms"],
)
pareto.to_csv(output_dir / "pareto_frontier.csv", index=False)
print(f"パレートフロンティアのベンダー:{list(pareto['model'])}")
return report
# 実行例
# vendors = [
# ClaudeAdapter("claude-sonnet-4-5-20241022"),
# ClaudeAdapter("claude-opus-4-5-20250219"),
# OpenAIAdapter("gpt-4o-2024-08-06"),
# OpenAIAdapter("gpt-4o-mini-2024-07-18"),
# GeminiAdapter("gemini-2.5-pro"),
# GeminiAdapter("gemini-2.5-flash"),
# MeditronAdapter("epfl-llm/meditron-7b"),
# ]
# report = asyncio.run(full_benchmark(dataset, vendors, Path("./bench_output")))
## パフォーマンス・コスト・既知の失敗事例
### パフォーマンスの参照(公開されたベンチマークの引用)
臨床IEタスクにおける概算のパフォーマンス(複数の論文や企業のベンチマークにおける参照):
| ベンダー・モデル | F1(臨床IE) | コスト/1000リクエスト(参照) | レイテンシp50 | ソース |
|----------------|:----------------:|:------------------------:|:-----------:|--------|
| Claude Opus 4.5 | 0.870.91 | 3050 USD | 35 秒 | Anthropicベンチマーク [1] |
| Claude Sonnet 4.5 | 0.840.88 | 510 USD | 12 秒 | Anthropicベンチマーク [1] |
| GPT-4o | 0.830.87 | 510 USD | 23 秒 | Agrawal et al., NEJM AI 2024 [4] |
| GPT-4o-mini | 0.750.80 | 0.51 USD | 12 秒 | 独自のベンチマーク [2] |
| o1 | 0.880.92 | 4060 USD | 1030 秒(推論) | OpenAIベンチマーク [2] |
| o3 | 0.890.92 | 2545 USD | 820 秒 | OpenAIベンチマーク [2] |
| Gemini 2.5 Pro | 0.850.89 | 37 USD | 12 秒 | Google Researchベンチマーク [3] |
| Med-Gemini(特化) | 0.870.91 | 37 USD | 12 秒 | Google Research 2024 [5] |
| Meditron 7B(ローカル) | 0.720.78 | 0(ローカルGPU) | 515 秒(7B、RTX 4090) | Chen et al., EPFL 2023 [6] |
| Meditron 70B | 0.800.85 | 0(ローカルGPU) | 3060 秒(70B、24GB+ VRAM) | Chen et al. 2023 [6] |
### 推定される学習者による再現にかかるコスト
- 8ベンダー × 500サンプルの実行:合計で約30150 USD(各ベンダーの価格は、その時点で確認する必要があります)。
- Meditronのローカル実行における、追加のGPU時間。
- Claude・OpenAI・Geminiからのプロンプトキャッシュを使用することで、半分のコスト、またはそれ以上に削減できます。
### 5つの既知の失敗事例(コミュニティ/論文から収集)
1. **ベンダーごとのJSON解析の失敗率の違い**
症状:同じプロンプトを使用した場合、GPT-4o(json_mode)はJSONを99%以上解析し、GeminiはMarkdownフェンスで515%、Meditronは30%以上で失敗します。
原因:ベンダーごとのトレーニングデータにおける応答形式の偏り。ベンダーの構造化出力機能に対するサポートの違い。
対策:(a)寛容なJSONパーサー(`{...}`を抽出するための正規表現、Markdownフェンスの処理)、(b)ベンダーごとの構造化出力機能を使用してベンチマークを再設計する(OpenAI json_mode・Claude tool_use・Gemini response_schema)、(c)解析の成功率もベンチマークの指標として報告する、(d)失敗したケースの生データを保存して、事後分析を行う。
出典:「structured output」に関するOpenAI・Gemini・Anthropicのデベロッパーフォーラムのスレッド[7]。
2. **モデルバージョンの静かな更新が、ベンチマークの再現性を損なう**
症状:ベンダーが`gpt-4o`エイリアスを新しいチェックポイントに置き換える→パフォーマンスが以前のベンチマークの結果と異なる。
原因:ベンダーの最新のエイリアスは常に更新される。
対策:(a)具体的なスナップショットIDを指定する(例:`gpt-4o-2024-08-06`、`claude-sonnet-4-5-20241022`)、(b)ベンチマークの実行時間・モデルバージョンを記録する、(c)定期的に(月次)再現ベンチマークを繰り返す、(d)ベンダーごとのモデルバージョンの非推奨ポリシーを監視する。
出典:OpenAIのモデルバージョンに関するドキュメント[8]。
3. **レート制限がレイテンシを歪める**
症状:多数の並行リクエストがある場合、レート制限の429応答→再試行の待機時間がレイテンシに含まれ、p95が急増する。
原因:ベンダーごとのレート制限は、ティアによって異なる(使用ティア・組織ティア)。
対策:(a)ベンダーごとの`max_concurrent`を調整する、(b)指数バックオフ+429での再試行、(c)失敗したリクエストをレイテンシの統計から分離して処理する、(d)有料ティアにアップグレードする、(e)事前にベンダーごとのレート制限ダッシュボードを確認する。
出典:Anthropic・OpenAIのレート制限に関するドキュメント[9]。
4. **プロンプトの専門化による偏り(ベンダーごとに最適なプロンプトが異なる)**
症状:ベンダーに依存しないプロンプトを使用した場合、ベンダーAが最適に機能し、ベンダーBは過小評価される。各ベンダーの最適なプロンプトを使用すると、公平性の議論が生じる。
原因:ベンダーごとのトレーニングデータ・RLHFポリシーの違い。Claudeは自然言語の指示に強く、GPTはFew-Shotに強く、GeminiはXMLタグに強い(一般的な傾向)。
対策:(a)ベンダーに依存しないプロンプト(この記事のデフォルト)+ベンダーに最適化されたプロンプトによる別のベンチマーク(両方の結果を並べて報告する)、(b)プロンプト・最適化の取り組みを学習者に明確に開示する、(c)DSPyなどの自動プロンプト最適化フレームワークを使用する。
出典:Anthropicのプロンプトエンジニアリングガイド・OpenAIのプロンプトエンジニアリングドキュメント・DSPy論文[10]。
5. **Meditron・Llama-3-Medローカル実行における環境変動**
症状:同じMeditron 70Bが、別の人のGPUではF1 0.85を達成するが、学習者のローカルでは0.72しか達成しない。
原因:(a)量子化方法の違い(fp16 vs int8 vs int4)、(b)アテンションの実装の違い(flash-attentionの有無)、(c)トークナイザーのバージョン、(d)サンプリングパラメータ(temperature・top_p)。
対策:(a)正確なfp16・flash-attention 2の条件を指定する、(b)トークナイザー・transformersのバージョンを固定する、(c)サンプリングパラメータ(temperature=0.1、top_p=1.0など)を再現する、(d)標準の推論フレームワーク(vLLMなど)を使用する。
出典:Meditron GitHub・HuggingFace transformersの再現性に関するディスカッション[11]。
## 拡張のアイデア
- **マルチタスク拡張**: 臨床IEに加えて、診断予測・要約・翻訳・CoT推論ベンチマークを並行して実行する。
- **A/Bプロンプトチューニング**: ベンダーごとの最適なプロンプトを自動的に検索する(DSPy・APE・promptbreeder)。
- **継続的なベンチマーク**: 毎週/毎月の自動ベンチマーク実行→モデルの更新を追跡し、回帰を検出する。
- **コスト-精度のパレート最適化**: タスクごとのパレートフロンティアの可視化→ベンダー選択ガイド。
- **オフラインとオンラインのオープンモデル**: Meditron・Llama-Med・MedGemma・Med42などのオープンソースのモデルを継続的に追加する。
- **ハイブリッドルーティング**: 容易なケースを安価なモデル(Haiku・GPT-4o-mini)に、難しいケースをトップティアのモデル(Opus・o1)にルーティングする。
## 次のトピック
- トピック10 `med-llm-reproduction`: HealthBench、MedQAベンチマークの再現を行います(このトピックが現実世界のタスク比較である場合、トピック10は標準ベンチマークの再現性を検証します)。
- トピック14 `bio-mcp-agent`: ベンチマークの結果をMCPツールとして公開し、エージェントが「このタスクに最適なベンダーはどれか」を自律的に判断するようにします。
- トピック01 `clinical-notes-ie-llm`: この記事のベンチマークの結果を基に、トピック01のパイプラインでベンダーを選択します。
## 参考文献
1. Anthropic Claude APIの価格とベンチマーク: `https://www.anthropic.com/pricing` · `https://www.anthropic.com/news/claude-3-family`
2. OpenAI APIの価格: `https://openai.com/api/pricing/`
3. Google AI Studioの価格(Gemini):`https://ai.google.dev/gemini-api/docs/pricing`
4. Agrawal M, Hegselmann S, Lang H, et al. "Large Language Models are Few-Shot Clinical Information Extractors." NEJM AI 2024.
5. Saab K, Tu T, Weng W-H, et al. "Capabilities of Gemini Models in Medicine (Med-Gemini)." Nature Medicine 2024. `https://www.nature.com/articles/s41591-024-03246-6`
6. Chen Z, Cano AH, Romanou A, et al. "Meditron-70B: Scaling Medical Pretraining for Large Language Models." EPFL 2023. `https://arxiv.org/abs/2311.16079`
7. OpenAI/Anthropic/Gemini開発者フォーラムにおける構造化出力に関する議論
8. OpenAIモデルのバージョン管理: `https://platform.openai.com/docs/models`
9. レート制限に関するドキュメント: `https://docs.anthropic.com/en/api/rate-limits` · `https://platform.openai.com/docs/guides/rate-limits`
10. DSPy (プロンプト最適化): `https://github.com/stanfordnlp/dspy` · Khattab O et al. 2023
11. Meditron GitHub: `https://github.com/epfLLM/meditron`
12. HuggingFace transformers: `https://huggingface.co/docs/transformers/`
13. Anthropic Python SDK: `https://github.com/anthropics/anthropic-sdk-python`
14. OpenAI Python SDK: `https://github.com/openai/openai-python`
15. google-generativeai Python: `https://github.com/google/generative-ai-python`
16. vLLM (高性能なオープンLLMサービング): `https://github.com/vllm-project/vllm`
17. Text Generation Inference (TGI, HuggingFace): `https://github.com/huggingface/text-generation-inference`
18. Med42 (オープンな医療LLM): `https://huggingface.co/m42-health/med42-70b`
19. Anthropicプロンプトエンジニアリングガイド: `https://docs.anthropic.com/en/docs/build-with-claude/prompt-engineering/overview`
20. HELM (言語モデルの包括的な評価): `https://crfm.stanford.edu/helm/`

💬 質問・コメント

0件のコメント

ログインせずに投稿できます。ゲスト投稿は投稿者自身で編集・削除できません。

0/2000

読み込み中...