dspy-framework

DSPy Framework

---

progressive_disclosure: entry_point: summary: "用于自动优化提示词的声明式框架，将提示词视为代码" when_to_use: - "使用评估数据系统地优化提示词时" - "构建需要提升准确性的生产级大模型系统时" - "实现 RAG、分类或结构化提取任务时" - "需要对提示词进行版本控制并确保结果可复现时" quick_start: - "pip install dspy-ai" - "定义签名：class QA(dspy.Signature): question = dspy.InputField(); answer = dspy.OutputField()" - "创建模块：qa = dspy.ChainOfThought(QA)" - "优化：optimizer.compile(qa, trainset=examples)" token_estimate: entry: 75 full: 5500

核心理念

DSPy（Declarative Self-improving Python，声明式自我改进 Python）将重点从手动提示词工程转向编程语言模型。将提示词视为代码，并采用：

• 声明式签名定义输入/输出
• 通过编译器进行自动优化
• 版本控制和系统化测试
• 模型变更后仍能获得可复现的结果

核心原则：不要手动编写提示词——定义任务规范，让 DSPy 来优化。

核心概念

签名：定义任务接口

签名指定大模型模块应完成的任务（输入 → 输出），而不说明具体实现方式。

基本签名：

import dspy

# Inline signature (quick)
qa_module = dspy.ChainOfThought("question -> answer")

# Class-based signature (recommended for production)
class QuestionAnswer(dspy.Signature):
    """Answer questions with short factual answers."""

    question = dspy.InputField()
    answer = dspy.OutputField(desc="often between 1 and 5 words")

# Use signature
qa = dspy.ChainOfThought(QuestionAnswer)
response = qa(question="What is the capital of France?")
print(response.answer)  # "Paris"

带类型提示的高级签名：

from typing import List

class DocumentSummary(dspy.Signature):
    """Generate concise document summaries."""

    document: str = dspy.InputField(desc="Full text to summarize")
    key_points: List[str] = dspy.OutputField(desc="3-5 bullet points")
    summary: str = dspy.OutputField(desc="2-3 sentence summary")
    sentiment: str = dspy.OutputField(desc="positive, negative, or neutral")

# Type hints provide strong typing and validation
summarizer = dspy.ChainOfThought(DocumentSummary)
result = summarizer(document="Long document text...")

字段描述：

• 使用简短、描述性短语（而非完整句子）
• 示例：desc="often between 1 and 5 words"、desc="JSON format"
• 优化器会利用这些描述来提升提示词质量

模块：构建模块

模块是 DSPy 的推理模式，可替代手动提示词工程。

思维链（CoT）：

# Zero-shot reasoning
class Reasoning(dspy.Signature):
    """Solve complex problems step by step."""
    problem = dspy.InputField()
    solution = dspy.OutputField()

cot = dspy.ChainOfThought(Reasoning)
result = cot(problem="Roger has 5 tennis balls. He buys 2 cans of 3 balls each. How many total?")
print(result.solution)  # Includes reasoning steps automatically
print(result.rationale)  # Access the chain-of-thought reasoning

检索模块（RAG）：

class RAGSignature(dspy.Signature):
    """Answer questions using retrieved context."""
    question = dspy.InputField()
    context = dspy.InputField(desc="relevant passages")
    answer = dspy.OutputField(desc="answer based on context")

# Combine retrieval + reasoning
retriever = dspy.Retrieve(k=3)  # Retrieve top 3 passages
rag = dspy.ChainOfThought(RAGSignature)

# Use in pipeline
question = "What is quantum entanglement?"
context = retriever(question).passages
answer = rag(question=question, context=context)

ReAct（推理 + 行动）：

class ResearchTask(dspy.Signature):
    """Research a topic using tools."""
    topic = dspy.InputField()
    findings = dspy.OutputField()

# ReAct interleaves reasoning with tool calls
react = dspy.ReAct(ResearchTask, tools=[web_search, calculator])
result = react(topic="Apple stock price change last month")
# Automatically uses tools when needed

ProgramOfThought：

# Generate and execute Python code
class MathProblem(dspy.Signature):
    """Solve math problems by writing Python code."""
    problem = dspy.InputField()
    code = dspy.OutputField(desc="Python code to solve problem")
    result = dspy.OutputField(desc="final numerical answer")

pot = dspy.ProgramOfThought(MathProblem)
answer = pot(problem="Calculate compound interest on $1000 at 5% for 10 years")

自定义模块：

class MultiStepRAG(dspy.Module):
    """Custom module combining retrieval and reasoning."""

    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        # Retrieve relevant passages
        context = self.retrieve(question).passages

        # Generate answer with context
        prediction = self.generate(context=context, question=question)

        # Return with metadata
        return dspy.Prediction(
            answer=prediction.answer,
            context=context,
            rationale=prediction.rationale
        )

# Use custom module
rag = MultiStepRAG(num_passages=5)
optimized_rag = optimizer.compile(rag, trainset=examples)

优化器：自动改进提示词

优化器会将高级程序编译为优化后的提示词或微调后的权重。

