AI 应用评测体系教程

SEO 元信息

名称：AI应用评测体系教程

描述：零基础AI应用评测体系教程，涵盖LLM评测框架、评测指标设计、RAGAS评测、Agent评测、A/B测试、评测数据集构建、企业评测平台搭建等核心技能，适合AI开发者和产品经理系统学习。

关键词：AI评测, LLM评测, RAGAS, Agent评测, A/B测试

长尾关键词：AI应用评测体系搭建教程, LLM自动化评测教程, RAGAS RAG系统评测实战, 企业AI评测平台开发教程

1. 为什么 AI 应用需要评测体系

1.1 没有评测的 AI 应用，就是在"盲飞"

传统软件有明确的输入输出——给定条件 A，期望结果 B，跑个单元测试就知道对不对。AI 应用不同：同一个问题问两次，可能得到不同但都正确的回答。这种输出的不确定性和主观性，让传统的"期望值对比"式测试几乎失效。

如果你的 AI 应用没有评测体系，你将无法回答这些关键问题：

模型升级后效果变好了还是变差了？
Prompt 修改后哪些场景受益、哪些场景受损？
RAG 系统召回的文档是否真的有用？
Agent 是否在正确地调用工具？
用户满意度是在提升还是下降？

1.2 评测体系的核心价值

1. 质量守护：每次模型/Prompt/数据变更后，自动验证质量不退化 2. 决策支撑：用数据而非直觉选择模型、优化 Prompt、调整策略 3. 持续改进：发现薄弱环节，针对性优化 4. 合规保障：确保 AI 输出符合安全、伦理、法规要求

1.3 评测的三个层次

┌─────────────────────────────────────┐
│  L3: 业务指标层                      │
│  用户满意度、任务完成率、留存率         │
├─────────────────────────────────────┤
│  L2: 质量指标层                      │
│  准确率、相关性、忠实度、安全性         │
├─────────────────────────────────────┤
│  L1: 技术指标层                      │
│  困惑度、延迟、吞吐量、成本             │
└─────────────────────────────────────┘

好的评测体系需要跨越这三个层次，将技术指标与业务价值关联起来。

2. LLM 评测框架概览

2.1 主流评测框架

框架	开发者	特点	适用场景
lm-evaluation-harness	EleutherAI	学术标准，400+ 基准	模型能力全面评估
OpenCompass	上海AI实验室	中文评测强，可视化好	中文模型评测
HELM	Stanford	多维度、可复现	全面的模型评估
RAGAS	Exploding Gradients	RAG 专用	RAG 系统评测
DeepEval	Confident AI	LLM-as-Judge，CI/CD 集成	生产环境持续评测
LangSmith	LangChain	链路追踪 + 评测	LangChain 生态
Promptfoo	Promptfoo	命令行、多模型对比	Prompt 工程迭代
TruLens	TruEra	可观测性 + 评测	RAG/Agent 反馈循环

2.2 lm-evaluation-harness 快速上手

# 安装
pip install lm-eval

# 评测一个模型在多个基准上的表现
lm_eval --model hf \
    --model_args pretrained=meta-llama/Llama-2-7b-hf \
    --tasks mmlu,hellaswag,arc_challenge \
    --device cuda:0 \
    --batch_size 8 \
    --output_path ./eval_results/

# 评测 HuggingFace 上的量化模型
lm_eval --model hf \
    --model_args pretrained=./llama2-7b-gptq-4bit,dtype=float16 \
    --tasks mmlu \
    --device cuda:0

2.3 DeepEval 快速上手

# pip install deepeval

from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    ContextualPrecisionMetric,
    ContextualRecallMetric,
)
from deepeval.test_case import LLMTestCase

# 定义测试用例
test_case = LLMTestCase(
    input="什么是机器学习？",
    actual_output="机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出决策，而无需被明确编程。",
    retrieval_context=[
        "机器学习（ML）是人工智能的一个子领域，专注于让计算机系统从数据中学习和改进。",
        "深度学习是机器学习的一个子集，使用多层神经网络处理复杂数据。"
    ]
)

# 定义评测指标
metrics = [
    AnswerRelevancyMetric(threshold=0.7),    # 回答相关性
    FaithfulnessMetric(threshold=0.7),        # 忠实度（不幻觉）
    ContextualPrecisionMetric(threshold=0.7), # 上下文精确率
    ContextualRecallMetric(threshold=0.7),    # 上下文召回率
]

# 执行评测
evaluate(test_cases=[test_case], metrics=metrics)

3. 评测指标体系设计

3.1 通用 LLM 评测指标

准确性指标：

Pass@k：在 k 次采样中至少有一次正确的概率（代码生成常用）
Exact Match (EM)：答案是否完全匹配
F1 Score：部分匹配时的精确率和召回率调和平均

语言质量指标：

困惑度（Perplexity）：模型对文本的"惊讶程度"，越低越好
BLEU / ROUGE：与参考答案的 n-gram 重叠度（自动摘要、翻译常用）
BERTScore：基于语义嵌入的相似度，比 BLEU 更符合人类判断

安全性指标：

毒性（Toxicity）：输出是否包含有害、歧视性内容
偏见（Bias）：对不同群体是否存在系统性偏见
拒答率：面对不当问题时是否正确拒绝

3.2 RAG 系统专用指标

指标	定义	评估对象
Context Precision	检索文档中相关文档的排名	检索质量
Context Recall	回答所需信息是否都被检索到	检索完整性
Faithfulness	回答是否忠实于检索到的上下文	生成可靠性
Answer Relevancy	回答与问题的相关程度	生成质量

3.3 Agent 评测指标

指标	定义
Tool Selection Accuracy	是否选择了正确的工具
Tool Call Success Rate	工具调用是否成功执行
Task Completion Rate	最终任务是否完成
Step Efficiency	完成任务的步骤数是否合理
Error Recovery	遇到错误时能否自我纠正

3.4 构建指标体系的实践建议

