Metric:把评估指标写成函数

Metric 的函数签名

def metric(example: dspy.Example, pred: dspy.Prediction, trace=None) -> float:
    """
    example:  标注数据,带输入 + 期望输出
    pred:     Module 的实际预测
    trace:    可选,调用过程的完整 trace(Optimizer 有时会传)
    返回:     float,通常 [0, 1],也可以是 bool(会被转成 1/0)
    """

最简单:精确匹配

def exact_match(ex, pred, trace=None):
    return ex.answer.strip().lower() == pred.answer.strip().lower()

适合分类、单词答案。问题:答案稍微多一个空格就判错。

更宽松:子串匹配 / F1

def contains(ex, pred, trace=None):
    return ex.answer.lower() in pred.answer.lower()

def token_f1(ex, pred, trace=None):
    import re
    a = set(re.findall(r"\w+", ex.answer.lower()))
    b = set(re.findall(r"\w+", pred.answer.lower()))
    if not a or not b: return 0.0
    p = len(a & b) / len(b)
    r = len(a & b) / len(a)
    return 2 * p * r / (p + r) if (p + r) else 0.0

SQuAD 问答常用 F1,适合"答案核心对就行,表达方式自由"的任务。

LLM-as-Judge:让另一个 LLM 评分

主观任务(风格、流畅度、事实性)没法用字符串匹配,请 LLM 当评委:

class Judge(dspy.Signature):
    """判断预测答案是否正确回答了问题,给 1-5 分。"""
    question: str = dspy.InputField()
    expected: str = dspy.InputField(desc="参考答案,可能不唯一")
    predicted: str = dspy.InputField()
    score: int = dspy.OutputField(desc="1=完全错,3=部分对,5=完全对")
    reason: str = dspy.OutputField()

judge = dspy.ChainOfThought(Judge)

def llm_metric(ex, pred, trace=None):
    out = judge(question=ex.question, expected=ex.answer, predicted=pred.answer)
    return out.score / 5.0

组合多指标:加权或 pass/fail

def combined(ex, pred, trace=None):
    # 1) 答案正确性 70%
    correctness = token_f1(ex, pred)
    # 2) 引用合规 30%(必须有 citation)
    has_cite = float(bool(getattr(pred, "citations", None)))
    return 0.7 * correctness + 0.3 * has_cite

或者分层(第二关必须先过第一关):

def gated(ex, pred, trace=None):
    if not pred.answer:
        return 0.0
    if ex.answer.lower() not in pred.answer.lower():
        return 0.0
    if len(pred.answer) > 500:
        return 0.5      # 太长扣分
    return 1.0

trace 参数:多步程序的细粒度打分

def mh_metric(ex, pred, trace=None):
    if trace is None:
        return exact_match(ex, pred)
    # trace 是 list[(predictor, inputs, Prediction)]
    # 可以对每一步单独打分,用于更细的优化
    queries = [str(call[1].get("search_query", "")) for call in trace]
    diverse = len(set(queries)) == len(queries)   # 每次 query 不同
    return exact_match(ex, pred) and diverse

BootstrapFewShot 会传 trace,让你可以对"过程合理性"而非仅"结果"打分。

Metric 设计的三大陷阱

陷阱 1:可被 hack 的指标

# 坏:模型只要拷贝 context 就能过
def bad(ex, pred, trace=None):
    return ex.answer in pred.answer

# 好:要求答案简短,强迫模型真的回答
def better(ex, pred, trace=None):
    ok = ex.answer.lower() in pred.answer.lower()
    short = len(pred.answer) < 200
    return float(ok and short)

陷阱 2:二值指标太稀疏

只返回 0/1 的指标,对 MIPRO 之类有一定探索随机性的 Optimizer 不友好——小幅改进看不出来。尽量用 [0,1] 连续值。

陷阱 3:Metric 本身不稳

LLM-as-Judge 每次调用分数可能不同,用 temperature=0 再加 3 次投票平均,分数波动 < 0.05 才算可用。

评估 Module

from dspy.evaluate import Evaluate

evaluator = Evaluate(
    devset=devset,
    metric=combined,
    num_threads=8,
    display_progress=True,
    display_table=10,   # 打印 10 条样例结果
)

score = evaluator(my_module)
print(f"分数: {score:.3f}")

常用 Metric 模板

# JSON 合法性
def json_valid(ex, pred, trace=None):
    import json
    try:
        json.loads(pred.output); return 1.0
    except: return 0.0

# 引用召回(RAG)
def citation_recall(ex, pred, trace=None):
    gold = set(ex.gold_doc_ids)
    got  = set(pred.doc_ids)
    return len(gold & got) / len(gold) if gold else 0

# 分类 macro-F1
def macro_f1(ex, pred, trace=None):
    return 1.0 if ex.label == pred.label else 0.0
# (batch 级再求 sklearn.metrics.f1_score(..., average='macro'))

打分和优化的关系

Metric 形状	Optimizer 建议
二值(0/1)	BootstrapFewShot 够用
连续 [0,1]	MIPROv2 能更好利用梯度感信号
多维(correctness, cost, latency)	先标量化再进 Optimizer,或用 Pareto 二次手选
LLM-Judge	先缓存 judge 调用,否则优化一轮成本爆炸

本章小结