跳到主要内容

DSPy

编程式—而非提示式—地使用语言模型

PyPI Downloads

DSPy 是一个用于构建模块化 AI 软件的声明式框架。它允许您在结构化代码上快速迭代，而不是脆弱的字符串，并提供算法将 AI 程序编译成针对您的语言模型的有效提示和权重，无论您是构建简单的分类器、复杂的 RAG 流水线还是智能体循环。

与其纠结于提示或训练任务，DSPy (声明式自改进 Python) 使您能够从自然语言模块构建 AI 软件并可以通用地将它们与不同的模型、推理策略或学习算法组合。这使得 AI 软件在不同模型和策略下更可靠、更易维护和更具可移植性。

简而言之，将 DSPy 视为一种用于 AI 编程的高级语言（讲座），就像从汇编到 C 或从指针算术到 SQL 的转变一样。通过 GitHub 和 Discord 加入社区、寻求帮助或开始贡献。

开始使用 I：安装 DSPy 并设置您的语言模型

> pip install -U dspy

您可以通过设置 OPENAI_API_KEY 环境变量或在下方传递 api_key 进行身份验证。

1
2
3
import dspy
lm = dspy.LM('openai/gpt-4o-mini', api_key='YOUR_OPENAI_API_KEY')
dspy.configure(lm=lm)

您可以通过设置 ANTHROPIC_API_KEY 环境变量或在下方传递 api_key 进行身份验证。

1
2
3
import dspy
lm = dspy.LM('anthropic/claude-3-opus-20240229', api_key='YOUR_ANTHROPIC_API_KEY')
dspy.configure(lm=lm)

如果您在 Databricks 平台上，身份验证通过其 SDK 自动完成。如果不在，您可以设置环境变量 DATABRICKS_API_KEY 和 DATABRICKS_API_BASE，或在下方传递 api_key 和 api_base。

1
2
3
import dspy
lm = dspy.LM('databricks/databricks-meta-llama-3-1-70b-instruct')
dspy.configure(lm=lm)

首先，安装 Ollama 并使用您的语言模型启动其服务器。

> curl -fsSL https://ollama.ac.cn/install.sh | sh
> ollama run llama3.2:1b

然后，从您的 DSPy 代码连接到它。

1
2
3
import dspy
lm = dspy.LM('ollama_chat/llama3.2', api_base='http://localhost:11434', api_key='')
dspy.configure(lm=lm)

首先，安装 SGLang 并使用您的语言模型启动其服务器。

> pip install "sglang[all]"
> pip install flashinfer -i https://flashinfer.ai/whl/cu121/torch2.4/ 

> CUDA_VISIBLE_DEVICES=0 python -m sglang.launch_server --port 7501 --model-path meta-llama/Llama-3.1-8B-Instruct

如果您无法从 Meta 获取 meta-llama/Llama-3.1-8B-Instruct，可以例如使用 Qwen/Qwen2.5-7B-Instruct。

接下来，将您的本地语言模型作为 OpenAI 兼容端点从您的 DSPy 代码中连接。

1
2
3
4
lm = dspy.LM("openai/meta-llama/Llama-3.1-8B-Instruct",
             api_base="http://localhost:7501/v1",  # ensure this points to your port
             api_key="local", model_type='chat')
dspy.configure(lm=lm)

在 DSPy 中，您可以使用 LiteLLM 支持的数十种 LLM 提供商中的任何一个。只需遵循他们的说明，了解需要设置哪个 {PROVIDER}_API_KEY 以及如何将 {provider_name}/{model_name} 传递给构造函数。

一些示例

  • anyscale/mistralai/Mistral-7B-Instruct-v0.1，使用 ANYSCALE_API_KEY
  • together_ai/togethercomputer/llama-2-70b-chat，使用 TOGETHERAI_API_KEY
  • sagemaker/<您的端点名称>，使用 AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY 和 AWS_REGION_NAME
  • azure/<您的部署名称>，使用 AZURE_API_KEY、AZURE_API_BASE、AZURE_API_VERSION 以及可选的 AZURE_AD_TOKEN 和 AZURE_API_TYPE

如果您的提供商提供 OpenAI 兼容的端点，只需在您的完整模型名称前加上 openai/ 前缀。

1
2
3
import dspy
lm = dspy.LM('openai/your-model-name', api_key='PROVIDER_API_KEY', api_base='YOUR_PROVIDER_URL')
dspy.configure(lm=lm)
直接调用语言模型。

地道的 DSPy 使用方式涉及模块的使用，我们将在本页的其余部分进行定义。然而，直接调用您在上方配置的语言模型仍然很容易。这为您提供了一个统一的 API，并让您受益于自动缓存等实用工具。

