DSPy 模块详解 #

┌─────────────────────────────────────────────────────────────┐
│                    模块的作用                                │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 封装逻辑：将 LLM 调用封装为可复用的组件                  │
│  2. 绑定签名：将签名与具体的处理方式结合                     │
│  3. 组合构建：多个模块可以组合成复杂程序                     │
│  4. 自动优化：模块可以被优化器自动改进                       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                    DSPy 内置模块                             │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ dspy.Predict                                         │   │
│  │ 最基础的预测模块，直接根据签名生成输出                │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ dspy.ChainOfThought                                  │   │
│  │ 思维链模块，先推理再输出                              │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ dspy.ReAct                                           │   │
│  │ 推理-行动模块，支持工具调用                           │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ dspy.ProgramOfThoughts                               │   │
│  │ 程序化思维模块，生成代码解决问题                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ dspy.Retrieve                                        │   │
│  │ 检索模块，从知识库获取相关内容                        │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ dspy.MultiChainComparison                            │   │
│  │ 多链比较模块，比较多个推理路径                        │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

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)

class ContextQA(dspy.Signature):
    """根据上下文回答问题"""
    context = dspy.InputField()
    question = dspy.InputField()
    answer = dspy.OutputField()

qa = dspy.Predict(ContextQA)
result = qa(
    context="Python 由 Guido van Rossum 于 1991 年创建。",
    question="Python 何时创建？"
)
print(result.answer)

class Analysis(dspy.Signature):
    """分析文本"""
    text = dspy.InputField()
    summary = dspy.OutputField()
    sentiment = dspy.OutputField()
    keywords = dspy.OutputField()

analyzer = dspy.Predict(Analysis)
result = analyzer(text="今天天气真好，阳光明媚！")
print(result.summary)
print(result.sentiment)
print(result.keywords)

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

translate = dspy.Predict("text, source_lang, target_lang -> translation")
result = translate(text="Hello", source_lang="English", target_lang="Chinese")

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.rationale)
print(result.answer)

┌─────────────────────────────────────────────────────────────┐
│                    ChainOfThought 工作流程                   │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  输入：problem = "如果 x + 5 = 12，那么 x 是多少？"          │
│                                                             │
│  ChainOfThought 处理：                                       │
│  1. 分析问题：这是一个简单的一元一次方程                     │
│  2. 移项：x = 12 - 5                                        │
│  3. 计算：x = 7                                             │
│  4. 验证：7 + 5 = 12 ✓                                      │
│                                                             │
│  输出：                                                      │
│  rationale: "首先，我需要解这个方程..."                      │
│  answer: "x = 7"                                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

class ReasoningTask(dspy.Signature):
    """需要推理的任务"""
    question = dspy.InputField()
    answer = dspy.OutputField()

reasoner = dspy.ChainOfThought(ReasoningTask)

result = reasoner(question="为什么天空是蓝色的？")
result = reasoner(question="请解释相对论的基本原理")
result = reasoner(question="分析这段代码的时间复杂度")

import dspy

class Task(dspy.Signature):
    """任务"""
    input = dspy.InputField()
    output = dspy.OutputField()

predict = dspy.Predict(Task)
cot = dspy.ChainOfThought(Task)

input_text = "解释量子纠缠现象"

result1 = predict(input=input_text)
result2 = cot(input=input_text)

print("Predict 输出:", result1.output)
print("CoT 推理:", result2.rationale)
print("CoT 输出:", result2.output)

import dspy

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

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

react = dspy.ReAct(SearchQA, tools=[search])
result = react(question="Python 的最新版本是什么？")
print(result.answer)

import dspy

class ToolQA(dspy.Signature):
    """使用工具回答问题"""
    question = dspy.InputField()
    answer = dspy.OutputField()

def calculator(expression):
    return str(eval(expression))

def search(query):
    return f"搜索: {query}"

def weather(city):
    return f"{city} 天气: 晴，25°C"

react = dspy.ReAct(ToolQA, tools=[calculator, search, weather])
result = react(question="北京今天天气怎么样？")

┌─────────────────────────────────────────────────────────────┐
│                    ReAct 工作流程                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  问题："北京今天天气怎么样？"                                │
│                                                             │
│  步骤 1: 思考                                               │
│  "我需要查询北京的天气信息"                                  │
│                                                             │
│  步骤 2: 行动                                               │
│  调用 weather("北京")                                       │
│                                                             │
│  步骤 3: 观察                                               │
│  "北京 天气: 晴，25°C"                                      │
│                                                             │
│  步骤 4: 回答                                               │
│  "北京今天天气晴朗，气温 25°C"                               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

import dspy

class MathCalculation(dspy.Signature):
    """数学计算"""
    problem = dspy.InputField()
    answer = dspy.OutputField()

solver = dspy.ProgramOfThoughts(MathCalculation)
result = solver(problem="计算 1 到 100 的和")
print(result.answer)

┌─────────────────────────────────────────────────────────────┐
│                    ProgramOfThoughts 工作流程                │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  问题："计算 1 到 100 的和"                                  │
│                                                             │
│  生成的代码：                                                │
│  total = 0                                                  │
│  for i in range(1, 101):                                    │
│      total += i                                             │
│  print(total)                                               │
│                                                             │
│  执行结果：5050                                             │
│                                                             │
│  输出：answer = "5050"                                      │
│                                                             │
└─────────────────────────────────────────────────────────────┘

class DataAnalysis(dspy.Signature):
    """数据分析"""
    data_description = dspy.InputField()
    analysis_result = dspy.OutputField()

analyzer = dspy.ProgramOfThoughts(DataAnalysis)

result = analyzer(data_description="计算列表 [1,2,3,4,5] 的平均值和标准差")

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)

import dspy

