← 返回
IT技术

如何构建自评估AI系统:LLM应用的自动化测试与评估流水线

✍️ zhirenhun 📅 2026/9/13 👁 88 阅读 ⏱ 94 分钟
如何构建自评估AI系统:LLM应用的自动化测试与评估流水线

你已经把 AI 功能发布了,在演示中运行良好。团队对此印象深刻。随后,用户提出了一个略微超出你测试用例的问题,模型却自信地返回了完全错误的答案。

使用大型语言模型构建系统的真相是,传统的软件测试方法在这里失效了。当系统每次运行都会生成不同的文本时,你无法仅仅写一个简单的 assert output == expected 来验证。

市面上大多数教程会教你如何搭建聊天机器人或接入 RAG 流水线,然后就戛然而止了。他们说“部署到生产环境”,仿佛艰难部分已经过去。但实际上,难点在于判断你的 AI 是否真的有效,并在它开始变差时及时发现。

在这篇文章中,我将带你一步步构建完整的评估流水线。我们还将介绍三种在不同成本和深度层次上适用的评估策略。

What We'll Cover:

What You'll Need

要跟随学习,你需要安装 Python 3.10+,并具备一些调用 LLM API 的基本经验。无论你使用的是 OpenAI、Anthropic 还是本地模型,评估模式都是相同的。

你还需要一个 OpenAI API key 来运行 LLM-as-judge 示例(我们使用 gpt-4o-mini,因为它成本低且足够用于评分)。

如果你已经有一个想要评估的 LLM 驱动的应用,哪怕只是一个小 demo,那就再好不过了。如果没有,示例是自包含的,你仍然可以完全跟随学习。

在这里获取依赖:

pip install openai numpy pandas scikit-learn python-dotenv

传统测试为何在LLM应用中失效

如果你曾为普通软件编写过测试,你就会知道这个套路。函数输入,得到输出,你断言它们匹配。简洁、直接,完成。

但LLM彻底打破了这种模式。不仅如此,问题不仅是单一的,而是多种因素相互叠加。

首先,输出不是确定的。你可以两次发送完全相同的提示,得到不同的措辞。即使将 temperature=0 也无法完全避免,因为模型提供方会在后台更新模型。同一次API调用在一月和三月之间可能表现不同。

其次,没有唯一的正确答案。如果你的应用是对文档进行摘要,什么才算正确的摘要?两个人可能会写出不同的摘要,但两者都可能是很好的。你无法通过 assertEqual 来解决这个问题。

第三,问题不会明显爆炸,日志中也没有错误、崩溃或红色警告。模型只是悄悄返回一个看起来很自信却错误的答案。你的正常运行时间仪表板显示100%,而用户却在收到胡言乱语。这就是LLM失效时最让人头疼的地方。

因此,你不能像测试REST API那样测试LLM应用。你需要的是评分而非简单的通过/失败。你需要对输出批次进行评估,而不是逐条检查。此外,你需要一个持续运行的机制,因为即使你没有改动一行代码,质量也可能随时间漂移。

LLM评估的三层结构

经过大量试错后,我形成了一种方法:将三层评估从廉价快速到昂贵彻底进行堆叠。

  1. 第一层是确定性检查。 可以把这些看作门口的保安。输出在应为JSON时是否是有效的JSON?它是否可疑地短或荒谬地长?它是否包含幻觉生成的URL?这些检查即时、免费,并且能捕获比你预期更多的问题。

  2. 第二层是LLM作为评判者。 这里的做法是使用一次独立的LLM调用来为主LLM的输出打分。“这个答案相关吗?准确吗?实际上有帮助吗?”像 gpt-4o-mini 这样的模型在给出明确评分标准的情况下,竟然能很好地为其他模型的工作打分。

  3. 第三层是人工评估。 真实的人审查真实的输出。你不会对每个响应都这样做,因为那样会异常缓慢。但你会定期进行,以确保你的自动化层没有偏离真正意义上的“好”。

关键在于知道何时使用哪一层,我们将逐个构建它们。

如何构建第一层:确定性检查

当我刚开始构建评估管道时,我直接跳过了基础,去玩花哨的东西:LLM 评判、嵌入相似度得分,应有尽有。与此同时,我的应用偶尔会返回完全空的字符串,而我竟然两周都没发现。两周!

这就是为什么我现在在每个项目一开始就加入确定性检查。它们极其简单:不需要机器学习,也不需要 API 调用,只用纯 Python 对输出提出基本的健全性问题。它存在吗?格式是否正确?是否异常短?模型是否幻觉出了一个 URL?

你可能会觉得这些检查太基础,根本不起作用。我当初也是这么想的。直到我把它们跑在一个月的生产日志上,发现我之前漏掉的错误输出中,大约有三分之一其实可以用五分钟就能写出来的检查捕获到。

