DSPy Python 教程：2026 年提示词优化的数据划分、评分指标、预算管理与程序保存实操指南

在正式运行提示词优化前，先验证训练样本没有混入验收集、评分器能拒绝错误输出，并且保存后的程序能重新加载。本教程用 DSPy 的真实接口和本地模拟模型完成这些检查，再说明如何为 GEPA 设计数据和预算。

验证使用 Python 3.11 与 dspy==3.4.1，样本均为合成客服消息。DummyLM 是框架里的测试辅助类，答案由脚本预设，不能测量模型能力或优化效果；测试期间阻断网络连接，没有调用付费模型，也没有运行真实 GEPA 搜索。

1. 固定版本与运行范围

在一个新目录创建环境：

bash

python3.11 -m venv .venv
source .venv/bin/activate
python -m pip install 'dspy==3.4.1'

下面直接使用 dspy 包，不安装历史名称的替代包。当前官方文档与项目入口见 DSPy和仓库。完整依赖、模型和 adapter 也应随实验固定，升级后重新验收。

2. 建立数据隔离和严格评分

保存为 workflow.py。每个样本保留唯一 id 和来源 source，供划分审计使用；模型输入只含 text，参考标签不会放进待预测输入。

验证器检查来源跨集合、重复 id，以及经 Unicode NFKC、大小写与空格规范化后的精确重复。这个指纹只用于去重，不能发现所有语义近重复，也不是敏感数据匿名化方法。

评分函数只接受合法且完全匹配的标签，返回布尔值。这样可以避免把“合法 JSON”或“非空字符串”误当正确，也避免在接受真值的接口里把低但非零的分数无意视为通过。不同优化器对 metric 返回值有不同要求，应按当前接口核对。

python

 hashlib
 os
 unicodedata

os.environ.setdefault(, )
 dspy
 dspy.utils  DummyLM

dspy.configure_cache(enable_disk_cache=, enable_memory_cache=)
LABELS = {, , }
SPLITS = {
    : [
        {: , : , : , : },
        {: , : , : , : },
    ],
    : [
        {: , : , : , : },
        {: , : , : , : },
    ],
    : [
        {: , : , : , : },
        {: , : , : , : },
    ],
}


 ():
    normalized = .join(unicodedata.normalize(, text).casefold().split())
     hashlib.sha256(normalized.encode()).hexdigest()


 ():
     (splits) != {, , }:
         ValueError()
    ids, source_owner, content_owner = (), {}, {}
     split, rows  splits.items():
          rows:
             ValueError()
         row  rows:
             ( (row.get(k), )   row[k].strip()
                    k  (, , , )):
                 ValueError()
             row[]   LABELS  row[]  ids:
                 ValueError()
            ids.add(row[])
             key, owner  [(row[], source_owner), (fingerprint(row[]), content_owner)]:
                 key  owner  owner[key] != split:
                     ValueError()
                owner[key] = split


 (dspy.Signature):
    
    text:  = dspy.InputField()
    label:  = dspy.OutputField()


 ():
    
     [dspy.Example(text=r[], label=r[]).with_inputs()  r  rows]


 ():
     gold.label   LABELS:
         ValueError()
    actual = (pred, , )
     (actual, )  actual  LABELS  actual == gold.label


 ():
    correct, errors = , 
     example  examples(rows):
        :
            prediction = program(**example.inputs())
            correct += (exact_label(example, prediction))
         Exception:
            errors += 
     {: (rows), : correct, : errors,
            : correct / (rows)  rows  ,
            : errors == }


 ():
    validate_splits(SPLITS)
    
    scripted = {r[]: {: r[]}  r  SPLITS[]}
     dspy.context(lm=DummyLM(scripted)):
         dspy.BootstrapFewShot(
            metric=exact_label, max_bootstrapped_demos=,
            max_labeled_demos=, max_rounds=, max_errors=,
        ).(dspy.Predict(Route), trainset=examples(SPLITS[]))


 ():
    passed = exact_label(gold, pred)
     dspy.Prediction(score=(passed), feedback=(
          passed  
    ))


 ():
    
     dspy.GEPA(
        metric=gepa_metric, auto=, max_metric_calls=,
        reflection_lm=reflection_lm, num_threads=, track_stats=,
    )

常见问题

教程会调用付费模型吗？

不会。本地运行使用 DSPy DummyLM 预设输出，测试期间阻断网络；没有运行真实 GEPA 搜索或付费模型请求。

九组测试通过是否证明模型优化有效？

不能。它们验证数据隔离、评分、原生编译、JSON 回读和预算配置。模拟答案已预设，不代表模型学习或泛化。

最终 test 为什么不传入 compile？

trainset 与 valset 会参与更新或候选选择。最终测试应在程序和指标冻结后单独运行，避免验收信息泄漏。

metric calls 等于所有模型调用吗？