# 推荐的指标分层设计
METRICS_FRAMEWORK = {
    "core": {
        # 必须评测，每次变更都要跑
        "answer_accuracy": {"weight": 0.3, "threshold": 0.85},
        "faithfulness": {"weight": 0.25, "threshold": 0.90},
        "safety_score": {"weight": 0.2, "threshold": 0.95},
        "latency_p95": {"weight": 0.1, "threshold_ms": 3000},
    },
    "extended": {
        # 定期评测，每周/每月
        "answer_relevancy": {"weight": 0.05, "threshold": 0.80},
        "context_precision": {"weight": 0.05, "threshold": 0.75},
        "context_recall": {"weight": 0.05, "threshold": 0.75},
    },
    "business": {
        # 与业务指标关联
        "user_satisfaction": {"target": 4.2},  # 5 分制
        "task_success_rate": {"target": 0.85},
    }
}

def compute_overall_score(results: dict, framework: dict) -> float:
    """计算综合评分"""
    total_score = 0.0
    total_weight = 0.0
    
    for category in framework.values():
        for metric_name, config in category.items():
            if metric_name in results:
                score = results[metric_name]
                weight = config.get("weight", 1.0)
                total_score += score * weight
                total_weight += weight
    
    return total_score / total_weight if total_weight > 0 else 0.0

4. 自动化评测流水线

4.1 CI/CD 集成

将评测嵌入开发流程，确保每次变更都经过质量验证。

# .github/workflows/ai-eval.yml
name: AI Model Evaluation

on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'models/**'
      - 'configs/**'

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.10'
      
      - name: Install Dependencies
        run: pip install -r requirements-eval.txt
      
      - name: Run Core Metrics
        run: |
          python -m eval.run_suite \
            --suite core \
            --test-data eval/test_data/core_v2.json \
            --output results/core.json \
            --threshold 0.85
      
      - name: Run Safety Checks
        run: |
          python -m eval.safety_check \
            --input results/core.json \
            --fail-on toxicity_bias
      
      - name: Compare with Baseline
        run: |
          python -m eval.compare \
            --current results/core.json \
            --baseline eval/baselines/latest.json \
            --max-regression 0.03
      
      - name: Upload Results
        uses: actions/upload-artifact@v4
        with:
          name: eval-results
          path: results/

4.2 评测流水线核心组件

import json
import time
from dataclasses import dataclass, field
from typing import Callable
from concurrent.futures import ThreadPoolExecutor

@dataclass
class EvalCase:
    """单个评测用例"""
    id: str
    input_text: str
    expected_output: str = ""
    context: list[str] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)

@dataclass
class EvalResult:
    """评测结果"""
    case_id: str
    actual_output: str
    scores: dict[str, float]
    latency_ms: float
    passed: bool

class EvalPipeline:
    """自动化评测流水线"""
    
    def __init__(self, model_fn: Callable, metrics: list, threshold: float = 0.8):
        self.model_fn = model_fn    # 模型推理函数
        self.metrics = metrics      # 评测指标列表
        self.threshold = threshold  # 通过阈值
    
    def evaluate_single(self, case: EvalCase) -> EvalResult:
        """评测单个用例"""
        start = time.time()
        actual_output = self.model_fn(case.input_text, case.context)
        latency = (time.time() - start) * 1000
        
        scores = {}
        for metric in self.metrics:
            scores[metric.name] = metric.compute(
                input_text=case.input_text,
                expected=case.expected_output,
                actual=actual_output,
                context=case.context
            )
        
        avg_score = sum(scores.values()) / len(scores) if scores else 0
        
        return EvalResult(
            case_id=case.id,
            actual_output=actual_output,
            scores=scores,
            latency_ms=latency,
            passed=avg_score >= self.threshold
        )
    
    def run_suite(self, cases: list[EvalCase], concurrency: int = 4) -> dict:
        """运行完整评测套件"""
        results = []
        
        with ThreadPoolExecutor(max_workers=concurrency) as executor:
            futures = [
                executor.submit(self.evaluate_single, case)
                for case in cases
            ]
            results = [f.result() for f in futures]
        
        # 汇总统计
        total = len(results)
        passed = sum(1 for r in results if r.passed)
        avg_scores = {}
        
        for metric_name in self.metrics:
            scores = [r.scores.get(metric_name.name, 0) for r in results]
            avg_scores[metric_name.name] = sum(scores) / len(scores)
        
        return {
            "total": total,
            "passed": passed,
            "pass_rate": passed / total,
            "avg_scores": avg_scores,
            "avg_latency_ms": sum(r.latency_ms for r in results) / total,
            "details": results,
        }

4.3 回归检测

def detect_regression(current: dict, baseline: dict, tolerance: float = 0.03):
    """
    检测评测结果是否相比基线有退化
    
    Args:
        current: 当前评测结果
        baseline: 基线评测结果
        tolerance: 允许的最大退化幅度（3%）
    
    Returns:
        regressions: 退化指标列表
    """
    regressions = []
    
    for metric, current_score in current["avg_scores"].items():
        baseline_score = baseline["avg_scores"].get(metric)
        if baseline_score is None:
            continue
        
        delta = current_score - baseline_score
        if delta < -tolerance:
            regressions.append({
                "metric": metric,
                "baseline": baseline_score,
                "current": current_score,
                "delta": delta,
                "severity": "critical" if delta < -0.1 else "warning"
            })
    
    return regressions

# 使用示例
regressions = detect_regression(current_results, baseline_results)
if regressions:
    print("⚠️  检测到质量退化：")
    for r in regressions:
        print(f"  [{r['severity'].upper()}] {r['metric']}: "
              f"{r['baseline']:.3f} → {r['current']:.3f} (Δ{r['delta']:+.3f})")

5. RAG 系统评测（RAGAS）

5.1 RAGAS 框架简介