下面是我现在会在每个项目第一天就加入的 DeterministicEvaluator 类:

import json
import re
from dataclasses import dataclass


@dataclass
class EvalResult:
    """Holds the result of a single evaluation check."""
    check_name: str
    passed: bool
    score: float  # 0.0 to 1.0
    details: str


class DeterministicEvaluator:
    """Layer 1: Fast, rule-based checks for LLM outputs."""

    def check_json_validity(self, output: str) -> EvalResult:
        """Verify the output is valid JSON when JSON is expected."""
        try:
            json.loads(output)
            return EvalResult("json_validity", True, 1.0, "Valid JSON")
        except json.JSONDecodeError as e:
            return EvalResult("json_validity", False, 0.0, f"Invalid JSON: {e}")

    def check_length_bounds(
        self, output: str, min_chars: int = 10, max_chars: int = 5000
    ) -> EvalResult:
        """Check that output length falls within acceptable bounds."""
        length = len(output)
        if length < min_chars:
            return EvalResult(
                "length_bounds", False, 0.0,
                f"Too short: {length} chars (minimum: {min_chars})"
            )
        if length > max_chars:
            return EvalResult(
                "length_bounds", False, 0.0,
                f"Too long: {length} chars (maximum: {max_chars})"
            )
        return EvalResult("length_bounds", True, 1.0, f"Length OK: {length} chars")

    def check_no_hallucinated_links(self, output: str) -> EvalResult:
        """Detect URLs in output that the model may have fabricated."""
        url_pattern = r'https?://[^\s\)\]\}\"\'<>]+'
        urls = re.findall(url_pattern, output)
        if urls:
            return EvalResult(
                "no_hallucinated_links", False, 0.0,
                f"Found {len(urls)} URLs that may be hallucinated: {urls[:3]}"
            )
        return EvalResult("no_hallucinated_links", True, 1.0, "No URLs found")

    def check_required_sections(
        self, output: str, required: list[str]
    ) -> EvalResult:
        """Verify that required sections or keywords appear in the output."""
        missing = [s for s in required if s.lower() not in output.lower()]
        if missing:
            score = 1.0 - (len(missing) / len(required))
            return EvalResult(
                "required_sections", False, score,
                f"Missing sections: {missing}"
            )
        return EvalResult("required_sections", True, 1.0, "All sections present")

    def check_no_refusal(self, output: str) -> EvalResult:
        """Detect if the model refused to answer when it should not have."""
        refusal_phrases = [
            "i cannot", "i can't", "i'm unable to", "as an ai",
            "i don't have access", "i'm not able to"
        ]
        output_lower = output.lower()
        for phrase in refusal_phrases:
            if phrase in output_lower:
                return EvalResult(
                    "no_refusal", False, 0.0,
                    f"Possible refusal detected: '{phrase}'"
                )
        return EvalResult("no_refusal", True, 1.0, "No refusal detected")

    def run_all(self, output: str, config: dict = None) -> list[EvalResult]:
        """Run all deterministic checks and return results."""
        config = config or {}
        results = [
            self.check_length_bounds(
                output,
                config.get("min_chars", 10),
                config.get("max_chars", 5000)
            ),
            self.check_no_hallucinated_links(output),
            self.check_no_refusal(output),
        ]
        if config.get("expect_json"):
            results.append(self.check_json_validity(output))
        if config.get("required_sections"):
            results.append(
                self.check_required_sections(output, config["required_sections"])
            )
        return results


if __name__ == "__main__":
    evaluator = DeterministicEvaluator()

    # Test with a normal output
    good_output = "Python is a high-level programming language known for its readability."
    results = evaluator.run_all(good_output)
    for r in results:
        print(f"  {r.check_name}: {'PASS' if r.passed else 'FAIL'} ({r.details})")

    # Test with a suspicious output
    bad_output = "Visit https://fake-docs.example.com/api for more details."
    results = evaluator.run_all(bad_output)
    for r in results:
        print(f"  {r.check_name}: {'PASS' if r.passed else 'FAIL'} ({r.details})")

这些检查每个都能在毫秒级完成,而且完全免费。但别被简单表象迷惑——仅凭‘幻觉链接’这一项检查,就已经多次帮我避免把虚构的文档 URL 发送给用户,次数之多甚至让我自己都有点不好意思。

值得强调的是,这些只是起点。上面的通用检查适用于任何 LLM 应用,但真正的收益来自领域特定的检查。如果你的应用会生成 SQL,加入语法解析器;如果它负责起草邮件,检查是否包含主题行和问候语;如果它输出代码,则尝试用 linter 运行一下。

你在这里添加的每一项检查,都意味着下游昂贵环节(甚至直接面向用户)会少收到一次错误输出。

如何构建 Layer 2:LLM 作为评判者的评估

好了,假设你的输出已经通过了基本检查:它是有效的 JSON,长度合适,没有虚构的链接。但这里有一个问题,第一层无法回答:这个回答实际上是否 有用

