跳至内容

模块

一个DSPy模块是使用语言模型的程序构建块。

  • 每个内置模块抽象化一种提示技术（如思维链或ReAct）。关键的是，它们被通用化以处理任何签名。

  • 一个DSPy模块具有可学习参数（即构成提示和语言模型权重的小片段），并且可以被调用（调用）以处理输入并返回输出。

  • 多个模块可以组合成更大的模块（程序）。DSPy模块直接受到PyTorch中神经网络模块的启发，但应用于语言模型程序。

如何使用内置模块，比如 dspy.Predict 或 dspy.ChainOfThought？

让我们从最基础的模块开始，dspy.Predict。在内部，所有其他DSPy模块都是使用dspy.Predict构建的。我们假设你已经至少对DSPy签名有了一些了解，这些是用于定义我们在DSPy中使用的任何模块行为的声明式规范。

要使用一个模块，我们首先通过赋予它一个签名来声明它。然后我们用输入参数调用该模块，并提取输出字段！

sentence = "it's a charming and often affecting journey."  # SST-2数据集中的示例。

# 1) 使用签名声明。
classify = dspy.Predict('sentence -> sentiment: bool')

# 2) 使用输入参数调用。
response = classify(sentence=sentence)

# 3) 访问输出。
print(response.sentiment)
输出: 
True

当我们声明一个模块时，可以向它传递配置键。

下面，我们将传递 n=5 来请求五个补全。我们也可以传递 temperature 或 max_len 等参数。

让我们使用dspy.ChainOfThought。在许多情况下，只需将dspy.ChainOfThought替换dspy.Predict就能提高质量。

question = "关于ColBERT检索模型有什么优点？"

# 1) 使用签名声明，并传递一些配置。
classify = dspy.ChainOfThought('question -> answer', n=5)

# 2) 使用输入参数调用。
response = classify(question=question)

# 3) 访问输出结果。
response.completions.answer
可能的输出： 
['ColBERT检索模型的一大优点是与其他模型相比具有更高的效率和效果。',
 '它能够从大型文档集合中高效检索相关信息。',
 'ColBERT检索模型的一大优点是与其他模型相比具有更优越的性能，并且有效利用了预训练语言模型。',
 'ColBERT检索模型的一大优点是与其他模型相比具有更高的效率和准确性。',
 'ColBERT检索模型的一大优点是能够整合用户反馈并支持复杂查询。']

让我们在这里讨论输出对象。dspy.ChainOfThought 模块通常会在你的签名输出字段前注入一个reasoning。

让我们检查（第一个）推理和答案！

print(f"推理过程: {response.reasoning}")
print(f"答案: {response.answer}")
可能的输出: 
推理过程: 我们可以考虑到ColBERT在效率和效果方面已证明优于其他最先进的检索模型。它使用上下文嵌入并以既准确又可扩展的方式执行文档检索。
答案: ColBERT检索模型的一个很棒的特点是它相比其他模型具有卓越的效率和效果。

无论我们请求一个还是多个完成项，这都可以访问。

我们也可以将不同的补全结果作为Prediction列表访问，或者作为多个列表访问，每个字段一个列表。

response.completions[3].reasoning == response.completions.reasoning[3]
输出: 
True

还有哪些其他DSPy模块？我该如何使用它们？

其他的都非常相似。它们主要改变实现你签名时的内部行为！

  1. dspy.Predict: 基础预测器。不修改签名。处理关键形式的学习（即存储指令和演示以及对LM的更新）。

  2. dspy.ChainOfThought: 教导语言模型在确定签名响应之前进行逐步思考。

  3. dspy.ProgramOfThought: 教导语言模型输出代码，其执行结果将决定响应内容。

  4. dspy.ReAct: 一个可以使用工具来实现给定签名的智能体。

  5. dspy.MultiChainComparison: 可以比较来自ChainOfThought的多个输出，以生成最终预测。

我们还有一些函数式模块：

  1. dspy.majority: 可以进行基本投票，从一组预测中返回最受欢迎的响应。

DSPy 模块在简单任务上的几个示例。

配置好你的lm后，尝试下面的示例。调整字段以探索你的语言模型开箱即用能胜任哪些任务。

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='当两个骰子被投掷时，每个骰子有6个面，总共产生6 x 6 = 36种可能结果。两个骰子上的数字之和等于二仅当两个骰子都显示1。这只是一个特定结果：(1, 1)。因此，只有一个有利结果。和为二的概率是有利结果数除以总可能结果数，即1/36。',
    answer=0.0277776
)


1
2
3
4
5
6
7
8
9
def search(query: str) -> list[str]:
    """Retrieves abstracts from Wikipedia."""
    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(question), question=question)