RAGAS（Retrieval Augmented Generation Assessment）是专为 RAG 系统设计的评测框架，核心理念是用 LLM-as-Judge 实现无需人工标注的自动化评测。

RAGAS 评测 RAG 的四个核心维度：

Faithfulness（忠实度）：回答是否基于检索到的上下文，而非模型自身知识"幻觉"
Answer Relevancy（回答相关性）：回答是否与问题相关
Context Precision（上下文精确率）：检索到的文档中，有多少是真正有用的
Context Recall（上下文召回率）：回答问题所需的信息是否都被检索到了

5.2 RAGAS 实战

pip install ragas langchain langchain-openai datasets

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

# 准备评测数据
# 每条数据包含：question, answer, contexts, ground_truth
eval_data = {
    "question": [
        "Llama-2 模型有多少参数版本？",
        "什么是 RLHF？",
        "Transformer 的注意力机制是如何工作的？",
    ],
    "answer": [
        "Llama-2 有 7B、13B 和 70B 三个参数版本。",
        "RLHF（Reinforcement Learning from Human Feedback）是一种通过人类反馈来微调语言模型的技术。",
        "Transformer 的注意力机制通过计算 Query 和 Key 的点积来分配 Value 的权重，实现对不同位置信息的动态关注。"
    ],
    "contexts": [
        [
            "Llama 2 是 Meta 发布的开源大语言模型，提供 7B、13B 和 70B 三种规模。",
            "Llama 2 使用了 2 万亿 token 进行预训练，上下文窗口为 4096。"
        ],
        [
            "RLHF 是将强化学习与人类偏好反馈相结合的训练方法，用于对齐语言模型的行为与人类价值观。",
            "ChatGPT 使用了 RLHF 技术来提升对话质量。"
        ],
        [
            "自注意力机制计算公式为 Attention(Q,K,V) = softmax(QK^T/√d_k)V。",
            "Transformer 架构由编码器和解码器组成，每层包含多头注意力和前馈网络。"
        ]
    ],
    "ground_truth": [
        "Llama-2 提供 7B、13B 和 70B 三种参数规模。",
        "RLHF 是基于人类反馈的强化学习，用于语言模型对齐。",
        "注意力机制通过 QKV 矩阵运算，计算 softmax(QK^T/√d_k)V 来动态分配权重。"
    ]
}

dataset = Dataset.from_dict(eval_data)

# 执行评测
result = evaluate(
    dataset=dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ],
)

# 查看结果
print("=== RAGAS 评测结果 ===")
print(f"Faithfulness:      {result['faithfulness']:.4f}")
print(f"Answer Relevancy:  {result['answer_relevancy']:.4f}")
print(f"Context Precision: {result['context_precision']:.4f}")
print(f"Context Recall:    {result['context_recall']:.4f}")
print(f"Overall Score:     {result['faithfulness']:.4f}")

# 导出详细结果
df = result.to_pandas()
df.to_csv("ragas_eval_results.csv", index=False)

5.3 RAG 评测最佳实践

1. 分层评测

# 先评测检索，再评测生成
def evaluate_rag_pipeline(rag_system, test_cases):
    results = {"retrieval": [], "generation": []}
    
    for case in test_cases:
        # 评测检索阶段
        retrieved_docs = rag_system.retrieve(case.question)
        retrieval_score = evaluate_retrieval(
            retrieved=retrieved_docs,
            relevant=case.relevant_doc_ids
        )
        results["retrieval"].append(retrieval_score)
        
        # 评测生成阶段
        answer = rag_system.generate(case.question)
        generation_score = evaluate_generation(
            question=case.question,
            answer=answer,
            context=[doc.text for doc in retrieved_docs],
            ground_truth=case.ground_truth
        )
        results["generation"].append(generation_score)
    
    return {
        "avg_retrieval": sum(results["retrieval"]) / len(results["retrieval"]),
        "avg_generation": sum(results["generation"]) / len(results["generation"]),
    }

2. 关注失败模式

检索到了但没用上 → 生成模型能力不足
没检索到但答对了 → 模型在"幻觉"（可能正确但不可靠）
检索到了且答错了 → 检索质量差或生成理解错误

6. Agent 评测方法论

6.1 Agent 评测的独特挑战

Agent（智能体）不同于简单的问答系统，它具有自主决策、工具调用、多步推理的能力。这带来了独特的评测挑战：

非确定性路径：完成同一个任务可能有多种合理的工具调用序列
状态依赖：后续步骤的正确性依赖于前面步骤的结果
环境交互：涉及外部 API 调用、数据库查询等，难以完全模拟
部分成功：任务可能部分完成，需要评估"完成了多少"

6.2 Agent 评测维度

AGENT_EVAL_DIMENSIONS = {
    "planning": {
        "description": "任务分解与规划能力",
        "metrics": ["step_count", "plan_quality", "dependency_accuracy"]
    },
    "tool_use": {
        "description": "工具选择与调用能力",
        "metrics": ["tool_selection_accuracy", "param_accuracy", "call_success_rate"]
    },
    "reasoning": {
        "description": "中间推理与决策能力",
        "metrics": ["reasoning_coherence", "error_recovery", "adaptability"]
    },
    "task_completion": {
        "description": "最终任务完成情况",
        "metrics": ["completion_rate", "output_accuracy", "efficiency"]
    }
}

6.3 Agent 评测实战

from dataclasses import dataclass
from typing import Any

@dataclass
class AgentStep:
    """Agent 执行的一步"""
    tool_name: str
    tool_input: dict
    tool_output: Any
    reasoning: str

@dataclass
class AgentTestCase:
    """Agent 评测用例"""
    id: str
    task: str
    expected_tools: list[str]           # 期望使用的工具
    expected_steps_range: tuple[int, int]  # 期望步骤数范围
    ground_truth_output: Any            # 期望最终输出
    available_tools: list[str]          # 可用工具列表

