为什么需要评估流水线

大语言模型迭代速度极快，每次模型更新或 Prompt 修改都需要回答一个核心问题：新的版本到底比旧的好多少？ 靠人工试几个案例远远不够，你需要一条系统化的评估流水线。

评估流水线的核心价值：

  • 可复现：同一套数据集和指标，任何人任何时候跑都能得到一致结果
  • 可比较：不同模型版本之间的差异被量化为具体数字
  • 可扩展：从 100 条测试用例扩展到 10000 条只需改一个参数
  • 可追踪：历史评估结果存档，形成模型演进的时间线

流水线架构总览

一条完整的 LLM 评估流水线包含五个核心阶段：

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  数据集构建  │ -> │  评估执行    │ -> │  指标计算    │ -> │  结果分析    │ -> │  报告生成    │
│  Dataset    │    │  Execution  │    │  Metrics    │    │  Analysis   │    │  Report     │
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘

阶段一：数据集构建

数据集是评估的基石。一个高质量的评估数据集应具备以下特征：

特征说明示例
代表性覆盖实际使用场景的主要类型问答、摘要、翻译、代码生成
多样性包含不同难度和长度的输入简单事实题 vs 多步推理题
无泄漏不包含训练数据中的内容使用人工新写的题目
可验证有标准答案或明确的评判标准精确匹配 / 人工评分标准
可扩展能方便地增加新类别模块化的数据结构

数据集格式设计

推荐使用 JSONL 格式，每行一个测试样本：

{
  "id": "qa_001",
  "category": "knowledge_qa",
  "difficulty": "easy",
  "input": "光合作用的化学方程式是什么？",
  "expected_output": "6CO₂ + 6H₂O → C₆H₁₂O₆ + 6O₂（在光照条件下）",
  "eval_method": "semantic_match",
  "metadata": {
    "language": "zh",
    "domain": "biology",
    "tags": ["chemistry", "photosynthesis"]
  }
}

数据集构建脚本

import json
import random
from pathlib import Path

class EvalDatasetBuilder:
    def __init__(self, name: str):
        self.name = name
        self.samples = []

    def add_sample(self, sample: dict):
        required_fields = {"id", "category", "input", "expected_output", "eval_method"}
        if not required_fields.issubset(sample.keys()):
            raise ValueError(f"Missing required fields: {required_fields - set(sample.keys())}")
        self.samples.append(sample)

    def add_batch(self, samples: list[dict]):
        for s in samples:
            self.add_sample(s)

    def split(self, ratios: dict = None) -> dict:
        """按比例拆分数据集"""
        if ratios is None:
            ratios = {"test": 0.8, "dev": 0.1, "heldout": 0.1}
        total = len(self.samples)
        shuffled = random.sample(self.samples, total)
        result = {}
        start = 0
        for name, ratio in ratios.items():
            end = start + int(total * ratio)
            result[name] = shuffled[start:end]
            start = end
        return result

    def save(self, output_dir: str):
        path = Path(output_dir) / f"{self.name}.jsonl"
        path.parent.mkdir(parents=True, exist_ok=True)
        with open(path, "w", encoding="utf-8") as f:
            for sample in self.samples:
                f.write(json.dumps(sample, ensure_ascii=False) + "\n")
        print(f"Saved {len(self.samples)} samples to {path}")

    def stats(self):
        from collections import Counter
        cats = Counter(s["category"] for s in self.samples)
        diffs = Counter(s.get("difficulty", "unknown") for s in self.samples)
        print(f"Dataset: {self.name}")
        print(f"Total samples: {len(self.samples)}")
        print(f"Categories: {dict(cats)}")
        print(f"Difficulty distribution: {dict(diffs)}")

阶段二：评估执行

评估执行阶段负责将数据集送入模型并收集输出。核心设计要点是并发控制和错误重试。

import asyncio
import aiohttp
import time
from dataclasses import dataclass, field

@dataclass
class EvalConfig:
    model_name: str = "gpt-4o"
    api_base: str = "https://api.openai.com/v1"
    max_concurrent: int = 10
    max_retries: int = 3
    retry_delay: float = 1.0
    timeout: float = 60.0
    temperature: float = 0.0  # 评估时使用确定性输出

@dataclass
class EvalResult:
    sample_id: str
    input: str
    expected: str
    actual: str
    latency_ms: float
    success: bool
    error: str = ""

