dspy

DSPy：声明式大模型编程

何时使用此技能

在以下情况下使用 DSPy：

• 构建复杂的 AI 系统，包含多个组件和工作流
• 以声明式方式编写大模型程序，而不是手动设计提示词
• 使用数据驱动的方法自动优化提示词
• 创建模块化 AI 流水线，使其易于维护且可移植
• 使用优化器系统地改进模型输出
• 构建 RAG 系统、Agent 或分类器，提高可靠性

安装

# Stable release
pip install dspy

# Latest development version
pip install git+https://github.com/stanfordnlp/dspy.git

# With specific LM providers
pip install dspy[openai]        # OpenAI
pip install dspy[anthropic]     # Anthropic Claude
pip install dspy[all]           # All providers

快速入门

基础示例：问答

import dspy

# Configure your language model
lm = dspy.Claude(model="claude-sonnet-4-5-20250929")
dspy.settings.configure(lm=lm)

# Define a signature (input → output)
class QA(dspy.Signature):
    """Answer questions with short factual answers."""
    question = dspy.InputField()
    answer = dspy.OutputField(desc="often between 1 and 5 words")

# Create a module
qa = dspy.Predict(QA)

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

思维链推理

import dspy

lm = dspy.Claude(model="claude-sonnet-4-5-20250929")
dspy.settings.configure(lm=lm)

# Use ChainOfThought for better reasoning
class MathProblem(dspy.Signature):
    """Solve math word problems."""
    problem = dspy.InputField()
    answer = dspy.OutputField(desc="numerical answer")

# ChainOfThought generates reasoning steps automatically
cot = dspy.ChainOfThought(MathProblem)

response = cot(problem="If John has 5 apples and gives 2 to Mary, how many does he have?")
print(response.rationale)  # Shows reasoning steps
print(response.answer)     # "3"

核心概念

1. 签名

签名定义了 AI 任务的结构（输入 → 输出）：

# Inline signature (simple)
qa = dspy.Predict("question -> answer")

# Class signature (detailed)
class Summarize(dspy.Signature):
    """Summarize text into key points."""
    text = dspy.InputField()
    summary = dspy.OutputField(desc="bullet points, 3-5 items")

summarizer = dspy.ChainOfThought(Summarize)

何时使用：

• 内联：快速原型设计、简单任务
• 类：复杂任务、类型提示、更完善的文档

2. 模块

模块是可重复使用的组件，可将输入转换为输出：

dspy.Predict

基础预测模块：

predictor = dspy.Predict("context, question -> answer")
result = predictor(context="Paris is the capital of France",
                   question="What is the capital?")

dspy.ChainOfThought

在回答前生成推理步骤：

cot = dspy.ChainOfThought("question -> answer")
result = cot(question="Why is the sky blue?")
print(result.rationale)  # Reasoning steps
print(result.answer)     # Final answer

dspy.ReAct

使用工具进行类 Agent 推理：

from dspy.predict import ReAct

class SearchQA(dspy.Signature):
    """Answer questions using search."""
    question = dspy.InputField()
    answer = dspy.OutputField()

def search_tool(query: str) -> str:
    """Search Wikipedia."""
    # Your search implementation
    return results

react = ReAct(SearchQA, tools=[search_tool])
result = react(question="When was Python created?")

dspy.ProgramOfThought

生成并执行代码来进行推理：

pot = dspy.ProgramOfThought("question -> answer")
result = pot(question="What is 15% of 240?")
# Generates: answer = 240 * 0.15

3. 优化器

优化器使用训练数据自动改进模块：

BootstrapFewShot

从示例中学习：

from dspy.teleprompt import BootstrapFewShot

# Training data
trainset = [
    dspy.Example(question="What is 2+2?", answer="4").with_inputs("question"),
    dspy.Example(question="What is 3+5?", answer="8").with_inputs("question"),
]

# Define metric
def validate_answer(example, pred, trace=None):
    return example.answer == pred.answer

# Optimize
optimizer = BootstrapFewShot(metric=validate_answer, max_bootstrapped_demos=3)
optimized_qa = optimizer.compile(qa, trainset=trainset)

# Now optimized_qa performs better!

MIPRO（最重要提示词优化）

迭代改进提示词：

from dspy.teleprompt import MIPRO

optimizer = MIPRO(
    metric=validate_answer,
    num_candidates=10,
    init_temperature=1.0
)

optimized_cot = optimizer.compile(
    cot,
    trainset=trainset,
    num_trials=100
)

BootstrapFinetune

为模型微调创建数据集：

from dspy.teleprompt import BootstrapFinetune

optimizer = BootstrapFinetune(metric=validate_answer)
optimized_module = optimizer.compile(qa, trainset=trainset)

# Exports training data for fine-tuning

4. 构建复杂系统

多阶段流水线

import dspy

class MultiHopQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=3)
        self.generate_query = dspy.ChainOfThought("question -> search_query")
        self.generate_answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        # Stage 1: Generate search query
        search_query = self.generate_query(question=question).search_query

        # Stage 2: Retrieve context
        passages = self.retrieve(search_query).passages
        context = "\n".join(passages)

        # Stage 3: Generate answer
        answer = self.generate_answer(context=context, question=question).answer
        return dspy.Prediction(answer=answer, context=context)

# Use the pipeline
qa_system = MultiHopQA()
result = qa_system(question="Who wrote the book that inspired the movie Blade Runner?")

带优化的 RAG 系统

import dspy
from dspy.retrieve.chromadb_rm import ChromadbRM

# Configure retriever
retriever = ChromadbRM(
    collection_name="documents",
    persist_directory="./chroma_db"
)

class RAG(dspy.Module):
    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):
        context = self.retrieve(question).passages
        return self.generate(context=context, question=question)