BootstrapFewShot

最适合：小型数据集（10-50 个示例），快速优化 优化内容：仅少样本示例

from dspy.teleprompt import BootstrapFewShot

# Define metric function
def accuracy_metric(example, prediction, trace=None):
    """Evaluate prediction correctness."""
    return example.answer.lower() == prediction.answer.lower()

# Configure optimizer
optimizer = BootstrapFewShot(
    metric=accuracy_metric,
    max_bootstrapped_demos=4,    # Max examples to bootstrap
    max_labeled_demos=16,         # Max labeled examples to consider
    max_rounds=1,                 # Bootstrapping rounds
    max_errors=10                 # Stop after N errors
)

# Training examples
trainset = [
    dspy.Example(question="What is 2+2?", answer="4").with_inputs("question"),
    dspy.Example(question="Capital of France?", answer="Paris").with_inputs("question"),
    # ... more examples
]

# Compile program
qa_module = dspy.ChainOfThought("question -> answer")
optimized_qa = optimizer.compile(
    student=qa_module,
    trainset=trainset
)

# Save optimized program
optimized_qa.save("qa_optimized.json")

工作原理：

1. 使用你的程序在训练数据上生成输出
2. 使用你的指标筛选成功的执行轨迹
3. 选取具有代表性的示例作为演示
4. 返回包含最佳少样本示例的优化程序

BootstrapFewShotWithRandomSearch

最适合：中型数据集（50-300 个示例），需要更充分探索时 优化内容：通过候选方案探索来优化少样本示例

from dspy.teleprompt import BootstrapFewShotWithRandomSearch

config = dict(
    max_bootstrapped_demos=4,
    max_labeled_demos=4,
    num_candidate_programs=10,   # Explore 10 candidate programs
    num_threads=4                # Parallel optimization
)

optimizer = BootstrapFewShotWithRandomSearch(
    metric=accuracy_metric,
    **config
)

optimized_program = optimizer.compile(
    qa_module,
    trainset=training_examples,
    valset=validation_examples  # Optional validation set
)

# Compare candidates
print(f"Best program score: {optimizer.best_score}")

优势：并行探索多个候选程序，并通过随机搜索选出表现最佳的程序。

MIPROv2（2025 年最先进技术）

最适合：大型数据集（300 个以上示例）、生产系统 优化内容：通过贝叶斯优化联合优化指令和少样本示例

import dspy
from dspy.teleprompt import MIPROv2

# Initialize language model
lm = dspy.LM('openai/gpt-4o-mini', api_key='YOUR_API_KEY')
dspy.configure(lm=lm)

# Define comprehensive metric
def quality_metric(example, prediction, trace=None):
    """Multi-dimensional quality scoring."""
    correct = example.answer.lower() in prediction.answer.lower()
    reasonable_length = 10 < len(prediction.answer) < 200
    has_reasoning = hasattr(prediction, 'rationale') and len(prediction.rationale) > 20

    # Weighted composite score
    score = (
        correct * 1.0 +
        reasonable_length * 0.2 +
        has_reasoning * 0.3
    )
    return score / 1.5  # Normalize to [0, 1]

# Initialize MIPROv2 with auto-configuration
teleprompter = MIPROv2(
    metric=quality_metric,
    auto="medium",  # Options: "light", "medium", "heavy"
    num_candidates=10,  # Number of instruction candidates to explore
    init_temperature=1.0  # Temperature for instruction generation
)

# Optimize program
optimized_program = teleprompter.compile(
    dspy.ChainOfThought("question -> answer"),
    trainset=training_examples,
    num_trials=100,  # Bayesian optimization trials
    max_bootstrapped_demos=4,
    max_labeled_demos=8
)

# Save for production
optimized_program.save("production_qa_model.json")

MIPROv2 自动配置模式：

• light：快速优化，约 20 次试验，最适合迭代（15-30 分钟）
• medium：均衡优化，约 50 次试验，推荐默认选项（30-60 分钟）
• heavy：穷举搜索，约 100 次以上试验，质量最高（1-3 小时）

MIPROv2 的工作原理：

1. 引导生成候选项：根据训练数据生成少样本示例候选项
2. 提出指令：基于任务动态特征生成不同的指令
3. 贝叶斯优化：使用代理模型寻找最佳的指令与示例组合
4. 联合优化：同时优化两个组件（而非分别优化），以发挥协同效应

性能提升（2025 年研究）：

• 提示词评估：准确率提升 +38.5%（46.2% → 64.0%）
• 护栏执行：准确率提升 +16.9%（72.1% → 84.3%）
• 代码生成：准确率提升 +21.9%（58.4% → 71.2%）
• 幻觉检测：准确率提升 +20.8%（65.8% → 79.5%）
• Agent 路由：准确率提升 +18.5%（69.3% → 82.1%）

KNN 少样本选择器

最适合：根据查询相似度动态选择示例

from dspy.teleprompt import KNNFewShot

# Requires embeddings for examples
knn_optimizer = KNNFewShot(
    k=3,  # Select 3 most similar examples
    trainset=training_examples
)