class LLMEvaluator:
    def __init__(self, config: EvalConfig):
        self.config = config
        self.semaphore = asyncio.Semaphore(config.max_concurrent)

    async def query_model(self, session: aiohttp.ClientSession, prompt: str) -> tuple[str, float]:
        headers = {"Authorization": f"Bearer {self.config.api_key}"}
        payload = {
            "model": self.config.model_name,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": self.config.temperature,
        }
        start = time.perf_counter()
        async with session.post(f"{self.config.api_base}/chat/completions",
                                json=payload, headers=headers,
                                timeout=aiohttp.ClientTimeout(total=self.config.timeout)) as resp:
            data = await resp.json()
            latency = (time.perf_counter() - start) * 1000
            return data["choices"][0]["message"]["content"], latency

    async def evaluate_single(self, session, sample: dict) -> EvalResult:
        async with self.semaphore:
            for attempt in range(self.config.max_retries):
                try:
                    output, latency = await self.query_model(session, sample["input"])
                    return EvalResult(
                        sample_id=sample["id"],
                        input=sample["input"],
                        expected=sample["expected_output"],
                        actual=output,
                        latency_ms=latency,
                        success=True,
                    )
                except Exception as e:
                    if attempt == self.config.max_retries - 1:
                        return EvalResult(
                            sample_id=sample["id"],
                            input=sample["input"],
                            expected=sample["expected_output"],
                            actual="",
                            latency_ms=0,
                            success=False,
                            error=str(e),
                        )
                    await asyncio.sleep(self.config.retry_delay * (2 ** attempt))

    async def run(self, dataset: list[dict]) -> list[EvalResult]:
        async with aiohttp.ClientSession() as session:
            tasks = [self.evaluate_single(session, s) for s in dataset]
            results = await asyncio.gather(*tasks)
            return list(results)

阶段三：指标计算

不同任务需要不同的评估指标，以下是常用指标的分类：

指标类别适用场景代表指标特点
精确匹配事实问答、数学计算Exact Match, Accuracy二值判断，简单直接
文本相似度摘要、翻译BLEU, ROUGE, BERTScore基于词重叠或语义嵌入
语义匹配开放问答、创意写作Embedding Cosine, LLM-as-Judge更灵活但计算成本高
结构化代码生成、JSON 输出AST Match, Schema Validation语法正确性验证
多维度综合能力评估LLM-as-Judge (多维 rubric)可定制评分标准

指标计算引擎

from abc import ABC, abstractmethod
import numpy as np

class Metric(ABC):
    @abstractmethod
    def compute(self, predicted: str, reference: str) -> float:
        pass

class ExactMatch(Metric):
    def compute(self, predicted: str, reference: str) -> float:
        return 1.0 if predicted.strip().lower() == reference.strip().lower() else 0.0

class ContainsAnswer(Metric):
    def compute(self, predicted: str, reference: str) -> float:
        return 1.0 if reference.strip().lower() in predicted.strip().lower() else 0.0

class BertScore(Metric):
    def __init__(self, model_name: str = "bert-base-chinese"):
        from transformers import AutoTokenizer, AutoModel
        import torch
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModel.from_pretrained(model_name)
        self.model.eval()

    def compute(self, predicted: str, reference: str) -> float:
        import torch
        with torch.no_grad():
            p_inputs = self.tokenizer(predicted, return_tensors="pt", truncation=True, max_length=512)
            r_inputs = self.tokenizer(reference, return_tensors="pt", truncation=True, max_length=512)
            p_emb = self.model(**p_inputs).last_hidden_state.mean(dim=1)
            r_emb = self.model(**r_inputs).last_hidden_state.mean(dim=1)
            cosine = torch.nn.functional.cosine_similarity(p_emb, r_emb)
            return cosine.item()

class LLMAsJudge(Metric):
    """使用另一个 LLM 作为评判者"""
    def __init__(self, judge_model: str, rubric: str):
        self.judge_model = judge_model
        self.rubric = rubric

    def compute(self, predicted: str, reference: str) -> float:
        prompt = f"""请根据以下评分标准对回答进行打分（0-10分）：
评分标准：{self.rubric}
参考答案：{reference}
待评回答：{predicted}
请只输出一个数字（0-10），表示得分。"""
        # 调用 judge_model 获取分数
        score_str = call_llm(self.judge_model, prompt)
        try:
            return float(score_str.strip()) / 10.0
        except ValueError:
            return 0.0

class MetricEngine:
    def __init__(self):
        self.metrics: dict[str, Metric] = {}

    def register(self, name: str, metric: Metric):
        self.metrics[name] = metric

    def evaluate(self, results: list[EvalResult], metric_name: str) -> dict:
        metric = self.metrics[metric_name]
        scores = []
        for r in results:
            if r.success:
                scores.append(metric.compute(r.actual, r.expected))
        return {
            "metric": metric_name,
            "count": len(scores),
            "mean": np.mean(scores) if scores else 0,
            "std": np.std(scores) if scores else 0,
            "min": np.min(scores) if scores else 0,
            "max": np.max(scores) if scores else 0,
            "p50": np.percentile(scores, 50) if scores else 0,
            "p95": np.percentile(scores, 95) if scores else 0,
        }

阶段四：结果分析

评估结果不能只看平均分。深入的分析需要关注：

  1. 分类别表现：模型在哪些类别上强/弱？
  2. 难度分布：简单题全对但难题全错？还是均匀分布？
  3. 错误模式：失败的案例有什么共性？
  4. 延迟分布：P50/P95/P99 延迟分别是多少？
import pandas as pd

