如何构建一个具备自我评估功能的人工智能系统：为大型语言模型应用开发自动化测试与评估流程

你将自己的AI功能部署出去，演示效果也确实不错，团队成员们都感到非常满意。然而，当有用户提出一些超出测试范围的问题时，模型却会给出完全错误的回答。

使用大型语言模型进行开发的现实是：传统的软件测试方法根本行不通。当你的系统每次运行都会生成不同的输出结果时，你根本无法使用“output == expected”这样的简单判断语句来进行测试。

大多数教程都会教你如何构建聊天机器人或设置RAG评估流程，但之后就戛然而止了。它们只会告诉你“将其部署到生产环境中”，仿佛最困难的部分已经结束了。但实际上，真正关键的是要判断你的AI模型是否真的有效，以及在它性能下降时及时发现这个问题。

在这篇文章中，我会带你一步步构建一个完整的评估体系。同时，我们还会探讨三种不同类型的评估方法，这些方法适用于不同成本和深度的需求。

我们将涵盖的内容：

• 为什么传统测试方法不适用于大型语言模型应用

• 大型语言模型的三层评估体系

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

• 如何构建第二层：以大型语言模型作为评估工具

• 如何构建第三层：人工评估环节

• 如何构建回归测试流程

• 如何判断你的AI模型是否真的取得了进步：统计显著性分析

• 如何将所有环节整合起来：完整的评估架构

• 我真希望自己能早点了解这些

为了顺利跟随本文的学习，你需要安装Python 3.10或更高版本，并具备调用大型语言模型API的基本经验。无论你使用的是OpenAI、Anthropic还是其他本地模型，这些评估方法都是通用的。

在进行以大型语言模型作为评估工具的示例时，你还需要一个OpenAI API密钥（我们这里使用了gpt-4o-mini模型，因为它价格便宜且性能足够满足评估需求）。

如果你已经有一个基于大型语言模型的应用想要进行评估，哪怕是一个非常简单的应用，那也是再理想不过了。如果没有这样的应用，这些示例也可以让你独立完成学习过程。

你可以在这里下载所需的依赖库：

pip install openai numpy pandas scikit-learn python-dotenv

为什么传统测试方法不适用于大型语言模型应用

如果你曾经为常规软件编写过测试用例，你就知道其中的套路：输入数据，得到结果，然后验证这两者是否一致。简单明了，流程清晰。

但大语言模型彻底打破了这种模式。而且这种破坏不是以一种方式发生的，而是通过多种相互叠加的因素共同作用的结果。

首先，它们的输出结果并不是确定性的。你即使两次输入完全相同的指令，也可能得到不同的结果。即便设置了temperature=0这样的参数，也无法完全保证结果的一致性，因为模型提供商会在幕后不断更新他们的模型。同一API在1月和3月的调用结果也可能有所不同。

其次，不存在唯一的正确答案。如果你的应用程序需要总结一份文档，那么什么是正确的总结呢？两个人可能会写出不同的总结，但两者都可能是合理的。你无法通过简单的assertEqual方法来验证这些结果的正确性。

第三，这些模型并不会出现明显的错误或故障，日志中也不会显示任何异常信息。它们只是会默默地返回一些看似合理但实际上错误的答案。你的系统运行时间监控数据显示100%正常，但用户得到的却是一些毫无意义的内容。当大语言模型出现问题时，这种隐蔽性反而会更加令人困扰。

因此，你不能用测试REST API的方法来测试大语言模型应用。你需要采用评分机制，而不是简单的通过/失败判断。你需要评估大量的输出结果，而不仅仅是单个样本。此外，你还需要一种能够持续运行的评估系统，因为模型的质量可能会随时间发生变化，而你却无需修改任何代码。

大语言模型评估的三个层次

经过多次尝试和探索，我最终采用了一种分为三个层次的评估方法，这些层次从成本低廉、速度快的方法逐渐过渡到成本高昂、细致入微的方法。

1. 第一层是确定性检查。可以把这些检查想象成门口的守卫。输出的结果是否是有效的JSON格式？它的长度是否异常短或过长？其中是否包含一些莫名其妙的链接？这类检查速度快、成本低，而且能发现很多你意想不到的问题。

2. 第二层是利用另一款大语言模型来进行评估。你可以让另一款大语言模型来评价你的主要模型产生的输出结果。“这个答案相关吗？准确吗？真的有帮助吗？”只要给出明确的评分标准，像gpt-4o-mini这样的模型在评估其他模型的表现时表现得相当出色。