一个输出可能结构完美、通过所有的确定性检查,却对阅读者完全毫无用处。‘法国的首都是柏林’这句话本身是合法的文本,长度合适,也没有幻觉 URL……但它显然是错误的。

这就涉及一点‘元’的概念。LLM-as-judge 的思路是:再发起一次独立的 LLM 调用,它的唯一任务是读取主模型的输出并给出评分。没错,这就是用 AI 来评 AI。乍一听就像让一个学生去评另一个学生的作业,但实际效果出乎意料地好。Anthropic、Google 等实验室的研究表明,只要给出明确的评分标准,LLM 评判者与人工评估者之间的相关性就会非常高。

关键在于‘明确的评分标准’。如果没有这一点,整个方法就会土崩瓦解。

如何设计评分规则

如果你直接让 LLM ‘给这个打 1~10 分’,你会得到五花八门的分数——这一次可能是 7,下一次就变成 5。这些分数基本上毫无意义,因为模型对每个数字的含义没有共同的定义。

解决办法是使用带有具体锚点描述的评分规则。下面是一个关于‘有用性’的示例。

Score 1 - The response is completely irrelevant, incorrect, or harmful.
Score 2 - The response addresses the topic but contains major errors or omissions.
Score 3 - The response is partially correct but misses key information.
Score 4 - The response is correct and helpful with minor issues.
Score 5 - The response is comprehensive, accurate, and directly addresses the question.

现在注意,每个层级描述的是你可以在输出中指向的具体事物,而不是一种感觉。"完全离题"是可观察的。"有点不好"则不是。正是这种具体性,使得评判者在多次运行中保持一致。

如何实现评判者

下面是完整的 LLMJudge 类。随后我将讲解其中的重要设计决策。

import json
import os
from openai import OpenAI
from dataclasses import dataclass

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


@dataclass
class JudgeResult:
    """Holds the result of an LLM judge evaluation."""
    criterion: str
    score: int
    max_score: int
    reasoning: str


RUBRICS = {
    "relevance": {
        "description": "Does the response directly address the user's question?",
        "levels": {
            1: "Completely off-topic or addresses a different question entirely.",
            2: "Tangentially related but misses the core question.",
            3: "Addresses the question but includes significant irrelevant content.",
            4: "Directly addresses the question with minor tangents.",
            5: "Precisely and completely addresses the question asked.",
        },
    },
    "accuracy": {
        "description": "Is the factual content of the response correct?",
        "levels": {
            1: "Contains critical factual errors that would mislead the reader.",
            2: "Multiple factual errors on important points.",
            3: "Mostly accurate but contains one notable error.",
            4: "Accurate with only trivial imprecisions.",
            5: "Completely accurate with no factual errors.",
        },
    },
    "completeness": {
        "description": "Does the response cover all important aspects of the question?",
        "levels": {
            1: "Addresses less than 20 percent of what the question requires.",
            2: "Covers some aspects but misses major required components.",
            3: "Covers the basics but lacks depth on important points.",
            4: "Comprehensive coverage with minor gaps.",
            5: "Thoroughly covers all aspects the question requires.",
        },
    },
}


class LLMJudge:
    """Layer 2: Uses a separate LLM to evaluate response quality."""

    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model

    def evaluate(
        self, question: str, response: str, criterion: str
    ) -> JudgeResult:
        """Evaluate a single response on a single criterion."""
        rubric = RUBRICS[criterion]
        levels_text = "\n".join(
            f"Score {score}: {desc}"
            for score, desc in rubric["levels"].items()
        )

        judge_prompt = f"""You are an expert evaluator. Your job is to score an AI assistant's response.

CRITERION: {rubric['description']}

SCORING RUBRIC:
{levels_text}

USER QUESTION:
{question}

AI RESPONSE:
{response}

Evaluate the response on the criterion above. You must respond with valid JSON only:
{{"score": , "reasoning": "<2-3 sentence explanation>"}}"""

        judge_response = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": judge_prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )

        result = json.loads(judge_response.choices[0].message.content)
        return JudgeResult(
            criterion=criterion,
            score=result["score"],
            max_score=5,
            reasoning=result["reasoning"],
        )

    def evaluate_all(
        self, question: str, response: str, criteria: list[str] = None
    ) -> list[JudgeResult]:
        """Evaluate a response across all specified criteria."""
        criteria = criteria or list(RUBRICS.keys())
        return [self.evaluate(question, response, c) for c in criteria]


if __name__ == "__main__":
    judge = LLMJudge()

    question = "What is a Python decorator and when should you use one?"
    good_response = (
        "A Python decorator is a function that takes another function as input "
        "and extends its behavior without modifying it. You define a decorator "
        "with the @decorator_name syntax above a function definition. Use "
        "decorators when you need to add cross-cutting concerns like logging, "
        "authentication checks, or caching to multiple functions without "
        "duplicating code in each one."
    )

    results = judge.evaluate_all(question, good_response)
    for r in results:
        print(f"  {r.criterion}: {r.score}/{r.max_score} - {r.reasoning}")

