🧪 AI 评估 — LLM-as-Judge、Benchmark 与 A/B 测试

同样的输入，不同模型、不同 prompt、不同温度，输出千差万别。你不能用 assertEquals 来验证一封邮件的"礼貌程度"。

改了 prompt 或换了模型，你怎么知道变好了还是变差了？没有评估，每次改动都是盲飞。团队里就会出现"我觉得好一点了"的主观争论。

GPT-4 比 GPT-4o-mini 贵 15 倍。如果你的应用 90% 的场景用便宜模型就够了，但你不评估，你就只敢用最贵的——因为"至少不会太差"。

计算 1-gram 到 4-gram 的精确率，取几何平均，乘以长度惩罚（防止短文本作弊）：

BLEU = BP × exp(Σ wn × log(pn))

# pn = n-gram 精确率
# wn = 权重（通常各 0.25）
# BP = brevity penalty = min(1, exp(1 - ref_len/gen_len))

# BLEU/ROUGE: 表面匹配
参考: "这部电影非常精彩"
生成: "此片十分出色"  → BLEU ≈ 0.0 (无 n-gram 重叠)

# BERTScore: 语义匹配
参考: "这部电影非常精彩"
生成: "此片十分出色"  → BERTScore ≈ 0.92 (语义高度相似)

有标准答案时的 LLM 评判——结合了传统指标的"有参考"优势和 LLM 的"语义理解"优势。

Prompt: "请对比以下回答与参考答案，评估其准确性。

问题: {question}
参考答案: {reference}
模型回答: {answer}

评估维度:
1. 事实准确性 (1-5): 是否与参考一致？
2. 完整性 (1-5): 是否覆盖参考的所有要点？
3. 额外信息 (1-5): 是否添加了有价值的信息？"

{
  "system": "你是一个专业的 AI 输出质量评估员。你必须严格按照评分标准评估，先写出评估理由，再给出分数。",

  "user": """请评估以下 AI 回答的质量。

## 问题
{question}

## AI 回答
{answer}

## 评估标准

### 1. 准确性 (accuracy): 1-5
- 5: 所有事实陈述正确，无幻觉
- 4: 主要事实正确，轻微不精确
- 3: 有 1-2 个事实错误或未经验证的声称
- 2: 多个事实错误
- 1: 严重幻觉或捏造

### 2. 完整性 (completeness): 1-5
- 5: 全面覆盖问题的所有方面
- 3: 覆盖了主要方面但遗漏了重要的细节
- 1: 只回答了问题的一小部分

### 3. 指令遵循 (instruction_following): 1-5
- 5: 完全遵守所有格式、长度、风格要求
- 3: 大致遵守但有偏差
- 1: 忽视了明确的指令

## 输出格式（JSON）
```json
{
  "accuracy": {score},
  "completeness": {score},
  "instruction_following": {score},
  "reasoning": "你的评估理由...",
  "overall": {weighted_score}
}
```"""
}

• 用 Benchmark 做模型初筛——从 50 个模型缩到 3 个候选
• 选与你任务最相关的 Benchmark（代码 → HumanEval/SWE-bench，对话 → MT-Bench）
• 关注细分分数而非总分（MMLU 里的编程/数学/历史分开看）
• Benchmark 分数只做参考下限——Benchmark 好不一定产品好
• 定期检查数据泄露——训练集是否包含测试集

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

# 准备评估数据
data = {
    "question": ["什么是向量数据库？", "RAG 的优势是什么？"],
    "answer": ["向量数据库是专门存储和检索向量嵌入的数据库...", "RAG 可以减少幻觉、提供可追溯来源..."],
    "contexts": [
        ["向量数据库如 Pinecone 和 Weaviate 专门为高维向量优化..."],
        ["RAG 通过检索相关文档来增强生成，降低幻觉率..."],
    ],
    "ground_truth": ["向量数据库是存储向量嵌入的专用数据库，支持相似性搜索", "RAG 结合检索和生成，减少幻觉并提供引用"],
}

