dspy-evaluation-harness

DSPy 评估工具集（3.2.x）

指标通常比程序更重要。尤其对于 dspy.GEPA，指标中文本反馈的质量决定了优化能否收敛。

两条规则

1. 返回 dspy.Prediction(score=..., feedback=...)，而不是字典。 dspy.Evaluate 的并行执行器通过求和聚合分数，这在输出为字典时会出错（TypeError: unsupported operand type(s) for +: 'int' and 'dict'）。dspy.Prediction 支持 __float__/__add__，也是 GEPA 适配器原生支持解包的类型。单独返回浮点数仍适用于纯 dspy.Evaluate 评分，但 GEPA 需要分数和反馈这两个值。
2. 使用独立的验证集。 切勿在同一批示例上进行优化和评估。优化器很容易过拟合。

标准的丰富反馈指标

import dspy

def rich_metric(gold: dspy.Example, pred: dspy.Prediction, trace=None,
                pred_name: str | None = None, pred_trace=None):
    # 1. Compute sub-scores — multi-axis beats scalar
    correctness = 1.0 if _normalize(pred.answer) == _normalize(gold.answer) else 0.0
    cited = _has_citation(pred.answer, gold.sources) if hasattr(gold, "sources") else 1.0
    concise = 1.0 if len(pred.answer.split()) <= 50 else 0.5
    score = 0.6 * correctness + 0.25 * cited + 0.15 * concise

    # 2. Write feedback that teaches the optimizer
    parts = []
    if correctness < 1.0:
        parts.append(
            f"Answer mismatch. Predicted: {pred.answer!r}. Expected: {gold.answer!r}. "
            f"Likely cause: reasoning skipped the units/quantity in the question."
        )
    if cited < 1.0:
        parts.append("Did not ground the claim in the provided sources. Quote a source fragment.")
    if concise < 1.0:
        parts.append("Answer exceeded 50 words — tighten to one sentence.")
    if not parts:
        parts.append("Correct, grounded, and concise.")
    feedback = " ".join(parts)

    return dspy.Prediction(score=score, feedback=feedback)

标准评估工具集

evaluator = dspy.Evaluate(
    devset=valset,
    metric=rich_metric,
    num_threads=8,
    display_progress=True,
    display_table=10,          # pretty-print first 10 rows
    provide_traceback=True,    # surface exceptions, don't swallow them
    max_errors=5,
    failure_score=0.0,
    save_as_json="eval_runs/baseline.json",
)
result = evaluator(program)
print("Overall:", result.score)
for example_result in result.results[:3]:
    print(example_result)

dspy.Evaluate 会返回一个 EvaluationResult，其中包含 .score（聚合浮点数）和 .results（由 (example, pred, score) 元组组成的列表）。

数据集规范

• 规模：对于 GEPA 的反思循环，20–50 个示例就足够；对于 MIPROv2 风格的引导，建议使用 100–500 个示例。
• 划分：手动整理两个互不重叠的集合——trainset（用于优化）和 valset（用于评估优化后的程序）。在开发过程中从不查看的测试集最为理想。
• 代表性比规模更重要。请纳入边缘案例、歧义情况和对抗性输入。
• 构建 dspy.Example(...).with_inputs("question", "context")——with_inputs 调用会标记哪些字段是输入，哪些字段是标准输出。

trainset = [
    dspy.Example(question="…", answer="…").with_inputs("question"),
    ...
]

多维指标（推荐）

综合正确性、忠实度、格式遵循度、延迟和成本。每个维度都应是一个定义明确的 0–1 浮点数。明确设置各项权重；不要把权重藏在神秘数字中——应将它们定义为常量，以便告知优化器需要如何权衡。

适用于 CI 的评估套件

# tests/test_dspy_eval.py
import dspy, pytest
from my_program import program, valset, rich_metric

@pytest.fixture(scope="module")
def evaluator():
    return dspy.Evaluate(devset=valset, metric=rich_metric, num_threads=8,
                         display_progress=False, provide_traceback=True)

def test_program_meets_threshold(evaluator):
    result = evaluator(program)
    assert result.score >= 0.75, f"Regression: {result.score:.3f}"

在 CI 中使用缓存的 LM（dspy.LM(..., cache=True)）和预先填充的 DSPY_CACHEDIR 离线运行。

跟踪与可观测性

• 在 dspy.configure 中设置 track_usage=True，即可累计预测中的 token 数（pred.get_lm_usage()）。
• MLflow：import mlflow; mlflow.dspy.autolog() → 跟踪每次预测。
• W&B：向 dspy.GEPA 传入 use_wandb=True，即可记录帕累托前沿。
• 始终将评估结果记录到带版本号的 JSON 文件中（save_as_json=...），以便对比运行结果。

反模式

• 使用 GEPA 时，只返回标量的指标（有浮点数，但没有反馈）——浪费信号。
• return {"score": s, "feedback": f}（字典）——会导致 dspy.Evaluate 的并行聚合器崩溃。请使用 dspy.Prediction(score=s, feedback=f)。
• 对开放式生成任务使用精确匹配指标——请使用语义评分或 LM-as-judge 评分。
• 在训练集上评估——结果会乐观 10–30 分。
• 静默吞掉异常（provide_traceback=False）——最终你会误以为是 LM 导致了 KeyError。
• 在实验中途更改指标，却不重新建立基线——此前的结果将无法比较。

下一步

• 将此指标接入 dspy-gepa-optimizer。
• 完整工具集模式 → reference.md。
• 可运行示例 → example_metric.py。