这段代码中有几点值得注意。

  1. 温度为零: 你并不是在要求评判者发挥创造力。你希望相同的输入每次都能得到相同的得分,或尽可能接近。

  2. 输出是结构化的 JSON: 我是通过痛苦的方式学到这一点的。如果你让评判者以自由文本形式回复,你最终会编写脆弱的解析代码来提取分数。强制使用 JSON 输出,你的生活会变得轻松很多。

  3. 评分标准被嵌入到每个提示中: 评判者永远不会使用其自身对“好”意味着什么的概念。它总是根据你的评分标准进行打分,这正是它可重复的原因。

如何处理评判者的可靠性

即使考虑了所有这些因素,单次评判调用仍可能产生噪声。我曾看到同一个响应在一次调用中得到 4 分,在下一次调用中得到 3 分。如果你的决策依赖于这些得分,这种方差就很重要。有两种方法可以帮助你。

第一种是 多评判者共识。这意味着你对同一评估运行三次并取中位数。是的,这会增加三倍的成本。但得分会变得更加稳定,而在 CI/CD 门控决策中,稳定性比省下几美分更重要。

第二种是 校准集。你保留一小组响应(也许 20-30 条),在这些响应上你已经有可靠的人工得分。定期在这些数据上运行你的评判者。如果评判者开始与人工得分不一致,说明出现了变化,你需要进行调查。

我们可以查看这个共识实现,它展示了如何处理这一点:

import numpy as np


def evaluate_with_consensus(
    judge: LLMJudge,
    question: str,
    response: str,
    criterion: str,
    num_judges: int = 3,
) -> JudgeResult:
    """Run multiple judge evaluations and return the median."""
    results = [
        judge.evaluate(question, response, criterion)
        for _ in range(num_judges)
    ]
    scores = [r.score for r in results]
    median_score = int(np.median(scores))
    median_result = min(results, key=lambda r: abs(r.score - median_score))
    return JudgeResult(
        criterion=criterion,
        score=median_score,
        max_score=5,
        reasoning=f"Consensus ({scores}): {median_result.reasoning}",
    )

如何构建第3层:人工评估循环

我曾经让一个LLM评判器对一个回答在准确性上打了5/5,相关性5/5,完整性4/5。这些分数看起来很完美,直到一位同事真的读了这个回答并说:“这在技术上是正确的,但对非专业人士来说会让人摸不着头脑。”他说得对。

该回答使用了用户不熟悉的术语,把关键点埋在了三段之后,读起来像教科书而不是有帮助的回复。

这就是自动评估的上限。LLM评判器在发现事实错误和结构问题方面表现出色,但在语气、针对特定受众的清晰度以及“正确”与“确实有帮助”之间的细微差别上存在盲区。这些盲区正是人工评估发挥作用的地方。

需要澄清的是,这并不意味着要雇佣一个团队来逐条审查每个回答。那样做既不可扩展也没有必要。目标更为具体:定期获取少量人工评分,并将这些分数作为对自动化层的现实检验。

如何构建轻量级标注界面

为此,你其实不需要Label Studio或其他花哨的标注平台。只需要一个能够展示回答并请求评分的Python脚本即可。

从高层次来看,其工作原理是:脚本获取一个问题-回答对,在终端中显示它,让审阅者在1-5的范围内打分,并将结果保存到文件。每条标注都以JSONL格式(即单行JSON)存储,这样以后就能轻松加载、进行分析或送入仪表盘。

import json
import random
from pathlib import Path
from dataclasses import dataclass, asdict


@dataclass
class Annotation:
    """A single human annotation for an LLM response."""
    question: str
    response: str
    annotator: str
    score: int
    notes: str


class AnnotationCollector:
    """Collects and stores human evaluations."""

    def __init__(self, output_file: str = "annotations.jsonl"):
        self.output_path = Path(output_file)

    def collect_annotation(
        self, question: str, response: str, annotator: str
    ) -> Annotation:
        """Present a question-response pair and collect a human score."""
        print("\n" + "=" * 60)
        print(f"QUESTION: {question}")
        print("-" * 60)
        print(f"RESPONSE: {response}")
        print("-" * 60)
        print("Score this response (1-5):")
        print("  1 = Terrible  2 = Poor  3 = Acceptable  4 = Good  5 = Excellent")

        while True:
            try:
                score = int(input("Score: "))
                if 1 <= score <= 5:
                    break
                print("Please enter a number between 1 and 5.")
            except ValueError:
                print("Please enter a valid number.")

        notes = input("Notes (optional, press Enter to skip): ").strip()

        annotation = Annotation(
            question=question,
            response=response,
            annotator=annotator,
            score=score,
            notes=notes,
        )
        self.save(annotation)
        return annotation

    def save(self, annotation: Annotation) -> None:
        """Append annotation to JSONL file."""
        with open(self.output_path, "a") as f:
            f.write(json.dumps(asdict(annotation)) + "\n")

    def load_all(self) -> list[Annotation]:
        """Load all saved annotations."""
        annotations = []
        if self.output_path.exists():
            with open(self.output_path) as f:
                for line in f:
                    data = json.loads(line)
                    annotations.append(Annotation(**data))
        return annotations