optimized_program = knn_optimizer.compile(qa_module)

# Automatically selects relevant examples at inference time
# Math query → retrieves math examples
# Geography query → retrieves geography examples

SignatureOptimizer

最适合：优化签名描述和字段规范

from dspy.teleprompt import SignatureOptimizer

sig_optimizer = SignatureOptimizer(
    metric=accuracy_metric,
    breadth=10,  # Number of variations to generate
    depth=3      # Optimization iterations
)

optimized_signature = sig_optimizer.compile(
    initial_signature=QuestionAnswer,
    trainset=trainset
)

# Use optimized signature
qa = dspy.ChainOfThought(optimized_signature)

顺序优化策略

组合使用优化器，以获得最佳效果：

# Step 1: Bootstrap few-shot examples (fast)
bootstrap = dspy.BootstrapFewShot(metric=accuracy_metric)
bootstrapped_program = bootstrap.compile(qa_module, trainset=train_examples)

# Step 2: Optimize instructions with MIPRO (comprehensive)
mipro = dspy.MIPROv2(metric=accuracy_metric, auto="medium")
final_program = mipro.compile(
    bootstrapped_program,
    trainset=train_examples,
    num_trials=50
)

# Step 3: Fine-tune signature descriptions
sig_optimizer = dspy.SignatureOptimizer(metric=accuracy_metric)
production_program = sig_optimizer.compile(final_program, trainset=train_examples)

# Save production model
production_program.save("production_optimized.json")

Teleprompters：编译流水线

Teleprompters 用于编排优化过程（“优化器”的旧称）。

自定义 Teleprompter：

class CustomTeleprompter:
    """Custom optimization pipeline."""

    def __init__(self, metric):
        self.metric = metric

    def compile(self, student, trainset, valset=None):
        # Stage 1: Bootstrap examples
        bootstrap = BootstrapFewShot(metric=self.metric)
        stage1 = bootstrap.compile(student, trainset=trainset)

        # Stage 2: Optimize instructions
        mipro = MIPROv2(metric=self.metric, auto="light")
        stage2 = mipro.compile(stage1, trainset=trainset)

        # Stage 3: Validate on held-out set
        if valset:
            score = self._evaluate(stage2, valset)
            print(f"Validation score: {score:.2%}")

        return stage2

    def _evaluate(self, program, dataset):
        correct = 0
        for example in dataset:
            prediction = program(**example.inputs())
            if self.metric(example, prediction):
                correct += 1
        return correct / len(dataset)

# Use custom teleprompter
custom_optimizer = CustomTeleprompter(metric=accuracy_metric)
optimized = custom_optimizer.compile(
    student=qa_module,
    trainset=train_examples,
    valset=val_examples
)

指标与评估

自定义指标

二元准确率：

def exact_match(example, prediction, trace=None):
    """Exact match metric."""
    return example.answer.lower().strip() == prediction.answer.lower().strip()

模糊匹配：

from difflib import SequenceMatcher

def fuzzy_match(example, prediction, trace=None):
    """Fuzzy string matching."""
    similarity = SequenceMatcher(
        None,
        example.answer.lower(),
        prediction.answer.lower()
    ).ratio()
    return similarity > 0.8  # 80% similarity threshold

多标准评估：

def comprehensive_metric(example, prediction, trace=None):
    """Evaluate on multiple criteria."""
    # Correctness
    correct = example.answer.lower() in prediction.answer.lower()

    # Length appropriateness
    length_ok = 10 < len(prediction.answer) < 200

    # Has reasoning (if CoT)
    has_reasoning = (
        hasattr(prediction, 'rationale') and
        len(prediction.rationale) > 30
    )

    # Citation quality (if RAG)
    has_citations = (
        hasattr(prediction, 'context') and
        len(prediction.context) > 0
    )

    # Composite score
    score = sum([
        correct * 1.0,
        length_ok * 0.2,
        has_reasoning * 0.3,
        has_citations * 0.2
    ]) / 1.7

    return score

LLM 评审：

def llm_judge_metric(example, prediction, trace=None):
    """Use LLM to evaluate quality."""
    judge_prompt = f"""
    Question: {example.question}
    Expected Answer: {example.answer}
    Predicted Answer: {prediction.answer}

    Evaluate the predicted answer on a scale of 0-10 for:
    1. Correctness
    2. Completeness
    3. Clarity

    Return only a number 0-10.
    """

    judge_lm = dspy.LM('openai/gpt-4o-mini')
    response = judge_lm(judge_prompt)
    score = float(response.strip()) / 10.0

    return score > 0.7  # Pass if score > 7/10

评估流水线

class Evaluator:
    """Comprehensive evaluation system."""

    def __init__(self, program, metrics):
        self.program = program
        self.metrics = metrics

    def evaluate(self, dataset, verbose=True):
        """Evaluate program on dataset."""
        results = {name: [] for name in self.metrics.keys()}

        for example in dataset:
            prediction = self.program(**example.inputs())

            for metric_name, metric_fn in self.metrics.items():
                score = metric_fn(example, prediction)
                results[metric_name].append(score)

        # Aggregate results
        aggregated = {
            name: sum(scores) / len(scores)
            for name, scores in results.items()
        }

        if verbose:
            print("\nEvaluation Results:")
            print("=" * 50)
            for name, score in aggregated.items():
                print(f"{name:20s}: {score:.2%}")

        return aggregated

