◈ API / SDK/2026-04-08上級

Gemini 2.5 の Thinking Budget を握る — コストと精度の釣り合いを取る実装

Gemini 2.5 の Thinking Budget パラメータを制御し、推論コストと精度の釣り合いを取る実装をまとめました。Thinking モデルの内部アーキテクチャ、コスト構造、タスク別の最適設定、動的調整システム、思考プロセスの可視化とデバッグまで解説します。

thinking-budget² gemini-2-5 cost-optimization²² gemini-api²⁸⁵ advanced¹²

✦ プレミアム記事

「考えすぎ」と「考えなさすぎ」の間で

Gemini 2.5 シリーズに搭載された「Thinking（思考）」機能は、モデルが回答を生成する前に内部で推論プロセスを実行する仕組みです。この思考プロセスに割り当てるトークン数を制御するパラメータが Thinking Budget（thinkingBudget）です。

2026年時点、Gemini 2.5 Pro および Gemini 2.5 Flash はどちらも Thinking モデルとして提供されており、このパラメータの適切な設定が APIコストと出力品質の両方に大きく影響します。

思考予算を理解せずに使い続けると、次のような問題が発生します。

単純な質問に対して数万トークンの思考が走り、コストが爆発的に増加する
逆に複雑な推論タスクに予算を絞りすぎて、精度が著しく低下する
思考トークンの課金仕様を誤解し、予算オーバーが続く

Gemini 2.5 Flash Thinking 完全ガイドで基本的な思考モードの使い方を学んだ方が、次のステップとして取り組むのに最適な内容です。

Thinking Modelの内部アーキテクチャを理解する

思考トークンとは

Thinking モデルは、ユーザーのプロンプトを受け取ると、まず「思考フェーズ」として内部推論を行います。この過程で生成されるトークンが「思考トークン（thinking tokens）」と呼ばれます。

思考フェーズの流れは以下の通りです。

ユーザーのプロンプトを受信
モデルが問題を分解・分析する内部モノローグを生成（思考トークン）
思考結果をもとに最終回答を生成（出力トークン）
ユーザーへのレスポンス送信

重要なのは、思考トークンはデフォルトではAPIレスポンスに含まれないという点です。思考の内容をデバッグや分析のために取得するには、別途設定が必要です。

通常モデルとの根本的な違い

通常モデルとThinkingモデルの比較：

推論方式: 通常モデルはシングルパス生成、Thinkingモデルは思考フェーズ→回答生成の2段階
得意タスク: 通常モデルは高速な情報取得・要約、Thinkingモデルは複雑な推論・数学・コード設計
コスト特性: 通常モデルは入出力トークンのみ課金、Thinkingモデルは思考トークン（入力として換算）も課金
精度: 通常モデルは単純タスクで十分、Thinkingモデルは多段階推論が必要なタスクで大幅に向上

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦Thinking Budgetの仕組みと課金体系を完全理解し、コストを最大70%削減できる

✦タスク複雑度に応じた動的Budget調整システムをPythonで実装できる

✦本番環境での思考トークン使用量モニタリングとA/Bテスト設計ができる

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

Thinking Budget パラメータの詳細仕様

設定方法とデフォルト値

Thinking Budget は generationConfig.thinkingConfig.thinkingBudget として設定します。

import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
model = genai.GenerativeModel(
    model_name="gemini-2.5-pro-preview-05-06",
    generation_config={
        "thinkingConfig": {
            "thinkingBudget": 8192,  # 思考トークン上限を指定
        }
    }
)
 
response = model.generate_content("素数判定アルゴリズムをPythonで実装し、計算量を解説してください。")
print(response.text)
 
# 思考トークン使用量を確認
if hasattr(response, 'usage_metadata'):
    usage = response.usage_metadata
    print(f"思考トークン: {usage.thoughts_token_count}")
    print(f"出力トークン: {usage.candidates_token_count}")
    print(f"合計トークン: {usage.total_token_count}")

モデル別の設定可能範囲

Gemini 2.5 Flash の場合：

最小値: 0（Thinking完全無効）
最大値: 24576（約24Kトークン）
デフォルト: dynamicThinkingが有効な場合はモデルが自動調整

Gemini 2.5 Pro の場合：

最小値: 0（Thinking完全無効）
最大値: 32768（約32Kトークン）
デフォルト: 動的調整（タスクの複雑さに応じてモデルが判断）