让我带你看看这个脚本里发生了什么。

Annotation 数据类只是一个容器,用于保存单条评审的所有信息:原始问题、模型的响应、谁进行了评审、他们给出的得分以及他们添加的任何备注。虽然没什么花哨的功能,但拥有结构化的格式意味着以后可以轻松地在不同评审者之间比较得分。

collect_annotation 方法是实际评审发生的地方。它会在终端中用一些视觉分隔符打印出问题和响应,以便评审者能够清晰阅读,然后提示输入得分。

这里的带有输入验证的 while true 循环非常重要。它会一直询问,直到评审者给出 1 到 5 之间的有效数字,这样你就不会在标注文件中得到垃圾数据。

save 方法会把每条标注作为单独的一行 JSON 写入到 annotations.jsonl 文件中。我使用 JSONL(每行一个 JSON 对象)而不是普通的 JSON 数组,因为它更适合追加写入。你可以在不读取和重写整个文件的情况下添加新的标注,这在长时间收集数百条评审时尤为重要。

load_all 则会读取全部内容,把每一行解析为一个 Annotation 对象。当你想要分析标注、将它们与 LLM 判断得分进行比较,或计算评审者之间的一致性时,就会调用这个方法。

在实际使用中,你可以把生产日志或黄金数据集中的一批问题-响应对喂给它。你可能会在每周的评审会话中运行它,让团队成员花 30 分钟对 20‑30 条响应进行评分。这小小的投入能为你提供一个可靠的基准 truth,用于校准你的自动化层。

如何计算标注者间一致性

接下来你很快就会遇到的一个问题是:让两个人对同一个响应进行评分,但他们给出的得分不同。这是响应本身有歧义,还是你的评分标准有歧义?

你需要一种方法来衡量这一点,而 Cohen's Kappa系数 是标准工具。它基本上告诉你两个标注者之间的一致程度,并扣除仅凭偶然就能达到的一致性水平。

from sklearn.metrics import cohen_kappa_score


def measure_agreement(
    scores_annotator_1: list[int], scores_annotator_2: list[int]
) -> dict:
    """Calculate inter-annotator agreement using Cohen's Kappa."""
    kappa = cohen_kappa_score(scores_annotator_1, scores_annotator_2)

    interpretation = "poor"
    if kappa > 0.8:
        interpretation = "almost perfect"
    elif kappa > 0.6:
        interpretation = "substantial"
    elif kappa > 0.4:
        interpretation = "moderate"
    elif kappa > 0.2:
        interpretation = "fair"

    exact_agreement = sum(
        a == b for a, b in zip(scores_annotator_1, scores_annotator_2)
    ) / len(scores_annotator_1)

    return {
        "cohens_kappa": round(kappa, 3),
        "interpretation": interpretation,
        "exact_agreement": round(exact_agreement, 3),
    }


if __name__ == "__main__":
    # two annotators scored the same 10 responses
    annotator_a = [5, 4, 3, 4, 5, 2, 3, 4, 5, 4]
    annotator_b = [5, 4, 4, 4, 5, 3, 3, 4, 5, 3]

    agreement = measure_agreement(annotator_a, annotator_b)
    print(f"Cohen's Kappa: {agreement['cohens_kappa']}")
    print(f"Interpretation: {agreement['interpretation']}")
    print(f"Exact Agreement: {agreement['exact_agreement']:.0%}")

您希望 Kappa 高于 0.6。低于此值时,问题出在评分标准而非标注人员。返回并在每个得分级别添加更具体的示例。持续改进,直至人们的一致意见趋于稳定。通常需要两到三轮迭代。

如何构建回归测试流程

我们可以看看一种很可能发生在您或您认识的人身上的情形:您调整提示词以修复您注意到的某个错误输出。修复后该输出变好了。随后您发布它,一周后发现此更改实际上导致了三个您本来不会去检查的其他响应出现问题。

这种情况非常常见。唯一的出路是进行回归测试。如果您有传统软件开发经验,可能已经知道回归测试的含义。它指的是在每次更改后重新运行一套固定的测试,专门用来确认您没有破坏原本正常工作的功能。

“回归”一词字面意思是倒退:您的系统之前能够正确处理某个问题,但在您的更改之后却无法做到。

