DSPy^[1] （Declarative Self-improving Python）是一个开源框架，它允许用户通过编程而非简单的提示来与语言模型交互。

这个框架致力于快速迭代构建模块化的AI系统，并提供优化提示和权重的算法，无论是构建简单的分类器、复杂的RAG管道还是代理循环。

主要特点

• 编程而非提示：DSPy允许用户编写组合式的Python代码，而不是脆弱的提示。
• 模块化AI系统：快速迭代构建模块化的AI系统。
• 优化算法：提供优化提示和权重的算法。
• 高质量输出：使用DSPy的工具教会语言模型提供高质量的输出。

使用场景

DSPy适用于需要构建语言模型的各种场景，包括但不限于：

• 简单的分类任务。
• 复杂的RAG（Retrieval-Augmented Generation）管道。
• 代理循环等高级应用。

项目使用

安装

使用pip安装DSPy：

pip install dspy

或者，安装最新版本：

pip install git+https://github.com/stanfordnlp/dspy.git

使用

• 配置语言模型和检索模型

import dspy
# 配置语言模型，例如使用OpenAI
lm = dspy.OpenAI(model='gpt-3.5-turbo')
# 配置检索模型
rm = dspy.WeaviateRM("MyExampleIndex", weaviate_client=client)
# 设置DSPy默认使用的语言模型和检索模型
dspy.settings.configure(lm=lm, rm=rm)

• 加载数据集

from dspy.datasets import DataLoader
dl = DataLoader()
dataset = dl.from_csv("cmrc2018_sampled.csv", fields=("question", "answer"), input_keys=("Title", "question"))
splits = dl.train_test_split(dataset, train_size=0.8)
trainset = splits['train']
devset = splits['test']

• 构建DSPy模块

用户可以通过配置好的语言模型尝试不同的DSPy模块，如dspy.Predict、dspy.ChainOfThought或dspy.ReAct，每个模块都有特定的任务签名。

class GenerateAnswer(dspy.Signature):
    """Answer questions with short factoid answers."""
    context = dspy.InputField(desc="may contain relevant facts")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="often between 1 and 5 words")

class RAG(dspy.Module):
    def __init__(self, num_passages=3):
        super().__init__()
        self.retrieve = dspy.Retrieve(k=num_passages)
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question):
        context = self.retrieve(question).passages
        prediction = self.generate_answer(context=context, question=question)
        return dspy.Prediction(context=context, answer=prediction.answer)

• 编译DSPy程序：DSPy编译器将考虑训练数据、程序、优化器和验证度量，以优化程序。

示例

示例展示了如何设置和优化一个dspy.ReAct代理，该代理通过搜索Wikipedia回答问题，并使用dspy.MIPROv2进行优化。

import dspy
from dspy.datasets import HotPotQA

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

def search_wikipedia(query: str) -> list[str]:
    results = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')(query, k=3)
    return [x['text'] for x in results]

trainset = [x.with_inputs('question') for x in HotPotQA(train_seed=2024, train_size=500).train]
react = dspy.ReAct("question -> answer", tools=[search_wikipedia])

tp = dspy.MIPROv2(metric=dspy.evaluate.answer_exact_match, auto="light", num_threads=24)
optimized_react = tp.compile(react, trainset=trainset)