AI AinoCode AI 工具与基础设施
AI教程 7 分钟

Eval-Driven Development 实战:用评估数据集驱动 Prompt/模型/架构迭代

从构建 Golden Dataset → CI 集成 → 回归检测 → 自动选优,搭建完整的 AI 质量门禁 pipeline,覆盖代码生成/客服问答/信息抽取三类任务的完整实践。

AinoCode 编辑部

Eval-Driven Development Pipeline 架构

Eval-Driven Development 实战:用评估数据集驱动 Prompt/模型/架构迭代

传统软件开发有单元测试兜底——代码改了跑一遍 pytest,绿了就合并。但 AI 应用开发呢?改了一行 prompt,效果变了还是差了,只能靠肉眼看输出。

这不是工程,这是玄学。

Eval-Driven Development(EDD) 的核心思路:把评测集(Golden Dataset)当成 AI 应用的”单元测试”,每次改 prompt/模型/架构,先跑评测,过了才能合并。

本文完整搭建一套 EDD Pipeline,覆盖:

  1. Golden Dataset 构建方法论
  2. 评测指标设计(不只是准确率)
  3. CI 集成与回归检测
  4. 自动选优策略
  5. 三类真实任务的完整评测方案

一、为什么需要 EDD

先看一个真实场景:

你的客服 Agent 上线后效果不错。某天同事改了 system prompt 里的一句话:

- "请用专业但友好的语气回答用户问题。"
+ "请简洁直接地回答用户问题,避免客套话。"

看起来只是语气微调。但在生产环境中:

  • 原本 92% 的用户满意度跌到 78%
  • 投诉率上升了 3 倍
  • 原因是”简洁直接”让一些需要安抚情绪的愤怒用户更生气了

如果有 Golden Dataset + 自动化评测,这个改动在 PR 阶段就会被拦截。

EDD 解决的正是这个问题:让 AI 应用的每一次变更都有数据支撑,可度量、可回滚。


二、Golden Dataset 构建方法论

2.1 数据来源

Golden Dataset 不是凭空写的测试用例,而是从真实生产数据中提取的。三种主要来源:

来源提取方式适用场景
历史工单人工标注 500+ 条代表性问答对客服/技术支持
代码 Review 记录从 Git log 提取 feature request + 对应实现代码生成
业务文档从产品手册/合同/报告中抽取结构化信息信息抽取

2.2 数据集规模

经验法则:每个任务类型至少 50-200 条标注样本。太少统计显著性不够,太多标注成本过高。

任务复杂度建议样本量标注成本(人时)
简单分类(意图识别)50-1002-4
问答/摘要100-2008-16
代码生成50-10010-20
多步推理/Agent30-5015-30

2.3 标注规范

每条样本至少包含:

from dataclasses import dataclass
from enum import Enum

class TaskType(str, Enum):
    CODE_GENERATION = "code_generation"
    QA = "qa"
    EXTRACTION = "extraction"

@dataclass
class GoldenSample:
    id: str                          # "qa_001"
    input: str                       # 输入(用户问题 / 需求描述)
    expected_output: str             # 期望输出
    rubric: dict                     # 评分标准(多维度)
    difficulty: str = "medium"       # easy/medium/hard
    tags: list[str] = None           # ["error_handling", "api_design"]
    metadata: dict = None            # 额外上下文

标注质量是 EDD 的生命线。几条铁律:

  • ✅ 每条样本必须有明确的评价标准(rubric),不是”对/错”二分
  • ✅ 至少 2 人交叉标注,不一致的讨论确认
  • ✅ 覆盖边界 case:恶意输入、模糊需求、多语言混合
  • ✅ 标注文档必须版本化,和评测集一起存 Git

2.4 三类任务的 Golden Dataset 示例

任务 A:代码生成(Python 函数)