在普通软件中,回归测试通常是单元测试或集成测试。对于 LLM 应用,其工作方式略有不同。我们不再检查确切的输出,而是对一批响应进行评分,并将这些得分与之前的运行结果进行比较。如果得分下降,则表明出现了回归。核心思想相同,但机制围绕评分展开,而不是通过/失败的断言。

如何创建黄金数据集

黄金数据集只不过是一份经过精心挑选的问题列表,这些问题代表了您的应用实际需要处理的内容。每当发生变化时(新提示词、新模型或更新的检索逻辑),您都会用该列表测试您的系统,并将得分与上次运行的结果进行比较。

import json
from pathlib import Path
from dataclasses import dataclass, asdict


@dataclass
class GoldenExample:
    """A single test case in the golden dataset."""
    id: str
    question: str
    reference_answer: str
    category: str
    difficulty: str  # "easy", "medium", "hard"
    criteria: list[str]  # which criteria to evaluate


class GoldenDataset:
    """Manages a curated evaluation dataset."""

    def __init__(self, filepath: str = "golden_dataset.json"):
        self.filepath = Path(filepath)
        self.examples: list[GoldenExample] = []
        if self.filepath.exists():
            self.load()

    def add(self, example: GoldenExample) -> None:
        """Add a new example to the dataset."""
        self.examples.append(example)
        self.save()

    def get_by_category(self, category: str) -> list[GoldenExample]:
        """Filter examples by category."""
        return [e for e in self.examples if e.category == category]

    def save(self) -> None:
        """Persist dataset to disk."""
        data = [asdict(e) for e in self.examples]
        with open(self.filepath, "w") as f:
            json.dump(data, f, indent=2)

    def load(self) -> None:
        """Load dataset from disk."""
        with open(self.filepath) as f:
            data = json.load(f)
            self.examples = [GoldenExample(**item) for item in data]

    def summary(self) -> dict:
        """Return dataset statistics."""
        categories = {}
        for e in self.examples:
            categories[e.category] = categories.get(e.category, 0) + 1
        return {
            "total_examples": len(self.examples),
            "categories": categories,
        }

我在构建这些时,学到的一些经验是,应该从 50 到 100 个示例开始。这样既能捕捉到有意义的回归,又不会让每次评估运行耗时过长。

此外,一定要加入边界情况——那些以前让你的模型跌倒的奇怪问题。如果数据集中 90% 是简单问题,那么难题开始出错时你可能察觉不到。

最后,把这份文件当作活文档来维护。每次在生产环境中出现问题时,就把它转化为黄金数据集的一个示例。几个月后,你的数据集会从通用测试题演变为一份详细的图谱,精准标示出应用脆弱的位置。

如何在 CI/CD 中运行评估

现在让我们把所有东西串联起来。RegressionPipeline 类会在黄金数据集上运行你的系统,对每个响应进行评分,并将结果与之前的运行进行比较。

import json
from datetime import datetime, timezone
from dataclasses import dataclass, asdict


@dataclass
class EvalRun:
    """Records the results of one full evaluation run."""
    run_id: str
    timestamp: str
    model: str
    prompt_version: str
    total_examples: int
    avg_scores: dict  # criterion -> average score
    pass_rate: float  # percentage of examples above threshold
    failures: list[dict]  # examples that scored below threshold


