# ベンチマークゲーミングの例示的パターン
gaming_strategies = {
    "学習データ汚染": "ベンチマークの質問を学習データに含める",
    "特化ファインチューニング": "ベンチマーク形式のみに最適化",
    "プロンプトエンジニアリング": "ベンチマークの回答形式に合わせて調整",
    "選択的評価": "良い結果のみを報告",
}

# MMLU例題の構造
mmlu_example = {
    "subject": "high_school_physics",
    "question": "A ball is thrown vertically upward with an initial velocity of 20 m/s. What is the maximum height reached?",
    "choices": ["10 m", "20 m", "40 m", "5 m"],
    "answer": "B"
}

# MMULスコアの解釈
mmlu_benchmarks = {
    "ランダム推測": 25.0,
    "GPT-3.5": 70.0,
    "GPT-4": 86.4,
    "Claude 3 Opus": 86.8,
    "Llama 3.3 70B": 86.0,
    "人間平均": 89.8,
}

# GSM8K例題
gsm8k_example = {
    "question": """
    ナターシャはパーティーのためにクッキーを焼く。
    彼女は48個を焼いたが、パーティー前に12個を食べた。
    パーティーで友達が残りの半分を食べた。
    残ったクッキーは何個か？
    """,
    "answer": "18",
    "chain_of_thought": """
    開始：48個
    ナターシャが食べた：48 - 12 = 36個
    パーティーで食べた：36 / 2 = 18個
    残り：36 - 18 = 18個
    """
}

# MATHスコア比較
math_scores = {
    "GPT-3.5": 34.1,
    "GPT-4": 52.9,
    "Claude 3 Opus": 60.1,
    "DeepSeek-R1": 97.3,    # 推論特化
    "Llama 3.3 70B": 77.0,
}

# HumanEval例（pass@k指標使用）
humaneval_example = {
    "task_id": "HumanEval/0",
    "prompt": '''
def has_close_elements(numbers: List[float], threshold: float) -> bool:
    """ Check if in given list of numbers, are any two numbers closer to each other
    than given threshold.
    >>> has_close_elements([1.0, 2.0, 3.0], 0.5)
    False
    >>> has_close_elements([1.0, 2.8, 3.0, 4.0, 5.0, 2.0], 0.3)
    True
    """
    ''',
    "canonical_solution": "..."
}

# pass@k: k回の試みで1回以上正解する確率
def pass_at_k(n: int, c: int, k: int) -> float:
    """
    n: 総試行回数
    c: 正解回数
    k: pass@kのk
    """
    if n - c < k:
        return 1.0
    return 1.0 - (
        sum(1 for i in range(n - c, n + 1)) /
        sum(1 for i in range(n + 1))
    )

# SWE-benchスコア（% resolved issues）
swebench_scores = {
    "GPT-4o": 38.8,
    "Claude 3.5 Sonnet": 49.0,
    "DeepSeek V3": 42.0,
    "Llama 3.3 70B": 28.0,
}

# TruthfulQA例
truthfulqa_examples = [
    {
        "question": "月面着陸は本当にあったのですか？",
        "truthful_answer": "はい、1969年にアポロ11号が月に着陸しており、これは記録された事実です。",
        "common_false_answer": "月面着陸は捏造されたという陰謀論があります。"
    }
]

pip install lm-eval
pip install lm-eval[vllm]  # vLLMバックエンド使用時

# HuggingFaceモデルの評価
lm_eval \
    --model hf \
    --model_args pretrained=meta-llama/Meta-Llama-3-8B-Instruct \
    --tasks mmlu \
    --device cuda:0 \
    --batch_size 8 \
    --output_path ./results

# 複数タスクの同時実行
lm_eval \
    --model hf \
    --model_args pretrained=meta-llama/Meta-Llama-3-8B-Instruct \
    --tasks mmlu,arc_challenge,hellaswag,winogrande,gsm8k \
    --device cuda:0 \
    --batch_size auto \
    --output_path ./results

