← 文章 / AI技术
freeCodeCamp 1小时前 · 2026-09-13 02:28:23 · 3 阅读

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

你的 AI 功能上线了,演示效果也无可挑剔,团队对此印象深刻。然而,当用户提出一个略微超出测试用例范围的问题时,模型却信心满满地给出了一个完全错误的回答。

使用大语言模型(Large Language Model, LLM)构建应用的残酷现实是,传统的软件测试方法在此失效。当系统每次运行都生成不同的文本时,你无法简单地写入断言 `output == expected` 来验证结果。

市面上大多数教程只教你如何搭建聊天机器人或配置 RAG 管道,然后就戛然而止了。它们轻描淡写地说“部署到生产环境”,仿佛最难的部分已经结束。但实际上,真正的难点在于判断你的 AI 表现是否合格,并在它出错时及时捕捉这些问题。

在本文中,我将带你从零开始构建一套完整的评估流水线。同时,我们将介绍三种不同成本和深度的评估策略。

本文涵盖内容:

前置准备

要跟随本文实操,你需要 Python 3.10 或更高版本,以及调用 LLM API 的基础经验。无论是使用 OpenAI、Anthropic 还是本地模型,评估模式的逻辑都是相通的。

此外,你还需要一个 OpenAI API 密钥来运行 LLM-as-judge 示例(我们选用 gpt-4o-mini,因为它成本低廉且评分能力足够)。

如果你手头有一个基于 LLM 的应用(哪怕很小)想要评估,那再好不过了。如果没有也没关系,示例是自包含的,你依然可以跟进全部内容。

在这里获取依赖项:

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-as-judge。用另一次 LLM 调用来给主模型的输出打分:"这个回答相关吗?准确吗?真的有帮助吗?"像 gpt-4o-mini 这样的模型在给其他模型打分方面出奇地好,只要给它一个清晰的评分标准。

  3. 第三层是人工评估。让真人来审查真实输出。不可能对每条回复都这么做,那太慢了,但可以定期抽查,确保自动化层没有偏离"好"的真正含义。

关键在于知道什么时候该用哪一层,接下来我们会把这三层都搭建出来。

LLM 评估的三层:确定性检查、LLM-as-Judge 和人工评估

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

我刚开始做评估流水线时,直接跳到了那些花哨的东西:LLM judge、embedding 相似度分数等等。与此同时,我的应用偶尔会返回完全空白的字符串,而我整整两周都没发现。整整两周!

所以现在每个项目我都从确定性检查入手。它们非常简单:不用机器学习,不用调用 API,就是用普通 Python 对输出做基本的健全性检查——输出存在吗?格式对吗?是不是短得可疑?模型是不是编造了一个 URL?

你可能会觉得这些检查太基础、不值得做。我以前也这么想。后来我在一个月的生产日志上跑了一遍,发现我漏掉的那些糟糕输出里,大约三分之一本来就能被这些五分钟就能写好的检查捕获。

下面是我现在每个项目第一天就会放进代码里的 DeterministicEvaluator 类:

import json
import re
from dataclasses import dataclass

@dataclass
class EvalResult:
    """保存单个评估检查的结果"""
    check_name: str
    passed: bool
    score: float  # 取值范围 0.0 到 1.0
    details: str

class DeterministicEvaluator:
    """第一层:针对 LLM 输出的快速规则校验"""

    def check_json_validity(self, output: str) -> EvalResult:
        """当预期输出为 JSON 时,校验其格式是否合法"""
        try:
            json.loads(output)
            return EvalResult("json_validity", True, 1.0, "合法 JSON")
        except json.JSONDecodeError as e:
            return EvalResult("json_validity", False, 0.0, f"非法 JSON: {e}")

    def check_length_bounds(
        self, output: str, min_chars: int = 10, max_chars: int = 5000
    ) -> EvalResult:
        """检查输出长度是否落在允许范围内"""
        length = len(output)
        if length < min_chars:
            return EvalResult(
                "length_bounds", False, 0.0,
                f"过短:{length} 字符(下限 {min_chars})"
            )
        if length > max_chars:
            return EvalResult(
                "length_bounds", False, 0.0,
                f"过长:{length} 字符(上限 {max_chars})"
            )
        return EvalResult("length_bounds", True, 1.0, f"长度正常:{length} 字符")

    def check_no_hallucinated_links(self, output: str) -> EvalResult:
        """检测输出中可能被模型臆造的 URL"""
        url_pattern = r'https?://[^\s\)\]\}\"\'<>]+'
        urls = re.findall(url_pattern, output)
        if urls:
            return EvalResult(
                "no_hallucinated_links", False, 0.0,
                f"发现 {len(urls)} 个疑似臆造的 URL: {urls[:3]}"
            )
        return EvalResult("no_hallucinated_links", True, 1.0, "未发现 URL")

    def check_required_sections(
        self, output: str, required: list[str]
    ) -> EvalResult:
        """校验必选章节或关键词是否出现在输出中"""
        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}"
            )
        return EvalResult("required_sections", True, 1.0, "所有章节齐全")

    def check_no_refusal(self, output: str) -> EvalResult:
        """检测模型是否在不该拒绝的情况下拒绝作答"""
        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"检测到疑似拒绝话术:'{phrase}'"
                )
        return EvalResult("no_refusal", True, 1.0, "未检测到拒绝")

    def run_all(self, output: str, config: dict = None) -> list[EvalResult]:
        """运行所有确定性检查并返回结果列表"""
        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()

    # 使用正常输出测试
    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}: {'通过' if r.passed else '失败'} ({r.details})")

    # 使用可疑输出测试
    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}: {'通过' if r.passed else '失败'} ({r.details})")