class RegressionPipeline:
    """Runs evaluation against golden dataset and detects regressions."""

    def __init__(
        self,
        deterministic_eval: "DeterministicEvaluator",
        llm_judge: "LLMJudge",
        threshold: float = 3.5,
    ):
        self.det_eval = deterministic_eval
        self.judge = llm_judge
        self.threshold = threshold

    def run(
        self,
        golden_dataset: "GoldenDataset",
        generate_fn: callable,
        model_name: str,
        prompt_version: str,
    ) -> EvalRun:
        """Run full evaluation pipeline against golden dataset.

        Args:
            golden_dataset: The dataset to evaluate against.
            generate_fn: A function that takes a question string and
                         returns the model's response string.
            model_name: Identifier for the model being tested.
            prompt_version: Identifier for the prompt version.
        """
        all_scores = {}
        failures = []

        for example in golden_dataset.examples:
            # Generate response
            response = generate_fn(example.question)

            # Layer 1: Deterministic checks
            det_results = self.det_eval.run_all(response)
            det_failures = [r for r in det_results if not r.passed]

            if det_failures:
                failures.append({
                    "id": example.id,
                    "question": example.question,
                    "layer": "deterministic",
                    "details": [r.details for r in det_failures],
                })
                continue

            # Layer 2: LLM judge
            judge_results = self.judge.evaluate_all(
                example.question, response, example.criteria
            )

            for result in judge_results:
                if result.criterion not in all_scores:
                    all_scores[result.criterion] = []
                all_scores[result.criterion].append(result.score)

                if result.score < self.threshold:
                    failures.append({
                        "id": example.id,
                        "question": example.question,
                        "layer": "llm_judge",
                        "criterion": result.criterion,
                        "score": result.score,
                        "reasoning": result.reasoning,
                    })

        avg_scores = {
            criterion: sum(scores) / len(scores)
            for criterion, scores in all_scores.items()
        }

        total_evaluated = len(golden_dataset.examples)
        pass_count = total_evaluated - len(failures)

        return EvalRun(
            run_id=f"eval_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}",
            timestamp=datetime.now(timezone.utc).isoformat(),
            model=model_name,
            prompt_version=prompt_version,
            total_examples=total_evaluated,
            avg_scores=avg_scores,
            pass_rate=pass_count / total_evaluated if total_evaluated else 0,
            failures=failures,
        )

    def compare_runs(self, baseline: EvalRun, current: EvalRun) -> dict:
        """Compare two evaluation runs to detect regressions."""
        regressions = {}
        improvements = {}

        for criterion in current.avg_scores:
            if criterion in baseline.avg_scores:
                diff = current.avg_scores[criterion] - baseline.avg_scores[criterion]
                if diff < -0.2:  # Score dropped by more than 0.2
                    regressions[criterion] = {
                        "baseline": baseline.avg_scores[criterion],
                        "current": current.avg_scores[criterion],
                        "change": round(diff, 3),
                    }
                elif diff > 0.2:
                    improvements[criterion] = {
                        "baseline": baseline.avg_scores[criterion],
                        "current": current.avg_scores[criterion],
                        "change": round(diff, 3),
                    }

        return {
            "verdict": "REGRESSION" if regressions else "PASS",
            "regressions": regressions,
            "improvements": improvements,
            "pass_rate_change": current.pass_rate - baseline.pass_rate,
        }

现在你可以将此钩入你的 CI/CD 流水线,以便在有人更改提示词或模型配置时自动运行。如果 compare_runs 返回 REGRESSION,则构建失败。在弄清楚问题所在之前,不会有人进行部署。

如何判断你的 AI 实际上是否变得更好:统计显著性

你对提示词进行了微调,平均得分从 3.8 提升到了 4.0。这时候该庆祝了,对吧?也许。或者也许这 0.2 的提升只是随机噪声。

使用包含 50-100 条示例的黄金数据集,仅凭方差就可能轻易产生如此大的得分差异。你需要进行实际的统计检验,以判断变化是否真实。

如果你已经有一段时间没接触统计了,这里作个快速入门。配对 t 检验 是一种比较两组相关测量值的方法。在我们的情况下,每一对都是同一个问题在系统两个不同版本下的得分:旧提示词和新提示词。

该检验会查看每一对,计算每个问题的得分变化量,然后问: "这些变化是否始终朝同一个方向,还是散布得毫无规律?"

如果这些变化具有一致性(大多数问题在新提示词下得分更高),检验会给出一个低 p 值,这表明提升很可能是真实的。如果这些变化散落无章(有些题目变好,有些变差,没有明显规律),p 值会很高,这意味着你无法有把握地说新版本实际上更好。

我们使用 配对 t 检验而非常规 t 检验的原因在于,它会考虑题目难度。有些题目本身就比其他题目更难,配对能够确保我们衡量的是 每题的变化,而不仅仅是比较两组无关的得分批次。

以下是实现此方法的步骤:

from scipy import stats
import numpy as np


def is_improvement_significant(
    scores_before: list[float],
    scores_after: list[float],
    alpha: float = 0.05,
) -> dict:
    """Test whether a score improvement is statistically significant.

    Uses a paired t-test since the same questions are evaluated in both runs.
    """
    t_stat, p_value = stats.ttest_rel(scores_after, scores_before)
    mean_diff = np.mean(scores_after) - np.mean(scores_before)

    return {
        "mean_before": round(np.mean(scores_before), 3),
        "mean_after": round(np.mean(scores_after), 3),
        "mean_difference": round(mean_diff, 3),
        "p_value": round(p_value, 4),
        "is_significant": p_value < alpha,
        "direction": "improvement" if mean_diff > 0 else "regression",
        "recommendation": (
            "Safe to deploy"
            if p_value < alpha and mean_diff > 0
            else "Do not deploy - change is not a significant improvement"
        ),
    }


if __name__ == "__main__":
    # scores on 20 golden examples, before and after a prompt change
    before = [3, 4, 3, 5, 4, 3, 4, 4, 3, 5, 4, 3, 4, 3, 4, 5, 3, 4, 4, 3]
    after =  [4, 4, 4, 5, 5, 3, 4, 5, 4, 5, 4, 4, 4, 4, 5, 5, 4, 4, 5, 4]

    result = is_improvement_significant(before, after)
    print(f"Mean: {result['mean_before']} -> {result['mean_after']}")
    print(f"p-value: {result['p_value']}")
    print(f"Significant: {result['is_significant']}")
    print(f"Recommendation: {result['recommendation']}")