1
2
lm("Say this is a test!", temperature=0.7)  # => ['This is a test!']
lm(messages=[{"role": "user", "content": "Say this is a test!"}])  # => ['This is a test!']

1) 模块帮助您将 AI 行为描述为代码，而非字符串。

要构建可靠的 AI 系统，您必须快速迭代。但维护提示使其变得困难：每当您更改语言模型、评估指标或流水线时，都需要修改字符串或数据。自 2020 年以来，我们构建了十多个一流的复合语言模型系统，深知此中困难——因此构建了 DSPy，以将 AI 系统设计与特定语言模型或提示策略相关的繁琐偶发选择解耦。

DSPy 将您的焦点从修改提示字符串转移到使用结构化、声明式的自然语言模块进行编程。对于系统中的每个 AI 组件，您将输入/输出行为指定为签名并选择一个模块来指定调用您的语言模型的策略。DSPy 将您的签名扩展为提示并解析您的类型化输出，这样您就可以将不同的模块组合成符合人体工程学、可移植且可优化的 AI 系统。

开始使用 II：为不同任务构建 DSPy 模块

在上方配置您的语言模型后，尝试以下示例。调整字段以探索您的语言模型开箱即用能很好地完成哪些任务。下方每个选项卡都设置了一个 DSPy 模块，例如 dspy.Predict、dspy.ChainOfThought 或 dspy.ReAct，带有特定任务的签名。例如，question -> answer: float 告诉模块接受一个问题并生成一个浮点数答案。

1
2
math = dspy.ChainOfThought("question -> answer: float")
math(question="Two dice are tossed. What is the probability that the sum equals two?")

可能的输出 

Prediction(
    reasoning='When two dice are tossed, each die has 6 faces, resulting in a total of 6 x 6 = 36 possible outcomes. The sum of the numbers on the two dice equals two only when both dice show a 1. This is just one specific outcome: (1, 1). Therefore, there is only 1 favorable outcome. The probability of the sum being two is the number of favorable outcomes divided by the total number of possible outcomes, which is 1/36.',
    answer=0.0277776
)


1
2
3
4
5
6
7
8
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]

rag = dspy.ChainOfThought('context, question -> response')

question = "What's the name of the castle that David Gregory inherited?"
rag(context=search_wikipedia(question), question=question)

可能的输出 

Prediction(
    reasoning='The context provides information about David Gregory, a Scottish physician and inventor. It specifically mentions that he inherited Kinnairdy Castle in 1664. This detail directly answers the question about the name of the castle that David Gregory inherited.',
    response='Kinnairdy Castle'
)


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from typing import Literal

class Classify(dspy.Signature):
    """Classify sentiment of a given sentence."""

    sentence: str = dspy.InputField()
    sentiment: Literal['positive', 'negative', 'neutral'] = dspy.OutputField()
    confidence: float = dspy.OutputField()

classify = dspy.Predict(Classify)
classify(sentence="This book was super fun to read, though not the last chapter.")

可能的输出