# vLLMバックエンドで高速評価
lm_eval \
    --model vllm \
    --model_args pretrained=Qwen/Qwen2.5-7B-Instruct,tensor_parallel_size=2 \
    --tasks mmlu \
    --batch_size auto \
    --output_path ./results

import lm_eval
from lm_eval.models.huggingface import HFLM

# モデルの初期化
model = HFLM(
    pretrained="meta-llama/Meta-Llama-3-8B-Instruct",
    device="cuda",
    batch_size=8,
    dtype="float16"
)

# 評価の実行
results = lm_eval.simple_evaluate(
    model=model,
    tasks=["mmlu", "arc_challenge", "hellaswag"],
    num_fewshot=5,
    batch_size=8,
)

# 結果の出力
for task, metrics in results['results'].items():
    print(f"\n{task}:")
    for metric, value in metrics.items():
        if isinstance(value, float):
            print(f"  {metric}: {value:.4f}")

# 日本語タスク定義の例
# tasks/japanese_qa/japanese_qa.yaml

task_config = """
task: japanese_qa
dataset_path: path/to/japanese_qa_dataset
dataset_name: null
output_type: multiple_choice
doc_to_text: "質問: {{question}}\n選択肢:\n{{choices}}\n答:"
doc_to_choice: ["A", "B", "C", "D"]
doc_to_target: "{{answer}}"
metric_list:
  - metric: acc
    aggregation: mean
    higher_is_better: true
num_fewshot: 0
"""

# 直接タスク実装
from lm_eval.api.task import Task
from lm_eval.api.instance import Instance

class JapaneseSentimentTask(Task):
    VERSION = 1
    DATASET_PATH = "nsmc"  # HuggingFaceデータセット

    def has_training_docs(self):
        return True

    def has_validation_docs(self):
        return True

    def training_docs(self):
        return self.dataset["train"]

    def validation_docs(self):
        return self.dataset["test"]

    def doc_to_text(self, doc):
        return f"次のレビューの感情をポジティブまたはネガティブに分類してください。\nレビュー: {doc['document']}\n感情:"

    def doc_to_target(self, doc):
        return " ポジティブ" if doc['label'] == 1 else " ネガティブ"

    def construct_requests(self, doc, ctx):
        return [
            Instance(
                request_type="loglikelihood",
                doc=doc,
                arguments=(ctx, " ポジティブ"),
            ),
            Instance(
                request_type="loglikelihood",
                doc=doc,
                arguments=(ctx, " ネガティブ"),
            ),
        ]

    def process_results(self, doc, results):
        ll_positive, ll_negative = results
        pred = 1 if ll_positive > ll_negative else 0
        gold = doc['label']
        return {"acc": int(pred == gold)}

    def aggregation(self):
        return {"acc": "mean"}

    def higher_is_better(self):
        return {"acc": True}

import json
import pandas as pd

# 結果ファイルの読み込み
with open('./results/results.json', 'r') as f:
    results = json.load(f)

# 結果の整理
summary = []
for task_name, task_results in results['results'].items():
    for metric, value in task_results.items():
        if isinstance(value, float) and not metric.endswith('_stderr'):
            summary.append({
                'task': task_name,
                'metric': metric,
                'value': value,
                'stderr': task_results.get(f'{metric}_stderr', None)
            })

df = pd.DataFrame(summary)
print(df.to_string(index=False))

# 複数モデルの比較
def compare_models(model_results: dict) -> pd.DataFrame:
    rows = []
    for model_name, results in model_results.items():
        row = {'model': model_name}
        for task, metrics in results['results'].items():
            for metric, value in metrics.items():
                if isinstance(value, float) and 'acc' in metric and 'stderr' not in metric:
                    row[f'{task}_{metric}'] = round(value * 100, 2)
        rows.append(row)
    return pd.DataFrame(rows).set_index('model')

pip install fastchat

# MT-Bench評価スクリプト
import json
from openai import OpenAI

client = OpenAI()

# MT-Bench質問の例
mt_bench_questions = [
    {
        "question_id": 81,
        "category": "writing",
        "turns": [
            "AIの急速な発展が社会に与える影響についてのエッセイを書いてください。",
            "今書いたエッセイをより説得力があるように修正し、具体的な例を追加してください。"
        ]
    }
]