Dynamic Thinking との関係

thinkingBudget を明示的に設定しない場合、モデルは dynamicThinking モードで動作します。このモードでは、モデルが入力プロンプトの複雑さを自動判断し、必要な思考量を自動的に決定します。

# Dynamic Thinking（デフォルト）の確認
model_dynamic = genai.GenerativeModel(
    model_name="gemini-2.5-flash-preview-04-17",
    generation_config={
        "thinkingConfig": {
            # thinkingBudget を指定しない = dynamicThinking が有効
        }
    }
)
 
# 特定の上限内で動的調整させる場合
model_capped = genai.GenerativeModel(
    model_name="gemini-2.5-flash-preview-04-17",
    generation_config={
        "thinkingConfig": {
            "thinkingBudget": 4096,  # 最大4096トークンの範囲で動的調整
        }
    }
)

コスト構造の完全理解

思考トークンの課金メカニズム

Thinking Budgetを理解する上で最も重要なのが、思考トークンの課金方式です。

重要: 思考トークンは「入力トークン」として課金されます。つまり、同じ入力プロンプトでも、Thinking Budgetが大きければ大きいほどコストが増加します。

2026年4月時点の Gemini 2.5 Pro の料金（参考値）：

入力トークン（200K以下）: $1.25 / 1M tokens
入力トークン（200K超）: $2.50 / 1M tokens
思考トークン: $3.50 / 1M tokens（入力より高い）
出力トークン: $10.00 / 1M tokens

思考トークンが通常の入力より高い料金設定になっているのは、推論処理の計算コストが高いためです。

実際のコスト試算

簡単な例で計算してみましょう。

def calculate_thinking_cost(
    input_tokens: int,
    thinking_tokens: int,
    output_tokens: int,
    model: str = "2.5-pro"
) -> dict:
    """思考コストの試算（概算、USD）"""
    if model == "2.5-pro":
        input_rate = 1.25 / 1_000_000   # 200K以下の場合
        thinking_rate = 3.50 / 1_000_000
        output_rate = 10.00 / 1_000_000
    else:  # 2.5-flash
        input_rate = 0.15 / 1_000_000
        thinking_rate = 3.50 / 1_000_000  # Flashも思考は高い
        output_rate = 0.60 / 1_000_000
 
    input_cost = input_tokens * input_rate
    thinking_cost = thinking_tokens * thinking_rate
    output_cost = output_tokens * output_rate
    total = input_cost + thinking_cost + output_cost
 
    return {
        "input_cost_usd": round(input_cost, 6),
        "thinking_cost_usd": round(thinking_cost, 6),
        "output_cost_usd": round(output_cost, 6),
        "total_usd": round(total, 6),
        "thinking_ratio": round(thinking_cost / total * 100, 1)
    }
 
# 例: 1000トークンのプロンプト + 16000トークンの思考 + 500トークンの回答
cost = calculate_thinking_cost(
    input_tokens=1000,
    thinking_tokens=16000,
    output_tokens=500
)
print(cost)
# 出力例:
# {
#   "input_cost_usd": 0.00125,
#   "thinking_cost_usd": 0.056,
#   "output_cost_usd": 0.005,
#   "total_usd": 0.062125,
#   "thinking_ratio": 90.1  # ← コストの90%が思考トークン！
# }

このように、思考トークンがコスト全体の大部分を占める場合があります。だからこそ、適切なBudget設定が重要なのです。

タスク別 Thinking Budget 最適設定ガイド

レベル1: Thinking不要のタスク（Budget = 0）

以下のタスクでは、Thinking機能は不要もしくは逆効果です。

単純な情報検索・質問応答
テキストの要約・翻訳
テンプレートに基づくコンテンツ生成
感情分析・分類タスク

# Thinking完全無効（Budget = 0）
model_no_thinking = genai.GenerativeModel(
    model_name="gemini-2.5-flash-preview-04-17",
    generation_config={
        "thinkingConfig": {
            "thinkingBudget": 0,  # 0 = Thinking完全無効
        }
    }
)
 
# 単純な翻訳タスク: Thinking不要
response = model_no_thinking.generate_content(
    "次の文章を英語に翻訳してください: 今日は良い天気ですね。"
)
print(response.text)
# 期待出力: "It's nice weather today."
# 思考トークン: 0（コストゼロ）