# Use evaluator
evaluator = Evaluator(
    program=optimized_qa,
    metrics={
        "accuracy": exact_match,
        "fuzzy_match": fuzzy_match,
        "quality": comprehensive_metric
    }
)

scores = evaluator.evaluate(test_dataset)

语言模型配置

支持的提供商

OpenAI：

import dspy

lm = dspy.LM('openai/gpt-4o', api_key='YOUR_API_KEY')
dspy.configure(lm=lm)

# With custom settings
lm = dspy.LM(
    'openai/gpt-4o-mini',
    api_key='YOUR_API_KEY',
    temperature=0.7,
    max_tokens=1024
)

Anthropic Claude：

lm = dspy.LM(
    'anthropic/claude-3-5-sonnet-20241022',
    api_key='YOUR_ANTHROPIC_KEY',
    max_tokens=4096
)
dspy.configure(lm=lm)

# Claude Opus for complex reasoning
lm_opus = dspy.LM('anthropic/claude-3-opus-20240229', api_key=key)

本地模型（Ollama）：

# Requires Ollama running locally
lm = dspy.LM('ollama/llama3.1:70b', api_base='http://localhost:11434')
dspy.configure(lm=lm)

# Mixtral
lm = dspy.LM('ollama/mixtral:8x7b')

多个模型：

# Use different models for different stages
strong_lm = dspy.LM('openai/gpt-4o')
fast_lm = dspy.LM('openai/gpt-4o-mini')

# Configure per module
class HybridPipeline(dspy.Module):
    def __init__(self):
        super().__init__()
        # Fast model for retrieval
        self.retrieve = dspy.Retrieve(k=5)

        # Strong model for reasoning
        with dspy.context(lm=strong_lm):
            self.reason = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        context = self.retrieve(question).passages
        return self.reason(context=context, question=question)

模型选择策略

def select_model(task_complexity, budget):
    """Select appropriate model based on task and budget."""
    models = {
        "simple": [
            ("openai/gpt-4o-mini", 0.15),  # (model, cost per 1M tokens)
            ("anthropic/claude-3-haiku-20240307", 0.25),
        ],
        "medium": [
            ("openai/gpt-4o", 2.50),
            ("anthropic/claude-3-5-sonnet-20241022", 3.00),
        ],
        "complex": [
            ("anthropic/claude-3-opus-20240229", 15.00),
            ("openai/o1-preview", 15.00),
        ]
    }

    candidates = models[task_complexity]
    affordable = [m for m, cost in candidates if cost <= budget]

    return affordable[0] if affordable else candidates[0][0]

# Use in optimization
task = "complex"
model = select_model(task, budget=10.0)
lm = dspy.LM(model)
dspy.configure(lm=lm)

程序组合

链接模块

class MultiStepPipeline(dspy.Module):
    """Chain multiple reasoning steps."""

    def __init__(self):
        super().__init__()
        self.step1 = dspy.ChainOfThought("question -> subtasks")
        self.step2 = dspy.ChainOfThought("subtask -> solution")
        self.step3 = dspy.ChainOfThought("solutions -> final_answer")

    def forward(self, question):
        # Break down question
        decomposition = self.step1(question=question)

        # Solve each subtask
        solutions = []
        for subtask in decomposition.subtasks.split('\n'):
            if subtask.strip():
                sol = self.step2(subtask=subtask)
                solutions.append(sol.solution)

        # Synthesize final answer
        combined = '\n'.join(solutions)
        final = self.step3(solutions=combined)

        return dspy.Prediction(
            answer=final.final_answer,
            subtasks=decomposition.subtasks,
            solutions=solutions
        )

# Optimize entire pipeline
pipeline = MultiStepPipeline()
optimizer = MIPROv2(metric=quality_metric, auto="medium")
optimized_pipeline = optimizer.compile(pipeline, trainset=examples)

条件分支

class AdaptivePipeline(dspy.Module):
    """Adapt reasoning based on query type."""

    def __init__(self):
        super().__init__()
        self.classifier = dspy.ChainOfThought("question -> category")
        self.math_solver = dspy.ProgramOfThought("problem -> solution")
        self.fact_qa = dspy.ChainOfThought("question -> answer")
        self.creative = dspy.ChainOfThought("prompt -> response")

    def forward(self, question):
        # Classify query type
        category = self.classifier(question=question).category.lower()

        # Route to appropriate module
        if "math" in category or "calculation" in category:
            return self.math_solver(problem=question)
        elif "creative" in category or "story" in category:
            return self.creative(prompt=question)
        else:
            return self.fact_qa(question=question)

# Optimize each branch independently
adaptive = AdaptivePipeline()
optimized_adaptive = optimizer.compile(adaptive, trainset=diverse_examples)