dataset = Dataset.from_dict(data)

# 运行评估
result = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=your_llm,          # 用于评估的 LLM
    embeddings=your_embeds, # 用于 answer_relevancy
)

print(result)
# {'faithfulness': 0.85, 'answer_relevancy': 0.92, 
#  'context_precision': 0.78, 'context_recall': 0.70}

# 理想轨迹 (4 步完成)
用户: "查一下北京明天的天气并发邮件给 alice@example.com"
1. 💭 思考: 需要先查天气，再发邮件
2. 🔧 调用: get_weather(city="北京", date="明天") → 25°C 晴
3. 🔧 调用: send_email(to="alice@example.com", subject="北京天气", body="...")
4. ✅ 完成

# 低效轨迹 (7 步，2 次回退)
用户: "查一下北京明天的天气并发邮件给 alice@example.com"
1. 💭 思考: 我要发邮件
2. 🔧 调用: send_email(to="alice@example.com", ...)  ← ❌ 还没查天气！
3. 💭 思考: 我需要天气信息
4. 🔧 调用: get_weather(city="Beijing")  ← ❌ 拼写错误
5. 💭 思考: 中文城市名
6. 🔧 调用: get_weather(city="北京") → 25°C 晴
7. 🔧 调用: send_email(...) ← 重新发

# 轨迹评分
理想轨迹: Step Efficiency = 4/4 = 1.0, Tool Accuracy = 2/2 = 100%
低效轨迹: Step Efficiency = 4/7 = 0.57, Tool Accuracy = 2/4 = 50%

class AgentEvalCase:
    """单个评估用例"""
    task: str                    # 用户任务描述
    environment: dict            # 初始环境状态（文件、数据库、API mock）
    expected_outcome: str        # 期望结果描述
    success_criteria: Callable   # 验证函数（检查文件内容、API 调用等）
    max_steps: int               # 最大允许步数
    timeout_seconds: int         # 超时限制

# 示例：代码修复 Agent 评估
case = AgentEvalCase(
    task="修复 utils.py 中的除零错误",
    environment={"files": {"utils.py": "def divide(a, b): return a / b"}},
    expected_outcome="添加 b == 0 的检查并返回 None 或抛出异常",
    success_criteria=lambda env: (
        "b ==" in env["files"]["utils.py"] or 
        "b ==" in env["files"]["utils.py"]
    ) and "ZeroDivisionError" not in env.get("test_output", ""),
    max_steps=10,
    timeout_seconds=120,
)

# 运行评估
results = []
for case in eval_set:
    trajectory = agent.run(case.task, env=case.environment)
    success = case.success_criteria(trajectory.final_state)
    efficiency = case.max_steps / len(trajectory.steps)
    results.append({
        "task": case.task,
        "success": success,
        "steps": len(trajectory.steps),
        "efficiency": efficiency,
        "tool_calls": [s for s in trajectory.steps if s.type == "tool"],
        "errors": [s for s in trajectory.steps if s.type == "error"],
    })

# 1. 定义假设
"Hypothesis: 使用 GPT-4o 替代 GPT-4，在保持质量不变的前提下，
成本降低 60%，延迟减少 40%"

# 2. 选择指标
primary_metric = "user_satisfaction"    # 主要：用户满意度
guardrail_metrics = [
    "hallucination_rate",               # 护栏：幻觉率不恶化
    "task_success_rate",                # 护栏：任务完成率不下降
    "p95_latency",                      # 护rail：延迟不恶化
]

# 3. 计算样本量
# AI 系统方差大 → 需要更大样本
# 经验法则：至少 1000 次交互/组
# 如果满意度提升 5% → ~2000/组 (α=0.05, β=0.2)

# 4. 分流策略
# ⚠️ 关键：同一个用户的对话必须始终在同一组
# 否则用户会感到体验不一致（忽好忽坏）
split_strategy = "user_id_hash"  # 按用户 ID 哈希分流