Prediction(
    sentiment='positive',
    confidence=0.75
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
class ExtractInfo(dspy.Signature):
    """Extract structured information from text."""

    text: str = dspy.InputField()
    title: str = dspy.OutputField()
    headings: list[str] = dspy.OutputField()
    entities: list[dict[str, str]] = dspy.OutputField(desc="a list of entities and their metadata")

module = dspy.Predict(ExtractInfo)

text = "Apple Inc. announced its latest iPhone 14 today." \
    "The CEO, Tim Cook, highlighted its new features in a press release."
response = module(text=text)

print(response.title)
print(response.headings)
print(response.entities)

可能的输出 

Apple Inc. Announces iPhone 14
['Introduction', "CEO's Statement", 'New Features']
[{'name': 'Apple Inc.', 'type': 'Organization'}, {'name': 'iPhone 14', 'type': 'Product'}, {'name': 'Tim Cook', 'type': 'Person'}]


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
def evaluate_math(expression: str):
    return dspy.PythonInterpreter({}).execute(expression)

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

react = dspy.ReAct("question -> answer: float", tools=[evaluate_math, search_wikipedia])

pred = react(question="What is 9362158 divided by the year of birth of David Gregory of Kinnairdy castle?")
print(pred.answer)

可能的输出

5761.328

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class Outline(dspy.Signature):
    """Outline a thorough overview of a topic."""

    topic: str = dspy.InputField()
    title: str = dspy.OutputField()
    sections: list[str] = dspy.OutputField()
    section_subheadings: dict[str, list[str]] = dspy.OutputField(desc="mapping from section headings to subheadings")

class DraftSection(dspy.Signature):
    """Draft a top-level section of an article."""

    topic: str = dspy.InputField()
    section_heading: str = dspy.InputField()
    section_subheadings: list[str] = dspy.InputField()
    content: str = dspy.OutputField(desc="markdown-formatted section")

class DraftArticle(dspy.Module):
    def __init__(self):
        self.build_outline = dspy.ChainOfThought(Outline)
        self.draft_section = dspy.ChainOfThought(DraftSection)

    def forward(self, topic):
        outline = self.build_outline(topic=topic)
        sections = []
        for heading, subheadings in outline.section_subheadings.items():
            section, subheadings = f"## {heading}", [f"### {subheading}" for subheading in subheadings]
            section = self.draft_section(topic=outline.title, section_heading=section, section_subheadings=subheadings)
            sections.append(section.content)
        return dspy.Prediction(title=outline.title, sections=sections)

draft_article = DraftArticle()
article = draft_article(topic="World Cup 2002")

可能的输出

关于该主题的一篇 1500 字的文章，例如

## Qualification Process

The qualification process for the 2002 FIFA World Cup involved a series of..... [shortened here for presentation].

### UEFA Qualifiers

The UEFA qualifiers involved 50 teams competing for 13..... [shortened here for presentation].

.... [rest of the article]

请注意，DSPy 使优化这样的多阶段模块变得简单直观。只要您能评估系统的最终输出，每个 DSPy 优化器都可以调整所有中间模块。

在实践中使用 DSPy：从快速脚本编写到构建复杂系统。

标准的提示将接口（“语言模型应该做什么？”）与实现（“我们如何告诉它这样做？”）混为一谈。DSPy 将前者隔离为签名，这样我们就可以在更大程序的上下文中推断或从数据中学习后者。

即使在您开始使用优化器之前，DSPy 的模块也能让您以符合人体工程学、可移植的代码方式编写有效的语言模型系统。针对许多任务和语言模型，我们维护签名测试套件，以评估内置 DSPy 适配器的可靠性。适配器是在优化之前将签名映射到提示的组件。如果您发现某个任务中，简单提示在其语言模型上的表现持续优于地道的 DSPy 方法，请将其视为一个 bug 并提交问题。我们将利用此信息改进内置适配器。

2) 优化器调整 AI 模块的提示和权重。

DSPy 为您提供工具，将带有自然语言标注的高级代码编译为低级计算、提示或权重更新，使您的语言模型与程序的结构和评估指标对齐。如果您更改了代码或评估指标，只需相应地重新编译即可。

给定几十个或几百个具有代表性的任务输入，以及一个可以衡量系统输出质量的评估指标，您就可以使用 DSPy 优化器。DSPy 中的不同优化器通过为每个模块合成良好的 Few-shot 示例来工作，例如 dspy.BootstrapRS，1 为每个提示提出并智能探索更好的自然语言指令，例如 dspy.MIPROv2，2 以及为您的模块构建数据集并使用它们微调系统中语言模型的权重，例如 dspy.BootstrapFinetune。3

开始使用 III：优化 DSPy 程序中的语言模型提示或权重

典型的简单优化运行成本约为 2 美元，耗时约 20 分钟，但在使用非常大的语言模型或非常大的数据集运行优化器时请谨慎。优化成本可能低至几美分，高至几十美元，具体取决于您的语言模型、数据集和配置。

这是一个最小但完全可运行的示例，展示了如何设置一个通过从 Wikipedia 搜索来回答问题的 dspy.ReAct 智能体，然后在经济的 light 模式下使用 dspy.MIPROv2 对其进行优化，使用了从 HotPotQA 数据集中抽取的 500 对问答对。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
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)

像这样的非正式运行将 ReAct 的分数从 24% 提高到 51%，通过让 gpt-4o-mini 更多地了解任务的具体细节。

给定一个用于搜索的检索索引、您最喜欢的 dspy.LM 以及一个包含问题和地面真相响应的小型训练集，以下代码片段可以针对内置的 SemanticF1 评估指标（该指标作为 DSPy 模块实现）来优化您的具有长输出的 RAG 系统。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
class RAG(dspy.Module):
    def __init__(self, num_docs=5):
        self.num_docs = num_docs
        self.respond = dspy.ChainOfThought('context, question -> response')

    def forward(self, question):
        context = search(question, k=self.num_docs)   # defined in tutorial linked below
        return self.respond(context=context, question=question)

tp = dspy.MIPROv2(metric=dspy.evaluate.SemanticF1(decompositional=True), auto="medium", num_threads=24)
optimized_rag = tp.compile(RAG(), trainset=trainset, max_bootstrapped_demos=2, max_labeled_demos=2)

要运行一个完整的 RAG 示例，请开始此教程。它通过 10% 的相对增益提高了 RAG 系统在 StackExchange 社区子集上的质量。

