构建自动化评估工具
创建可重复的评估流程，让完整的 RAG 系统在测试集上运行，计算所有指标并生成报告，以便持续跟踪改进情况。
构建自动化评估工具 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
什么是评估工具？
评估工具是一条可重复运行的自动化流水线，它会针对标准化测试集运行完整的 RAG 系统，计算所有指标并生成报告。关键在于“可重复”：每次您修改分块策略、嵌入模型、提示词或 LLM 时，都运行同一个评估工具，并将结果与基线进行比较。这会把 RAG 开发从主观试错转变为数据驱动的工程实践。
评估工具架构
一个设计良好的评估工具包含四个层次：测试数据管理（加载黄金数据集并进行版本管理）、流水线执行（让每个测试问题通过完整的 RAG 流水线）、指标计算（计算所有检索和生成指标），以及报告生成（保存带有版本信息的结果，并与之前的基线生成差异）。每一层都应能够独立测试和配置。
class RAGEvaluationHarness:
def __init__(self, retriever, llm_client, config):
self.retriever = retriever
self.llm_client = llm_client
self.config = config # chunk_size, top_k, model, threshold, etc.
self.results = []
def run(self, golden_dataset):
for item in golden_dataset:
result = self._evaluate_single(item)
self.results.append(result)
metrics = self._compute_metrics()
self._save_report(metrics)
return metrics运行每个测试用例
对于黄金数据集中的每个问题，运行完整的 RAG 流水线，并捕获所有中间输出：检索到的片段 ID 和分数、格式化后的上下文、生成的答案以及令牌数量。保存这些中间值对于调试故障至关重要——当某个问题得分较低时，您可以准确检查检索到了哪些片段以及答案为何错误，而无需重新运行成本高昂的流水线。
import time
def _evaluate_single(self, item):
start = time.perf_counter()
query_vector = embed_query(item['question'])
chunks = self.retriever.retrieve(query_vector, top_k=self.config['top_k'])
filtered_chunks = filter_by_score(chunks, self.config['threshold'])
context = format_context(filtered_chunks)
answer_result = generate_answer(item['question'], context, self.llm_client)
latency_ms = (time.perf_counter() - start) * 1000
return {
'question': item['question'],
'expected_answer': item['answer'],
'generated_answer': answer_result['answer'],
'retrieved_chunk_ids': [c['id'] for c in filtered_chunks],
'retrieved_scores': [c['score'] for c in filtered_chunks],
'relevant_chunk_ids': item['relevant_chunk_ids'],
'context_texts': [c['text'] for c in filtered_chunks],
'tokens_used': answer_result['tokens_used'],
'latency_ms': round(latency_ms)
}一次性计算所有指标
收集完所有测试用例的输出后，只需遍历一次结果即可计算完整的指标集。将检索指标（根据片段 ID 计算）与生成指标（通过调用评审 LLM 计算）分开。批量执行评审 LLM 调用以提高效率——将忠实性评估分组，并使用 asyncio 并行发送，而不是按顺序执行。由于生成指标可能需要数分钟来处理 100 个以上的测试用例，请记录进度。
def _compute_metrics(self):
# Retrieval metrics (no LLM calls needed)
hit_rates = []
mrr_scores = []
for r in self.results:
retrieved = r['retrieved_chunk_ids']
relevant = set(r['relevant_chunk_ids'])
hit = any(rid in relevant for rid in retrieved)
hit_rates.append(1.0 if hit else 0.0)
for rank, rid in enumerate(retrieved, 1):
if rid in relevant:
mrr_scores.append(1.0 / rank)
break
else:
mrr_scores.append(0.0)
metrics = {
'hit_rate_at_5': sum(hit_rates) / len(hit_rates),
'mrr': sum(mrr_scores) / len(mrr_scores),
'mean_latency_ms': sum(r['latency_ms'] for r in self.results) / len(self.results),
'mean_tokens': sum(r['tokens_used'] for r in self.results) / len(self.results)
}
return metrics保存带有版本信息的结果
每次评估运行都应保存版本元数据，以便比较不同配置下的结果。请包含代码的 git 提交哈希、配置参数（嵌入模型、片段大小、K 值、阈值、LLM 模型）、时间戳以及便于人类理解的运行说明。将结果存储在 JSONL 文件或数据库表中。这样就能永久记录系统的演进过程。
import json
import subprocess
from datetime import datetime
def _save_report(self, metrics):
git_hash = subprocess.check_output(
['git', 'rev-parse', '--short', 'HEAD']
).decode().strip()
report = {
'run_id': datetime.utcnow().strftime('%Y%m%d_%H%M%S'),
'git_commit': git_hash,
'config': self.config,
'metrics': metrics,
'n_test_cases': len(self.results),
'timestamp': datetime.utcnow().isoformat()
}
with open('eval_history.jsonl', 'a') as f:
f.write(json.dumps(report) + '\n')
print(f'Saved evaluation run: {report["run_id"]}')
print(json.dumps(metrics, indent=2))与基线进行比较
每次运行后，自动与之前的基线进行比较并标记回归。指标下降超过某个阈值（例如 2 个百分点）即可视为回归。打印显示指标变化的差异表。如果任何指标出现显著回归，评估运行应以非零退出代码失败，从而使持续集成/持续交付流水线阻止部署该变更。
def compare_to_baseline(current_metrics, baseline_file='best_eval.json'):
import json
from pathlib import Path
if not Path(baseline_file).exists():
print('No baseline yet. Saving current as baseline.')
Path(baseline_file).write_text(json.dumps(current_metrics, indent=2))
return True
baseline = json.loads(Path(baseline_file).read_text())
regressions = []
print('\nMetric comparison (current vs baseline):')
for metric, current_val in current_metrics.items():
baseline_val = baseline.get(metric, 0)
delta = current_val - baseline_val
status = 'OK' if delta >= -0.02 else 'REGRESSION'
print(f' {metric}: {current_val:.3f} vs {baseline_val:.3f} ({delta:+.3f}) {status}')
if status == 'REGRESSION':
regressions.append(metric)
return len(regressions) == 0集成到持续集成/持续交付流程
评估工具集成到持续集成/持续交付流水线后，才能发挥最大作用。请将其配置为：每当修改分块逻辑、嵌入模型配置、提示词模板或检索参数的拉取请求提交时，自动运行。只有当所有指标达到最低阈值，且没有指标相对于主分支基线发生回归时，流水线才会通过。这样可以防止意外的质量回归进入生产环境。
# GitHub Actions workflow (eval.yml)
# on:
# pull_request:
# paths:
# - 'rag/**'
# - 'prompts/**'
# - 'config/**'
# jobs:
# evaluate:
# runs-on: ubuntu-latest
# steps:
# - uses: actions/checkout@v3
# - name: Install dependencies
# run: pip install -r requirements.txt
# - name: Run evaluation harness
# run: |
# python eval/run_harness.py \
# --test-set eval/golden_dataset.json \
# --config config/rag_config.yaml \
# --fail-on-regression
# env:
# OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}生成便于人类阅读的报告
除了原始指标文件外，还应生成一份便于人类阅读的 HTML 或 Markdown 报告，供团队在拉取请求评论中查看。报告应包含所有指标的摘要表、失败测试用例列表（包括问题、预期答案、生成的答案以及检索到的片段），以及展示最近 10 次运行指标变化趋势的图表。可视化报告有助于非技术相关人员了解系统是否正在改进。
def generate_markdown_report(metrics, failed_cases, run_id):
lines = [
f'# RAG Evaluation Report — {run_id}\n',
'## Summary Metrics',
'| Metric | Score | Target |',
'|--------|-------|--------|',
f'| Hit Rate@5 | {metrics["hit_rate_at_5"]:.1%} | > 80% |',
f'| MRR | {metrics["mrr"]:.3f} | > 0.70 |',
f'| Mean Latency | {metrics["mean_latency_ms"]:.0f}ms | < 500ms |',
'',
f'## Failed Cases ({len(failed_cases)} failures)'
]
for case in failed_cases[:10]: # show first 10
lines += [
f'**Q:** {case["question"]}',
f'**Expected:** {case["expected_answer"]}',
f'**Generated:** {case["generated_answer"]}\n'
]
return '\n'.join(lines)跟踪每次评估运行的成本
评估运行需要成本——它会调用嵌入 API、LLM API 和评审 LLM。请将每次评估运行的成本与质量指标一起跟踪。对 100 个测试用例进行全面评估，通常需要花费 0.50 至 2.00 美元，具体取决于所使用的模型。评审调用可以使用成本更低的模型（例如使用 GPT-4o-mini 进行忠实性评分），而将昂贵的模型留给生成任务。请在保存的报告中包含预计运行成本，以便将评估成本纳入开发周期预算。
def estimate_run_cost(results, config):
# Embedding cost
embed_tokens = sum(len(r['question'].split()) * 1.3 for r in results)
embed_cost = (embed_tokens / 1_000_000) * 0.02 # $0.02/1M tokens
# Generation cost
total_gen_tokens = sum(r['tokens_used'] for r in results)
gen_cost = (total_gen_tokens / 1_000_000) * 5.0 # gpt-4o approx
# Judge cost (faithfulness evals)
judge_cost = len(results) * 0.001 # ~$0.001 per eval with gpt-4o-mini
total = embed_cost + gen_cost + judge_cost
print(f'Evaluation cost estimate: ${total:.2f}')
print(f' Embedding: ${embed_cost:.3f}')
print(f' Generation: ${gen_cost:.3f}')
print(f' Judgment: ${judge_cost:.3f}')
return total安排生产环境评估以进行监控
除了针对代码变更执行持续集成/持续交付评估外，还应按计划在生产环境中运行评估工具——每天或每周一次——使用从日志中抽取的真实用户查询进行测试。这可以检测数据漂移：随着文档语料库不断变化、用户查询模式发生转移，即使代码没有任何变更，系统质量也可能下降。请安排每周评估运行，抽取最近的 50 个用户查询，对其进行评估，并自动将质量摘要发送到团队的 Slack 频道。
# Example scheduled evaluation (cron job or scheduled cloud function)
import random
def sample_production_queries(query_log_file, n=50):
with open(query_log_file) as f:
all_queries = [json.loads(line) for line in f]
sample = random.sample(all_queries, min(n, len(all_queries)))
# Convert to golden dataset format (without expected answers — use LLM judge)
return [
{'question': q['user_question'], 'relevant_chunk_ids': []}
for q in sample
]
# Run weekly evaluation against production queries
if __name__ == '__main__':
prod_queries = sample_production_queries('/var/log/rag_queries.jsonl')
harness = RAGEvaluationHarness(retriever, llm_client, config)
metrics = harness.run(prod_queries)
send_slack_digest(metrics)可视化随时间变化的指标趋势
JSONL 文件中的原始数字很难一眼看懂。请构建一个简单的趋势可视化，用折线图绘制最近 20 次评估运行中的每项指标。使用运行时间戳作为横轴，指标分数作为纵轴。在图中绘制一条位于最低可接受阈值处的水平线。当指标跌破阈值线时，无需阅读原始数据即可立即发现问题。Matplotlib 或简单的网页仪表板（Grafana、Streamlit）等工具都很适合完成这项工作。
import json
import matplotlib.pyplot as plt
from pathlib import Path
def plot_metric_trends(history_file='eval_history.jsonl', metric='hit_rate_at_5'):
records = [
json.loads(line)
for line in Path(history_file).read_text().strip().split('\n')
]
timestamps = [r['timestamp'][:10] for r in records[-20:]]
scores = [r['metrics'].get(metric, 0) for r in records[-20:]]
plt.figure(figsize=(10, 4))
plt.plot(timestamps, scores, marker='o', label=metric)
plt.axhline(y=0.80, color='r', linestyle='--', label='Min threshold')
plt.title(f'{metric} over last 20 evaluations')
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig(f'eval_trend_{metric}.png')
print(f'Saved trend chart for {metric}')快速检查
请检验您对本课 AI 工程概念的理解。
课程回顾
在本课中，您学习了：如何构建完整的评估工具，包括测试数据管理、流水线执行、指标计算和报告生成；如何将运行结果与基线进行比较，并在发生回归时使持续集成/持续交付流程失败；如何生成便于人类阅读的报告供团队审查；以及如何运行定期的生产环境监控，在不修改代码的情况下检测数据漂移。现在，您已经具备构建和评估生产级 RAG 系统的完整基础。
常见问题解答
「构建自动化评估工具」课时是免费的吗？
是的 — 「构建自动化评估工具」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「构建自动化评估工具」这节课中我会学到什么？
创建可重复的评估流程，让完整的 RAG 系统在测试集上运行，计算所有指标并生成报告，以便持续跟踪改进情况。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「构建自动化评估工具」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。