DSPy 基础概念 #

安装与配置 #

安装 DSPy #

bash

# 使用 pip 安装
pip install dspy-ai

# 安装完整版（包含更多功能）
pip install dspy-ai[all]

# 使用 poetry 安装
poetry add dspy-ai

验证安装 #

python

import dspy
print(dspy.__version__)

配置语言模型 #

DSPy 支持多种语言模型后端，配置方式统一且简单。

OpenAI 配置 #

python

import dspy

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

Claude 配置 #

python

import dspy

lm = dspy.LM('anthropic/claude-3-opus-20240229', api_key='your-api-key')
dspy.configure(lm=lm)

本地模型配置 #

python

import dspy

# Ollama
lm = dspy.LM('ollama/llama3', api_base='http://localhost:11434')
dspy.configure(lm=lm)

# vLLM
lm = dspy.LM('vllm/meta-llama/Llama-3-70b', api_base='http://localhost:8000')
dspy.configure(lm=lm)

环境变量配置 #

python

import os
import dspy

os.environ['OPENAI_API_KEY'] = 'your-api-key'
os.environ['ANTHROPIC_API_KEY'] = 'your-api-key'

lm = dspy.LM('openai/gpt-4o-mini')
dspy.configure(lm=lm)

核心概念概览 #

text

┌─────────────────────────────────────────────────────────────┐
│                    DSPy 核心概念                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  签名 (Signature)                                   │   │
│  │  定义模块的输入输出接口                              │   │
│  └─────────────────────────────────────────────────────┘   │
│                           │                                  │
│                           ▼                                  │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  模块 (Module)                                      │   │
│  │  实现具体的功能逻辑                                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                           │                                  │
│                           ▼                                  │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  优化器 (Optimizer)                                 │   │
│  │  自动优化模块性能                                    │   │
│  └─────────────────────────────────────────────────────┘   │
│                           │                                  │
│                           ▼                                  │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  评估器 (Metric)                                    │   │
│  │  评估模块输出质量                                    │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

签名（Signature） #

签名是 DSPy 中最基础的概念，它定义了模块的输入和输出。

签名的定义 #

python

import dspy

class QuestionAnswer(dspy.Signature):
    """根据问题给出答案"""
    question = dspy.InputField(desc="用户的问题")
    answer = dspy.OutputField(desc="问题的答案")

签名的组成 #

text

┌─────────────────────────────────────────────────────────────┐
│                    签名的组成                                │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  class MySignature(dspy.Signature):                         │
│      """文档字符串：描述签名的用途"""                        │
│      input_field = dspy.InputField(desc="输入描述")         │
│      output_field = dspy.OutputField(desc="输出描述")       │
│                                                             │
│  组成部分：                                                  │
│  1. 类名：签名名称，应具有描述性                             │
│  2. 文档字符串：描述签名的功能                               │
│  3. 输入字段：定义输入参数                                   │
│  4. 输出字段：定义输出结果                                   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

多输入多输出 #

python

class SummarizeDocument(dspy.Signature):
    """将文档总结为摘要并提取关键词"""
    document = dspy.InputField(desc="需要总结的文档")
    max_length = dspy.InputField(desc="摘要最大长度")
    summary = dspy.OutputField(desc="文档摘要")
    keywords = dspy.OutputField(desc="关键词列表")

字段描述的重要性 #

python

class Translate(dspy.Signature):
    """翻译文本"""
    text = dspy.InputField(desc="需要翻译的文本")
    source_lang = dspy.InputField(desc="源语言，如 '中文'")
    target_lang = dspy.InputField(desc="目标语言，如 '英文'")
    translation = dspy.OutputField(desc="翻译结果，保持原文风格和语气")

内联签名 #

python

# 使用字符串定义简单签名
signature = "question -> answer"
predictor = dspy.Predict(signature)

# 带描述的内联签名
signature = "question: str -> answer: str (简洁准确的回答)"

模块（Module） #