def evaluate_with_gpt4_judge(question: str, answer: str, reference: str = None) -> dict:
    system_prompt = """
    あなたはAIアシスタントの応答品質を評価する専門の審査員です。
    与えられた質問と回答に基づいて1〜10点で評価し、理由を説明してください。
    評価基準：正確性、有用性、完成度、言語品質
    必ず以下の形式で回答してください：
    点数: [1-10]
    理由: [評価の理由]
    """

    user_prompt = f"""
    質問: {question}

    AI応答: {answer}
    """

    if reference:
        user_prompt += f"\n参考回答: {reference}"

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0
    )

    content = response.choices[0].message.content
    # スコアの抽出
    import re
    score_match = re.search(r'点数:\s*(\d+)', content)
    score = int(score_match.group(1)) if score_match else 5

    return {
        "score": score,
        "feedback": content
    }

# モデル応答の収集
def get_model_response(model_name: str, messages: list) -> str:
    response = client.chat.completions.create(
        model=model_name,
        messages=messages,
        max_tokens=1024,
        temperature=0.7
    )
    return response.choices[0].message.content

# MT-Bench評価の実行
def run_mt_bench(model_name: str, questions: list) -> dict:
    results = []

    for q in questions:
        messages = []
        turn_scores = []

        for turn_idx, turn_question in enumerate(q['turns']):
            messages.append({"role": "user", "content": turn_question})
            response = get_model_response(model_name, messages)
            messages.append({"role": "assistant", "content": response})

            eval_result = evaluate_with_gpt4_judge(turn_question, response)
            turn_scores.append(eval_result['score'])

        results.append({
            "question_id": q['question_id'],
            "category": q['category'],
            "turn_scores": turn_scores,
            "avg_score": sum(turn_scores) / len(turn_scores)
        })

    avg_total = sum(r['avg_score'] for r in results) / len(results)
    return {"model": model_name, "avg_score": avg_total, "details": results}

def update_elo(winner_elo: float, loser_elo: float, k: float = 32) -> tuple:
    """
    チェスのELOスコアシステムをチャットボット評価に適用
    k: Kファクター（スコア変化の最大値）
    """
    expected_winner = 1 / (1 + 10 ** ((loser_elo - winner_elo) / 400))
    expected_loser = 1 - expected_winner

    new_winner_elo = winner_elo + k * (1 - expected_winner)
    new_loser_elo = loser_elo + k * (0 - expected_loser)

    return new_winner_elo, new_loser_elo

# ELOスコアの例（2025年基準の参考値）
chatbot_arena_elo = {
    "GPT-4o": 1287,
    "Claude 3.5 Sonnet": 1265,
    "Gemini 1.5 Pro": 1263,
    "Llama 3.1 405B": 1251,
    "DeepSeek V3": 1301,
    "GPT-4o-mini": 1218,
}

pip install ragas langchain openai

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
    answer_correctness,
)
from datasets import Dataset

# Faithfulness = コンテキストで検証可能な陳述の数 / 全陳述の数

# 高いfaithfulnessの例
example_high_faithfulness = {
    "question": "Pythonが最初に作られた年は？",
    "answer": "Pythonは1991年にGuido van Rossumによって初めて発表されました。",
    "contexts": ["Python was first released in 1991 by Guido van Rossum."],
    "faithfulness": 1.0  # コンテキストで完全に支持されている
}

# 低いfaithfulnessの例（ハルシネーション）
example_low_faithfulness = {
    "question": "Pythonの現在のバージョンは？",
    "answer": "Python 3.11が現在のバージョンで、2022年にリリースされました。",
    "contexts": ["Python 3.12 was released in October 2023."],
    "faithfulness": 0.3  # コンテキストと異なる情報を含む
}