class AgentEvaluator:
    """Agent 评测器"""
    
    def evaluate(self, test_case: AgentTestCase, 
                 actual_steps: list[AgentStep], 
                 actual_output: Any) -> dict:
        scores = {}
        
        # 1. 工具选择准确率
        actual_tools = [s.tool_name for s in actual_steps]
        tool_overlap = set(actual_tools) & set(test_case.expected_tools)
        tool_precision = len(tool_overlap) / len(actual_tools) if actual_tools else 0
        tool_recall = len(tool_overlap) / len(test_case.expected_tools)
        scores["tool_selection_f1"] = (
            2 * tool_precision * tool_recall / (tool_precision + tool_recall)
            if (tool_precision + tool_recall) > 0 else 0
        )
        
        # 2. 步骤效率
        min_steps, max_steps = test_case.expected_steps_range
        actual_step_count = len(actual_steps)
        if min_steps <= actual_step_count <= max_steps:
            scores["step_efficiency"] = 1.0
        elif actual_step_count < min_steps:
            scores["step_efficiency"] = max(0, 1 - (min_steps - actual_step_count) / min_steps)
        else:
            scores["step_efficiency"] = max(0, 1 - (actual_step_count - max_steps) / max_steps)
        
        # 3. 工具调用成功率
        successful_calls = sum(1 for s in actual_steps if s.tool_output is not None)
        scores["call_success_rate"] = (
            successful_calls / len(actual_steps) if actual_steps else 0
        )
        
        # 4. 错误恢复能力（连续失败后能否恢复）
        max_consecutive_failures = 0
        current_failures = 0
        for step in actual_steps:
            if step.tool_output is None:
                current_failures += 1
                max_consecutive_failures = max(max_consecutive_failures, current_failures)
            else:
                current_failures = 0
        scores["error_recovery"] = 1.0 if max_consecutive_failures <= 1 else max(0, 1 - max_consecutive_failures * 0.3)
        
        # 5. 最终输出正确性（使用 LLM-as-Judge）
        scores["output_accuracy"] = self._judge_output(
            test_case.task, actual_output, test_case.ground_truth_output
        )
        
        # 综合评分
        weights = {
            "tool_selection_f1": 0.25,
            "step_efficiency": 0.15,
            "call_success_rate": 0.15,
            "error_recovery": 0.10,
            "output_accuracy": 0.35,
        }
        scores["overall"] = sum(scores[k] * weights[k] for k in weights)
        
        return scores
    
    def _judge_output(self, task: str, actual: Any, expected: Any) -> float:
        """使用 LLM 评判输出正确性"""
        # 实际实现中调用 LLM 进行评判
        # 这里简化为字符串匹配
        actual_str = str(actual).strip().lower()
        expected_str = str(expected).strip().lower()
        if actual_str == expected_str:
            return 1.0
        # 部分匹配
        from difflib import SequenceMatcher
        return SequenceMatcher(None, actual_str, expected_str).ratio()

7. A/B 测试与人类评测

7.1 A/B 测试设计

A/B 测试是验证 AI 改进是否真正有效的"金标准"。在 AI 应用中，A/B 测试通常用于对比不同模型、不同 Prompt 或不同 RAG 策略的效果。

import random
import hashlib
from datetime import datetime

class ABTestRouter:
    """A/B 测试路由器"""
    
    def __init__(self, experiments: dict):
        """
        experiments = {
            "model_upgrade": {
                "control": {"model": "llama2-70b", "prompt_v": 1},
                "treatment": {"model": "llama3-70b", "prompt_v": 2},
                "traffic_split": 0.1,  # 10% 流量给 treatment
            }
        }
        """
        self.experiments = experiments
    
    def get_variant(self, experiment_name: str, user_id: str) -> dict:
        """根据用户 ID 确定分配到哪个变体（保证同一用户始终在同一组）"""
        exp = self.experiments[experiment_name]
        
        # 用哈希保证一致性
        hash_input = f"{experiment_name}:{user_id}"
        hash_val = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
        bucket = (hash_val % 1000) / 1000
        
        if bucket < exp["traffic_split"]:
            variant = "treatment"
        else:
            variant = "control"
        
        return {
            "variant": variant,
            "config": exp[variant],
            "experiment": experiment_name,
        }

class ABTestLogger:
    """A/B 测试日志记录"""
    
    def log_event(self, event: dict):
        """记录用户交互事件"""
        event["timestamp"] = datetime.utcnow().isoformat()
        # 实际生产中写入数据库或消息队列
        print(f"[AB Event] {event}")

# 使用示例
router = ABTestRouter({
    "model_upgrade_v2": {
        "control": {"model": "gpt-4o-mini", "temperature": 0.7},
        "treatment": {"model": "gpt-4o", "temperature": 0.7},
        "traffic_split": 0.2,
    }
})

# 模拟用户请求
for user_id in ["user_001", "user_002", "user_003"]:
    variant = router.get_variant("model_upgrade_v2", user_id)
    print(f"{user_id} → {variant['variant']}: {variant['config']}")

7.2 人类评测方案

from enum import Enum
from dataclasses import dataclass

class RatingScale(Enum):
    """评分量表"""
    LIKERT_5 = [1, 2, 3, 4, 5]           # 5 分制
    LIKERT_7 = [1, 2, 3, 4, 5, 6, 7]     # 7 分制
    BINARY = [0, 1]                        # 是/否
    ELO = None                             # 对比式

@dataclass
class HumanEvalTask:
    """人类评测任务"""
    id: str
    prompt: str
    response_a: str
    response_b: str  # 盲评：不告知来源
    criteria: list[str]  # 评测维度