# 5. 运行周期
min_days = 7  # 至少一周，覆盖工作日+周末
max_days = 21 # 最长三周，避免历史效应

import scipy.stats as stats
import numpy as np

# AI 系统的特殊考虑：
# 1. 同一用户多次交互不独立 → 用用户级聚合
# 2. 输出随机性 → 需要更多样本
# 3. 多个指标同时看 → 需要多重检验校正

def ab_test_significance(control_scores, treatment_scores, alpha=0.05):
    """
    AI 系统的 A/B 显著性检验
    control_scores: 对照组满意度列表 (per-user 平均)
    treatment_scores: 实验组满意度列表
    """
    # Welch's t-test (不假设等方差)
    t_stat, p_value = stats.ttest_ind(
        control_scores, treatment_scores, 
        equal_var=False
    )
    
    # 效应量 (Cohen's d)
    pooled_std = np.sqrt(
        (np.var(control_scores) + np.var(treatment_scores)) / 2
    )
    cohens_d = (np.mean(treatment_scores) - np.mean(control_scores)) / pooled_std
    
    # 置信区间
    diff = np.mean(treatment_scores) - np.mean(control_scores)
    se = np.sqrt(np.var(control_scores)/len(control_scores) + 
                 np.var(treatment_scores)/len(treatment_scores))
    ci = (diff - 1.96*se, diff + 1.96*se)
    
    return {
        "significant": p_value < alpha,
        "p_value": p_value,
        "effect_size": cohens_d,
        "confidence_interval": ci,
        "recommendation": "Ship" if (p_value < alpha and cohens_d > 0.2) else "Hold"
    }

# 多重检验校正 (Bonferroni)
def bonferroni_correction(p_values, alpha=0.05):
    """当同时看 N 个指标时，提高显著性阈值"""
    n_tests = len(p_values)
    adjusted_alpha = alpha / n_tests
    return [p < adjusted_alpha for p in p_values]

• 覆盖分布：你的真实用户查询分布是什么？评估集应匹配这个分布，而非全是简单/困难案例
• 分层采样：按任务类型、难度、领域分层，确保每层都有代表性
• 持续更新：从生产日志中采样新 case，防止分布漂移
• 金标答案：至少 20% 的 case 需要人工标注的参考答案
• 对抗样本：包含故意设计的边界 case（诱导幻觉、注入攻击、模糊指令）
• 最小规模：起步 100-200 条，成熟后 1000+ 条

import random
from datetime import datetime, timedelta

def build_eval_set_from_logs(production_logs, target_size=200):
    """
    从生产日志中采样构建评估集
    """
    # 1. 按意图分类
    intent_counts = {}
    for log in production_logs:
        intent = classify_intent(log["user_query"])
        intent_counts[intent] = intent_counts.get(intent, 0) + 1
    
    # 2. 分层采样
    eval_set = []
    for intent, count in intent_counts.items():
        proportion = count / len(production_logs)
        n_samples = max(5, int(target_size * proportion))  # 每个意图至少 5 条
        intent_logs = [l for l in production_logs if classify_intent(l["user_query"]) == intent]
        eval_set.extend(random.sample(intent_logs, min(n_samples, len(intent_logs))))
    
    # 3. 补充边界 case
    edge_cases = [
        l for l in production_logs
        if l.get("user_feedback") == "negative"  # 用户不满意的
        or len(l.get("model_output", "")) < 20    # 太短的
        or "error" in l.get("model_output", "").lower()  # 出错的
    ]
    eval_set.extend(random.sample(edge_cases, min(20, len(edge_cases))))
    
    # 4. 去重
    seen = set()
    unique = []
    for item in eval_set:
        key = item["user_query"][:100]
        if key not in seen:
            seen.add(key)
            unique.append(item)
    
    return unique[:target_size]

# .github/workflows/eval.yml
name: AI Evaluation

