dspy-advanced-workflow

DSPy 高级工作流（2026）

此技能运行七步循环，将自然语言任务描述转化为经过优化、已保存且可部署的 DSPy 程序。按顺序执行相关步骤。原型只需达到经过验证的基线；运行优化器需要适当的授权预算和需求依据。导出本地工件并不代表获准部署。

七个步骤

1. 规格说明

用一句话重新表述用户的任务。确定输入、输出、重要的质量维度和任何约束（延迟、成本、工具访问权限、上下文大小）。选择预测器类型：

任务类型	预测器
单步结构化输入／输出	dspy.Predict / dspy.ChainOfThought
工具使用／多步骤	dspy.ReAct
代码执行	dspy.ProgramOfThought
长上下文／代码库	dspy.RLM → dspy-rlm-module

2. 程序

根据 dspy-fundamentals 编写带类型的 dspy.Signature 和 dspy.Module 子类。不要硬编码提示词。为预测器命名，以便 GEPA 能定位它们。

3. 数据

将 trainset 和单独的 valset 构建为 dspy.Example(...).with_inputs(...)。对于 GEPA，尽量扩大训练集，并让验证集只大到足以体现下游行为；保留的 testset 只在最后报告。参见 dspy-evaluation-harness。

4. 丰富指标

编写 rich_metric(gold, pred, trace=None, pred_name=None, pred_trace=None)，返回 dspy.Prediction(score=0..1, feedback="natural-language critique")。反馈至关重要——GEPA 的反思 LM 正是从中学习。使用相同字段的字典会导致 dspy.Evaluate 崩溃；只有 dspy.Prediction 能正确聚合。参见 dspy-evaluation-harness。

5. 基线

evaluator = dspy.Evaluate(devset=valset, metric=rich_metric,
                          num_threads=8, display_progress=True,
                          provide_traceback=True,
                          save_as_json="runs/baseline.json")
baseline = evaluator(program)
print("Baseline:", baseline.score)

6. 使用 GEPA 优化

reflection_lm = dspy.LM("openai/gpt-5", temperature=1.0, max_tokens=32000)
optimizer = dspy.GEPA(
    metric=rich_metric,
    auto="medium",
    reflection_lm=reflection_lm,
    candidate_selection_strategy="pareto",
    track_stats=True,
    track_best_outputs=True,
    log_dir="./gepa_logs",
    num_threads=8,
    seed=0,
)
optimized = optimizer.compile(student=program, trainset=trainset, valset=valset)
print("Optimized:", evaluator(optimized).score)

先运行 auto="light" 进行健全性检查；最终运行时再改用 auto="medium"/"heavy"。参见 dspy-gepa-optimizer。

如果需要有意进行多阶段编译循环，DSPy 3.2.x 还提供 dspy.BetterTogether(metric=..., bootstrap=..., gepa=...)，可在获得干净的 GEPA 基线配置后串联多个命名优化器。

7. 导出与部署

optimized.save("artifacts/program.json", save_program=False)     # state, portable
# or for full deployment artifact:
optimized.save("artifacts/program_dir/", save_program=True)

部署：

• 使用 dspy.load("artifacts/program_dir/") 加载，或重建程序后调用 .load("program.json")。
• 封装为 FastAPI／命令行工具。
• 启用 track_usage=True，以便观测成本和延迟。
• 在 CI 中使用 MLflow（mlflow.dspy.autolog()）或 W&B 记录日志。
• 保留一项离线回归测试，让它针对已保存的程序运行 evaluator，并在结果低于阈值时使 CI 失败。

完整编排模板

"""DSPy end-to-end pipeline — spec → optimize → deploy."""

import dspy
from pathlib import Path

# ----- 1–2. Spec & program (dspy-fundamentals) -----
class MyTask(dspy.Signature):
    """<one-line instruction from the spec>."""
    input_field: str = dspy.InputField()
    output_field: str = dspy.OutputField()

class MyProgram(dspy.Module):
    def __init__(self):
        super().__init__()
        self.step = dspy.ChainOfThought(MyTask)
    def forward(self, **kw):
        return self.step(**kw)

# ----- 3. Data (dspy-evaluation-harness) -----
trainset = [...]   # list[dspy.Example(...).with_inputs(...)]
valset   = [...]

# ----- 4. Rich metric (dspy-evaluation-harness) -----
def rich_metric(gold, pred, trace=None, pred_name=None, pred_trace=None):
    score = ...          # compute 0..1
    feedback = ...       # detailed critique
    return dspy.Prediction(score=score, feedback=feedback)  # NOT a dict

# ----- 5. Baseline -----
dspy.configure(lm=dspy.LM("openai/gpt-4o"), track_usage=True)
evaluator = dspy.Evaluate(devset=valset, metric=rich_metric, num_threads=8,
                          display_progress=True, provide_traceback=True,
                          save_as_json="runs/baseline.json")
program = MyProgram()
print("Baseline:", evaluator(program).score)

# ----- 6. GEPA optimize (dspy-gepa-optimizer) -----
optimizer = dspy.GEPA(
    metric=rich_metric,
    auto="medium",
    reflection_lm=dspy.LM("openai/gpt-5", temperature=1.0, max_tokens=32000),
    candidate_selection_strategy="pareto",
    track_stats=True, track_best_outputs=True,
    log_dir="./gepa_logs", num_threads=8, seed=0,
)
optimized = optimizer.compile(student=program, trainset=trainset, valset=valset)
print("Optimized:", evaluator(optimized).score)

# ----- 7. Export (dspy-fundamentals) -----
Path("artifacts").mkdir(exist_ok=True)
optimized.save("artifacts/program.json", save_program=False)

防护措施

• 在优化前于步骤 4 定义指标；根据任务提供有参考价值的反馈。
• 优化前务必先建立基线——没有基线，就不能宣称优化有效。
• 将优化前后的指标都保存为 JSON，以便审计。
• 如果保留的测试集得分下降，保留该结果，并根据训练／验证证据进行诊断，不要臆测原因。根据测试集信息做出更改后，应使用新的、未触碰过的最终留出集，或将后续结果标记为探索性结果；不要反复针对原测试集调优。
• 多阶段重新编译前，将优化后的程序冻结：设置 module._compiled = True。

可运行脚手架 → example_pipeline.py