生产部署

保存和加载模型

# Save optimized program
optimized_program.save("models/qa_v1.0.0.json")

# Load in production
production_qa = dspy.ChainOfThought("question -> answer")
production_qa.load("models/qa_v1.0.0.json")

# Use loaded model
response = production_qa(question="What is quantum computing?")

版本控制

import json
from datetime import datetime

class ModelRegistry:
    """Version control for DSPy models."""

    def __init__(self, registry_path="models/registry.json"):
        self.registry_path = registry_path
        self.registry = self._load_registry()

    def register(self, name, version, model_path, metadata=None):
        """Register a model version."""
        model_id = f"{name}:v{version}"

        self.registry[model_id] = {
            "name": name,
            "version": version,
            "path": model_path,
            "created_at": datetime.utcnow().isoformat(),
            "metadata": metadata or {}
        }

        self._save_registry()
        return model_id

    def get_model(self, name, version="latest"):
        """Load model by name and version."""
        if version == "latest":
            versions = [
                v for k, v in self.registry.items()
                if v["name"] == name
            ]
            if not versions:
                raise ValueError(f"No versions found for {name}")

            latest = max(versions, key=lambda x: x["created_at"])
            model_path = latest["path"]
        else:
            model_id = f"{name}:v{version}"
            model_path = self.registry[model_id]["path"]

        # Load model
        module = dspy.ChainOfThought("question -> answer")
        module.load(model_path)
        return module

    def _load_registry(self):
        try:
            with open(self.registry_path, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {}

    def _save_registry(self):
        with open(self.registry_path, 'w') as f:
            json.dump(self.registry, f, indent=2)

# Use registry
registry = ModelRegistry()

# Register new version
registry.register(
    name="qa_assistant",
    version="1.0.0",
    model_path="models/qa_v1.0.0.json",
    metadata={
        "accuracy": 0.87,
        "optimizer": "MIPROv2",
        "training_examples": 500
    }
)

# Load for production
qa = registry.get_model("qa_assistant", version="latest")

监控和日志记录

import logging
from datetime import datetime

class DSPyMonitor:
    """Monitor DSPy program execution."""

    def __init__(self, program, log_file="logs/dspy.log"):
        self.program = program
        self.logger = self._setup_logger(log_file)
        self.metrics = []

    def __call__(self, **kwargs):
        """Wrap program execution with monitoring."""
        start_time = datetime.utcnow()

        try:
            # Execute program
            result = self.program(**kwargs)

            # Log success
            duration = (datetime.utcnow() - start_time).total_seconds()
            self._log_execution(
                status="success",
                inputs=kwargs,
                outputs=result,
                duration=duration
            )

            return result

        except Exception as e:
            # Log error
            duration = (datetime.utcnow() - start_time).total_seconds()
            self._log_execution(
                status="error",
                inputs=kwargs,
                error=str(e),
                duration=duration
            )
            raise

    def _log_execution(self, status, inputs, duration, outputs=None, error=None):
        """Log execution details."""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "status": status,
            "inputs": inputs,
            "duration_seconds": duration
        }

        if outputs:
            log_entry["outputs"] = str(outputs)
        if error:
            log_entry["error"] = error

        self.logger.info(json.dumps(log_entry))
        self.metrics.append(log_entry)

    def _setup_logger(self, log_file):
        """Setup logging."""
        logger = logging.getLogger("dspy_monitor")
        logger.setLevel(logging.INFO)

        handler = logging.FileHandler(log_file)
        handler.setFormatter(
            logging.Formatter('%(asctime)s - %(message)s')
        )
        logger.addHandler(handler)

        return logger

    def get_stats(self):
        """Get execution statistics."""
        if not self.metrics:
            return {}

        successes = [m for m in self.metrics if m["status"] == "success"]
        errors = [m for m in self.metrics if m["status"] == "error"]

        return {
            "total_calls": len(self.metrics),
            "success_rate": len(successes) / len(self.metrics),
            "error_rate": len(errors) / len(self.metrics),
            "avg_duration": sum(m["duration_seconds"] for m in self.metrics) / len(self.metrics),
            "errors": [m["error"] for m in errors]
        }

# Use monitor
monitored_qa = DSPyMonitor(optimized_qa)
result = monitored_qa(question="What is AI?")

# Check stats
stats = monitored_qa.get_stats()
print(f"Success rate: {stats['success_rate']:.2%}")

与 LangSmith 集成

使用 LangSmith 评估 DSPy 程序：

import os
from langsmith import Client
from langsmith.evaluation import evaluate

# Setup
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"

client = Client()

# Wrap DSPy program for LangSmith
def dspy_wrapper(inputs: dict) -> dict:
    """Wrapper for LangSmith evaluation."""
    question = inputs["question"]
    result = optimized_qa(question=question)
    return {"answer": result.answer}

# Define evaluator
def dspy_evaluator(run, example):
    """Evaluate DSPy output."""
    predicted = run.outputs["answer"]
    expected = example.outputs["answer"]

    return {
        "key": "correctness",
        "score": 1.0 if expected.lower() in predicted.lower() else 0.0
    }