class ResultAnalyzer:
    def __init__(self, results: list[EvalResult], dataset: list[dict]):
        self.df = pd.DataFrame([
            {
                "id": r.sample_id,
                "category": next((s["category"] for s in dataset if s["id"] == r.sample_id), "unknown"),
                "difficulty": next((s.get("difficulty", "unknown") for s in dataset if s["id"] == r.sample_id), "unknown"),
                "success": r.success,
                "latency_ms": r.latency_ms,
                "actual": r.actual,
                "expected": r.expected,
            } for r in results
        ])

    def category_breakdown(self) -> pd.DataFrame:
        return self.df.groupby("category").agg(
            total=("id", "count"),
            success_rate=("success", "mean"),
            avg_latency=("latency_ms", "mean"),
            p95_latency=("latency_ms", lambda x: x.quantile(0.95)),
        ).round(3)

    def difficulty_breakdown(self) -> pd.DataFrame:
        return self.df.groupby("difficulty").agg(
            total=("id", "count"),
            success_rate=("success", "mean"),
        ).round(3)

    def latency_stats(self) -> dict:
        lat = self.df[self.df["success"]]["latency_ms"]
        return {
            "mean_ms": round(lat.mean(), 1),
            "p50_ms": round(lat.median(), 1),
            "p95_ms": round(lat.quantile(0.95), 1),
            "p99_ms": round(lat.quantile(0.99), 1),
        }

    def failed_samples(self) -> pd.DataFrame:
        return self.df[~self.df["success"]][["id", "category", "actual", "expected"]]

阶段五：报告生成

最终报告应该同时面向技术人员和决策者。推荐使用 HTML 格式，包含可视化图表。

import matplotlib.pyplot as plt
import base64
from datetime import datetime

class ReportGenerator:
    def __init__(self, analyzer: ResultAnalyzer, model_name: str):
        self.analyzer = analyzer
        self.model_name = model_name
        self.timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")

    def generate_category_chart(self) -> str:
        cat_df = self.analyzer.category_breakdown()
        fig, ax = plt.subplots(figsize=(10, 5))
        cat_df["success_rate"].plot(kind="bar", ax=ax, color="steelblue")
        ax.set_title(f"{self.model_name} - 各类别成功率")
        ax.set_ylabel("成功率")
        ax.set_ylim(0, 1)
        plt.tight_layout()
        path = f"/tmp/eval_report_{self.model_name}.png"
        plt.savefig(path)
        plt.close()
        return path

    def generate_html(self, metric_results: list[dict]) -> str:
        cat_breakdown = self.analyzer.category_breakdown().to_html()
        diff_breakdown = self.analyzer.difficulty_breakdown().to_html()
        latency = self.analyzer.latency_stats()
        chart_path = self.generate_category_chart()

        html = f"""<!DOCTYPE html>
<html>
<head><meta charset="utf-8"><title>评估报告 - {self.model_name}</title></head>
<body>
<h1>LLM 评估报告</h1>
<p>模型: {self.model_name} | 时间: {self.timestamp}</p>

<h2>核心指标</h2>
<table>
<tr><th>指标</th><th>样本数</th><th>均值</th><th>标准差</th><th>P50</th><th>P95</th></tr>
"""
        for m in metric_results:
            html += f"""<tr>
<td>{m['metric']}</td><td>{m['count']}</td><td>{m['mean']:.4f}</td>
<td>{m['std']:.4f}</td><td>{m['p50']:.4f}</td><td>{m['p95']:.4f}</td>
</tr>"""
        html += f"""</table>

<h2>延迟统计</h2>
<p>平均: {latency['mean_ms']}ms | P50: {latency['p50_ms']}ms |
P95: {latency['p95_ms']}ms | P99: {latency['p99_ms']}ms</p>

<h2>分类别表现</h2>
{cat_breakdown}

<h2>难度分布</h2>
{diff_breakdown}

<h2>可视化</h2>
<img src="file://{chart_path}" width="800"/>

</body></html>"""
        return html

CI/CD 集成

评估流水线应集成到 CI/CD 中，每次模型或 Prompt 变更自动触发评估：

# .github/workflows/eval.yml
name: LLM Evaluation Pipeline
on:
  pull_request:
    paths: ["prompts/**", "models/**"]
  schedule:
    - cron: "0 2 * * *"  # 每天凌晨2点执行

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run evaluation
        run: python -m eval_pipeline --config configs/eval.yaml
      - name: Upload report
        uses: actions/upload-artifact@v4
        with:
          name: eval-report
          path: reports/
      - name: Compare with baseline
        run: python -m eval_compare --current reports/latest.json --baseline reports/baseline.json

关键实践总结

实践说明收益
版本化数据集数据集用 Git 管理，每次变更有 diff评估结果可追溯
温度设为 0评估时使用 temperature=0结果可复现
多次运行取均值对非确定性模型运行 3-5 次减少随机性影响
分桶分析至少按类别和难度两个维度分析避免平均分掩盖短板
设立基线维护一个基线模型的结果作为参照新版本有比较对象
自动告警指标下降超过阈值时触发告警及时发现问题

结语

评估流水线不是一次性工程，而是持续演进的基础设施。随着模型能力提升和业务场景扩展，数据集要更新，指标要迭代，分析维度要深化。把评估流水线当作产品的"质检车间"来建设，才能让每一次模型迭代都有据可依。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。