class HumanEvalPlatform:
    """人类评测平台"""
    
    def create_eval_batch(self, tasks: list[HumanEvalTask], 
                          num_evaluators: int = 3) -> dict:
        """创建评测批次"""
        batch = {
            "tasks": tasks,
            "evaluators_per_task": num_evaluators,
            "total_annotations": len(tasks) * num_evaluators,
            "criteria": [
                "准确性：回答是否事实正确",
                "完整性：是否充分回答了问题",
                "流畅性：语言是否自然流畅",
                "有用性：对用户是否有实际帮助",
            ],
            "instructions": (
                "请对比两个回答（A 和 B），对每个评测维度打分（1-5分）。"
                "1=很差 2=较差 3=一般 4=较好 5=很好。"
                "请根据你的专业判断独立评分。"
            )
        }
        return batch
    
    def compute_inter_annotator_agreement(self, annotations: list[dict]) -> float:
        """计算标注者一致性（Cohen's Kappa / Fleiss' Kappa）"""
        # 简化实现：计算成对一致率
        agreements = 0
        total = 0
        
        for i in range(len(annotations)):
            for j in range(i + 1, len(annotations)):
                if annotations[i]["winner"] == annotations[j]["winner"]:
                    agreements += 1
                total += 1
        
        return agreements / total if total > 0 else 0

7.3 统计显著性检验

from scipy import stats
import numpy as np

def ab_test_significance(control_scores: list[float], 
                         treatment_scores: list[float],
                         alpha: float = 0.05) -> dict:
    """A/B 测试统计显著性检验"""
    
    control = np.array(control_scores)
    treatment = np.array(treatment_scores)
    
    # 描述性统计
    control_mean = control.mean()
    treatment_mean = treatment.mean()
    lift = (treatment_mean - control_mean) / control_mean * 100
    
    # 双样本 t 检验
    t_stat, p_value = stats.ttest_ind(control, treatment, equal_var=False)
    
    # 效应量（Cohen's d）
    pooled_std = np.sqrt((control.std()**2 + treatment.std()**2) / 2)
    cohens_d = (treatment_mean - control_mean) / pooled_std if pooled_std > 0 else 0
    
    # 判断显著性
    is_significant = p_value < alpha
    
    return {
        "control_mean": round(control_mean, 4),
        "treatment_mean": round(treatment_mean, 4),
        "lift_pct": round(lift, 2),
        "p_value": round(p_value, 6),
        "cohens_d": round(cohens_d, 4),
        "is_significant": is_significant,
        "sample_sizes": {"control": len(control), "treatment": len(treatment)},
        "recommendation": (
            "✅ 显著差异，建议采用 treatment"
            if is_significant and lift > 0
            else "❌ 无显著差异或 treatment 更差，保持 control"
        )
    }

# 示例
control = [4.1, 3.8, 4.3, 4.0, 3.9, 4.2, 3.7, 4.1, 4.0, 3.8]
treatment = [4.5, 4.3, 4.6, 4.4, 4.7, 4.2, 4.5, 4.6, 4.4, 4.3]
result = ab_test_significance(control, treatment)
print(result)

8. 评测数据集构建

8.1 数据集构建流程

需求分析 → 数据收集 → 标注规范 → 人工标注 → 质量审核 → 数据发布
    ↑                                                    ↓
    └──────────── 持续迭代 ←── 错误分析 ←── 评测反馈 ←──┘

8.2 数据集构建实战

import json
import uuid
from datetime import datetime

class EvalDatasetBuilder:
    """评测数据集构建器"""
    
    def __init__(self, name: str, version: str):
        self.name = name
        self.version = version
        self.cases = []
        self.metadata = {
            "name": name,
            "version": version,
            "created_at": datetime.utcnow().isoformat(),
            "cases_count": 0,
        }
    
    def add_case(self, input_text: str, expected_output: str = "",
                 context: list[str] = None, category: str = "general",
                 difficulty: str = "medium", tags: list[str] = None):
        """添加评测用例"""
        case = {
            "id": str(uuid.uuid4()),
            "input": input_text,
            "expected_output": expected_output,
            "context": context or [],
            "category": category,
            "difficulty": difficulty,
            "tags": tags or [],
            "created_at": datetime.utcnow().isoformat(),
        }
        self.cases.append(case)
        return case["id"]
    
    def add_from_real_queries(self, queries: list[dict], 
                              source: str = "production_logs"):
        """从真实用户查询中构建评测用例"""
        for q in queries:
            self.add_case(
                input_text=q["query"],
                expected_output=q.get("good_response", ""),
                category=q.get("category", "real_world"),
                difficulty=q.get("difficulty", "medium"),
                tags=[source, q.get("intent", "unknown")]
            )
    
    def ensure_coverage(self, min_per_category: int = 10):
        """检查各类别覆盖度"""
        from collections import Counter
        categories = Counter(c["category"] for c in self.cases)
        
        underrepresented = {
            cat: count for cat, count in categories.items()
            if count < min_per_category
        }
        
        if underrepresented:
            print("⚠️ 以下类别用例不足：")
            for cat, count in underrepresented.items():
                print(f"  {cat}: {count}/{min_per_category}")
        
        return underrepresented
    
    def export(self, filepath: str):
        """导出数据集"""
        self.metadata["cases_count"] = len(self.cases)
        
        dataset = {
            "metadata": self.metadata,
            "cases": self.cases,
        }
        
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(dataset, f, ensure_ascii=False, indent=2)
        
        print(f"✅ 数据集已导出: {filepath}")
        print(f"   共 {len(self.cases)} 条用例")
        
        # 统计信息
        from collections import Counter
        cats = Counter(c["category"] for c in self.cases)
        diffs = Counter(c["difficulty"] for c in self.cases)
        print(f"   类别分布: {dict(cats)}")
        print(f"   难度分布: {dict(diffs)}")

# 使用示例
builder = EvalDatasetBuilder("customer_service_v2", "2.0")

# 添加不同类别的用例
builder.add_case(
    input_text="我想退款，订单号是 12345",
    expected_output="好的，我来帮您查询订单 12345 的退款流程...",
    category="refund",
    difficulty="easy",
    tags=["refund", "order"]
)

builder.add_case(
    input_text="你们的产品和竞品 X 相比有什么优势？",
    expected_output="我们的产品在以下几个方面有明显优势：...",
    category="comparison",
    difficulty="hard",
    tags=["sales", "comparison"]
)

