如何构建可自我评估的 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 评估的三层架构
经过大量试错,我确定了一种方法,采用三层堆叠架构,从廉价快速到昂贵详尽。
第一层是确定性检查。把这视为门口的保安。输出在需要 JSON 时是否为合法 JSON?是否短得可疑或长得不合常理?是否包含幻觉生成的 URL?这些检查即时、免费,且能捕捉到比你预期更多的错误。
第二层是 LLM-as-judge。用另一次 LLM 调用来给主模型的输出打分:"这个回答相关吗?准确吗?真的有帮助吗?"像
gpt-4o-mini这样的模型在给其他模型打分方面出奇地好,只要给它一个清晰的评分标准。第三层是人工评估。让真人来审查真实输出。不可能对每条回复都这么做,那太慢了,但可以定期抽查,确保自动化层没有偏离"好"的真正含义。
关键在于知道什么时候该用哪一层,接下来我们会把这三层都搭建出来。
如何搭建第一层:确定性检查
我刚开始做评估流水线时,直接跳到了那些花哨的东西: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}")
这段代码里有几个细节值得特别强调。
温度设为零:你不是在让评判模型搞创作。你要的是相同的输入每次都产出相同的分数,或者尽可能接近。
输出是结构化 JSON:这一条我是踩坑才学到的。如果让评判模型自由回答文本,你最后还得写一堆脆弱的解析代码来提取分数。强制它输出 JSON,日子会好过得多。
评分标准内置于每个 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 产品的团队,往往不是拥有最花哨模型的团队,而是那些搭建了支撑体系、能在模型失效时及时察觉,并抢在用户之前捕捉到问题的团队。