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

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

import hashlib
import os
import unicodedata

os.environ.setdefault("LITELLM_LOCAL_MODEL_COST_MAP", "True")
import dspy
from dspy.utils import DummyLM

dspy.configure_cache(enable_disk_cache=False, enable_memory_cache=False)
LABELS = {"billing", "technical", "other"}
SPLITS = {
    "train": [
        {"id": "tr1", "source": "case-a", "text": "合成消息：想修改账单抬头", "label": "billing"},
        {"id": "tr2", "source": "case-b", "text": "合成消息：客户端启动失败", "label": "technical"},
    ],
    "validation": [
        {"id": "va1", "source": "case-c", "text": "合成消息：咨询发票状态", "label": "billing"},
        {"id": "va2", "source": "case-d", "text": "合成消息：附件无法打开", "label": "technical"},
    ],
    "test": [
        {"id": "te1", "source": "case-e", "text": "合成消息：核对上期扣款", "label": "billing"},
        {"id": "te2", "source": "case-f", "text": "合成消息：询问营业时间", "label": "other"},
    ],
}


def fingerprint(text):
    normalized = " ".join(unicodedata.normalize("NFKC", text).casefold().split())
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def validate_splits(splits):
    if set(splits) != {"train", "validation", "test"}:
        raise ValueError("Require train, validation and test")
    ids, source_owner, content_owner = set(), {}, {}
    for split, rows in splits.items():
        if not rows:
            raise ValueError("Empty split")
        for row in rows:
            if any(not isinstance(row.get(k), str) or not row[k].strip()
                   for k in ("id", "source", "text", "label")):
                raise ValueError("Invalid dataset row")
            if row["label"] not in LABELS or row["id"] in ids:
                raise ValueError("Invalid label or duplicate id")
            ids.add(row["id"])
            for key, owner in [(row["source"], source_owner), (fingerprint(row["text"]), content_owner)]:
                if key in owner and owner[key] != split:
                    raise ValueError("Cross-split leakage")
                owner[key] = split


class Route(dspy.Signature):
    """Return exactly one label: billing, technical, or other."""
    text: str = dspy.InputField()
    label: str = dspy.OutputField()


def examples(rows):
    # Source/id are for split auditing; only text is a model input.
    return [dspy.Example(text=r["text"], label=r["label"]).with_inputs("text") for r in rows]


def exact_label(gold, pred, trace=None):
    if gold.label not in LABELS:
        raise ValueError("Invalid reference label")
    actual = getattr(pred, "label", None)
    return isinstance(actual, str) and actual in LABELS and actual == gold.label


def audit(program, rows):
    correct, errors = 0, 0
    for example in examples(rows):
        try:
            prediction = program(**example.inputs())
            correct += int(exact_label(example, prediction))
        except Exception:
            errors += 1
    return {"total": len(rows), "correct": correct, "errors": errors,
            "accuracy_all": correct / len(rows) if rows else None,
            "measurement_ok": errors == 0}


def compile_mock():
    validate_splits(SPLITS)
    # Scripted responses exercise interfaces, not actual model learning.
    scripted = {r["text"]: {"label": r["label"]} for r in SPLITS["train"]}
    with dspy.context(lm=DummyLM(scripted)):
        return dspy.BootstrapFewShot(
            metric=exact_label, max_bootstrapped_demos=1,
            max_labeled_demos=0, max_rounds=1, max_errors=1,
        ).compile(dspy.Predict(Route), trainset=examples(SPLITS["train"]))


def gepa_metric(gold, pred, trace=None, pred_name=None, pred_trace=None):
    passed = exact_label(gold, pred)
    return dspy.Prediction(score=float(passed), feedback=(
        "Correct allowed label." if passed else "Return the correct single allowed label."
    ))


def build_gepa(reflection_lm):
    # Construction only. Calling compile with a real LM can incur charges.
    return dspy.GEPA(
        metric=gepa_metric, auto=None, max_metric_calls=40,
        reflection_lm=reflection_lm, num_threads=1, track_stats=True,
    )

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)

export DSPY_CACHEDIR="$PWD/.dspy-demo-cache"
export LITELLM_LOCAL_MODEL_COST_MAP=True
python test_workflow.py

import json
import workflow as w

compiled = w.compile_mock()
rows = w.SPLITS['validation']
scripted = {
    rows[0]['text']: {'label': 'billing'},
    rows[1]['text']: {'label': 'other'},
}
with w.dspy.context(lm=w.DummyLM(scripted)):
    report = w.audit(compiled, rows)
assert report == {
    'total': 2, 'correct': 1, 'errors': 0,
    'accuracy_all': 0.5, 'measurement_ok': True,
}
compiled.save('program.json')
print(json.dumps(report, ensure_ascii=False))

def optimize_with_gepa(program, reflection_lm, splits):
    import workflow as w
    w.validate_splits(splits)
    optimizer = w.build_gepa(reflection_lm)
    return optimizer.compile(
        program,
        trainset=w.examples(splits['train']),
        valset=w.examples(splits['validation']),
    )