# 从生产日志添加
real_queries = [
    {"query": "怎么修改收货地址？", "category": "account", "difficulty": "easy"},
    {"query": "这个产品支持分期付款吗？", "category": "payment", "difficulty": "medium"},
]
builder.add_from_real_queries(real_queries)

# 检查覆盖度并导出
builder.ensure_coverage(min_per_category=5)
builder.export("eval_dataset_v2.json")

8.3 数据集质量保障

多样性：确保覆盖不同类别、难度、语言、长度的用例 代表性：用例应来自真实用户场景，而非"拍脑袋"构造 时效性：定期更新，淘汰过时用例，补充新场景 标注质量：多人标注 + 一致性检验（Kappa > 0.7） 黄金标准：保留一批"必对"用例，作为底线保障

9. LMSYS/Chatbot Arena 式评测

9.1 Arena 评测模式

LMSYS Chatbot Arena 开创了一种革命性的评测模式：用户盲评 + Elo 排名。

核心机制：

用户提交问题，系统随机分配两个匿名模型生成回答
用户选择更好的回答（或平局）
基于投票结果更新 Elo 评分
积累足够投票后得到可信的模型排名

这种模式的优势在于：

依赖真实用户的偏好，而非预设标准
盲评消除了品牌偏见
Elo 评分系统收敛稳定
可持续扩展

9.2 自建 Arena 系统

import random
import math
from dataclasses import dataclass
from typing import Callable

@dataclass
class ModelEntry:
    """参赛模型"""
    name: str
    elo: float = 1000.0
    matches: int = 0
    wins: int = 0
    losses: int = 0
    draws: int = 0

class ArenaSystem:
    """Chatbot Arena 评测系统"""
    
    def __init__(self, models: dict[str, Callable], k_factor: int = 32):
        """
        Args:
            models: {模型名: 推理函数} 的字典
            k_factor: Elo 评分的 K 因子
        """
        self.entries = {name: ModelEntry(name=name) for name in models}
        self.model_fns = models
        self.k_factor = k_factor
        self.match_history = []
    
    def create_match(self, prompt: str) -> dict:
        """创建一场比赛"""
        # 随机选择两个不同的模型
        names = random.sample(list(self.entries.keys()), 2)
        model_a, model_b = names
        
        # 生成回答（打乱顺序以消除位置偏见）
        response_a = self.model_fns[model_a](prompt)
        response_b = self.model_fns[model_b](prompt)
        
        # 随机决定展示顺序
        if random.random() > 0.5:
            left, right = response_a, response_b
            left_model, right_model = model_a, model_b
        else:
            left, right = response_b, response_a
            left_model, right_model = model_b, model_a
        
        match_id = len(self.match_history)
        match = {
            "id": match_id,
            "prompt": prompt,
            "left_model": left_model,  # 不暴露给用户
            "right_model": right_model,
            "left_response": left,
            "right_response": right,
        }
        self.match_history.append(match)
        
        return {
            "match_id": match_id,
            "prompt": prompt,
            "response_left": left,
            "response_right": right,
            # 不返回模型名称！
        }
    
    def record_vote(self, match_id: int, vote: str):
        """
        记录用户投票
        vote: "left" | "right" | "tie"
        """
        match = self.match_history[match_id]
        left_name = match["left_model"]
        right_name = match["right_model"]
        
        # 计算期望得分
        entry_left = self.entries[left_name]
        entry_right = self.entries[right_name]
        
        e_left = 1 / (1 + 10 ** ((entry_right.elo - entry_left.elo) / 400))
        e_right = 1 - e_left
        
        # 实际得分
        if vote == "left":
            s_left, s_right = 1.0, 0.0
            entry_left.wins += 1
            entry_right.losses += 1
        elif vote == "right":
            s_left, s_right = 0.0, 1.0
            entry_left.losses += 1
            entry_right.wins += 1
        else:  # tie
            s_left, s_right = 0.5, 0.5
            entry_left.draws += 1
            entry_right.draws += 1
        
        # 更新 Elo
        entry_left.elo += self.k_factor * (s_left - e_left)
        entry_right.elo += self.k_factor * (s_right - e_right)
        entry_left.matches += 1
        entry_right.matches += 1
    
    def get_leaderboard(self) -> list[dict]:
        """获取排行榜"""
        sorted_entries = sorted(
            self.entries.values(),
            key=lambda e: e.elo,
            reverse=True
        )
        
        return [
            {
                "rank": i + 1,
                "model": e.name,
                "elo": round(e.elo),
                "matches": e.matches,
                "wins": e.wins,
                "losses": e.losses,
                "win_rate": round(e.wins / e.matches * 100, 1) if e.matches > 0 else 0,
            }
            for i, e in enumerate(sorted_entries)
        ]

# 使用示例
def model_a_fn(prompt): return f"[Model A] 回答: {prompt}"
def model_b_fn(prompt): return f"[Model B] 回答: {prompt}"
def model_c_fn(prompt): return f"[Model C] 回答: {prompt}"

arena = ArenaSystem({
    "GPT-4o": model_a_fn,
    "Claude-3.5": model_b_fn,
    "Llama-3-70B": model_c_fn,
})

# 模拟比赛
for _ in range(100):
    match_info = arena.create_match("请解释量子计算的基本原理")
    # 模拟投票（实际中由用户投票）
    vote = random.choice(["left", "right", "tie"])
    arena.record_vote(match_info["match_id"], vote)

# 查看排行榜
leaderboard = arena.get_leaderboard()
for entry in leaderboard:
    print(f"#{entry['rank']} {entry['model']}: "
          f"Elo={entry['elo']} Win Rate={entry['win_rate']}%")

10. 实战：构建企业 AI 应用评测平台

10.1 平台架构设计