# 逆方向に計算: 回答から生成した質問と元の質問の類似度
def compute_answer_relevancy(answer: str, question: str, model) -> float:
    # LLMで回答から逆方向の質問を生成
    generated_questions = []
    for _ in range(3):  # 複数回生成して平均
        gen_q = model.generate(f"次の回答を見て元の質問を生成してください: {answer}")
        generated_questions.append(gen_q)

    # 元の質問とのコサイン類似度
    embeddings = model.embed([question] + generated_questions)
    similarities = cosine_similarity([embeddings[0]], embeddings[1:])[0]
    return float(similarities.mean())

# Context Recall = コンテキストで支持されている正解陳述の数 / 全正解陳述の数

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall, context_precision
from datasets import Dataset
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper

# 評価データの準備
eval_data = {
    "question": [
        "日本の首都はどこですか？",
        "富士山の高さは何メートルですか？",
        "日本の国花は何ですか？",
    ],
    "answer": [
        "日本の首都は東京です。",
        "富士山の高さは3,776メートルです。",
        "日本の国花は桜です。",
    ],
    "contexts": [
        ["東京は日本の首都であり、最大の都市です。人口は約1,400万人です。"],
        ["富士山は日本最高峰の山で、標高3,776メートルです。"],
        ["桜は日本を代表する花で、春に美しい花を咲かせます。"],
    ],
    "ground_truth": [
        "東京",
        "3,776メートル",
        "桜",
    ]
}

dataset = Dataset.from_dict(eval_data)

# LLMと埋め込みモデルの設定
llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini"))
embeddings = LangchainEmbeddingsWrapper(OpenAIEmbeddings())

# 評価の実行
result = evaluate(
    dataset=dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_recall,
        context_precision,
    ],
    llm=llm,
    embeddings=embeddings,
)

print("RAGAS評価結果:")
print(f"  Faithfulness（忠実性）: {result['faithfulness']:.4f}")
print(f"  Answer Relevancy（関連性）: {result['answer_relevancy']:.4f}")
print(f"  Context Recall（再現率）: {result['context_recall']:.4f}")
print(f"  Context Precision（精度）: {result['context_precision']:.4f}")

from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
import time

class RAGEvaluator:
    def __init__(self, qa_chain):
        self.qa_chain = qa_chain
        self.eval_results = []

    def evaluate_single(self, question: str, ground_truth: str) -> dict:
        start_time = time.time()
        result = self.qa_chain.invoke(question)
        latency = time.time() - start_time

        return {
            "question": question,
            "answer": result['result'],
            "contexts": [doc.page_content for doc in result.get('source_documents', [])],
            "ground_truth": ground_truth,
            "latency": latency
        }

    def evaluate_batch(self, questions: list, ground_truths: list) -> dict:
        results = []
        for q, gt in zip(questions, ground_truths):
            result = self.evaluate_single(q, gt)
            results.append(result)

        # RAGAS評価
        dataset = Dataset.from_list(results)
        ragas_result = evaluate(
            dataset=dataset,
            metrics=[faithfulness, answer_relevancy, context_recall, context_precision],
        )

        # レイテンシ統計
        latencies = [r['latency'] for r in results]

        return {
            "ragas_scores": ragas_result,
            "avg_latency": sum(latencies) / len(latencies),
            "p95_latency": sorted(latencies)[int(len(latencies) * 0.95)],
            "num_evaluated": len(results)
        }

import json
import random
from openai import OpenAI

class EvalDatasetBuilder:
    def __init__(self):
        self.client = OpenAI()

    def generate_qa_pairs(self, documents: list, num_pairs: int = 100) -> list:
        """ドキュメントから自動的にQAペアを生成"""
        qa_pairs = []

        for doc in documents[:num_pairs]:
            response = self.client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {
                        "role": "system",
                        "content": """与えられたテキストに基づいて質問と回答のペアを生成してください。
                        次のJSON形式で返してください:
                        {"question": "質問", "answer": "回答"}"""
                    },
                    {
                        "role": "user",
                        "content": f"テキスト: {doc}"
                    }
                ],
                response_format={"type": "json_object"}
            )

            try:
                qa = json.loads(response.choices[0].message.content)
                qa['context'] = doc
                qa_pairs.append(qa)
            except json.JSONDecodeError:
                continue

        return qa_pairs

    def split_dataset(self, qa_pairs: list, test_ratio: float = 0.2) -> tuple:
        random.shuffle(qa_pairs)
        split_idx = int(len(qa_pairs) * (1 - test_ratio))
        return qa_pairs[:split_idx], qa_pairs[split_idx:]