3. 第三层是人工评估。让真实的人类来审查这些输出结果。你不需要对每一个响应都进行人工评估，因为那样会非常耗时，但你需要定期进行这样的检查，以确保自动化评估系统没有偏离真正的评估标准。

关键在于知道在什么情况下应该使用哪一层评估方法，而我们接下来将会逐一构建这些评估层次。



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

当我刚开始构建评估流程时，直接选择了那些高级的功能：使用大语言模型进行判断、计算嵌入相似度得分等等。然而，我的应用程序有时会返回完全空的字符串，而我竟然两周都没有注意到这一点——整整两周啊！

因此，现在我在每个项目中都会先进行一些简单的检查。这些检查非常容易实现：不需要使用机器学习技术，也不需要调用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: """第一层检查：针对大语言模型输出结果的快速、基于规则的验证。""" def check_json_validity(self, output: str) -> EvalResult: """当预期输出是JSON格式时，验证其是否确实为有效的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 seksiions", 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 seksiions(output, config["required seksiions"]) ) return results if __name__ == "__main__": evaluator = DeterministicEvaluator() # 使用正常的输出进行测试 good_output = "Python是一种高级编程语言，以其易读性而闻名。" results = evaluator.run_all(good_output) for r in results: print(f" {r.check_name}: {'PASS' if r.passed else 'FAIL'} ({r.details})") # 使用可疑的输出进行测试 bad_output = "访问https://fake-docs.example.com/api以获取更多信息。" results = evaluator.run_all(bad_output) for r in results: print(f" {r.check_name}: {'PASS' if r.passed else 'FAIL'} ({r.details})")

这些检查每项的运行时间都不到一毫秒，而且完全不耗费任何成本。但千万不要被这种简单性所迷惑——仅仅是通过那些针对虚假链接的检查，我就多次避免向用户发送伪造的文档链接，而这些次的数量多得我都不愿意承认。

还需要强调的一点是，这些检查其实只是起点而已。上面提到的通用检查方法适用于任何大型语言模型应用。但真正的成效往往来自于针对特定领域的个性化检查。如果你的应用程序会生成SQL代码，那就添加一个语法解析器；如果它用于编写电子邮件，就要确保邮件中包含主题行和问候语；如果它输出的是代码，那就尝试使用代码检查工具来检测其中存在的问题。

每当你添加一项检查措施，就能减少一份错误结果传达到后续处理环节，甚至更糟的是，避免这些错误结果到达用户手中。

如何构建第二层评估机制：利用大型语言模型作为评判工具

好了，现在你的输出已经通过了所有的基本检查：它是有效的JSON格式，长度也合适，并且没有包含任何虚假链接。但这里有一个第一层检查无法解决的问题：这个回答到底是否有帮助呢？

一个输出结果可能结构完整，也能通过所有定性的检查，但对阅读它的人来说却可能毫无用处。“法国的首都是柏林”这样的表述虽然格式正确、长度合适，也没有虚假链接……但它确实是错误的。

这时就需要运用一些元层面的思考方法了。利用大型语言模型作为评判工具的核心思想就是：让另一个大型语言模型来专门负责阅读你的主模型的输出结果并对其进行评分。没错，你是在用人工智能来评估另一套人工智能系统。这听起来有点像让学生来批改其他学生的作业，但实际上效果出奇地好。Anthropic和谷歌等机构的研究也表明，当为这些评判工具提供明确的评分标准时，它们给出的评分结果与人类评估者的判断结果具有很高的相关性。

这里的关键在于“明确的评分标准”。如果没有这个标准，整个评估机制就会失效。

如何设计评分标准

如果你让一个大型语言模型来“给这个答案打分（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": , "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的语法来定义一个装饰器。
        当你需要为多个函数添加诸如日志记录、身份验证检查或缓存等功能时，就可以使用装饰器，
        这样就不必在每个函数中重复编写这些代码了。」
    )

    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分；如果你是根据这样的评分结果来做出决策的，那么这种波动性确实会影响到你的判断结果。有两点方法可以帮助你解决这个问题。

第一点是多评估者共识机制。也就是说，你需要对同一个问题进行三次评估，并取这些评分的中值作为最终结果。虽然这样会增加工作量，但评分结果的稳定性会大大提高；对于那些需要基于稳定结果来做出决策的场景来说，稳定性显然比节省一些成本更为重要。

第二点是校准数据集。你可以准备一组包含20到30个回答的样本，这些样本都已经得到了可靠的人类评分结果。定期使用评估系统对这些样本进行评估；如果系统的评分结果开始与人类评估者的评分结果出现偏差，那就说明某些因素发生了变化，你需要及时进行调查。

下面这个代码示例展示了如何实现这种共识机制：

import numpy as np


def evaluate_with_consensus(
    judge: LLMJudge,
    question: str,
    response: str,
    criterion: str,
    num_judges: int = 3,
) -> JudgeResult:
    """对同一问题进行多次评估，并取评分结果的中值作为最终结果。"""
    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"共识评分结果为：{median_result.reasoning}",
    )

如何构建第三层评估机制：人类评估循环

我曾经遇到过这样一个案例：某个大型语言模型在准确性、相关性以及完整性这三个方面的评分都是5/5。这些评分看起来非常完美，但直到一位同事仔细阅读了该模型的回答后，他才指出：“虽然这个答案在技术上是正确的，但对于那些不是该领域专家的人来说，它反而会让人感到困惑。”他的观点确实很有道理。

那个模型的回答中使用了用户根本不懂的专业术语，关键信息被隐藏在了文章的深处，整篇回答读起来更像是一份教科书，而不是一个有帮助的回复。

这就是自动化评估所能达到的极限。大型语言模型在检测事实性错误与结构问题方面表现极为出色，但它们在语气处理、针对特定受众的表述清晰度，以及“正确”与“真正有帮助”之间的细微差别这些方面存在盲点。而正是这些盲点，需要人类来进行评估才能弥补。

需要明确的是，这并不意味着要雇用一个团队来审查每一条反馈信息。这样做既不现实，也没有必要。我们的目标其实很简单：定期让一小部分人对这些反馈信息进行评分，然后利用这些评分结果来检验我们自动化的处理流程是否有效。

如何构建一个轻量级的注释系统

实际上，你根本不需要使用Label Studio或任何复杂的注释工具。你只需要一个Python脚本，这个脚本能够显示一条反馈信息，并要求用户对其进行评分即可。

其工作原理大致如下：该脚本会获取一条问题-回答对，将其显示在终端中，然后请求用户按照1到5的等级给这条反馈信息打分，最后将评分结果保存到一个文件中。每条注释都会以JSONL格式存储，这种格式使得后续重新加载这些数据、进行分析或将其导入仪表板变得非常方便。

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


@dataclass
class Annotation:
    """针对大型语言模型的反馈信息，由人类提供的注释。"""
    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}")
        print("-" * 60)
        print(f"回答：{response}")
        print("-" * 60)
        print("请给这个回答打分（1-5分）：")
        print("  1 = 非常差  2 = 差  3 = 一般  4 = 良好  5 = 优秀")

        while True:
            try:
                score = int(input("评分："))
                if 1 <= score <= 5:
                    break
                print("请输入1到5之间的数字。")
            except ValueError:
                print("请输入有效的数字。")

        notes = input("备注（可选，按Enter键跳过）：").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对象。当你想要分析这些评论、将它们的评分结果与大型语言模型的评估结果进行对比，或者计算不同评审者之间的评分一致性时，就可以使用这个方法。

在实际应用中，你可以将生产日志或黄金数据集中的一批问题-回答对作为输入来使用这个脚本。例如，在每周的评审会议中，可以让一名团队成员花费30分钟的时间来为20到30条评论打分。这样的投入能够为你提供可靠的基准数据，从而帮助你校准自动化评分系统。

如何计算评审者间的一致性

现在你会遇到这样一个问题：当你让两个人对同一条评论进行评分时，他们给出的分数往往不同。这是由于这条评论本身存在歧义，还是因为你的评分标准不够明确呢？

你需要一种方法来衡量这种差异，而Cohen's Kappa系数正是用于解决这个问题的标准工具。它能够告诉你两个评审者之间的评分一致性程度，并同时考虑到这种一致性在随机情况下出现的概率。

from sklearn.metrics import cohen_kappa_score


def measure_agreement(
    scores_annotator_1: list[int], scores_annotator_2: list[int]
) -> dict:
    """使用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"
    else:
        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__":
    # 两位评审者对相同的10条评论给出了相同的评分
    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"评审者间的一致性程度: {agreement['interpretation']}")
    print(f>精确一致性比例: {agreement['exact_agreement']:.0%}")

你希望Kappa值高于0.6。如果低于这个数值，问题出在你的评分标准上，而不是注释者身上。回去为每个评分等级添加更多具体的例子，不断优化这些标准，直到所有人都能达成一致意见。通常需要经过两到三轮迭代才能达到目标。

如何构建回归测试流程

我们可以举一个你或其他人可能遇到过的情景：你修改了一个提示语，目的是修复某个错误结果。修改后，那个特定的结果确实得到了改善。但后来你在实际部署系统中发现，这个改动反而破坏了另外三个你之前完全没有考虑到的功能。

这种情况非常普遍。解决这类问题的唯一方法就是进行回归测试。如果你有过传统的软件开发经验，应该已经了解什么是回归测试了。回归测试就是在每次对代码进行修改后，重新运行一系列预先设定好的测试用例，目的是确保这些改动没有破坏那些原本能够正常运行的功能。

“回归”这个词的字面意思就是“倒退”：在修改之前，你的系统能够正确处理某个问题，但修改之后却不能了。

在常规的软件开发中，回归测试通常包括单元测试或集成测试。但对于大型语言模型应用来说，测试的方式略有不同。在这种情况下，我们不是检查具体的输出结果，而是对一批响应进行评分，并将这些评分与之前的测试结果进行比较。如果评分下降了，那就说明系统出现了退步。虽然原理相同，但测试的方法是基于评分来进行的，而不是通过判断“通过”或“失败”来确定结果的。

如何创建黄金数据集

黄金数据集其实就是一份精心挑选的问题列表，这些问题能够反映你的应用程序实际需要处理的各种情况。每当系统中的某些部分发生变化（比如新的提示语、新的模型或更新后的检索逻辑），你就使用这份数据集来重新测试系统，并将当前的测试结果与之前的结果进行比较。

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


@dataclass
class GoldenExample:
    """黄金数据集中的单个测试用例。"""
    id: str
    question: str
    reference_answer: str
    category: str
    difficulty: str  # "容易", "中等", "困难"
    criteria: list[str]  # 需要评估的标准


class GoldenDataset:
    """用于管理精心挑选的测试数据集。"""

    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:
        """向数据集中添加一个新的测试用例。"""
        self/examples.append(example)
        self.save()

    def get_by_category(self, category: str) -> list[GoldenExample]:
        """按类别筛选测试用例。"""
        return [e for e in self.examples if e.category == category]

    def save(self) -> None:
        """将数据集保存到磁盘上。"""
        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:
        """从磁盘上加载数据集。"""
        with open(self.filepath) as f:
            data = json.load(f)
            self.examples = [GoldenExample(**item) for item in data]

    def summary(self) -> dict:
        """返回数据集的统计信息。"""
        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  # 评估标准及其对应的平均分
    pass_rate: float  # 高于阈值的示例所占的比例
    failures: list[dict]  # 分数低于阈值的示例


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:
        """使用黄金数据集运行完整的评估流程。

        参数：
            golden_dataset：用于评估的数据集。
            generate_fn：一个函数，它接受一个问题字符串作为输入，
                         并返回模型对应的响应字符串。
            model_name：被测试模型的名称。
            prompt_version：所使用的提示版本。
        """
        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,
        }