{
    "id": "code_gen_012",
    "task_type": "code_generation",
    "input": "写一个函数,读取 CSV 文件中 'email' 列,验证格式,返回有效邮箱列表。要求:处理空值、去重、支持自定义分隔符。",
    "expected_output": """def extract_valid_emails(csv_path: str, delimiter: str = ',') -> list[str]:
    import csv, re
    pattern = re.compile(r'^[\\w.-]+@[\\w.-]+\\.\\w+$')
    valid = set()
    with open(csv_path, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f, delimiter=delimiter)
        for row in reader:
            email = row.get('email', '').strip()
            if email and pattern.match(email):
                valid.add(email.lower())
    return list(valid)""",
    "rubric": {
        "correctness": "函数能正确处理有效/无效邮箱",
        "edge_cases": "处理空值、去重、大小写",
        "api_design": "参数可配置分隔符",
        "type_hints": "有类型注解",
        "error_handling": "文件不存在时抛出合理异常"
    },
    "difficulty": "medium",
    "tags": ["csv", "validation", "regex"]
}

任务 B:客服问答

{
    "id": "qa_047",
    "task_type": "qa",
    "input": "我买了你们的 Pro 套餐,但现在用不上了,能退款吗?我是上周 3 号买的。",
    "expected_output": "您好!根据我们的退款政策,购买后 7 天内可申请全额退款。您的购买日期是本月 3 号,目前还在退款期内。请提供订单号,我将为您发起退款流程。退款将在 3-5 个工作日内原路返回。如需帮助,随时联系我。",
    "rubric": {
        "policy_accuracy": "正确引用 7 天退款政策",
        "empathy": "语气友好,体现理解",
        "actionable": "给出明确下一步(提供订单号)",
        "timeline": "告知退款时间"
    },
    "difficulty": "medium",
    "tags": ["refund", "policy", "empathy"]
}

任务 C:信息抽取

{
    "id": "extract_023",
    "task_type": "extraction",
    "input": "合同编号:HT-2026-0458。甲方:北京科技有限公司,统一社会信用代码:91110105MA01XXXX。乙方:上海贸易集团股份有限公司。合同金额:人民币壹佰伍拾万元整(¥1,500,000.00)。签订日期:2026年3月15日。有效期:自签订之日起一年。",
    "expected_output": {
        "contract_id": "HT-2026-0458",
        "party_a": {"name": "北京科技有限公司", "credit_code": "91110105MA01XXXX"},
        "party_b": {"name": "上海贸易集团股份有限公司"},
        "amount": {"currency": "CNY", "value": 1500000.00, "text": "壹佰伍拾万元整"},
        "sign_date": "2026-03-15",
        "validity": "1年"
    },
    "rubric": {
        "completeness": "所有字段均正确提取",
        "format": "金额转为数值型,日期标准化",
        "nested_structure": "甲乙方信息嵌套结构正确"
    },
    "difficulty": "hard",
    "tags": ["contract", "legal", "nested"]
}

三、评测指标设计

3.1 不要只用准确率

AI 输出不是非对即错的二进制判断。评测指标必须是多维度的

维度度量方式适用场景
正确性精确匹配 / 语义相似度 / LLM-as-Judge所有场景
完整性字段覆盖率 / 步骤覆盖率信息抽取、多步任务
格式合规JSON Schema 校验 / 正则匹配结构化输出
安全性Prompt Injection 检测 / 敏感词过滤所有场景
延迟P50/P95 响应时间生产环境
成本Token 消耗 × 单价生产环境
鲁棒性对抗样本通过率安全关键场景

3.2 LLM-as-Judge 的正确姿势

用 LLM 当裁判是目前最灵活的评测方式,但有三个陷阱必须避开:

陷阱 1:Position Bias

LLM 倾向于给先出现的输出更高分。

解决方案:AB 交叉评估——同一份样本跑两遍,交换 A/B 顺序,取平均。

def llm_judge_ab(input_text, output_a, output_b, rubric, judge_model):
    # 第一次:A 在前
    score_1 = judge(input_text, output_a, output_b, rubric)
    # 第二次:B 在前
    score_2 = judge(input_text, output_b, output_a, rubric)
    # 交换回来
    return {
        "score_a": (score_1["score_a"] + score_2["score_b"]) / 2,
        "score_b": (score_1["score_b"] + score_2["score_a"]) / 2
    }

陷阱 2:长度偏见

LLM 倾向于给更长的输出更高分。

解决方案:在 rubric 中明确”简洁性”权重,或归一化分数。

陷阱 3:自我偏好

用 GPT-4 评估 GPT-4 的输出会偏高。

解决方案:用不同模型做 Judge(如用 Claude 评估 GPT 输出)。

3.3 综合评分公式