レベル2: 軽量思考が有効なタスク（Budget = 512〜2048）

簡単なコードのバグ修正
構造化データの生成（JSON・CSV）
比較的単純なロジック設計
短めの文章の推敲・改善

model_light = genai.GenerativeModel(
    model_name="gemini-2.5-flash-preview-04-17",
    generation_config={
        "thinkingConfig": {
            "thinkingBudget": 1024,
        }
    }
)
 
# バグ修正タスク: 軽量思考で十分
buggy_code = """
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-1)  # バグあり
"""
 
response = model_light.generate_content(
    f"以下のPythonコードのバグを修正してください:\n{buggy_code}"
)
print(response.text)

レベル3: 標準的な思考が必要なタスク（Budget = 2048〜8192）

中程度の複雑さのアルゴリズム設計
APIの統合設計・アーキテクチャ提案
データ分析・インサイト抽出
複数の選択肢を比較した意思決定支援

model_standard = genai.GenerativeModel(
    model_name="gemini-2.5-pro-preview-05-06",
    generation_config={
        "thinkingConfig": {
            "thinkingBudget": 6144,
        }
    }
)
 
# アーキテクチャ設計: 標準思考
response = model_standard.generate_content("""
マイクロサービス構成のECサービスを設計してください。
要件:
- 日次100万リクエスト
- 決済処理は外部API連携
- リアルタイム在庫管理が必要
- Cloudflare Workers環境で動作
 
アーキテクチャ図（Mermaid記法）とサービス分割の根拠を示してください。
""")
print(response.text)

レベル4: 高度な思考が必要なタスク（Budget = 8192〜24576）

複雑な数学的証明・最適化問題
大規模なシステムのリファクタリング計画
セキュリティ脆弱性の分析と対策提案
複雑なビジネスロジックのコード実装

model_heavy = genai.GenerativeModel(
    model_name="gemini-2.5-pro-preview-05-06",
    generation_config={
        "thinkingConfig": {
            "thinkingBudget": 16384,  # 16K思考トークン
        }
    }
)
 
# 競技プログラミングレベルの問題
response = model_heavy.generate_content("""
以下の問題をO(n log n)以下の計算量で解くPythonコードを実装してください:
 
N個の整数からなる配列Aがある。
配列の連続部分列のうち、要素の和が最大となるものの和を求めよ。
ただし、空の部分列も許可し、その場合の和は0とする。
 
例: A = [-2, 1, -3, 4, -1, 2, 1, -5, 4]
答え: 6 (部分列 [4, -1, 2, 1])
 
実装方針の説明と計算量証明も含めること。
""")
print(response.text)

動的 Thinking Budget 調整システムの実装

本番環境では、タスクの複雑さを自動判定してBudgetを動的に割り当てるシステムが非常に有効です。コスト効率を最大化しながら、必要な精度を確保できます。

import google.generativeai as genai
from enum import Enum
from dataclasses import dataclass
import re
 
class TaskComplexity(Enum):
    SIMPLE = "simple"
    MODERATE = "moderate"
    COMPLEX = "complex"
    VERY_COMPLEX = "very_complex"
 
@dataclass
class BudgetConfig:
    thinking_budget: int
    model_name: str
    reason: str
 
def classify_task_complexity(prompt: str) -> TaskComplexity:
    """
    プロンプトの特徴からタスク複雑度を判定する
    """
    prompt_lower = prompt.lower()
    word_count = len(prompt.split())
 
    # 高複雑度のキーワード
    complex_keywords = [
        "証明", "最適化", "アーキテクチャ", "セキュリティ", "脆弱性",
        "分散システム", "アルゴリズム", "計算量", "複雑", "詳細に",
        "prove", "optimize", "architecture", "security", "vulnerability",
        "distributed", "algorithm", "complexity", "comprehensive"
    ]
 
    # 単純タスクのキーワード
    simple_keywords = [
        "翻訳", "要約", "感情分析", "分類", "変換",
        "translate", "summarize", "classify", "convert", "list"
    ]
 
    complex_score = sum(1 for kw in complex_keywords if kw in prompt_lower)
    simple_score = sum(1 for kw in simple_keywords if kw in prompt_lower)
 
    # コードブロックが含まれているか
    has_code = "```" in prompt or "def " in prompt or "class " in prompt
 
    # 判定ロジック
    if simple_score > 0 and complex_score == 0 and word_count < 50:
        return TaskComplexity.SIMPLE
    elif complex_score >= 3 or (has_code and word_count > 200):
        return TaskComplexity.VERY_COMPLEX
    elif complex_score >= 1 or (has_code and word_count > 50):
        return TaskComplexity.COMPLEX
    else:
        return TaskComplexity.MODERATE
 