# Create dataset
dataset = client.create_dataset(
    dataset_name="dspy_qa_eval",
    description="DSPy QA evaluation dataset"
)

# Add examples
for example in test_examples:
    client.create_example(
        dataset_id=dataset.id,
        inputs={"question": example.question},
        outputs={"answer": example.answer}
    )

# Run evaluation
results = evaluate(
    dspy_wrapper,
    data="dspy_qa_eval",
    evaluators=[dspy_evaluator],
    experiment_prefix="dspy_v1.0"
)

print(f"Average correctness: {results['results']['correctness']:.2%}")

实际示例

RAG 流水线

class ProductionRAG(dspy.Module):
    """Production-ready RAG system."""

    def __init__(self, k=5):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=k)

        # Multi-stage reasoning
        self.rerank = dspy.ChainOfThought(
            "question, passages -> relevant_passages"
        )
        self.generate = dspy.ChainOfThought(
            "question, context -> answer, citations"
        )

    def forward(self, question):
        # Retrieve candidate passages
        candidates = self.retrieve(question).passages

        # Rerank for relevance
        reranked = self.rerank(
            question=question,
            passages="\n---\n".join(candidates)
        )

        # Generate answer with citations
        result = self.generate(
            question=question,
            context=reranked.relevant_passages
        )

        return dspy.Prediction(
            answer=result.answer,
            citations=result.citations,
            passages=candidates
        )

# Optimize RAG pipeline
rag = ProductionRAG(k=10)

def rag_metric(example, prediction, trace=None):
    """Evaluate RAG quality."""
    answer_correct = example.answer.lower() in prediction.answer.lower()
    has_citations = len(prediction.citations) > 0
    return answer_correct and has_citations

optimizer = MIPROv2(metric=rag_metric, auto="heavy")
optimized_rag = optimizer.compile(rag, trainset=rag_examples)
optimized_rag.save("models/rag_production.json")

分类

class SentimentClassifier(dspy.Module):
    """Multi-class sentiment classification."""

    def __init__(self, classes):
        super().__init__()
        self.classes = classes

        class ClassificationSig(dspy.Signature):
            text = dspy.InputField()
            reasoning = dspy.OutputField(desc="step-by-step reasoning")
            sentiment = dspy.OutputField(desc=f"one of: {', '.join(classes)}")
            confidence = dspy.OutputField(desc="confidence score 0-1")

        self.classify = dspy.ChainOfThought(ClassificationSig)

    def forward(self, text):
        result = self.classify(text=text)

        # Validate output
        if result.sentiment not in self.classes:
            result.sentiment = "neutral"  # Fallback

        return result

# Train classifier
classes = ["positive", "negative", "neutral"]
classifier = SentimentClassifier(classes)

def classification_metric(example, prediction, trace=None):
    return example.sentiment == prediction.sentiment

optimizer = BootstrapFewShot(metric=classification_metric)
optimized_classifier = optimizer.compile(
    classifier,
    trainset=sentiment_examples
)

# Use in production
result = optimized_classifier(text="This product is amazing!")
print(f"Sentiment: {result.sentiment} ({result.confidence})")

摘要生成

class DocumentSummarizer(dspy.Module):
    """Hierarchical document summarization."""

    def __init__(self):
        super().__init__()

        # Chunk-level summaries
        self.chunk_summary = dspy.ChainOfThought(
            "chunk -> summary"
        )

        # Document-level synthesis
        self.final_summary = dspy.ChainOfThought(
            "chunk_summaries -> final_summary, key_points"
        )

    def forward(self, document, chunk_size=1000):
        # Split document into chunks
        chunks = self._chunk_document(document, chunk_size)

        # Summarize each chunk
        chunk_summaries = []
        for chunk in chunks:
            summary = self.chunk_summary(chunk=chunk)
            chunk_summaries.append(summary.summary)

        # Synthesize final summary
        combined = "\n---\n".join(chunk_summaries)
        final = self.final_summary(chunk_summaries=combined)

        return dspy.Prediction(
            summary=final.final_summary,
            key_points=final.key_points.split('\n'),
            chunk_count=len(chunks)
        )

    def _chunk_document(self, document, chunk_size):
        """Split document into chunks."""
        words = document.split()
        chunks = []

        for i in range(0, len(words), chunk_size):
            chunk = ' '.join(words[i:i + chunk_size])
            chunks.append(chunk)

        return chunks

# Optimize summarizer
summarizer = DocumentSummarizer()

def summary_metric(example, prediction, trace=None):
    # Check key points coverage
    key_points_present = sum(
        1 for kp in example.key_points
        if kp.lower() in prediction.summary.lower()
    )
    coverage = key_points_present / len(example.key_points)

    # Check length appropriateness
    length_ok = 100 < len(prediction.summary) < 500

    return coverage > 0.7 and length_ok

optimizer = MIPROv2(metric=summary_metric, auto="medium")
optimized_summarizer = optimizer.compile(summarizer, trainset=summary_examples)