on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'config/model*.yaml'

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Run eval suite
        run: |
          python scripts/run_eval.py \
            --eval-set data/eval-set-v3.jsonl \
            --config config/production.yaml \
            --compare main \
            --metrics accuracy,faithfulness,relevance \
            --judge-model gpt-4o \
            --output results/pr-${{ github.event.number }}.json
      
      - name: Check for regression
        run: |
          python scripts/check_regression.py \
            --current results/pr-${{ github.event.number }}.json \
            --baseline results/main-latest.json \
            --threshold -0.05 \
            --fail-on-regression
      
      - name: Comment on PR
        uses: actions/github-script@v7
        with:
          script: |
            const results = require('./results/pr-${{ github.event.number }}.json');
            github.rest.issues.createComment({
              ...context.repo,
              issue_number: context.issue.number,
              body: `## 📊 Eval Results\n${formatResults(results)}`
            })

用传统指标评估开放式生成任务。你的客服机器人回答"很抱歉给您带来不便"和"抱歉造成了困扰"，BLEU 给 0 分，但语义完全等价。

正确做法：开放式任务用 LLM-as-Judge，有 ground truth 的任务才用传统指标。

设了一个 Judge，跑了 1000 条，发现准确率 95%，就直接上线了。但你没检查——Judge 本身就有 ~10% 的错误率，而且对你的特定 domain 可能更差。

正确做法：Judge 必须与人工标注做校准（至少 50 条），计算 Cohen's Kappa ≥ 0.6 才可用。

半年前建的评估集，到现在还在用。但你的用户 query 分布已经变了——新功能上线后出现了全新的问题类型。

正确做法：每月从生产日志中采样补充评估集，保持与真实分布对齐。

准确率从 85% 降到 82%，你只看数字觉得"差不多"。但仔细看 case 发现，降分全是高价值用户的关键场景。

正确做法：分数回归时必须做 case study——看变差的具体 case，判断影响范围和严重程度。

跑了 2 天 A/B 就做决策。但 AI 系统方差大，200 条样本根本不够得出显著结论。

正确做法：先算所需样本量（通常 1000-5000 次/组），跑到足够才做决策。宁可多跑几天，不要误判。

只评估质量，不看成本。GPT-4 比 GPT-4o 质量高 5%，但贵 15 倍。如果你的场景 95% 的 case 用 GPT-4o 就够了，那你浪费了 93% 的成本。

正确做法：评估时加入"成本效率"维度——质量 / 成本。引入路由策略：简单 case 用便宜模型，困难 case 用贵模型（详见 LLM 集成）。

"""
完整的 LLM-as-Judge 评估流水线
支持 Pointwise + Pairwise + 多维度评估
"""
import json
import random
from dataclasses import dataclass, field
from typing import Optional
from openai import OpenAI

client = OpenAI()

@dataclass
class EvalCase:
    """评估用例"""
    id: str
    question: str
    answer: str
    reference: Optional[str] = None  # 可选参考答案
    category: Optional[str] = None   # 分类标签

@dataclass
class JudgeResult:
    """评判结果"""
    case_id: str
    accuracy: int        # 1-5
    completeness: int    # 1-5
    relevance: int       # 1-5
    faithfulness: int    # 1-5 (需要 reference 或 contexts)
    reasoning: str
    overall: float       # 加权总分
    judge_model: str

JUDGE_SYSTEM = """你是一个专业的 AI 输出质量评估员。
你必须严格按照评分标准评估，先写出评估理由，再给出分数。
以 JSON 格式输出。"""