def get_optimal_budget(
    complexity: TaskComplexity,
    model_preference: str = "auto"
) -> BudgetConfig:
    """複雑度に応じた最適Budget設定を返す"""
    configs = {
        TaskComplexity.SIMPLE: BudgetConfig(
            thinking_budget=0,
            model_name="gemini-2.5-flash-preview-04-17",
            reason="単純タスクにはThinkingは不要。Flashで高速・低コスト処理"
        ),
        TaskComplexity.MODERATE: BudgetConfig(
            thinking_budget=2048,
            model_name="gemini-2.5-flash-preview-04-17",
            reason="中程度のタスク。Flash + 軽量思考でコスト効率を確保"
        ),
        TaskComplexity.COMPLEX: BudgetConfig(
            thinking_budget=8192,
            model_name="gemini-2.5-pro-preview-05-06",
            reason="複雑なタスク。Proモデルで標準的な思考量を確保"
        ),
        TaskComplexity.VERY_COMPLEX: BudgetConfig(
            thinking_budget=24576,
            model_name="gemini-2.5-pro-preview-05-06",
            reason="最高複雑度。Proモデルで最大限の思考力を発揮"
        ),
    }
    return configs[complexity]
 
class SmartGeminiClient:
    """動的Thinking Budget調整を行うスマートGeminiクライアント"""
 
    def __init__(self, api_key: str):
        genai.configure(api_key=api_key)
        self._models: dict = {}
        self._usage_log: list = []
 
    def _get_or_create_model(self, config: BudgetConfig) -> genai.GenerativeModel:
        """モデルをキャッシュして再利用"""
        cache_key = f"{config.model_name}_{config.thinking_budget}"
        if cache_key not in self._models:
            self._models[cache_key] = genai.GenerativeModel(
                model_name=config.model_name,
                generation_config={
                    "thinkingConfig": {
                        "thinkingBudget": config.thinking_budget,
                    }
                }
            )
        return self._models[cache_key]
 
    def generate(
        self,
        prompt: str,
        force_complexity: TaskComplexity = None,
        verbose: bool = False
    ) -> tuple[str, dict]:
        """
        プロンプトを送信し、(response_text, usage_info) を返す
 
        Args:
            prompt: ユーザープロンプト
            force_complexity: 強制的に指定する複雑度（Noneで自動判定）
            verbose: デバッグ情報を出力するか
        """
        # 複雑度判定
        complexity = force_complexity or classify_task_complexity(prompt)
        config = get_optimal_budget(complexity)
 
        if verbose:
            print(f"📊 タスク複雑度: {complexity.value}")
            print(f"🤖 使用モデル: {config.model_name}")
            print(f"💭 思考Budget: {config.thinking_budget} tokens")
            print(f"💡 選択理由: {config.reason}")
 
        # モデル取得・実行
        model = self._get_or_create_model(config)
        response = model.generate_content(prompt)
 
        # 使用量情報の収集
        usage_info = {}
        if hasattr(response, 'usage_metadata') and response.usage_metadata:
            usage = response.usage_metadata
            thinking_tokens = getattr(usage, 'thoughts_token_count', 0) or 0
            output_tokens = getattr(usage, 'candidates_token_count', 0) or 0
            input_tokens = getattr(usage, 'prompt_token_count', 0) or 0
 
            usage_info = {
                "complexity": complexity.value,
                "model": config.model_name,
                "thinking_budget": config.thinking_budget,
                "actual_thinking_tokens": thinking_tokens,
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "budget_utilization": (
                    round(thinking_tokens / config.thinking_budget * 100, 1)
                    if config.thinking_budget > 0 else 0
                )
            }
 
            if verbose:
                print(f"\n📈 実際の思考トークン: {thinking_tokens}")
                print(f"📈 Budget使用率: {usage_info['budget_utilization']}%")
 
            self._usage_log.append(usage_info)
 
        return response.text, usage_info
 
    def get_cost_report(self) -> dict:
        """累積コストレポートを生成"""
        if not self._usage_log:
            return {"message": "使用履歴なし"}
 
        total_thinking = sum(u.get("actual_thinking_tokens", 0) for u in self._usage_log)
        total_input = sum(u.get("input_tokens", 0) for u in self._usage_log)
        total_output = sum(u.get("output_tokens", 0) for u in self._usage_log)
 
        complexity_breakdown = {}
        for u in self._usage_log:
            c = u.get("complexity", "unknown")
            complexity_breakdown[c] = complexity_breakdown.get(c, 0) + 1
 
        return {
            "total_requests": len(self._usage_log),
            "total_thinking_tokens": total_thinking,
            "total_input_tokens": total_input,
            "total_output_tokens": total_output,
            "complexity_breakdown": complexity_breakdown,
            "avg_thinking_per_request": round(total_thinking / len(self._usage_log), 0)
        }
 