如果 p 值低于 0.05,说明改善仅是偶然的概率小于 5%。这时候就可以发布。如果超过这个阈值,提升可能只是噪声,无论平均数看起来多好,都不要部署。

如何将所有部分组合在一起:完整的评估架构

我们将三层结合成一个单一的编排器。这个类负责把一切串联起来。它先运行确定性检查,如果通过则升级到 LLM 判断,并在需要时引入人工评估以进行校准。

class EvaluationOrchestrator:
    """Coordinates all three evaluation layers into a single pipeline."""

    def __init__(self):
        self.det_eval = DeterministicEvaluator()
        self.llm_judge = LLMJudge()
        self.annotation_collector = AnnotationCollector()

    def evaluate_response(
        self,
        question: str,
        response: str,
        run_human_eval: bool = False,
    ) -> dict:
        """Run the complete evaluation pipeline on a single response."""

        # Layer 1: Deterministic (always runs, every request)
        det_results = self.det_eval.run_all(response)
        det_passed = all(r.passed for r in det_results)

        if not det_passed:
            return {
                "status": "FAIL",
                "layer": "deterministic",
                "details": [r for r in det_results if not r.passed],
                "recommendation": "Fix structural issues before deeper eval",
            }

        # Layer 2: LLM Judge (runs on sample or in CI)
        judge_results = self.llm_judge.evaluate_all(question, response)
        avg_score = sum(r.score for r in judge_results) / len(judge_results)

        if avg_score < 3.5:
            return {
                "status": "FAIL",
                "layer": "llm_judge",
                "avg_score": avg_score,
                "details": judge_results,
                "recommendation": "Response quality below threshold",
            }

        # Layer 3: Human eval (periodic calibration)
        if run_human_eval:
            annotation = self.annotation_collector.collect_annotation(
                question, response, annotator="reviewer"
            )
            return {
                "status": "PASS" if annotation.score >= 4 else "REVIEW",
                "layer": "human",
                "automated_score": avg_score,
                "human_score": annotation.score,
            }

        return {
            "status": "PASS",
            "layer": "llm_judge",
            "avg_score": avg_score,
            "details": judge_results,
        }

我希望早点知道的事

我想以一些我希望在开始构建评估系统之前有人告诉我的话来结束。

首先,不要一次构建所有三层。先从确定性检查开始,随后交付。你会惊讶于它们能自行捕获多少问题,编写它们的过程会迫使你真正定义你的应用程序什么算是正确的输出。在需要时加入LLM评判,随后再加入人工评估。

其次,每月一次用人工评估来检查你的判断模型。在已有人工评分的20-30条响应上运行你的LLM判断模型。如果判断模型平均漂移超过0.5分,说明有变化:可能是判断模型被更新了,或者你的评分规则未覆盖新的失效模式。无论哪种情况,你都需要重新校准。

第三,每一次生产故障都会成为一个测试用例。这也许是最有用的习惯。出现故障?太好了,这就是一个新的黄金数据集示例。几个月后,你的数据集不再是通用的测试套件,而是你的应用程序曾经失败的每一种方式的详细地图。

最后,不要追求完美的评估得分。我见过一些团队无休止地调整提示词,只为了把评估得分从4.2提升到4.5,结果发现他们的评分规则存在盲点,用户仍然不满意。得分只是工具,不是目标,因此人工评估存在是为了捕捉数字遗漏的问题。

总结

本文讨论了很多内容,下面我来做个总结。核心问题是,LLM应用程序的失败方式与传统软件不同。没有崩溃,没有错误日志,也没有堆栈跟踪。只是一个自信且格式良好的错误答案。

而且由于输出不是确定的,你不能用简单的断言来测试它们。你需要一种完全不同的方法。

这种方法就是分层评估管道:

在这三层之上,你学会了如何用黄金数据集构建回归测试流水线,以便在问题进入生产前发现质量下降。你还学会了如何用统计显著性检验确保改进真实有效,而不仅仅是噪声。

如果只能让你记住一件事,那就是:从小处着手。不要试图在一个周末内完成所有这些工作。今天就把 DeterministicEvaluator 类加入你的项目:只需五分钟,它就会立即开始捕捉你目前遗漏的问题。等你准备好进行更深入的评估时,再加入 LLM 评判器。随后,随着应用的成熟,逐步引入人工审查和回归测试。

能够交付可靠 AI 产品的团队,不是拥有最炫模型的团队。他们是搭建了脚手架、能够知道模型何时失效,并在用户发现之前先行捕捉问题的团队。

——

🧑‍💻

zhirenhun

一个热爱技术的程序员,喜欢分享前沿AI知识和开发经验。