┌──────────────────────────────────────────────────────┐
│                    评测平台前端                         │
│    (评测任务管理 / 结果可视化 / 排行榜 / 配置中心)        │
├──────────────────────────────────────────────────────┤
│                    API 网关                            │
├──────────┬──────────┬──────────┬─────────────────────┤
│ 评测引擎  │ 数据管理  │ 报告生成  │ 模型服务网关         │
│ ·指标计算 │ ·数据集   │ ·趋势图  │ ·模型路由           │
│ ·LLM裁判 │ ·版本管理  │ ·对比表  │ ·负载均衡           │
│ ·回归检测 │ ·标注管理  │ ·告警   │ ·A/B分流            │
├──────────┴──────────┴──────────┴─────────────────────┤
│              消息队列 / 任务调度                        │
├──────────────────────────────────────────────────────┤
│         数据库 (PostgreSQL) + 对象存储 (S3)             │
└──────────────────────────────────────────────────────┘

10.2 核心模块实现

# ==================== 评测平台核心代码 ====================

import json
import uuid
import asyncio
from datetime import datetime
from enum import Enum
from dataclasses import dataclass, field, asdict

# ---------- 数据模型 ----------

class TaskStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

class MetricType(Enum):
    AUTOMATED = "automated"   # 自动计算
    LLM_JUDGE = "llm_judge"  # LLM 裁判
    HUMAN = "human"          # 人工评测

@dataclass
class EvalTask:
    """评测任务"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    model_id: str = ""
    dataset_id: str = ""
    metrics: list[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    completed_at: str = ""
    results: dict = field(default_factory=dict)
    config: dict = field(default_factory=dict)

@dataclass
class EvalDataset:
    """评测数据集"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    version: str = "1.0"
    cases: list[dict] = field(default_factory=list)
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())

# ---------- 评测引擎 ----------