class GenerateAnswer(dspy.Signature):
    """根据上下文回答问题"""
    context = dspy.InputField()
    question = dspy.InputField()
    answer = dspy.OutputField()

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

import dspy

class MyModule(dspy.Module):
    def __init__(self):
        super().__init__()
        self.predict = dspy.Predict(MySignature)
    
    def forward(self, **kwargs):
        return self.predict(**kwargs)

import dspy

class Step1(dspy.Signature):
    """第一步：分析"""
    input = dspy.InputField()
    analysis = dspy.OutputField()

class Step2(dspy.Signature):
    """第二步：处理"""
    analysis = dspy.InputField()
    result = dspy.OutputField()

class Step3(dspy.Signature):
    """第三步：输出"""
    result = dspy.InputField()
    output = dspy.OutputField()

class Pipeline(dspy.Module):
    def __init__(self):
        super().__init__()
        self.analyze = dspy.Predict(Step1)
        self.process = dspy.ChainOfThought(Step2)
        self.output = dspy.Predict(Step3)
    
    def forward(self, input_data):
        analysis = self.analyze(input=input_data).analysis
        result = self.process(analysis=analysis).result
        output = self.output(result=result).output
        return dspy.Prediction(output=output)

import dspy

class SimpleQA(dspy.Signature):
    """简单问答"""
    question = dspy.InputField()
    answer = dspy.OutputField()

class ComplexQA(dspy.Signature):
    """复杂问答"""
    question = dspy.InputField()
    reasoning = dspy.OutputField()
    answer = dspy.OutputField()

class ClassifyQuestion(dspy.Signature):
    """分类问题"""
    question = dspy.InputField()
    complexity = dspy.OutputField(desc="simple 或 complex")

class AdaptiveQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.classifier = dspy.Predict(ClassifyQuestion)
        self.simple_qa = dspy.Predict(SimpleQA)
        self.complex_qa = dspy.ChainOfThought(ComplexQA)
    
    def forward(self, question):
        complexity = self.classifier(question=question).complexity
        
        if complexity == "simple":
            return self.simple_qa(question=question)
        else:
            return self.complex_qa(question=question)

import dspy

class ProcessItem(dspy.Signature):
    """处理单个项目"""
    item = dspy.InputField()
    result = dspy.OutputField()

class BatchProcessor(dspy.Module):
    def __init__(self):
        super().__init__()
        self.process = dspy.Predict(ProcessItem)
    
    def forward(self, items):
        results = []
        for item in items:
            result = self.process(item=item)
            results.append(result.result)
        return dspy.Prediction(results=results)

import dspy

class Extract(dspy.Signature):
    """提取信息"""
    text = dspy.InputField()
    entities = dspy.OutputField()

class Transform(dspy.Signature):
    """转换格式"""
    entities = dspy.InputField()
    formatted = dspy.OutputField()

class Load(dspy.Signature):
    """加载数据"""
    formatted = dspy.InputField()
    status = dspy.OutputField()

class ETL(dspy.Module):
    def __init__(self):
        super().__init__()
        self.extract = dspy.Predict(Extract)
        self.transform = dspy.ChainOfThought(Transform)
        self.load = dspy.Predict(Load)
    
    def forward(self, text):
        entities = self.extract(text=text).entities
        formatted = self.transform(entities=entities).formatted
        status = self.load(formatted=formatted).status
        return dspy.Prediction(status=status)

import dspy

class AnalyzeSentiment(dspy.Signature):
    """情感分析"""
    text = dspy.InputField()
    sentiment = dspy.OutputField()

class ExtractKeywords(dspy.Signature):
    """关键词提取"""
    text = dspy.InputField()
    keywords = dspy.OutputField()

class SummarizeText(dspy.Signature):
    """文本摘要"""
    text = dspy.InputField()
    summary = dspy.OutputField()

class TextAnalyzer(dspy.Module):
    def __init__(self):
        super().__init__()
        self.sentiment = dspy.Predict(AnalyzeSentiment)
        self.keywords = dspy.Predict(ExtractKeywords)
        self.summary = dspy.ChainOfThought(SummarizeText)
    
    def forward(self, text):
        sentiment = self.sentiment(text=text).sentiment
        keywords = self.keywords(text=text).keywords
        summary = self.summary(text=text).summary
        
        return dspy.Prediction(
            sentiment=sentiment,
            keywords=keywords,
            summary=summary
        )

import dspy

class SubModuleA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.step = dspy.Predict("input -> output")
    
    def forward(self, input_data):
        return self.step(input=input_data)

class SubModuleB(dspy.Module):
    def __init__(self):
        super().__init__()
        self.step = dspy.ChainOfThought("input -> output")
    
    def forward(self, input_data):
        return self.step(input=input_data)

class MainModule(dspy.Module):
    def __init__(self):
        super().__init__()
        self.module_a = SubModuleA()
        self.module_b = SubModuleB()
    
    def forward(self, input_data):
        result_a = self.module_a(input_data)
        result_b = self.module_b(result_a.output)
        return result_b

import dspy

class MyModule(dspy.Module):
    def __init__(self):
        super().__init__()
        self.predict = dspy.Predict("input -> output")

module = MyModule()

module.save("my_module.json")

loaded_module = MyModule()
loaded_module.load("my_module.json")

import copy

original = dspy.Predict("input -> output")
cloned = copy.deepcopy(original)

import dspy

class MyModule(dspy.Module):
    def __init__(self):
        super().__init__()
        self.step1 = dspy.Predict("input -> intermediate")
        self.step2 = dspy.ChainOfThought("intermediate -> output")

module = MyModule()
print(module.named_modules())
print(module.named_predictors())

import dspy

dspy.settings.configure(trace=True)

module = dspy.Predict("question -> answer")
result = module(question="测试")

dspy.inspect_trace()