以上每一项检查的运行时间都不到一毫秒,且成本为零。但别被这种简单性迷惑了,仅『虚构链接』这一项检查,就帮我避免了好几次把伪造的文档 URL 发给用户的尴尬,次数多到我都不好意思全说。

还有一点值得强调:这些只是起点。上述通用检查适用于任何 LLM 应用,但最大的收益来自领域特定的检查。如果你的应用生成 SQL,就加一个语法解析器;如果它起草邮件,就验证是否包含主题行和问候语;如果它输出代码,就试着跑一遍 Linter。

在这里每增加一项检查,就意味着少一条劣质输出流到下游昂贵的环节,或者更糟——流到你的用户手里。

如何构建第 2 层:LLM-as-Judge 评估

好,现在你的输出已经通过了 sanity check:有效的 JSON、合理的长度、没有虚构链接。但这里有个第 1 层回答不了的问题:这个回答真的有用吗?

一个输出可能结构完美,通过所有确定性检查,但对阅读者来说却毫无用处。比如『法国首都是柏林』,它是有效文本,长度正确,没有幻觉链接……但它是错的。

这里就有点『套娃』了。LLM-as-Judge 的核心思想是:发起一次独立的 LLM 调用,唯一任务就是读取主模型输出并打分。是的,你在用 AI 给 AI 打分。这听起来像让一个学生给另一个学生的作业打分,但实际上效果出奇地好。来自 Anthropic 和 Google 等实验室的研究表明,当你提供清晰的评分标准时,LLM 评委与人类评估者的评分具有高度相关性。

关键短语是『清晰的评分标准』。没有它,这种方法就彻底失效了。

如何设计评分量规

如果你只告诉 LLM『从 1 到 10 打个分』,你会得到杂乱无章的结果。一次 7 分,下次可能变 5 分。由于模型没有对每个数字含义的统一定义,这些分数基本没有意义。

解决办法是提供带有具体锚点描述的量规。这里是一个针对『有用性』的量规示例。

1 分 —— 回答完全跑题、错误或有害。
2 分 —— 回答切题,但存在重大错误或遗漏。
3 分 —— 回答部分正确,但缺少关键信息。
4 分 —— 回答正确且有帮助,只有小问题。
5 分 —— 回答全面、准确,直接回答了问题。

注意,每一档描述的都是你在输出中可以明确指出的问题,而不是一种模糊的感觉。“完全跑题”是可观察的,“感觉不太好”则不是。正是这种具体性让评审模型在多次运行中保持一致。

如何实现评审器

下面是完整的 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:
    """保存由 LLM 评审生成的评估结果。"""
    criterion: str
    score: int
    max_score: int
    reasoning: str

RUBRICS = {
    "relevance": {
        "description": "回答是否直接回应了用户的问题?",
        "levels": {
            1: "完全跑题,或者回答的是另一个问题。",
            2: "稍有关联,但未能抓住核心问题。",
            3: "触及了问题,但包含大量无关内容。",
            4: "直接回应问题,略有偏离。",
            5: "精准且完整地回答了所提问题。",
        },
    },
    "accuracy": {
        "description": "回答中的事实内容是否准确?",
        "levels": {
            1: "包含会导致读者误解的关键事实错误。",
            2: "在重要问题上存在多处事实错误。",
            3: "大体准确,但有一处明显错误。",
            4: "准确,仅存在细小的不严谨之处。",
            5: "完全准确,无任何事实错误。",
        },
    },
    "completeness": {
        "description": "回答是否覆盖了问题的所有重要方面?",
        "levels": {
            1: "仅涉及问题要求的不到 20% 内容。",
            2: "涵盖部分方面,但缺少主要必要组件。",
            3: "覆盖了基础内容,但关键要点缺乏深度。",
            4: "覆盖全面,仅有细微遗漏。",
            5: "全面深入地涵盖了问题所要求的所有方面。",
        },
    },
}