可能的输出: 

Prediction(
    reasoning='上下文提供了关于苏格兰医生兼发明家David Gregory的信息。特别提到他在1664年继承了Kinnairdy城堡。这一细节直接回答了关于David Gregory继承的城堡名称的问题。',
    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
text = "Apple Inc. announced its latest iPhone 14 today. The CEO, Tim Cook, highlighted its new features in a press release."

module = dspy.Predict("text -> title, headings: list[str], entities_and_metadata: list[dict[str, str]]")
response = module(text=text)

print(response.title)
print(response.headings)
print(response.entities_and_metadata)

可能的输出: 

苹果发布iPhone 14
['引言', '关键特性', "CEO声明"]
[{'entity': 'Apple Inc.', 'type': 'Organization'}, {'entity': 'iPhone 14', 'type': 'Product'}, {'entity': 'Tim Cook', 'type': 'Person'}]


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

def search_wikipedia(query: str) -> 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

如何将多个模块组合成一个更大的程序？

DSPy 只是使用模块的 Python 代码，你可以按任意控制流编写，内部在 compile 时使用一点魔法来追踪你的 LM 调用。这意味着你可以自由调用这些模块。

查看教程，如多跳搜索，其模块作为示例在下方重现。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
class Hop(dspy.Module):
    def __init__(self, num_docs=10, num_hops=4):
        self.num_docs, self.num_hops = num_docs, num_hops
        self.generate_query = dspy.ChainOfThought('claim, notes -> query')
        self.append_notes = dspy.ChainOfThought('claim, notes, context -> new_notes: list[str], titles: list[str]')

    def forward(self, claim: str) -> list[str]:
        notes = []
        titles = []

        for _ in range(self.num_hops):
            query = self.generate_query(claim=claim, notes=notes).query
            context = search(query, k=self.num_docs)
            prediction = self.append_notes(claim=claim, notes=notes, context=context)
            notes.extend(prediction.new_notes)
            titles.extend(prediction.titles)

        return dspy.Prediction(notes=notes, titles=list(set(titles)))

然后你可以创建一个自定义模块类 Hop 的实例，然后通过 __call__ 方法调用它：

hop = Hop()
print(hop(claim="Stephen Curry is the best 3 pointer shooter ever in the human history"))

如何追踪语言模型使用情况？

版本要求

LM 使用追踪功能在 DSPy 版本 2.6.16 及之后可用。

DSPy 提供内置的语言模型使用情况跟踪功能，适用于所有模块调用。要启用跟踪：

dspy.settings.configure(track_usage=True)

一旦启用，您可以从任何 dspy.Prediction 对象中访问使用统计信息：

usage = prediction_instance.get_lm_usage()

使用数据以字典形式返回，将每个语言模型名称映射到其使用统计信息。以下是一个完整示例：

import dspy

# Configure DSPy with tracking enabled
dspy.settings.configure(
    lm=dspy.LM("openai/gpt-4o-mini", cache=False),
    track_usage=True
)

# Define a simple program that makes multiple LM calls
class MyProgram(dspy.Module):
    def __init__(self):
        self.predict1 = dspy.ChainOfThought("question -> answer")
        self.predict2 = dspy.ChainOfThought("question, answer -> score")

    def __call__(self, question: str) -> str:
        answer = self.predict1(question=question)
        score = self.predict2(question=question, answer=answer)
        return score

# Run the program and check usage
program = MyProgram()
output = program(question="What is the capital of France?")
print(output.get_lm_usage())

这将输出使用统计信息，例如：

{
    'openai/gpt-4o-mini': {
        'completion_tokens': 61,
        'prompt_tokens': 260,
        'total_tokens': 321,
        'completion_tokens_details': {
            'accepted_prediction_tokens': 0,
            'audio_tokens': 0,
            'reasoning_tokens': 0,
            'rejected_prediction_tokens': 0,
            'text_tokens': None
        },
        'prompt_tokens_details': {
            'audio_tokens': 0,
            'cached_tokens': 0,
            'text_tokens': None,
            'image_tokens': None
        }
    }
}

使用DSPy的缓存功能时（无论是通过litellm在内存中还是磁盘上），缓存的响应不会计入使用统计。例如：

# Enable caching
dspy.settings.configure(
    lm=dspy.LM("openai/gpt-4o-mini", cache=True),
    track_usage=True
)

program = MyProgram()

# First call - will show usage statistics
output = program(question="What is the capital of Zambia?")
print(output.get_lm_usage())  # Shows token usage

# Second call - same question, will use cache
output = program(question="What is the capital of Zambia?")
print(output.get_lm_usage())  # Shows empty dict: {}
优云智算