# 使用例
client = SmartGeminiClient(api_key="YOUR_GEMINI_API_KEY")
 
# 単純タスク（Thinking不要と自動判定）
text, usage = client.generate("こんにちは を英語に翻訳してください", verbose=True)
 
# 複雑なタスク（高Budget自動割当）
code, usage = client.generate("""
スレッドセーフなキャッシュシステムをPythonで実装してください。
LRUアルゴリズムを使用し、TTL（有効期限）機能も含めてください。
ユニットテストも合わせて実装してください。
""", verbose=True)
 
# コストレポート確認
print(client.get_cost_report())

思考プロセスの可視化とデバッグ

思考プロセスをAPIレスポンスに含めて取得することで、モデルがどのように推論しているかを確認できます。これはデバッグや品質検証に非常に有効です。

import google.generativeai as genai
from google.generativeai.types import content_types
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
model = genai.GenerativeModel("gemini-2.5-pro-preview-05-06")
 
# ストリーミングで思考プロセスを取得
response_stream = model.generate_content(
    "2025年のAI業界のトレンドを分析し、2026年の予測を述べてください。",
    generation_config={
        "thinkingConfig": {
            "thinkingBudget": 8192,
            "includeThoughts": True,  # 思考プロセスをレスポンスに含める
        }
    },
    stream=True
)
 
print("=== 思考プロセス ===")
full_thinking = ""
full_response = ""
 
for chunk in response_stream:
    if chunk.candidates:
        for part in chunk.candidates[0].content.parts:
            if hasattr(part, 'thought') and part.thought:
                # 思考部分
                full_thinking += part.text
                print(f"💭 {part.text}", end="", flush=True)
            else:
                # 通常の回答部分
                full_response += part.text
 
print("\n\n=== 最終回答 ===")
print(full_response)
 
print(f"\n📊 思考量: {len(full_thinking)} 文字")
print(f"📊 回答量: {len(full_response)} 文字")

本番環境でのモニタリング戦略

思考トークン使用量のロギング

本番環境では、思考トークンの使用量を継続的にモニタリングし、コスト異常を早期検知する点が肝心です。

import logging
from datetime import datetime
import json
 
# 構造化ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(message)s',  # JSON形式で出力するため
)
logger = logging.getLogger("gemini_thinking")
 
def log_thinking_usage(
    request_id: str,
    prompt_preview: str,
    usage_metadata,
    model_name: str,
    thinking_budget: int,
    duration_ms: float
):
    """思考トークン使用量を構造化ログに記録"""
    if not usage_metadata:
        return
 
    thinking_tokens = getattr(usage_metadata, 'thoughts_token_count', 0) or 0
 
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "request_id": request_id,
        "event": "gemini_thinking_usage",
        "model": model_name,
        "thinking_budget": thinking_budget,
        "thinking_tokens_used": thinking_tokens,
        "input_tokens": getattr(usage_metadata, 'prompt_token_count', 0) or 0,
        "output_tokens": getattr(usage_metadata, 'candidates_token_count', 0) or 0,
        "budget_utilization_pct": round(
            thinking_tokens / thinking_budget * 100, 1
        ) if thinking_budget > 0 else 0,
        "duration_ms": duration_ms,
        "prompt_preview": prompt_preview[:100],  # 先頭100文字のみ
    }
 
    # 予算の80%超過時はWARNINGレベルで記録
    if log_entry["budget_utilization_pct"] > 80:
        logger.warning(json.dumps(log_entry, ensure_ascii=False))
    else:
        logger.info(json.dumps(log_entry, ensure_ascii=False))
 
    return log_entry
 