class LLMJudge:
    """第二层:使用独立的 LLM 来评估响应质量。"""

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

    def evaluate(
        self, question: str, response: str, criterion: str
    ) -> JudgeResult:
        """根据单一标准评估单一响应的质量。"""
        rubric = RUBRICS[criterion]
        levels_text = "\n".join(
            f"评分 {score}: {desc}"
            for score, desc in rubric["levels"].items()
        )

        judge_prompt = f"""你是一名专业评估员,负责为 AI 助手的回答打分。

评估标准:{rubric['description']}

评分细则:
{levels_text}

用户问题:
{question}

AI 回答:
{response}

请依据上述标准对回答进行评分。必须且只能返回有效的 JSON 格式:
{{"score": <1-5 的整数>, "reasoning": "<2-3 句的评分理由>"}}"""

        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]:
        """根据所有指定标准全面评估一个响应。"""
        criteria = criteria or list(RUBRICS.keys())
        return [self.evaluate(question, response, c) for c in criteria]

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

    question = "什么是 Python 装饰器?什么时候应该使用它?"
    good_response = (
        "Python 装饰器是一种函数,它接收另一个函数作为输入,"
        "并在不修改原函数的前提下扩展其功能。你使用 "
        "@decorator_name 语法在函数定义上方定义装饰器。当你需要为多个函数添加 "
        "日志记录、身份验证检查或缓存等横切关注点(cross-cutting concerns)时,"
        "使用装饰器可以避免在每个函数中重复编写代码。"
    )

    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. 评分标准内置于每个 prompt 中:评判模型永远不能靠它自己对「好」的理解来打分。它必须始终依据你的评分标准,这才是可重复性的来源。

如何处理评判模型可靠性

即便做了以上所有工作,单次评判调用仍可能有噪声。我见过同样的响应,一次打 4 分,下一次打 3 分。如果你的决策依赖这些分数,这种波动就是个大问题。两件事能帮到你。

第一是多评判共识。意思是同一项评估跑三遍,取中位数。是的,成本翻了 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}",
    )

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

我曾经遇到过一个 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:
    """针对 LLM 响应的单个数据标注。"""
    question: str
    response: str
    annotator: str
    score: int
    notes: str