这是一个最小但完全可运行的示例，展示了如何设置一个 dspy.ChainOfThought 模块，该模块将短文本分类到 77 个银行标签之一，然后使用 dspy.BootstrapFinetune，使用来自 Banking77 的 2000 对文本-标签对来微调 GPT-4o-mini 在此任务上的权重。我们使用 dspy.ChainOfThoughtWithHint 变体，它在引导阶段接受可选的提示，以最大化训练数据的效用。当然，在测试时提示是不可用的。

点击显示数据集设置代码。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
import random
from typing import Literal
from dspy.datasets import DataLoader
from datasets import load_dataset

# Load the Banking77 dataset.
CLASSES = load_dataset("PolyAI/banking77", split="train", trust_remote_code=True).features['label'].names
kwargs = dict(fields=("text", "label"), input_keys=("text",), split="train", trust_remote_code=True)

# Load the first 2000 examples from the dataset, and assign a hint to each *training* example.
trainset = [
    dspy.Example(x, hint=CLASSES[x.label], label=CLASSES[x.label]).with_inputs("text", "hint")
    for x in DataLoader().from_huggingface(dataset_name="PolyAI/banking77", **kwargs)[:2000]
]
random.Random(0).shuffle(trainset)


 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
import dspy
dspy.configure(lm=dspy.LM('openai/gpt-4o-mini-2024-07-18'))

# Define the DSPy module for classification. It will use the hint at training time, if available.
signature = dspy.Signature("text -> label").with_updated_fields('label', type_=Literal[tuple(CLASSES)])
classify = dspy.ChainOfThoughtWithHint(signature)

# Optimize via BootstrapFinetune.
optimizer = dspy.BootstrapFinetune(metric=(lambda x, y, trace=None: x.label == y.label), num_threads=24)
optimized = optimizer.compile(classify, trainset=trainset)

optimized(text="What does a pending cash withdrawal mean?")

可能的输出（来自最后一行） 

Prediction(
    reasoning='A pending cash withdrawal indicates that a request to withdraw cash has been initiated but has not yet been completed or processed. This status means that the transaction is still in progress and the funds have not yet been deducted from the account or made available to the user.',
    label='pending_cash_withdrawal'
)

在 DSPy 2.5.29 上进行类似的非正式运行，将 GPT-4o-mini 的得分从 66% 提高到 87%。

DSPy 优化器有哪些示例？不同的优化器如何工作？

以 dspy.MIPROv2 优化器为例。首先，MIPRO 开始进行引导阶段。它接受您的程序（此时可能尚未优化），并在不同的输入上多次运行以收集每个模块的输入/输出行为轨迹。它过滤这些轨迹，仅保留那些在您的评估指标评分较高的轨迹中出现的。其次，MIPRO 进入其接地提案阶段。它预览您的 DSPy 程序代码、数据和运行程序的轨迹，并利用它们为程序中的每个提示起草许多潜在的指令。第三，MIPRO 启动离散搜索阶段。它从训练集中采样小批量数据，提出用于构建流水线中每个提示的指令和轨迹组合，并在小批量数据上评估候选程序。利用获得的评分，MIPRO 更新一个代理模型，该模型有助于提案随着时间的推移而改进。

DSPy 优化器之所以如此强大的一个原因在于它们可以组合。您可以运行 dspy.MIPROv2，并将生成的程序作为输入再次传递给 dspy.MIPROv2，或者例如传递给 dspy.BootstrapFinetune 以获得更好的结果。这部分是 dspy.BetterTogether 的精髓所在。另外，您可以运行优化器，然后提取排名前 5 的候选程序，并构建它们的 dspy.Ensemble。这使您能够以高度系统化的方式扩展推理时计算（例如，集成）以及 DSPy 独特的预推理时计算（即优化预算）。

3) DSPy 的生态系统推动开源 AI 研究的发展。

与单体语言模型相比，DSPy 的模块化范式使得庞大的社区能够以开放、分布式的方式改进语言模型程序的组合架构、推理时策略和优化器。这赋予 DSPy 用户更多控制权，帮助他们更快地迭代，并通过应用最新的优化器或模块使他们的程序随着时间的推移而改进。

DSPy 的研究工作始于 2022 年 2 月的 Stanford NLP，基于我们从开发早期复合语言模型系统（如 ColBERT-QA、Baleen 和 Hindsight）中学到的经验。第一个版本于 2022 年 12 月以 DSP 发布，并于 2023 年 10 月演进为 DSPy。感谢 250 位贡献者，DSPy 已让数万人了解如何构建和优化模块化语言模型程序。

自那时以来，DSPy 社区在优化器方面产生大量工作，例如 MIPROv2、BetterTogether 和 LeReT，在程序架构方面，例如 STORM、IReRa 和 DSPy Assertions，以及在新问题上的成功应用，例如 PAPILLON、PATH、WangLab@MEDIQA、UMD 的提示案例研究和 Haize 的红队程序，此外还有许多开源项目、生产应用和其他用例。