def compute_score(rubric_scores: dict, weights: dict) -> float:
    """
    rubric_scores: {"correctness": 0.9, "completeness": 0.8, "format": 1.0}
    weights: {"correctness": 0.4, "completeness": 0.3, "format": 0.3}
    """
    assert set(rubric_scores.keys()) == set(weights.keys())
    return sum(rubric_scores[k] * weights[k] for k in weights)

四、CI 集成与回归检测

4.1 Pipeline 架构

┌──────────────────────────────────────────────────┐
│                   Git Push / PR                   │
└──────────────────────┬───────────────────────────┘


┌──────────────────────────────────────────────────┐
│  Step 1: 加载 Golden Dataset (JSONL)              │
└──────────────────────┬───────────────────────────┘


┌──────────────────────────────────────────────────┐
│  Step 2: 对每条样本运行当前版本的 Agent/Prompt     │
└──────────────────────┬───────────────────────────┘


┌──────────────────────────────────────────────────┐
│  Step 3: 多维度评分 (LLM-Judge + 规则校验)         │
└──────────────────────┬───────────────────────────┘


┌──────────────────────────────────────────────────┐
│  Step 4: 与 Baseline 版本对比(回归检测)          │
└──────────────────────┬───────────────────────────┘

              ┌────────┴────────┐
              ▼                 ▼
┌──────────────┐    ┌──────────────────┐
│  分数 ≥ 阈值  │    │  分数 < 阈值     │
│  CI Green ✅  │    │  CI Red ❌       │
│  允许合并     │    │  阻止合并        │
└──────────────┘    └──────────────────┘

4.2 完整 GitHub Actions 配置

# .github/workflows/eval.yml
name: AI Eval Pipeline

on:
  pull_request:
    paths:
      - 'prompts/**'
      - 'config/model_config.yaml'
      - 'src/agents/**'
      - '.github/workflows/eval.yml'

jobs:
  eval:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          
      - name: Install dependencies
        run: |
          pip install openai pytest pyyaml pytest-md
          
      - name: Run Eval Pipeline
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          CLAUDE_API_KEY: ${{ secrets.CLAUDE_API_KEY }}
        run: |
          python eval/run_eval.py \
            --dataset data/golden_dataset.jsonl \
            --config config/current_config.yaml \
            --baseline-config config/baseline_config.yaml \
            --output results/eval_report.json \
            --threshold 0.85 \
            --min-sample-pass-rate 0.80
          
      - name: Check Regression
        run: |
          python eval/check_regression.py \
            --current results/eval_report.json \
            --baseline results/baseline_report.json \
            --max-regression 0.05
          
      - name: Upload Report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: eval-report
          path: results/eval_report.json

4.3 评测引擎核心代码

# eval/run_eval.py
import json
import asyncio
from dataclasses import dataclass, asdict
from typing import List

@dataclass
class EvalResult:
    sample_id: str
    input: str
    output: str
    expected: str
    scores: dict  # {"correctness": 0.9, "completeness": 0.8}
    weighted_score: float
    latency_ms: int
    tokens_used: int
    cost_usd: float

async def evaluate_sample(sample: dict, config: dict) -> EvalResult:
    """评估单个样本"""
    import time
    from openai import AsyncOpenAI
    
    client = AsyncOpenAI(api_key=config["api_key"])
    start = time.time()
    
    response = await client.chat.completions.create(
        model=config["model"],
        messages=[
            {"role": "system", "content": config["system_prompt"]},
            {"role": "user", "content": sample["input"]}
        ],
        temperature=config.get("temperature", 0.0),
    )
    
    latency = int((time.time() - start) * 1000)
    output = response.choices[0].message.content
    tokens = response.usage.total_tokens
    cost = tokens * config.get("price_per_token", 0.0)
    
    # LLM-as-Judge 评分
    scores = await llm_judge(
        input_text=sample["input"],
        output=output,
        expected=sample["expected_output"],
        rubric=sample["rubric"],
        judge_model=config.get("judge_model", "claude-sonnet-4-20250514")
    )
    
    weights = config.get("weights", {"correctness": 0.4, "completeness": 0.3, "format": 0.3})
    weighted = sum(scores[k] * weights.get(k, 0) for k in scores)
    
    return EvalResult(
        sample_id=sample["id"],
        input=sample["input"],
        output=output,
        expected=sample["expected_output"],
        scores=scores,
        weighted_score=round(weighted, 3),
        latency_ms=latency,
        tokens_used=tokens,
        cost_usd=round(cost, 6)
    )