class AnnotationCollector:
    """收集和存储人类评估数据。"""

    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:
        """展示问题和回答,收集人工评分。"""
        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:
        """追加标注到 JSONL 文件。"""
        with open(self.output_path, "a") as f:
            f.write(json.dumps(asdict(annotation)) + "\n")

    def load_all(self) -> list[Annotation]:
        """加载所有已保存的标注。"""
        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 个回答打分。这项小投入能为你提供可靠的基准(ground 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 以上。低于这个数,问题出在你的评分标准上,而不是标注员。回去给每个分值档位补充更具体的示例,反复打磨,直到大家能稳定达成一致。这通常需要两三轮迭代。

如何搭建回归测试流水线

看看这个你可能经历过、或听别人说起过的场景:你改了一条 prompt,修复了发现的某个坏输出,效果不错,那个输出确实变好了。上线之后,一周后你才发现这次改动把另外三个从没检查过的输出搞坏了。

这种事太常见了,唯一的出路就是回归测试。做过传统软件开发的话,你大概已经知道回归测试是什么:每次改动后重新跑一遍固定的测试集,专门确认没有破坏原本正常的功能。

Regression(回归)这个词的字面意思就是倒退:系统原来能正确回答某个问题,改动之后就不行了。

在传统软件中,回归测试通常是单元测试或集成测试。但对于 LLM 应用,逻辑略有不同。我们不再检查输出的精确匹配,而是对一批响应进行打分,并与上一次运行结果对比。如果分数下降,说明出现了回归问题。核心思想相同,但机制围绕打分而非通过/失败的断言来构建。

如何创建黄金数据集

黄金数据集(Golden Dataset)本质上是一个精心挑选的问题列表,代表了应用实际需要处理的各种场景。每次系统发生变更(如新提示词、新模型或更新后的检索逻辑)时,都要用这份列表运行系统,并将评分与上一次运行结果进行对比。

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:
    """记录一次完整评估的结果。"""
    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:
    """在黄金数据集上运行评估并检测性能回退。"""

    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:
        """在黄金数据集上运行完整的评估流水线。

        Args:
            golden_dataset: 用于评估的数据集。
            generate_fn: 接收问题字符串、返回模型响应字符串的函数。
            model_name: 被测试模型的标识。
            prompt_version: prompt 版本的标识。
        """
        all_scores = {}
        failures = []

        for example in golden_dataset.examples:
            # 生成响应
            response = generate_fn(example.question)

            # 第一层:确定性检查
            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

            # 第二层:LLM 评审
            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:
        """对比两次评估结果,检测是否出现性能回退。"""
        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:  # 分数下降超过 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 流水线,每当有人修改 prompt 或模型配置时自动运行。如果 compare_runs 返回 REGRESSION,构建就会失败。在查明原因之前,谁也无法部署上线。

如何判断 AI 是否真的变好了:统计显著性

假设你调整了 prompt,平均分从 3.8 涨到了 4.0。该庆祝一下吗?也许吧。但也可能那 0.2 的涨幅只是随机噪声。

对于 50 到 100 条样本的黄金数据集,仅靠方差波动就足以产生如此大的分数差异。你需要通过实际的统计检验来判断变化是否真实存在。

如果你对统计学有些生疏,这里快速科普一下。配对 t 检验用于比较两组相互关联的测量数据。在我们这个场景中,每一对数据就是同一个问题在两个不同版本的系统下的得分:旧 prompt 和新 prompt。

该检验会考察每一对数据,计算每个问题的分数变化量,然后回答:“这些变化方向是否一致,还是随机散布?”

如果变化具有一致性(例如大多数问题在新 prompt 下得分更高),检验会得到一个较低的 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:
    """测试分数提升是否具有统计显著性。

    由于两次运行评估的是相同的问题,因此采用配对 t 检验。
    """
    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__":
    # 在 20 个黄金示例上的得分,分别为提示词修改前后的结果
    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%。此时即可上线部署。若 p 值高于此阈值,提升可能只是噪声,无论平均分看起来多么漂亮,都不要部署。

如何整合:完整评估架构

让我们将这三层连接成一个统一的编排器。这个类负责将所有部分串联起来:首先运行确定性检查,若通过则升级为 LLM 评判,并可选择引入人工评估进行校准。

class EvaluationOrchestrator:
    """将三层评估协调成一个完整的流水线。"""

    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:
        """对单个响应运行完整的评估流水线。"""

        # 第一层:确定性评估(每次请求都会运行)
        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": "先修复结构性问题,再进行更深入的评估",
            }

        # 第二层:LLM 评审(按抽样或 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": "响应质量低于阈值",
            }

        # 第三层:人工评估(定期校准)
        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 应用失效的方式与传统软件截然不同。没有崩溃,没有错误日志,也没有堆栈跟踪。有的只是一个自信、格式工整却错误的回答。

由于输出是非确定性的,你无法用简单的断言来测试。你需要完全不同的方法。

这种方法就是分层评估流水线:

  • 第一层(确定性检查)处理基础问题:输出是否有效、长度是否合适、是否包含幻觉链接?这些检查快速且免费,捕获的问题比你想象的多。

  • 第二层(LLM 作为裁判)引入语义评估:响应是否真正相关、准确且完整?通过给裁判模型提供清晰的评分标准和具体的打分依据,你可以获得令人惊讶的可靠且自动化的质量分数。

  • 第三层(人工评估)确保整个系统保持校准。定期抽样进行人工审核,能捕捉到自动化评分容易遗漏的细微问题,例如语气、清晰度,以及“正确”与“真正有用”之间的区别。

在上述三层基础上,你学习了如何基于黄金数据集构建回归测试流水线,以便在质量下降影响生产环境之前提前发现。你还掌握了如何使用统计显著性检验,确保改进是真实的而非随机波动。

如果只能带走一个核心观点,那就是:从小处着手。不要试图在一个周末内搞定所有功能。今天就可以将 DeterministicEvaluator 类集成到你的项目中,这只需五分钟,却能立刻帮你发现当前遗漏的问题。准备好深入评估时,再引入 LLM 评审;随着应用成熟,再逐步加入人工审核和回归测试。

那些能交付可靠 AI 产品的团队,往往不是拥有最花哨模型的团队,而是那些搭建了支撑体系、能在模型失效时及时察觉,并抢在用户之前捕捉到问题的团队。

原始来源: freeCodeCamp

评论 (0)