问答

class MultiHopQA(dspy.Module):
    """Multi-hop question answering."""

    def __init__(self):
        super().__init__()

        # Decompose complex questions
        self.decompose = dspy.ChainOfThought(
            "question -> subquestions"
        )

        # Answer subquestions with retrieval
        self.retrieve = dspy.Retrieve(k=3)
        self.answer_subq = dspy.ChainOfThought(
            "subquestion, context -> answer"
        )

        # Synthesize final answer
        self.synthesize = dspy.ChainOfThought(
            "question, subanswers -> final_answer, reasoning"
        )

    def forward(self, question):
        # Decompose into subquestions
        decomp = self.decompose(question=question)
        subquestions = [
            sq.strip()
            for sq in decomp.subquestions.split('\n')
            if sq.strip()
        ]

        # Answer each subquestion
        subanswers = []
        for subq in subquestions:
            context = self.retrieve(subq).passages
            answer = self.answer_subq(
                subquestion=subq,
                context="\n".join(context)
            )
            subanswers.append(answer.answer)

        # Synthesize final answer
        combined = "\n".join([
            f"Q: {sq}\nA: {sa}"
            for sq, sa in zip(subquestions, subanswers)
        ])

        final = self.synthesize(
            question=question,
            subanswers=combined
        )

        return dspy.Prediction(
            answer=final.final_answer,
            reasoning=final.reasoning,
            subquestions=subquestions,
            subanswers=subanswers
        )

# Optimize multi-hop QA
multihop_qa = MultiHopQA()

def multihop_metric(example, prediction, trace=None):
    # Check answer correctness
    correct = example.answer.lower() in prediction.answer.lower()

    # Check reasoning quality
    has_reasoning = len(prediction.reasoning) > 50

    # Check subquestion coverage
    has_subquestions = len(prediction.subquestions) >= 2

    return correct and has_reasoning and has_subquestions

optimizer = MIPROv2(metric=multihop_metric, auto="heavy")
optimized_multihop = optimizer.compile(multihop_qa, trainset=multihop_examples)

从手动提示迁移

之前：手动提示

# Manual prompt engineering
PROMPT = """
You are a helpful assistant. Answer questions accurately and concisely.

Examples:
Q: What is 2+2?
A: 4

Q: Capital of France?
A: Paris

Q: {question}
A: """

def manual_qa(question):
    response = llm.invoke(PROMPT.format(question=question))
    return response

之后：DSPy

# DSPy declarative approach
class QA(dspy.Signature):
    """Answer questions accurately and concisely."""
    question = dspy.InputField()
    answer = dspy.OutputField(desc="short factual answer")

qa = dspy.ChainOfThought(QA)

# Optimize automatically
optimizer = MIPROv2(metric=accuracy_metric, auto="medium")
optimized_qa = optimizer.compile(qa, trainset=examples)

def dspy_qa(question):
    result = optimized_qa(question=question)
    return result.answer

优势：

• 系统化优化，而非手动反复试错
• 版本控制和可复现性
• 自动适应新模型
• 性能提升：准确率提高 18-38%

最佳实践

数据准备

# Create high-quality training examples
def prepare_training_data(raw_data):
    """Convert raw data to DSPy examples."""
    examples = []

    for item in raw_data:
        example = dspy.Example(
            question=item["question"],
            answer=item["answer"],
            context=item.get("context", "")  # Optional fields
        ).with_inputs("question", "context")  # Mark input fields

        examples.append(example)

    return examples

# Split data properly
def train_val_test_split(examples, train=0.7, val=0.15, test=0.15):
    """Split data for optimization and evaluation."""
    import random
    random.shuffle(examples)

    n = len(examples)
    train_end = int(n * train)
    val_end = int(n * (train + val))

    return {
        "train": examples[:train_end],
        "val": examples[train_end:val_end],
        "test": examples[val_end:]
    }

# Use split data
data = train_val_test_split(all_examples)
optimized = optimizer.compile(
    program,
    trainset=data["train"],
    valset=data["val"]  # For hyperparameter tuning
)

# Final evaluation on held-out test set
evaluator = Evaluator(optimized, metrics={"accuracy": accuracy_metric})
test_results = evaluator.evaluate(data["test"])

指标设计

# Design metrics aligned with business goals
def business_aligned_metric(example, prediction, trace=None):
    """Metric aligned with business KPIs."""

    # Core correctness (must have)
    correct = example.answer.lower() in prediction.answer.lower()
    if not correct:
        return 0.0

    # Business-specific criteria
    is_concise = len(prediction.answer) < 100  # User preference
    is_professional = not any(
        word in prediction.answer.lower()
        for word in ["um", "like", "maybe", "dunno"]
    )
    has_confidence = (
        hasattr(prediction, 'confidence') and
        float(prediction.confidence) > 0.7
    )

    # Weighted score
    score = (
        correct * 1.0 +
        is_concise * 0.2 +
        is_professional * 0.3 +
        has_confidence * 0.2
    ) / 1.7

    return score

错误处理

