评估 DSPy 流程
使用指标、开发集和 evaluate() 函数进行自动化评估。
评估 DSPy 流程 是 CoddyKit 上的免费 AI Prompt Engineering 课时。 这是第 4 节课,共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Prompt Engineering 学习路径的一部分,你的进度在网页和 CoddyKit 应用中同步。 AI Prompt Engineering 课程共包含 4 节课。
为什么评估在 DSPy 中至关重要
DSPy 优化的效果取决于您的评估质量。薄弱的指标会产生一个在该指标上得分很高、但在生产环境中却无法正常工作的编译程序。适当的评估框架可以让您比较未优化程序与优化程序,并在更新管道时发现回归问题。
dspy.Evaluate 类
dspy.Evaluate 会在数据集上运行您的程序,应用指标并报告汇总分数。它通过 num_threads 支持并行处理,从而能够快速评估大型数据集。
import dspy
# Build a devset of labeled examples
devset = [
dspy.Example(question='What is 7 * 8?', answer='56').with_inputs('question'),
dspy.Example(question='Name the largest planet.', answer='Jupiter').with_inputs('question'),
# ... more examples
]
# Create evaluator
evaluate = dspy.Evaluate(
devset=devset,
metric=exact_match_metric, # Your metric function
num_threads=4, # Parallel evaluation
display_progress=True, # Show progress bar
display_table=True, # Show per-example results
)
# Run
score = evaluate(my_program)
print(f'Overall score: {score:.1%}')编写指标函数
指标函数的签名为 (example, prediction, trace=None) -> float。它会将程序的预测结果与 example 中的真实答案进行比较。
在优化期间(而不是评估期间),trace 参数不为空;您可以利用这一点,在编译和评估时应用不同的逻辑。
import dspy
def exact_match_metric(example, prediction, trace=None):
return float(
example.answer.strip().lower() == prediction.answer.strip().lower()
)
def contains_metric(example, prediction, trace=None):
"""Check if expected answer appears anywhere in prediction."""
return float(example.answer.lower() in prediction.answer.lower())
def length_penalized_metric(example, prediction, trace=None):
"""Reward correct answers, penalize overly long ones."""
correct = float(example.answer.lower() in prediction.answer.lower())
length_ok = float(len(prediction.answer.split()) <= 20)
return correct * (0.8 + 0.2 * length_ok)
# Use any of these as the metric parameter
evaluate = dspy.Evaluate(devset=devset, metric=contains_metric)通过/失败阈值模式
对于二元指标,您可以定义一个阈值:如果预测结果达到最低质量标准,就视为“通过”。在自举优化期间筛选少样本示例时,这非常有用。
import dspy
def quality_metric(example, prediction, trace=None):
"""
Multi-factor metric with pass/fail threshold.
Returns float 0.0 to 1.0.
During compilation (trace is not None), DSPy uses this to decide
which traces to bootstrap as demos.
"""
score = 0.0
# Factor 1: Factual correctness (0.6 weight)
if example.answer.lower() in prediction.answer.lower():
score += 0.6
# Factor 2: Conciseness (0.4 weight)
word_count = len(prediction.answer.split())
if word_count <= 15:
score += 0.4
elif word_count <= 30:
score += 0.2
# During optimization: only use examples scoring >= 0.6
if trace is not None:
return score >= 0.6
return score拆分数据:训练集、开发集、测试集
在 DSPy 中遵循标准的机器学习数据拆分实践:
- 训练集:供优化器自举生成示例(20–200 个示例)
- 开发集:供优化器在搜索过程中进行验证
- 测试集:完全留出,仅用于最终评估
import random
# All labeled examples
all_examples = load_examples() # Returns list of dspy.Example
random.shuffle(all_examples)
total = len(all_examples)
train_end = int(total * 0.6)
dev_end = int(total * 0.8)
trainset = all_examples[:train_end] # 60% for optimization
devset = all_examples[train_end:dev_end] # 20% for validation
testset = all_examples[dev_end:] # 20% held out
print(f'Train: {len(trainset)}, Dev: {len(devset)}, Test: {len(testset)}')比较优化程序与未优化程序
始终在同一个测试集上,将编译后的程序与基线(未编译)程序进行基准测试。这样可以证明优化确实带来了帮助,并量化改进幅度。
import dspy
evaluate = dspy.Evaluate(
devset=testset,
metric=exact_match_metric,
num_threads=4,
display_progress=True,
)
# Baseline: unoptimized program
baseline_score = evaluate(unoptimized_program)
print(f'Baseline (no optimization): {baseline_score:.1%}')
# BootstrapFewShot compiled
bs_score = evaluate(bootstrap_compiled_program)
print(f'BootstrapFewShot compiled: {bs_score:.1%}')
# MIPRO compiled
mipro_score = evaluate(mipro_compiled_program)
print(f'MIPRO compiled: {mipro_score:.1%}')
# Pick the winner
print(f'Best improvement: +{max(bs_score, mipro_score) - baseline_score:.1%}')使用 num_threads 实现并行处理
大型评估集如果按顺序处理,可能需要数小时。dspy.Evaluate 中的 num_threads 会并行运行预测,从而按比例缩短实际耗时。
请根据您的应用程序接口速率限制设置 num_threads;线程过多会触发速率限制错误。
import dspy
import time
devset = [...] # 200 examples
# Sequential evaluation
start = time.time()
evaluate_seq = dspy.Evaluate(devset=devset, metric=metric, num_threads=1)
score_seq = evaluate_seq(program)
print(f'Sequential: {time.time()-start:.0f}s')
# Parallel evaluation (4 threads)
start = time.time()
evaluate_par = dspy.Evaluate(devset=devset, metric=metric, num_threads=4)
score_par = evaluate_par(program)
print(f'Parallel (4 threads): {time.time()-start:.0f}s')
# Typically ~4x faster — same score, less wait time解读评估输出
当 display_table=True 时,DSPy 会显示一个详细表格,其中包含每个示例、预测结果以及是否通过指标。这对于诊断失败模式非常有价值。
请关注以下情况:某类问题上的系统性失败、指标的边界情况,或训练集未覆盖的示例。
import dspy
evaluate = dspy.Evaluate(
devset=devset,
metric=exact_match_metric,
num_threads=2,
display_progress=True,
display_table=10, # Show first 10 rows of results table
return_outputs=True, # Return (score, outputs) tuple
)
score, outputs = evaluate(program, return_all_scores=True)
# Find failing examples
failures = [
(ex, pred, s)
for ex, pred, s in outputs
if s == 0.0
]
print(f'Failures: {len(failures)}/{len(devset)}')
for ex, pred, _ in failures[:3]:
print(f'Q: {ex.question}')
print(f'Expected: {ex.answer}')
print(f'Got: {pred.answer}')使用 LLM 评分的指标
对于精确匹配无法处理的开放式输出,可以使用 LLM 对质量进行评分。DSPy 让这一点很容易实现——您的指标函数本身就可以调用 DSPy 预测器。
import dspy
class GradeAnswer(dspy.Signature):
"""Grade whether the predicted answer is correct given the reference."""
question: str = dspy.InputField()
reference_answer: str = dspy.InputField()
predicted_answer: str = dspy.InputField()
is_correct: bool = dspy.OutputField(
desc='True if the predicted answer is semantically correct'
)
grader = dspy.Predict(GradeAnswer)
def llm_graded_metric(example, prediction, trace=None):
result = grader(
question=example.question,
reference_answer=example.answer,
predicted_answer=prediction.answer,
)
return float(result.is_correct)
# Use this metric when answers can vary in phrasing
evaluate = dspy.Evaluate(devset=devset, metric=llm_graded_metric)使用评估进行回归测试
请将 DSPy 评估套件视为测试套件。每当您更新签名、模块架构或训练数据时,都应重新运行评估并比较分数,以发现回归问题。
import json
import dspy
def run_and_save_evaluation(program, program_name, testset, metric):
evaluate = dspy.Evaluate(
devset=testset,
metric=metric,
num_threads=4,
)
score = evaluate(program)
# Save score to history file
history_file = 'eval_history.json'
try:
with open(history_file) as f:
history = json.load(f)
except FileNotFoundError:
history = []
history.append({'program': program_name, 'score': score})
with open(history_file, 'w') as f:
json.dump(history, f, indent=2)
print(f'{program_name}: {score:.1%}')
return score评估最佳实践
DSPy 管道的评估原则:
- 严格留出测试集,绝不要在测试集上进行优化
- 至少使用 50–100 个测试示例,以获得可靠的分数
- 使指标与实际生产目标相匹配
- 比较多个优化器,因为结果会因任务而异
- 持续跟踪分数,以检测回归问题
- 手动检查失败案例,以改进训练数据
知识检查:指标函数的 Trace 参数
在 DSPy 指标函数中,不为空的 trace 参数表示什么?
回顾:评估 DSPy 管道
dspy.Evaluate 会在带标签的开发集上运行您的程序,应用指标函数并报告汇总分数。指标函数遵循 (example, prediction, trace=None) -> float 这一模式。使用 num_threads 进行并行评估,并使用 display_table=True 诊断失败。始终在留出的测试集上比较优化程序与未优化程序。对于开放式输出,LLM 评分指标的效果优于精确字符串匹配。
常见问题解答
「评估 DSPy 流程」课时是免费的吗?
是的 — 「评估 DSPy 流程」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 AI 导师)并解锁 AI Prompt Engineering 课程的其余内容,请升级到 CoddyKit PRO。 AI Prompt Engineering 课程共包含 4 节课。
「评估 DSPy 流程」这节课中我会学到什么?
使用指标、开发集和 evaluate() 函数进行自动化评估。 你通过在浏览器中直接运行的动手代码来练习 AI Prompt Engineering,全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Prompt Engineering 需要有经验吗?
无需任何先前经验。CoddyKit 上的 AI Prompt Engineering 课程适合初学者到高级学习者,你可以从这里开始或从头开始,按照自己的节奏学习。 这是第 4 节课,共 4 节。
「评估 DSPy 流程」课时需要多长时间?
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动,所以你能稳步进步,并在网页和应用中从离开的地方继续。
我能在这节 AI Prompt Engineering 课中编写并运行代码吗?
能。每节 AI Prompt Engineering 课都包含内置代码编辑器,你可以在浏览器中直接编写并运行真实代码,并获得即时 AI 反馈 — 无需本地设置。