模块是签名的具体实现，封装了处理逻辑。

Predict 模块 #

最基础的预测模块：

python

import dspy

class QA(dspy.Signature):
    """回答问题"""
    question = dspy.InputField()
    answer = dspy.OutputField()

qa = dspy.Predict(QA)
result = qa(question="什么是机器学习？")
print(result.answer)

ChainOfThought 模块 #

带思维链的预测模块：

python

import dspy

class MathProblem(dspy.Signature):
    """解决数学问题"""
    problem = dspy.InputField()
    answer = dspy.OutputField()

solver = dspy.ChainOfThought(MathProblem)
result = solver(problem="如果 x + 5 = 12，那么 x 是多少？")
print(result.answer)
print(result.rationale)

ReAct 模块 #

推理-行动模块，支持工具调用：

python

import dspy

class SearchQA(dspy.Signature):
    """搜索并回答问题"""
    question = dspy.InputField()
    answer = dspy.OutputField()

def search_tool(query):
    return f"搜索结果: {query}"

react = dspy.ReAct(SearchQA, tools=[search_tool])
result = react(question="Python 的创始人是谁？")

自定义模块 #

python

import dspy

class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate = dspy.ChainOfThought(GenerateAnswer)
    
    def forward(self, question):
        context = self.retrieve(question).passages
        return self.generate(context=context, question=question)

class GenerateAnswer(dspy.Signature):
    """根据上下文回答问题"""
    context = dspy.InputField(desc="相关上下文")
    question = dspy.InputField(desc="用户问题")
    answer = dspy.OutputField(desc="基于上下文的回答")

第一个 DSPy 程序 #

简单问答程序 #

python

import dspy

lm = dspy.LM('openai/gpt-4o-mini')
dspy.configure(lm=lm)

class QA(dspy.Signature):
    """回答用户问题"""
    question = dspy.InputField()
    answer = dspy.OutputField()

qa = dspy.Predict(QA)
result = qa(question="Python 的创始人是谁？")
print(result.answer)

带思维链的问答 #

python

import dspy

lm = dspy.LM('openai/gpt-4o-mini')
dspy.configure(lm=lm)

class ReasoningQA(dspy.Signature):
    """分析并回答问题"""
    question = dspy.InputField()
    reasoning = dspy.OutputField(desc="思考过程")
    answer = dspy.OutputField(desc="最终答案")

qa = dspy.ChainOfThought(ReasoningQA)
result = qa(question="为什么天空是蓝色的？")
print(f"推理: {result.reasoning}")
print(f"答案: {result.answer}")

多步骤流水线 #

python

import dspy

lm = dspy.LM('openai/gpt-4o-mini')
dspy.configure(lm=lm)

class Analyze(dspy.Signature):
    """分析文本情感"""
    text = dspy.InputField()
    sentiment = dspy.OutputField(desc="情感：正面/负面/中性")

class Summarize(dspy.Signature):
    """总结文本"""
    text = dspy.InputField()
    summary = dspy.OutputField(desc="简短摘要")

class TextPipeline(dspy.Module):
    def __init__(self):
        super().__init__()
        self.analyze = dspy.Predict(Analyze)
        self.summarize = dspy.ChainOfThought(Summarize)
    
    def forward(self, text):
        sentiment = self.analyze(text=text).sentiment
        summary = self.summarize(text=text).summary
        return dspy.Prediction(sentiment=sentiment, summary=summary)

pipeline = TextPipeline()
result = pipeline(text="今天天气真好，阳光明媚，心情很愉快！")
print(f"情感: {result.sentiment}")
print(f"摘要: {result.summary}")

检索器配置 #

DSPy 内置了检索器支持，可以轻松构建 RAG 应用。

ColBERTv2 检索器 #

python

import dspy

rm = dspy.ColBERTv2(url='http://localhost:8893/api/search')
dspy.configure(rm=rm)

retrieve = dspy.Retrieve(k=3)
results = retrieve("什么是机器学习？")
print(results.passages)