class RobustModule(dspy.Module):
    """Module with error handling."""

    def __init__(self):
        super().__init__()
        self.qa = dspy.ChainOfThought("question -> answer")

    def forward(self, question, max_retries=3):
        """Forward with retry logic."""
        for attempt in range(max_retries):
            try:
                result = self.qa(question=question)

                # Validate output
                if self._validate_output(result):
                    return result
                else:
                    logging.warning(f"Invalid output on attempt {attempt + 1}")

            except Exception as e:
                logging.error(f"Error on attempt {attempt + 1}: {e}")

                if attempt == max_retries - 1:
                    raise

        # Fallback
        return dspy.Prediction(
            answer="I'm unable to answer that question.",
            confidence=0.0
        )

    def _validate_output(self, result):
        """Validate output quality."""
        return (
            hasattr(result, 'answer') and
            len(result.answer) > 0 and
            len(result.answer) < 1000
        )

使用缓存提升效率

from functools import lru_cache
import hashlib

class CachedModule(dspy.Module):
    """Module with semantic caching."""

    def __init__(self, base_module):
        super().__init__()
        self.base_module = base_module
        self.cache = {}

    def forward(self, question):
        # Check cache
        cache_key = self._get_cache_key(question)

        if cache_key in self.cache:
            logging.info("Cache hit")
            return self.cache[cache_key]

        # Cache miss: execute module
        result = self.base_module(question=question)
        self.cache[cache_key] = result

        return result

    def _get_cache_key(self, question):
        """Generate cache key."""
        return hashlib.md5(question.lower().encode()).hexdigest()

# Use cached module
base_qa = dspy.ChainOfThought("question -> answer")
cached_qa = CachedModule(base_qa)

故障排查

常见问题

优化效果不佳：

• 增加训练数据量（目标为 100+ 个示例）
• 使用质量更高的指标（更具体）
• 尝试不同的优化器（MIPROv2 可设置 auto="heavy"）
• 检查指标中是否存在数据泄漏

优化耗时过长：

• 使用 auto="light"，而不是 "heavy"
• 减少 MIPROv2 的 num_trials
• 快速迭代时使用 BootstrapFewShot，而不是 MIPROv2
• 使用 num_threads 参数进行并行处理

结果不一致：

• 设置随机种子：dspy.configure(random_seed=42)
• 增大温度以提高多样性，或降低温度以提高一致性
• 使用多个优化后程序组成的集成
• 在更大的测试集上验证

内存不足：

• 减小优化过程中的批次大小
• 对大型数据集使用流式处理
• 定期清除缓存
• 使用较小的模型进行自举

优化调试

# Enable verbose logging
import logging
logging.basicConfig(level=logging.INFO)

# Custom teleprompter with debugging
class DebugTeleprompter:
    def __init__(self, metric):
        self.metric = metric
        self.history = []

    def compile(self, student, trainset):
        print(f"\nStarting optimization with {len(trainset)} examples")

        # Bootstrap with debugging
        bootstrap = BootstrapFewShot(metric=self.metric)

        for i, example in enumerate(trainset):
            prediction = student(**example.inputs())
            score = self.metric(example, prediction)

            self.history.append({
                "example_idx": i,
                "score": score,
                "prediction": str(prediction)
            })

            print(f"Example {i}: score={score}")

        # Continue with optimization
        optimized = bootstrap.compile(student, trainset=trainset)

        print(f"\nOptimization complete")
        print(f"Average score: {sum(h['score'] for h in self.history) / len(self.history):.2f}")

        return optimized

# Use debug teleprompter
debug_optimizer = DebugTeleprompter(metric=accuracy_metric)
optimized = debug_optimizer.compile(qa_module, trainset=examples)

性能基准

根据 2025 年的生产研究：

使用场景	基线	DSPy 优化后	提升幅度	使用的优化器
提示词评估	46.2%	64.0%	+38.5%	MIPROv2
护栏执行	72.1%	84.3%	+16.9%	MIPROv2
代码生成	58.4%	71.2%	+21.9%	MIPROv2
幻觉检测	65.8%	79.5%	+20.8%	BootstrapFewShot
Agent 路由	69.3%	82.1%	+18.5%	MIPROv2
RAG 准确率	54.0%	68.5%	+26.9%	BootstrapFewShot + MIPRO

生产环境采用者：JetBlue、Databricks、Walmart、VMware、Replit、Sephora、Moody's

资源

• 文档：https://dspy.ai/
• GitHub：https://github.com/stanfordnlp/dspy
• 论文：“DSPy：将声明式语言模型调用编译为自我改进的流水线”
• 2025 年研究：“是否该将提示词视为代码？”（arXiv:2507.03620）
• 社区：Discord、GitHub Discussions

相关技能

使用 Dspy 时，这些技能可以优化你的工作流：

• langgraph：用于多 Agent 编排的 LangGraph（配合 DSPy 优化后的提示词使用）
• test-driven-development：测试 DSPy 模块和提示词优化
• systematic-debugging：调试 DSPy 编译和优化失败问题

[部署到你的工具包后，可在这些技能中查阅完整文档]