JUDGE_PROMPT = """请评估以下 AI 回答的质量。

## 问题
{question}

## AI 回答
{answer}

{reference_section}

## 评估维度与标准

### 1. 准确性 (accuracy): 1-5
5 = 所有事实正确，无幻觉
4 = 主要正确，轻微不精确
3 = 1-2 个事实错误
2 = 多个事实错误
1 = 严重幻觉/捏造

### 2. 完整性 (completeness): 1-5
5 = 全面覆盖所有方面
4 = 覆盖主要方面，轻微遗漏
3 = 覆盖主要方面但遗漏重要细节
2 = 只覆盖部分
1 = 严重不完整

### 3. 相关性 (relevance): 1-5
5 = 完全针对问题
4 = 大部分相关
3 = 有一些偏题内容
2 = 明显偏题
1 = 完全不相关

### 4. 忠实度 (faithfulness): 1-5
5 = 所有内容都有依据支持
4 = 大部分有依据
3 = 有 1-2 个无依据的声称
2 = 多个无依据声称
1 = 大量幻觉

## 输出格式
```json
{{
  "accuracy": <1-5>,
  "completeness": <1-5>,
  "relevance": <1-5>,
  "faithfulness": <1-5>,
  "reasoning": "<评估理由>",
  "overall": <0-100>
}}
```"""

def judge_pointwise(case: EvalCase, model: str = "gpt-4o") -> JudgeResult:
    """Pointwise 单点评估"""
    ref_section = f"## 参考答案\n{case.reference}" if case.reference else ""
    
    prompt = JUDGE_PROMPT.format(
        question=case.question,
        answer=case.answer,
        reference_section=ref_section,
    )
    
    response = client.chat.completions.create(
        model=model,
        temperature=0,
        messages=[
            {"role": "system", "content": JUDGE_SYSTEM},
            {"role": "user", "content": prompt},
        ],
        response_format={"type": "json_object"},
    )
    
    result = json.loads(response.choices[0].message.content)
    
    return JudgeResult(
        case_id=case.id,
        accuracy=result["accuracy"],
        completeness=result["completeness"],
        relevance=result["relevance"],
        faithfulness=result["faithfulness"],
        reasoning=result["reasoning"],
        overall=result["overall"],
        judge_model=model,
    )