# Create and optimize
rag = RAG()

# Optimize with training data
from dspy.teleprompt import BootstrapFewShot

optimizer = BootstrapFewShot(metric=validate_answer)
optimized_rag = optimizer.compile(rag, trainset=trainset)

大模型提供方配置

详细信息、代码示例和参数表：references/lm-provider-configuration.md。适用此步骤时请阅读。

常见模式

详细信息、代码示例和参数表：references/common-patterns.md。适用此步骤时请阅读。

评估与指标

自定义指标

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

def f1_score(example, pred, trace=None):
    """F1 score for text overlap."""
    pred_tokens = set(pred.answer.lower().split())
    gold_tokens = set(example.answer.lower().split())

    if not pred_tokens:
        return 0.0

    precision = len(pred_tokens & gold_tokens) / len(pred_tokens)
    recall = len(pred_tokens & gold_tokens) / len(gold_tokens)

    if precision + recall == 0:
        return 0.0

    return 2 * (precision * recall) / (precision + recall)

评估

from dspy.evaluate import Evaluate

# Create evaluator
evaluator = Evaluate(
    devset=testset,
    metric=exact_match,
    num_threads=4,
    display_progress=True
)

# Evaluate model
score = evaluator(qa_system)
print(f"Accuracy: {score}")

# Compare optimized vs unoptimized
score_before = evaluator(qa)
score_after = evaluator(optimized_qa)
print(f"Improvement: {score_after - score_before:.2%}")

最佳实践

1. 从简单开始，逐步迭代

# Start with Predict
qa = dspy.Predict("question -> answer")

# Add reasoning if needed
qa = dspy.ChainOfThought("question -> answer")

# Add optimization when you have data
optimized_qa = optimizer.compile(qa, trainset=data)

2. 使用描述性签名

# ❌ Bad: Vague
class Task(dspy.Signature):
    input = dspy.InputField()
    output = dspy.OutputField()

# ✅ Good: Descriptive
class SummarizeArticle(dspy.Signature):
    """Summarize news articles into 3-5 key points."""
    article = dspy.InputField(desc="full article text")
    summary = dspy.OutputField(desc="bullet points, 3-5 items")

3. 使用有代表性的数据进行优化

# Create diverse training examples
trainset = [
    dspy.Example(question="factual", answer="...).with_inputs("question"),
    dspy.Example(question="reasoning", answer="...").with_inputs("question"),
    dspy.Example(question="calculation", answer="...").with_inputs("question"),
]

# Use validation set for metric
def metric(example, pred, trace=None):
    return example.answer in pred.answer

4. 保存和加载优化后的模型

# Save
optimized_qa.save("models/qa_v1.json")

# Load
loaded_qa = dspy.ChainOfThought("question -> answer")
loaded_qa.load("models/qa_v1.json")

5. 监控和调试

# Enable tracing
dspy.settings.configure(lm=lm, trace=[])

# Run prediction
result = qa(question="...")

# Inspect trace
for call in dspy.settings.trace:
    print(f"Prompt: {call['prompt']}")
    print(f"Response: {call['response']}")

与其他方法的比较

特性	手动提示词设计	LangChain	DSPy
提示词工程	手动	手动	自动
优化	反复试错	无	数据驱动
模块化	低	中	高
类型安全	否	有限	是（签名）
可移植性	低	中	高
学习曲线	低	中	中高

何时选择 DSPy：

• 你有训练数据，或能够生成训练数据
• 你需要系统地改进提示词
• 你正在构建复杂的多阶段系统
• 你希望针对不同的大模型进行优化

何时选择其他方案：

• 快速原型设计（手动设计提示词）
• 使用现有工具构建简单链（LangChain）
• 需要自定义优化逻辑

资源

• 文档：https://dspy.ai
• GitHub：https://github.com/stanfordnlp/dspy
• Discord：https://discord.gg/XCGy2WDCQB
• Twitter：@DSPyOSS
• 论文：“DSPy：将声明式大模型调用编译为自我改进的流水线”

另请参阅

• references/modules.md - 详细的模块指南（Predict、ChainOfThought、ReAct、ProgramOfThought）
• references/optimizers.md - 优化算法（BootstrapFewShot、MIPRO、BootstrapFinetune）
• references/examples.md - 实际示例（RAG、Agent、分类器）

Agent 操作流程

1. 检查环境。 确认框架版本、模型提供商、API 密钥和速率限制。
2. 明确输入。 根据数据或用户确认格式、标识符和参数。任何会影响结果的值都不要猜测，应先询问。
3. 先运行小规模版本。 使用已知输入测试单次调用或调用链，并检查原始输出。
4. 使用以上说明和参考资料执行完整任务。
5. 验证结果。 在小型标注数据集上评估；根据架构检查结构化输出；记录提示词和响应。
6. 报告。 说明运行了什么（版本、命令、参数）、检查了什么，以及还有哪些不确定之处。

如果发生这种情况	请这样做
输出与预期架构不符	添加验证和重试机制，收紧架构，或简化提示词。
已安装版本中缺少这些说明提到的函数、标志或端点	查阅已安装版本的文档（help()、--help、官方文档），进行调整并告知用户。切勿臆造 API。
必需的输入、标识符或参数含义不明确	询问用户，或在运行前明确说明所采用的假设。

诚信守则

• 切勿伪造结果、参数、标识符、引用或统计数据。若无法运行或验证某项内容，请如实说明。
• 未经用户同意，切勿将私人或敏感数据发送到外部 API。
• 此处与版本相关的细节可能已经过时：请根据已安装版本的官方文档进行确认。
• 对于需要付费、消耗共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，请先征求同意。