class EvalEngine:
    """评测引擎"""
    
    def __init__(self):
        self.metrics_registry = {}
        self.model_registry = {}
        self.datasets = {}
        self.tasks = {}
    
    def register_metric(self, name: str, metric_fn, metric_type: MetricType):
        """注册评测指标"""
        self.metrics_registry[name] = {
            "fn": metric_fn,
            "type": metric_type,
        }
    
    def register_model(self, model_id: str, inference_fn):
        """注册模型"""
        self.model_registry[model_id] = inference_fn
    
    def create_dataset(self, name: str, cases: list[dict]) -> str:
        """创建评测数据集"""
        ds = EvalDataset(name=name, cases=cases)
        self.datasets[ds.id] = ds
        return ds.id
    
    async def run_evaluation(self, task: EvalTask) -> dict:
        """执行评测任务"""
        task.status = TaskStatus.RUNNING
        
        try:
            model_fn = self.model_registry[task.model_id]
            dataset = self.datasets[task.dataset_id]
            metrics = [self.metrics_registry[m] for m in task.metrics]
            
            results = []
            for case in dataset.cases:
                # 模型推理
                start_time = datetime.utcnow()
                actual_output = model_fn(case["input"], case.get("context", []))
                latency = (datetime.utcnow() - start_time).total_seconds() * 1000
                
                # 计算指标
                scores = {}
                for metric_name, metric_info in zip(task.metrics, metrics):
                    score = metric_info["fn"](
                        input_text=case["input"],
                        expected=case.get("expected_output", ""),
                        actual=actual_output,
                        context=case.get("context", [])
                    )
                    scores[metric_name] = score
                
                results.append({
                    "case_id": case.get("id", ""),
                    "input": case["input"],
                    "actual_output": actual_output,
                    "expected_output": case.get("expected_output", ""),
                    "scores": scores,
                    "latency_ms": latency,
                })
            
            # 汇总
            summary = self._compute_summary(results, task.metrics)
            
            task.status = TaskStatus.COMPLETED
            task.completed_at = datetime.utcnow().isoformat()
            task.results = {
                "summary": summary,
                "details": results,
            }
            
            return task.results
            
        except Exception as e:
            task.status = TaskStatus.FAILED
            task.results = {"error": str(e)}
            raise
    
    def _compute_summary(self, results: list, metrics: list) -> dict:
        """计算汇总统计"""
        summary = {}
        
        for metric in metrics:
            scores = [r["scores"].get(metric, 0) for r in results]
            summary[metric] = {
                "mean": sum(scores) / len(scores),
                "min": min(scores),
                "max": max(scores),
                "std": (sum((s - sum(scores)/len(scores))**2 for s in scores) / len(scores)) ** 0.5,
            }
        
        latencies = [r["latency_ms"] for r in results]
        latencies.sort()
        summary["latency"] = {
            "mean": sum(latencies) / len(latencies),
            "p50": latencies[len(latencies) // 2],
            "p95": latencies[int(len(latencies) * 0.95)],
            "p99": latencies[int(len(latencies) * 0.99)],
        }
        
        return summary

# ---------- 报告生成 ----------

class ReportGenerator:
    """评测报告生成器"""
    
    def generate_html_report(self, task: EvalTask) -> str:
        """生成 HTML 评测报告"""
        summary = task.results.get("summary", {})
        
        html = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>评测报告 - {task.name}</title>
            <style>
                body {{ font-family: -apple-system, sans-serif; max-width: 900px; margin: 0 auto; padding: 20px; }}
                .header {{ background: #1a1a2e; color: white; padding: 30px; border-radius: 12px; }}
                .metric-card {{ background: #f8f9fa; padding: 20px; border-radius: 8px; margin: 10px 0; }}
                .metric-value {{ font-size: 2em; font-weight: bold; color: #2563eb; }}
                .good {{ color: #16a34a; }} .bad {{ color: #dc2626; }}
                table {{ width: 100%; border-collapse: collapse; margin: 20px 0; }}
                th, td {{ padding: 12px; text-align: left; border-bottom: 1px solid #e5e7eb; }}
                th {{ background: #f1f5f9; }}
            </style>
        </head>
        <body>
            <div class="header">
                <h1>📊 评测报告</h1>
                <p>任务: {task.name} | 模型: {task.model_id}</p>
                <p>时间: {task.created_at} → {task.completed_at}</p>
            </div>
            
            <h2>📈 指标概览</h2>
        """
        
        for metric, stats in summary.items():
            if metric == "latency":
                continue
            mean_val = stats["mean"]
            color_class = "good" if mean_val > 0.8 else "bad" if mean_val < 0.6 else ""
            html += f"""
            <div class="metric-card">
                <div>{metric}</div>
                <div class="metric-value {color_class}">{mean_val:.4f}</div>
                <div>min={stats['min']:.4f} max={stats['max']:.4f} std={stats['std']:.4f}</div>
            </div>
            """
        
        if "latency" in summary:
            lat = summary["latency"]
            html += f"""
            <h2>⏱️ 延迟统计</h2>
            <table>
                <tr><th>指标</th><th>值</th></tr>
                <tr><td>平均延迟</td><td>{lat['mean']:.1f} ms</td></tr>
                <tr><td>P50</td><td>{lat['p50']:.1f} ms</td></tr>
                <tr><td>P95</td><td>{lat['p95']:.1f} ms</td></tr>
                <tr><td>P99</td><td>{lat['p99']:.1f} ms</td></tr>
            </table>
            """
        
        html += "</body></html>"
        return html

# ---------- 使用示例 ----------

async def main():
    # 初始化引擎
    engine = EvalEngine()
    
    # 注册指标
    def accuracy_score(input_text, expected, actual, context):
        return 1.0 if expected.lower().strip() in actual.lower() else 0.0
    
    def relevancy_score(input_text, expected, actual, context):
        # 简化：基于长度比估算相关性
        if not actual:
            return 0.0
        return min(1.0, len(actual) / max(len(expected), 1))
    
    engine.register_metric("accuracy", accuracy_score, MetricType.AUTOMATED)
    engine.register_metric("relevancy", relevancy_score, MetricType.AUTOMATED)
    
    # 注册模型
    def my_model(input_text, context):
        return f"这是模型对'{input_text}'的回答。"
    
    engine.register_model("my-llm-v1", my_model)
    
    # 创建数据集
    dataset_id = engine.create_dataset("基础问答测试", [
        {"id": "q1", "input": "1+1等于几？", "expected_output": "2"},
        {"id": "q2", "input": "中国的首都是哪里？", "expected_output": "北京"},
        {"id": "q3", "input": "太阳从哪个方向升起？", "expected_output": "东方"},
    ])
    
    # 创建并运行评测
    task = EvalTask(
        name="基线评测-v1",
        model_id="my-llm-v1",
        dataset_id=dataset_id,
        metrics=["accuracy", "relevancy"],
    )
    
    results = await engine.run_evaluation(task)
    print(json.dumps(results["summary"], indent=2, ensure_ascii=False))
    
    # 生成报告
    reporter = ReportGenerator()
    html = reporter.generate_html_report(task)
    with open("eval_report.html", "w") as f:
        f.write(html)
    print("✅ 报告已生成: eval_report.html")

# 运行
asyncio.run(main())

10.3 平台扩展建议

与 CI/CD 集成：每次 Prompt 或模型变更自动触发评测 告警机制：质量指标低于阈值时自动通知 评测数据飞轮：收集线上 bad case 自动加入评测集 多环境支持：开发/预发/生产环境独立评测 权限管理：不同角色查看不同维度的评测结果

11. 总结与最佳实践

评测体系搭建清单

□ 基础设施
  □ 选定评测框架（DeepEval / RAGAS / 自建）
  □ 搭建评测数据集管理系统
  □ 集成 CI/CD 自动评测流水线
  
□ 指标体系
  □ 定义核心指标（准确性、忠实度、安全性）
  □ 定义扩展指标（相关性、检索质量、延迟）
  □ 设定各指标阈值和回归容忍度
  
□ 数据集
  □ 构建初始评测集（100+ 用例，覆盖各场景）
  □ 建立数据集版本管理
  □ 制定数据集更新流程
  
□ 运营机制
  □ 每次模型/Prompt 变更触发自动评测
  □ 每周生成评测趋势报告
  □ 每月审视评测指标体系是否需要调整
  □ 建立 bad case 收集和修复流程

常见陷阱

只看平均分：平均分掩盖了长尾问题，要看分布和边界情况
评测集不更新：评测集过时会让评测失去意义
忽略安全性：只关注功能指标，忽视毒性、偏见、隐私
过度依赖自动评测：LLM-as-Judge 有偏见，需要定期人类校验
评测与生产脱节：评测场景应尽量覆盖真实用户使用场景

从 0 到 1 的实施路径

阶段 1（1-2 周）: 基础评测
  - 选择一个评测框架
  - 构建 50 条核心评测用例
  - 实现基本的自动化评测脚本

阶段 2（2-4 周）: 流水线集成
  - 将评测集成到 CI/CD
  - 添加回归检测和告警
  - 扩展评测集到 200+ 条

阶段 3（1-2 月）: 评测深化
  - 接入 RAGAS/Agent 评测
  - 搭建 A/B 测试框架
  - 建立人类评测流程

阶段 4（持续）: 评测飞轮
  - 线上 bad case 自动回流
  - 评测指标与业务指标关联
  - 构建评测可视化 dashboard
  - 建立模型评测排行榜

参考资源

RAGAS: https://github.com/explodinggradients/ragas

DeepEval: https://github.com/confident-ai/deepeval

lm-evaluation-harness: https://github.com/EleutherAI/lm-evaluation-harness

OpenCompass: https://github.com/open-compass/opencompass

LMSYS Chatbot Arena: https://chat.lmsys.org

HELM: https://crfm.stanford.edu/helm

Promptfoo: https://github.com/promptfoo/promptfoo

TruLens: https://github.com/truera/trulens

AI应用评测体系教程