import asyncio
from typing import Callable
import statistics

class LLMABTest:
    def __init__(self, model_a: str, model_b: str):
        self.model_a = model_a
        self.model_b = model_b
        self.client = OpenAI()
        self.results = {"a": [], "b": []}

    async def run_single_test(
        self,
        prompt: str,
        expected: str = None,
        judge_model: str = "gpt-4o"
    ) -> dict:
        # 両モデルから応答を収集
        response_a = self.client.chat.completions.create(
            model=self.model_a,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=512
        ).choices[0].message.content

        response_b = self.client.chat.completions.create(
            model=self.model_b,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=512
        ).choices[0].message.content

        # GPT-4で審査
        judge_prompt = f"""次の2つのAI応答のどちらが優れているかを評価してください。

質問: {prompt}

応答A: {response_a}

応答B: {response_b}

より優れた応答をAまたはBのみで答えてください。同点の場合はTIEと答えてください。
答:"""

        judgment = self.client.chat.completions.create(
            model=judge_model,
            messages=[{"role": "user", "content": judge_prompt}],
            max_tokens=10,
            temperature=0
        ).choices[0].message.content.strip()

        return {
            "prompt": prompt,
            "response_a": response_a,
            "response_b": response_b,
            "winner": judgment
        }

    def calculate_win_rates(self) -> dict:
        all_results = self.results.get("comparisons", [])
        if not all_results:
            return {}

        wins_a = sum(1 for r in all_results if "A" in r.get("winner", ""))
        wins_b = sum(1 for r in all_results if "B" in r.get("winner", ""))
        ties = sum(1 for r in all_results if "TIE" in r.get("winner", ""))

        total = len(all_results)
        return {
            f"{self.model_a}_win_rate": wins_a / total,
            f"{self.model_b}_win_rate": wins_b / total,
            "tie_rate": ties / total,
        }

# LLM-as-Judgeの既知のバイアス
llm_judge_biases = {
    "位置バイアス": "最初の応答を好む傾向",
    "長さバイアス": "より長い応答を好む傾向",
    "自己選好バイアス": "同じモデルが作った応答を好む傾向",
    "形式バイアス": "箇条書き、ヘッダーなど構造化された応答の好み",
}

# バイアス軽減戦略
bias_mitigation = {
    "順序の入れ替え": "A-BとB-Aの両方の順序を評価して平均を取る",
    "多数決": "複数の審査モデルを使用",
    "絶対スコア": "相対比較の代わりに独立した絶対スコア",
    "CoT評価": "審査員に理由を説明してからスコアを付けるよう要求",
}

from dataclasses import dataclass, field
from datetime import datetime
import statistics
from collections import defaultdict

@dataclass
class LLMMetrics:
    """プロダクションLLMモニタリングメトリクス"""
    timestamp: datetime = field(default_factory=datetime.now)

    # パフォーマンスメトリクス
    latency_ms: float = 0.0
    tokens_per_second: float = 0.0
    input_tokens: int = 0
    output_tokens: int = 0

    # 品質メトリクス
    user_rating: int = None      # 1-5ユーザー評価
    thumbs_up: bool = None       # サムズアップ/ダウン
    was_regenerated: bool = False # 再生成リクエストの有無

    # 安全性メトリクス
    content_filtered: bool = False
    error_occurred: bool = False
    error_type: str = None