# 使用例（FastAPI + ミドルウェアパターン）
import time
import uuid
 
async def gemini_with_monitoring(prompt: str, budget: int) -> str:
    """モニタリング付きGemini API呼び出し"""
    request_id = str(uuid.uuid4())
    start_time = time.time()
 
    model = genai.GenerativeModel(
        model_name="gemini-2.5-pro-preview-05-06",
        generation_config={
            "thinkingConfig": {"thinkingBudget": budget}
        }
    )
 
    response = model.generate_content(prompt)
 
    duration_ms = (time.time() - start_time) * 1000
    log_thinking_usage(
        request_id=request_id,
        prompt_preview=prompt[:100],
        usage_metadata=response.usage_metadata,
        model_name="gemini-2.5-pro-preview-05-06",
        thinking_budget=budget,
        duration_ms=duration_ms
    )
 
    return response.text

Budget使用率の分析と最適化サイクル

本番運用では、以下のサイクルで継続的にBudgetを最適化します。

収集フェーズ: 各リクエストの budget_utilization_pct を2週間記録
分析フェーズ: タスク種別ごとの平均使用率を集計
調整フェーズ: 平均使用率が50%未満のタスクはBudgetを半減
検証フェーズ: 精度の低下がないか品質チェック
固定フェーズ: 安定した設定値を本番コードに反映

Thinking Budget のA/Bテスト設計

Thinking Budget の最適化を最も厳密に行うのが A/B テストです。「このBudgetで十分だろう」という直感に頼るのではなく、実際のトラフィックでデータを収集し、コストと品質のトレードオフを定量化できます。

A/Bテストの設計方針

基本原則は「Budget変数を孤立させる」ことです。モデル・プロンプト・評価基準を固定したまま、Budgetの値だけを変えてテストします。

import hashlib
from dataclasses import dataclass
 
@dataclass
class ABTestVariant:
    name: str
    thinking_budget: int
    model_name: str
    weight: float  # トラフィック割り当て比率
 
# 実験バリアントの定義
EXPERIMENT_VARIANTS = [
    ABTestVariant("control", 8192, "gemini-2.5-pro-preview-05-06", 0.5),
    ABTestVariant("treatment_low", 4096, "gemini-2.5-pro-preview-05-06", 0.25),
    ABTestVariant("treatment_high", 16384, "gemini-2.5-pro-preview-05-06", 0.25),
]
 
def assign_variant(user_id: str, experiment_id: str) -> ABTestVariant:
    """
    決定論的なバリアント割り当て（同じユーザーは常に同じバリアントに）
    ハッシュで一貫性を保証
    """
    hash_input = f"{user_id}:{experiment_id}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
    normalized = (hash_value % 10000) / 10000.0  # 0.0〜1.0
 
    cumulative = 0.0
    for variant in EXPERIMENT_VARIANTS:
        cumulative += variant.weight
        if normalized < cumulative:
            return variant
 
    return EXPERIMENT_VARIANTS[-1]  # フォールバック
 
def run_with_ab_test(prompt: str, user_id: str, experiment_id: str = "thinking_budget_v1"):
    """A/Bテストバリアント割り当て付きリクエスト実行"""
    variant = assign_variant(user_id, experiment_id)
 
    model = genai.GenerativeModel(
        model_name=variant.model_name,
        generation_config={
            "thinkingConfig": {"thinkingBudget": variant.thinking_budget}
        }
    )
 
    response = model.generate_content(prompt)
 
    metadata = {
        "experiment_id": experiment_id,
        "variant": variant.name,
        "thinking_budget": variant.thinking_budget,
        "user_id": user_id,
    }
 
    if hasattr(response, 'usage_metadata') and response.usage_metadata:
        usage = response.usage_metadata
        metadata.update({
            "thinking_tokens": getattr(usage, 'thoughts_token_count', 0) or 0,
            "output_tokens": getattr(usage, 'candidates_token_count', 0) or 0,
        })
 
    return response.text, metadata