现在你可以将这个流程集成到你的持续集成/持续部署管道中，这样每当有人修改提示语或模型配置时，这个流程就会自动运行。如果compare_runs返回REGRESSION，那么构建过程就会失败。在弄清楚问题所在之前，任何人都不应该进行部署。

如何判断你的AI是否真的取得了进步：统计显著性分析

你修改了提示语，平均分从3.8提高到了4.0。这时候是不是该庆祝一下呢？也许吧……不过也有可能这种0.2分的提升只是偶然现象而已。

如果使用包含50到100个样本的黄金数据集，仅变异因素就足以导致如此大的分数差异。因此，你需要进行真正的统计测试才能确定这种变化是否真实存在。

如果你已经很久没有接触过统计学知识了，这里有一个简单的介绍：配对t检验是一种用于比较两组相互关联的数据的方法。在我们的例子中，每一组数据都代表在同一系统不同版本下对同一问题进行的测量结果：旧提示语版本和新提示语版本下的测试结果。

这种测试会逐一分析每组数据，计算每个问题的分数变化幅度，然后判断这些变化是呈一致的方向还是随机分布的。

如果变化趋势是一致的（即大多数问题的分数在新提示语下都提高了），那么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:
    """判断分数提升是否具有统计显著性。

    由于两次测试中评估的是相同的问题，因此使用配对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": (
            "可以安全部署"
            if p_value < alpha and mean_diff > 0
            else "不要部署——这种变化并不具有显著意义"
        ),
    }


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"平均分变化：{result['mean_before']} → {result['mean_after']}")
    print(f"P值：{result['p_value']}")
    print(f>"是否显著：{result['is_significant']}")
    print(f"建议：{result['recommendation']}")

如果p值低于0.05，那么这种改进纯粹是偶然发生的概率就小于5%。在这种情况下，就可以将其正式投入使用。而如果p值高于这个数值，那么这种改进可能只是随机现象，因此无论各项指标看起来有多好，都不应该将其应用实际环境中。

如何将所有环节整合起来：完整的评估架构

让我们把这三层评估机制连接成一个统一的整体。这个“协调器”类负责将所有环节串联起来。它首先会进行确定性检查，如果这些检查通过，就会交由大语言模型来进行进一步判断；必要时，还会引入人工评估来进行校准。

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": "在进行更深入的评估之前，先解决结构上的问题。",
            }

        # 第二层：大语言模型判断
        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,
        }

我希望早些时候能知道这些

最后，我想说一些希望在开始构建评估系统之前就能有人告诉我的事情。

首先，不要一次性构建所有三层功能。先从那些基于确定性规则的检查功能开始实现，然后立即将其投入使用。你会惊讶地发现，这些功能本身就能发现很多问题；而且编写这些检查代码的过程会迫使你明确界定“对于你的应用程序来说，什么是正确的输出结果”。在需要时再添加基于大语言模型的评估机制，之后再逐步引入人工评估环节。

其次，每月都要将你的评估模型与人类的评估结果进行一次对比。使用你的大型语言模型对那些已经有人类评分的20到30个样本进行评估。如果该模型的评分平均偏离了0.5分以上，那就说明某些情况发生了变化：可能是评估模型被更新了，也可能是你的评分标准遗漏了某些新的错误类型。无论哪种情况，你都需要重新调整评估模型。

第三，每一次应用程序出现的故障都应被视为一个测试案例。这或许是最有用的习惯。当某个功能出现故障时，其实就为你的数据集增添了一个新的测试样本。经过几个月的积累，你的数据集将不再是一个通用的测试工具，而会成为记录你的应用程序所有故障情况的详细档案。

最后，不要追求完美的评估分数。我见过一些团队不断调整输入内容，只为了让他们的评估分数从4.2提高到4.5，结果却发现他们的评分标准存在漏洞，用户的满意度依然没有提高。评估分数只是工具，而非目标；人类评估的存在正是为了弥补数字评估所无法发现的问题。

总结

在这篇文章中，我们讨论了很多内容。现在让我把这些要点归纳一下：核心问题在于，大型语言模型的故障表现与传统软件截然不同——它们不会崩溃，也不会生成错误日志或堆栈追踪信息，只会给出看似“正确”但实际上错误的答案。

由于这些输出结果具有不确定性，因此不能用简单的断言方法来测试它们，需要采取完全不同的评估方式。

这种评估方式就是分层评估流程：

• 第一层（确定性检查）用于处理最基本的问题：输出结果是否有效、长度是否合适，以及其中是否包含虚假的URL链接？这些检查速度快、成本低，而且能发现许多你意想不到的问题。

• 第二层（以大型语言模型作为评估工具）侧重于语义层面的评估：响应内容是否相关、准确且完整？通过为评估模型提供明确的评分标准，你可以获得相当可靠且自动化的质量评估结果。

• 第三层（人类评估）用于确保整个评估系统的准确性。定期安排少量人工审核，可以发现自动化评估所遗漏的细节问题，比如语言风格、表达清晰度，以及“正确”与“真正有帮助”之间的区别。

除了这三层评估机制外，你还学会了如何利用黄金数据集构建回归测试流程，从而在问题影响到生产环境之前就发现它们。同时，你也了解了如何通过统计显著性测试来验证你的改进措施是否真实有效，而非仅仅是偶然现象。

如果让我给大家提一条建议的话，那就是：从小处着手。不要试图在一个周末内完成所有这些工作。今天就把`DeterministicEvaluator`类添加到你的项目中吧——这只需要五分钟时间，就能立刻帮助你发现目前存在的问题。等你对更深入的评估做好准备时，再加入大型语言模型作为评估工具；随着应用程序的发展，逐步引入人类审核和回归测试环节。

那些能够推出可靠的人工智能产品的团队，并不是那些拥有最先进模型的团队。而是那些建立了相应的检测机制，能够及时发现这些模型出现的故障，并在用户察觉之前就解决问题的团队。