不等于。多模块执行、反思、judge 和工具会产生不同请求与 token，用量预算不能直接当美元硬上限。

指纹检查能发现所有近重复吗？

不能。它只检查经规范化的精确重复，还要按来源分组并人工或用适当方法检查语义近重复。

为什么异常不从分母删除？

只统计成功样本会掩盖无法处理或测量失败的情况。本例保留完整样本数、errors，并以 measurement_ok 标明测量是否正常。

import copy import json import tempfile import unittest from importlib.metadata import version from pathlib import Path from unittest.mock import patch import workflow as w class WorkflowTests(unittest.TestCase): def test_valid_splits_and_model_inputs(self): w.validate_splits(w.SPLITS) inputs = dict(w.examples(w.SPLITS["train"])[0].inputs()) self.assertEqual(set(inputs), {"text"}) def test_source_leakage_rejected(self): splits = copy.deepcopy(w.SPLITS) splits["test"][0]["source"] = splits["train"][0]["source"] with self.assertRaises(ValueError): w.validate_splits(splits) def test_normalized_content_leakage_rejected(self): splits = copy.deepcopy(w.SPLITS) splits["train"][0]["text"] = "SYNTHETIC BILLING" splits["validation"][0]["text"] = "ｓｙｎｔｈｅｔｉｃ billing" with self.assertRaises(ValueError): w.validate_splits(splits) def test_duplicate_id_rejected(self): splits = copy.deepcopy(w.SPLITS) splits["test"][0]["id"] = splits["train"][0]["id"] with self.assertRaises(ValueError): w.validate_splits(splits) def test_metric_rejects_wrong_or_invalid_output(self): gold = w.examples(w.SPLITS["train"])[0] self.assertTrue(w.exact_label(gold, w.dspy.Prediction(label="billing"))) for value in ("technical", "billing because...", "", None): self.assertFalse(w.exact_label(gold, w.dspy.Prediction(label=value))) def test_bad_reference_is_measurement_error(self): gold = w.dspy.Example(text="synthetic", label="not-a-label") with self.assertRaises(ValueError): w.exact_label(gold, w.dspy.Prediction(label="other")) def test_errors_keep_the_full_denominator(self): def broken(**kwargs): raise RuntimeError("Synthetic failure") result = w.audit(broken, w.SPLITS["validation"]) self.assertEqual(result, {"total": 2, "correct": 0, "errors": 2, "accuracy_all": 0.0, "measurement_ok": False}) def test_native_bootstrap_and_save_load(self): compiled = w.compile_mock() self.assertEqual(len(compiled.demos), 1) train_texts = {r["text"] for r in w.SPLITS["train"]} self.assertTrue(all(demo.text in train_texts for demo in compiled.demos)) with tempfile.TemporaryDirectory(prefix="dspy-workflow-") as directory: path = str(Path(directory) / "program.json") compiled.save(path) restored = w.dspy.Predict(w.Route) restored.load(path) self.assertEqual([dict(x) for x in compiled.demos], [dict(x) for x in restored.demos]) row = w.SPLITS["validation"][0] with w.dspy.context(lm=w.DummyLM({row["text"]: {"label": "billing"}})): self.assertEqual(restored(text=row["text"]).label, "billing") def test_native_gepa_budget_constructor(self): optimizer = w.build_gepa(w.DummyLM([{"label": "other"}])) self.assertEqual(optimizer.max_metric_calls, 40) with self.assertRaises(AssertionError): w.dspy.GEPA(metric=w.gepa_metric, auto="light", max_metric_calls=40, reflection_lm=w.DummyLM([{"label": "other"}])) if __name__ == "__main__": suite = unittest.defaultTestLoader.loadTestsFromTestCase(WorkflowTests) with patch("socket.socket.connect", side_effect=AssertionError("Network disabled for fixtures")): result = unittest.TextTestRunner(verbosity=2).run(suite) report = {"date": "2026-10-19", "python": "3.11", "dspy": version("dspy"), "tests_run": result.testsRun, "failures": len(result.failures), "errors": len(result.errors), "backend": "DSPy DummyLM scripted fixtures", "network_connections_blocked_during_tests": True, "gepa_compile_executed": False, "paid_model_calls": 0, "scope": "Split guards, metric contracts, native BootstrapFewShot compile, JSON save/load and GEPA constructor only; no real optimization benchmark"} Path(__file__).with_name("snippet-checks.json").write_text(json.dumps(report, ensure_ascii=False, indent=2)) raise SystemExit(0 if result.wasSuccessful() else 1)

DSPy Python 教程：提示词优化的数据划分、评分指标与预算控制

1. 固定版本与运行范围

2. 建立数据隔离和严格评分

常见问题

3. 用正反例验证工作流

4. 验证集评估和保存示例

5. 接入真实 GEPA 时怎样划分数据

6. 优化预算怎样控制

7. 怎样判断值得发布优化产物