async def run_eval(dataset_path: str, config_path: str) -> List[EvalResult]:
    with open(dataset_path) as f:
        samples = [json.loads(line) for line in f]
    with open(config_path) as f:
        config = json.load(f)
    
    results = []
    # 并发控制:避免 API 限流
    semaphore = asyncio.Semaphore(10)
    
    async def bounded_eval(s):
        async with semaphore:
            return await evaluate_sample(s, config)
    
    tasks = [bounded_eval(s) for s in samples]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # 过滤异常
    valid_results = [r for r in results if isinstance(r, EvalResult)]
    return valid_results

4.4 回归检测

# eval/check_regression.py
import json
import sys

def check_regression(current_path: str, baseline_path: str, 
                     max_regression: float = 0.05) -> bool:
    """
    检查当前版本相比 baseline 是否有显著退化。
    max_regression: 允许的最大退化幅度(如 0.05 = 5%)
    """
    with open(current_path) as f:
        current = json.load(f)
    with open(baseline_path) as f:
        baseline = json.load(f)
    
    current_avg = current["mean_weighted_score"]
    baseline_avg = baseline["mean_weighted_score"]
    
    delta = current_avg - baseline_avg
    
    print(f"Baseline: {baseline_avg:.3f}")
    print(f"Current:  {current_avg:.3f}")
    print(f"Delta:    {delta:+.3f}")
    
    # 分维度检查
    regressions = []
    for dim in baseline["mean_scores_by_dimension"]:
        b = baseline["mean_scores_by_dimension"][dim]
        c = current["mean_scores_by_dimension"].get(dim, 0)
        d = c - b
        if d < -max_regression:
            regressions.append(f"  {dim}: {b:.3f}{c:.3f} ({d:+.3f})")
    
    if delta < -max_regression:
        print(f"❌ REGRESSION DETECTED: 总体下降 {delta:.3f} (阈值: -{max_regression})")
        for r in regressions:
            print(r)
        return False
    else:
        print(f"✅ No significant regression. Delta: {delta:+.3f}")
        return True

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--current", required=True)
    parser.add_argument("--baseline", required=True)
    parser.add_argument("--max-regression", type=float, default=0.05)
    args = parser.parse_args()
    
    ok = check_regression(args.current, args.baseline, args.max_regression)
    sys.exit(0 if ok else 1)

五、自动选优策略

当同时测试多个 prompt 变体或模型时,EDD pipeline 可以自动选出最优版本

5.1 多版本并行评测

# eval/ab_test.py
import json
import asyncio
from typing import List, Dict

async def ab_test_variants(
    dataset_path: str,
    variants: Dict[str, dict]  # {"v1": config_a, "v2": config_b}
) -> dict:
    """
    对多个 prompt/model 变体并行评测,自动选出最优。
    """
    results = {}
    for name, config in variants.items():
        print(f"Evaluating variant: {name}")
        eval_results = await run_eval(dataset_path, config)
        results[name] = summarize(eval_results)
    
    # 排序
    ranking = sorted(
        results.items(),
        key=lambda x: x[1]["mean_weighted_score"],
        reverse=True
    )
    
    winner = ranking[0]
    print(f"\n🏆 Winner: {winner[0]} (score: {winner[1]['mean_weighted_score']:.3f})")
    
    for name, stats in ranking:
        print(f"  {name}: {stats['mean_weighted_score']:.3f} | "
              f"cost: ${stats['mean_cost']:.4f} | "
              f"latency: {stats['mean_latency_ms']}ms")
    
    return {"ranking": ranking, "winner": winner[0]}

5.2 选优决策树

不是分数高就一定选——需要考虑综合因素:

def select_best(results: dict, priority: str = "balanced") -> str:
    """
    priority: "quality" | "cost" | "balanced" | "latency"
    """
    ranked = results["ranking"]
    
    if priority == "quality":
        return ranked[0][0]  # 纯看分数
    
    elif priority == "cost":
        return min(ranked, key=lambda x: x[1]["mean_cost"])[0]
    
    elif priority == "balanced":
        # 分数在最优的 95% 以内且成本最低的
        best_score = ranked[0][1]["mean_weighted_score"]
        threshold = best_score * 0.95
        candidates = [(n, s) for n, s in ranked 
                      if s["mean_weighted_score"] >= threshold]
        return min(candidates, key=lambda x: x[1]["mean_cost"])[0]
    
    elif priority == "latency":
        best_score = ranked[0][1]["mean_weighted_score"]
        threshold = best_score * 0.95
        candidates = [(n, s) for n, s in ranked 
                      if s["mean_weighted_score"] >= threshold]
        return min(candidates, key=lambda x: x[1]["mean_latency_ms"])[0]

六、三类任务的完整评测方案

6.1 代码生成任务

Golden Dataset: 50 个 LeetCode Easy/Medium + 30 个真实业务函数需求

评测维度

维度检测方法权重
编译通过实际编译/解释执行0.3
测试通过运行预编写的单元测试0.3
代码风格pylint/ruff 检查0.1
边界处理特定边界输入测试0.15
复杂度AST 分析循环嵌套深度0.05
文档是否有 docstring0.1

关键技巧:对生成的代码必须实际运行,不能只靠 LLM-as-Judge。代码对错是二元的,编译器说了算。

6.2 客服问答任务

Golden Dataset: 200 条真实工单(覆盖咨询/投诉/退款/技术支持)

评测维度

维度检测方法权重
政策准确性规则校验(退款天数、套餐价格等)0.3
共情表达LLM-Judge 评估语气0.2
行动指引检查是否包含下一步操作0.2
安全性敏感信息泄露检测0.15
语言流畅度Perplexity 或 LLM-Judge0.15

6.3 信息抽取任务

Golden Dataset: 100 份合同/发票/报告,人工标注完整结构化字段

评测维度

维度检测方法权重
字段覆盖率提取字段数 / 期望字段数0.3
精确匹配率字段值完全正确的比例0.3
模糊匹配率语义等价但表述不同的比例0.15
JSON Schema 合规jsonschema.validate()0.15
嵌套结构正确性递归比较嵌套字段0.1

七、踩坑记录

坑 1:Golden Dataset 老化

现象:评测分数很高,但线上效果越来越差。

原因:产品政策变了、新增了业务场景,但 Golden Dataset 没更新。

解法

  • 每月从生产环境随机采样 50 条新工单,人工审核后加入数据集
  • 设置”数据新鲜度”指标——最近 30 天新增样本占比 ≥ 15%
  • 当线上投诉率突然上升时,立即分析新样本并补充到数据集

坑 2:LLM-as-Judge 不一致

现象:同一份输出,跑两次 Judge 分数差 0.2。

解法

  • 每个样本跑 3 次 Judge,取中位数
  • 设置 Judge 的 temperature=0.0
  • 在 rubric 中加入具体判断标准,减少主观性

坑 3:CI 超时

现象:200 条样本的评测跑 25 分钟,GitHub Actions 超时。

解法

  • 分层评测:PR 阶段只跑 50 条核心样本(5 分钟),merge 后跑全量
  • 提高并发:Semiconductor 调到 20,配合 API rate limit
  • 缓存 embedding:如果评测中包含 embedding 相似度计算,结果可缓存

坑 4:过拟合 Golden Dataset

现象:针对评测集手动调 prompt,分数涨了,但线上效果没变。

解法

  • 数据集分 train/eval/test 三组,prompt 调优只用 train
  • 留 20% 样本作为 hold-out set,永远不用于调优
  • 引入”对抗样本”——专门设计 prompt 无法轻易答对的 edge case

八、总结

EDD 不是银弹,但它是 AI 应用从”能跑”到”可靠”的必经之路。

核心原则

  1. 评测集必须来自真实数据,不是凭空编造的示例
  2. 多维度评分,不要只看一个”准确率”数字
  3. CI 集成是底线——没有自动化评测的 AI 代码变更,不应该允许合并
  4. Golden Dataset 是活文档,需要持续维护更新
  5. 自动选优不是只看分数,综合成本/延迟/质量做决策

这套 Pipeline 我们已经用在三个生产项目上,效果最明显的是:prompt 变更引发的线上事故从每月 3-4 次降到近 6 个月 0 次。


本文的评测 Pipeline 模板和 Golden Dataset 示例已开源,详见 GitHub 仓库。