def judge_pairwise(
    question: str, 
    answer_a: str, 
    answer_b: str,
    model: str = "gpt-4o",
    randomize: bool = True,
) -> dict:
    """Pairwise 两两比较，带位置随机化"""
    if randomize and random.random() > 0.5:
        answer_a, answer_b = answer_b, answer_a
        swapped = True
    else:
        swapped = False
    
    prompt = f"""比较以下两个回答，哪个更好地回答了问题？

## 问题
{question}

## 回答 A
{answer_a}

## 回答 B
{answer_b}

## 评估标准
- 准确性：事实是否正确
- 完整性：是否全面
- 相关性：是否针对问题
- 清晰度：是否易于理解

## 输出格式
```json
{{
  "winner": "A" | "B" | "Tie",
  "reasoning": "比较理由",
  "a_accuracy": <1-5>,
  "b_accuracy": <1-5>,
  "a_completeness": <1-5>,
  "b_completeness": <1-5>
}}
```"""
    
    response = client.chat.completions.create(
        model=model,
        temperature=0,
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    
    result = json.loads(response.choices[0].message.content)
    
    # 如果交换了位置，需要反转结果
    if swapped:
        if result["winner"] == "A":
            result["winner"] = "B"
        elif result["winner"] == "B":
            result["winner"] = "A"
        result["a_accuracy"], result["b_accuracy"] = result["b_accuracy"], result["a_accuracy"]
        result["a_completeness"], result["b_completeness"] = result["b_completeness"], result["a_completeness"]
    
    return result

def run_eval_suite(
    cases: list[EvalCase],
    judge_model: str = "gpt-4o",
    n_judges: int = 3,
) -> dict:
    """运行完整评估套件，多 Judge 投票"""
    all_results = []
    
    for case in cases:
        # 多次评判取平均
        judge_results = []
        for _ in range(n_judges):
            result = judge_pointwise(case, model=judge_model)
            judge_results.append(result)
        
        # 聚合结果
        avg_accuracy = sum(r.accuracy for r in judge_results) / n_judges
        avg_completeness = sum(r.completeness for r in judge_results) / n_judges
        avg_relevance = sum(r.relevance for r in judge_results) / n_judges
        avg_faithfulness = sum(r.faithfulness for r in judge_results) / n_judges
        avg_overall = sum(r.overall for r in judge_results) / n_judges
        
        all_results.append({
            "case_id": case.id,
            "category": case.category,
            "avg_accuracy": avg_accuracy,
            "avg_completeness": avg_completeness,
            "avg_relevance": avg_relevance,
            "avg_faithfulness": avg_faithfulness,
            "avg_overall": avg_overall,
            "individual_results": judge_results,
        })
    
    # 汇总统计
    summary = {
        "total_cases": len(cases),
        "judge_model": judge_model,
        "n_judges": n_judges,
        "avg_accuracy": sum(r["avg_accuracy"] for r in all_results) / len(all_results),
        "avg_completeness": sum(r["avg_completeness"] for r in all_results) / len(all_results),
        "avg_relevance": sum(r["avg_relevance"] for r in all_results) / len(all_results),
        "avg_faithfulness": sum(r["avg_faithfulness"] for r in all_results) / len(all_results),
        "avg_overall": sum(r["avg_overall"] for r in all_results) / len(all_results),
        "by_category": {},
    }
    
    # 按分类统计
    categories = set(r["category"] for r in all_results if r["category"])
    for cat in categories:
        cat_results = [r for r in all_results if r["category"] == cat]
        summary["by_category"][cat] = {
            "count": len(cat_results),
            "avg_overall": sum(r["avg_overall"] for r in cat_results) / len(cat_results),
        }
    
    return {"summary": summary, "details": all_results}

"""
RAG 评估流水线 — 结合 RAGAS 和自定义指标
"""
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

# 自定义评估指标
def calculate_hallucination_rate(answers: list[str], contexts: list[list[str]]) -> float:
    """计算幻觉率：答案中有多少 claim 没有 context 支持"""
    total_claims = 0
    unsupported_claims = 0
    
    for answer, ctx in zip(answers, contexts):
        # 用 LLM 拆分答案为独立 claims
        claims_prompt = f"""将以下回答拆分为独立的陈述。每行一个。
回答: {answer}
输出格式: 每行一个陈述，不要编号"""
        
        claims = llm_call(claims_prompt).strip().split('\n')
        total_claims += len(claims)
        
        # 检查每个 claim 是否有 context 支持
        for claim in claims:
            support_prompt = f"""判断以下陈述是否被给定上下文支持。
陈述: {claim}
上下文: {' '.join(ctx)}
只回答 YES 或 NO。"""
            
            support = llm_call(support_prompt).strip().upper()
            if support == "NO":
                unsupported_claims += 1
    
    return unsupported_claims / max(total_claims, 1)

def calculate_citation_accuracy(answers: list[str], contexts: list[list[str]]) -> float:
    """计算引用准确率"""
    correct_citations = 0
    total_citations = 0
    
    for answer, ctx in zip(answers, contexts):
        # 提取答案中的引用标记，如 [1], [2]
        import re
        citations = re.findall(r'\[(\d+)\]', answer)
        total_citations += len(citations)
        
        # 检查引用的上下文是否确实支持对应内容
        for cite_num in citations:
            idx = int(cite_num) - 1
            if 0 <= idx < len(ctx):
                # 简化：假设引用存在即正确
                # 实际应用 LLM 判断引用内容是否匹配
                correct_citations += 1
    
    return correct_citations / max(total_citations, 1)

def run_rag_eval(
    questions: list[str],
    answers: list[str],
    contexts: list[list[str]],
    ground_truths: list[str],
) -> dict:
    """运行完整 RAG 评估"""
    # RAGAS 标准指标
    dataset = Dataset.from_dict({
        "question": questions,
        "answer": answers,
        "contexts": contexts,
        "ground_truth": ground_truths,
    })
    
    ragas_results = evaluate(
        dataset=dataset,
        metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    )
    
    # 自定义指标
    halluc_rate = calculate_hallucination_rate(answers, contexts)
    citation_acc = calculate_citation_accuracy(answers, contexts)
    
    return {
        "ragas": ragas_results,
        "custom": {
            "hallucination_rate": halluc_rate,
            "citation_accuracy": citation_acc,
        },
        "overall_score": (
            ragas_results["faithfulness"] * 0.3 +
            ragas_results["answer_relevancy"] * 0.25 +
            ragas_results["context_precision"] * 0.2 +
            ragas_results["context_recall"] * 0.15 +
            (1 - halluc_rate) * 0.1
        ),
    }

"""
成本效率评估 — 找到质量-成本帕累托最优点
"""
from dataclasses import dataclass

@dataclass
class ModelEvalResult:
    model: str
    quality_score: float      # 0-100
    avg_latency_ms: float
    avg_cost_per_1k_tokens: float
    task_success_rate: float

def pareto_analysis(results: list[ModelEvalResult]) -> list[ModelEvalResult]:
    """找出帕累托最优的模型（质量不差且成本更低）"""
    # 按 cost 排序
    sorted_results = sorted(results, key=lambda r: r.avg_cost_per_1k_tokens)
    
    pareto_frontier = []
    best_quality_so_far = 0
    
    for result in sorted_results:
        if result.quality_score >= best_quality_so_far:
            pareto_frontier.append(result)
            best_quality_so_far = result.quality_score
    
    return pareto_frontier

# 示例：对比 5 个模型
model_results = [
    ModelEvalResult("gpt-4",         quality_score=92, avg_latency_ms=2200, avg_cost_per_1k_tokens=0.030, task_success_rate=0.94),
    ModelEvalResult("gpt-4o",        quality_score=89, avg_latency_ms=800,  avg_cost_per_1k_tokens=0.005, task_success_rate=0.91),
    ModelEvalResult("gpt-4o-mini",   quality_score=78, avg_latency_ms=400,  avg_cost_per_1k_tokens=0.00015, task_success_rate=0.82),
    ModelEvalResult("claude-3.5",    quality_score=90, avg_latency_ms=1200, avg_cost_per_1k_tokens=0.008, task_success_rate=0.92),
    ModelEvalResult("claude-3-haiku", quality_score=75, avg_latency_ms=300,  avg_cost_per_1k_tokens=0.00025, task_success_rate=0.80),
]

frontier = pareto_analysis(model_results)
print("🏆 帕累托最优模型:")
for m in frontier:
    print(f"  {m.model}: 质量={m.quality_score} 成本=${m.avg_cost_per_1k_tokens}/1k 成功率={m.task_success_rate}")

# 输出:
# 🏆 帕累托最优模型:
#   gpt-4o-mini: 质量=78 成本=$0.00015/1k 成功率=0.82
#   gpt-4o: 质量=89 成本=$0.005/1k 成功率=0.91
#   gpt-4: 质量=92 成本=$0.03/1k 成功率=0.94

def design_routing_strategy(results: list[ModelEvalResult]) -> dict:
    """设计路由策略：根据任务难度分配模型"""
    # 简单规则：用质量-成本比排序
    cost_efficiency = [
        (r.model, r.quality_score / r.avg_cost_per_1k_tokens)
        for r in results
    ]
    
    return {
        "easy_tasks": "gpt-4o-mini",    # 简单任务用便宜模型
        "medium_tasks": "gpt-4o",       # 中等任务用性价比模型
        "hard_tasks": "gpt-4",          # 困难任务用最强模型
        "routing_prompt": """判断以下问题的难度：
- Easy: 事实查询、简单翻译、格式转换
- Medium: 推理、分析、多步骤任务
- Hard: 复杂推理、代码生成、创意写作
只输出 Easy/Medium/Hard。""",
        "estimated_savings": "60-80%",  # 相比全部用最强模型
    }