自定义检索器 #

python

import dspy

class CustomRetriever(dspy.Retrieve):
    def __init__(self, k=3):
        super().__init__(k=k)
    
    def forward(self, query):
        results = my_search_function(query, k=self.k)
        return dspy.Prediction(passages=results)

dspy.configure(rm=CustomRetriever(k=5))

向量数据库集成 #

python

import dspy
from dspy.retrieve import ChromideRM

rm = ChromideRM(
    collection_name="my_docs",
    persist_directory="./chroma_db",
    embedding_function=my_embedding_func
)
dspy.configure(rm=rm)

示例（Example） #

DSPy 使用 Example 类来表示训练数据。

创建示例 #

python

from dspy import Example

qa_example = Example(
    question="Python 的创始人是谁？",
    answer="Python 的创始人是 Guido van Rossum"
)

print(qa_example.question)
print(qa_example.answer)

示例与输入字段 #

python

# 指定输入字段
qa_example = Example(
    question="Python 的创始人是谁？",
    answer="Python 的创始人是 Guido van Rossum"
).with_inputs("question")

# 只有 question 是输入，answer 是标签

创建训练集 #

python

trainset = [
    Example(question="Python 的创始人是谁？", answer="Guido van Rossum").with_inputs("question"),
    Example(question="JavaScript 的创始人是谁？", answer="Brendan Eich").with_inputs("question"),
    Example(question="Linux 的创始人是谁？", answer="Linus Torvalds").with_inputs("question"),
]

评估指标（Metric） #

评估指标用于衡量模型输出的质量。

定义评估指标 #

python

def validate_answer(example, pred, trace=None):
    return example.answer.lower() == pred.answer.lower()

def semantic_similarity(example, pred, trace=None):
    return compute_similarity(example.answer, pred.answer)

使用评估器 #

python

from dspy.evaluate import Evaluate

evaluator = Evaluate(
    devset=testset,
    metric=validate_answer,
    num_threads=4,
    display_progress=True
)

score = evaluator(my_program)
print(f"准确率: {score}")

调试与日志 #

查看调用历史 #

python

import dspy

lm = dspy.LM('openai/gpt-4o-mini', cache=False)
dspy.configure(lm=lm)

qa = dspy.Predict("question -> answer")
result = qa(question="测试问题")

lm.inspect_history(n=1)

启用详细日志 #

python

import dspy

dspy.settings.configure(trace=True)

qa = dspy.Predict("question -> answer")
result = qa(question="测试问题")

dspy.inspect_trace()

缓存控制 #

python

import dspy

lm = dspy.LM('openai/gpt-4o-mini', cache=True)
dspy.configure(lm=lm)

qa = dspy.Predict("question -> answer")
result = qa(question="测试问题")

dspy.clear_cache()

最佳实践 #

1. 签名设计 #

python

class GoodSignature(dspy.Signature):
    """清晰的文档字符串，描述签名用途"""
    input = dspy.InputField(desc="详细的输入描述")
    output = dspy.OutputField(desc="详细的输出描述，包括格式要求")

2. 模块组织 #

python

class WellOrganizedModule(dspy.Module):
    def __init__(self):
        super().__init__()
        self.step1 = dspy.Predict(Step1Signature)
        self.step2 = dspy.ChainOfThought(Step2Signature)
        self.step3 = dspy.Predict(Step3Signature)
    
    def forward(self, input_data):
        result1 = self.step1(input=input_data)
        result2 = self.step2(input=result1.output)
        result3 = self.step3(input=result2.output)
        return result3

3. 错误处理 #

python

class RobustModule(dspy.Module):
    def forward(self, question):
        try:
            result = self.predict(question=question)
            return result
        except Exception as e:
            return dspy.Prediction(answer="抱歉，处理出错", error=str(e))

下一步 #

现在你已经掌握了 DSPy 的基础概念，接下来学习签名详解，深入了解签名的各种用法！