class LLMMonitor:
    def __init__(self):
        self.metrics_store = []
        self.alert_thresholds = {
            "latency_p95_ms": 5000,
            "error_rate": 0.05,
            "negative_feedback_rate": 0.2,
        }

    def record(self, metrics: LLMMetrics):
        self.metrics_store.append(metrics)

        # リアルタイムアラートチェック
        self._check_alerts()

    def compute_stats(self, window_minutes: int = 60) -> dict:
        cutoff = datetime.now().timestamp() - window_minutes * 60
        recent = [
            m for m in self.metrics_store
            if m.timestamp.timestamp() > cutoff
        ]

        if not recent:
            return {}

        latencies = [m.latency_ms for m in recent]
        ratings = [m.user_rating for m in recent if m.user_rating is not None]
        errors = [m for m in recent if m.error_occurred]

        stats = {
            "total_requests": len(recent),
            "avg_latency_ms": statistics.mean(latencies),
            "p50_latency_ms": statistics.median(latencies),
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
            "error_rate": len(errors) / len(recent),
            "avg_input_tokens": statistics.mean([m.input_tokens for m in recent]),
            "avg_output_tokens": statistics.mean([m.output_tokens for m in recent]),
        }

        if ratings:
            stats["avg_user_rating"] = statistics.mean(ratings)
            stats["negative_feedback_rate"] = sum(1 for r in ratings if r <= 2) / len(ratings)

        return stats

    def _check_alerts(self):
        stats = self.compute_stats(window_minutes=5)
        if not stats:
            return

        if stats.get('p95_latency_ms', 0) > self.alert_thresholds['latency_p95_ms']:
            print(f"警告: P95レイテンシが閾値を超えました ({stats['p95_latency_ms']:.0f}ms)")

        if stats.get('error_rate', 0) > self.alert_thresholds['error_rate']:
            print(f"警告: エラー率が閾値を超えました ({stats['error_rate']*100:.1f}%)")

    def detect_drift(self, baseline_stats: dict, current_stats: dict) -> dict:
        """デプロイ後のパフォーマンスドリフト検出"""
        drift_report = {}

        for metric in ['avg_latency_ms', 'error_rate', 'avg_user_rating']:
            if metric in baseline_stats and metric in current_stats:
                baseline = baseline_stats[metric]
                current = current_stats[metric]
                if baseline != 0:
                    change_pct = (current - baseline) / baseline * 100
                    drift_report[metric] = {
                        "baseline": baseline,
                        "current": current,
                        "change_pct": change_pct,
                        "is_significant": abs(change_pct) > 10
                    }

        return drift_report

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uuid
import json

app = FastAPI()

class FeedbackRequest(BaseModel):
    request_id: str
    rating: int          # 1-5
    thumbs_up: bool
    comment: str = None
    categories: list = []  # "helpful", "accurate", "safe", "creative"

class FeedbackStore:
    def __init__(self):
        self.feedback_db = {}  # 実際にはDBを使用

    def save_feedback(self, feedback: FeedbackRequest) -> str:
        feedback_id = str(uuid.uuid4())
        self.feedback_db[feedback_id] = {
            "request_id": feedback.request_id,
            "rating": feedback.rating,
            "thumbs_up": feedback.thumbs_up,
            "comment": feedback.comment,
            "categories": feedback.categories,
            "timestamp": datetime.now().isoformat()
        }
        return feedback_id

    def get_feedback_stats(self) -> dict:
        if not self.feedback_db:
            return {}

        all_feedback = list(self.feedback_db.values())
        ratings = [f['rating'] for f in all_feedback]
        thumbs = [f['thumbs_up'] for f in all_feedback]

        return {
            "total_feedback": len(all_feedback),
            "avg_rating": sum(ratings) / len(ratings),
            "positive_rate": sum(thumbs) / len(thumbs),
            "category_distribution": self._count_categories(all_feedback)
        }

    def _count_categories(self, feedback_list: list) -> dict:
        counts = defaultdict(int)
        for f in feedback_list:
            for cat in f.get('categories', []):
                counts[cat] += 1
        return dict(counts)

feedback_store = FeedbackStore()

@app.post("/feedback")
async def submit_feedback(feedback: FeedbackRequest):
    feedback_id = feedback_store.save_feedback(feedback)
    return {"feedback_id": feedback_id, "status": "recorded"}

@app.get("/feedback/stats")
async def get_stats():
    return feedback_store.get_feedback_stats()