DSPy 实战：自动化 Prompt 框架快速入门

定义模型并加载数据

在官方教程中使用 LLM 为 gpt-3.5-turbo，数据集为在线的 ColBERTv2 服务器，托管维基百科 2017 年'摘要'搜索索引。问答数据集使用了 HotPotQA 数据集中的一个小样本。

import openai
openai.api_key = "YOUR_API_KEY"
turbo = dspy.OpenAI(model='gpt-3.5-turbo')
colbertv2_wiki17_abstracts = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')

dspy.settings.configure(lm=turbo, rm=colbertv2_wiki17_abstracts)

from dspy.datasets import HotPotQA

# Load the dataset.
dataset = HotPotQA(train_seed=1, train_size=20, eval_seed=2023, dev_size=50, test_size=0)

# Tell DSPy that the 'question' field is the input. Any other fields are labels and/or metadata.
trainset_q = [x.with_inputs('question') for x in dataset.train]
devset_q = [x.with_inputs('question') for x in dataset.dev]

print(len(trainset_q), len(devset_q))

在加载原始数据后，我们对每个示例应用了 x.with_inputs('question')，以告知 DSPy 我们在每个示例中的输入字段仅为 question。其他字段则是标签或元数据，不会提供给系统。

自定义大模型

• 基于 OLLAMA 的本地模型 需要先另开启一个终端开始服务 ollama serve

  lm = dspy.OllamaLocal(model='mistral')
  

• 基于 AzureOpenAI 的 API 由于 AzureOpenAI 和官方文档有一些不同，建议按照以下形式设置 AzureOpenAI 的参数

  turbo = dspy.AzureOpenAI(api_base="your azure endpoint",
                api_key="",
                api_version="2024-05-01-preview",
                api_provider="azure",
                model_type="chat",
                deployment_id="your deployment name",)
  

• 基于 ChatGPT

  turbo = dspy.OpenAI(model='gpt-3.5-turbo')
  import openai
  openai.api_key = "YOUR_API_KEY"
  

自定义数据库

文档给出了多种自定义加载数据的方案，这里主要说明的是，实际需要三步：第一步加载自然文本数据库，第二步使用句子向量模型转为向量，第三步使用向量数据库进行加载。

需要特别注意：数据库最终加载进入的方式要求是 List。

import dspy
from dspy.datasets import DataLoader
from dspy.retrieve.faiss_rm import FaissRM
from dsp.modules.sentence_vectorizer import SentenceTransformersVectorizer
import pandas as pd

dl = DataLoader()
turbo = dspy.OpenAI(model='gpt-3.5-turbo')

# 从 CSV 文件加载数据集
dataset = dl.from_csv(
    f"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']

# 将训练和测试数据集生成数据库
document_chunks = df['Context Text'].drop_duplicates().tolist()

vectorizer = SentenceTransformersVectorizer(
    model_name_or_path="distiluse-base-multilingual-cased-v2",  # 模型名称或路径
)

# 使用 Faiss 进行检索
frm = FaissRM(document_chunks, vectorizer=vectorizer)
dspy.settings.configure(lm=turbo, rm=frm)

trainset_q = [x.with_inputs('question') for x in trainset]
devset_q = [x.with_inputs('question') for x in devset]

2.2 定义程序流程

在 DSPy 中，我们将以声明方式定义模块与在 pipeline 中调用它们以解决任务之间保持清晰的分离。这让你能够专注于 pipeline 的信息流。DSPy 会接管你的程序，并自动优化如何提示（或微调）语言模型以适应你的特定 pipeline，使其效果良好。

使用语言模型：签名与预测器

在 DSPy 程序中，每次调用语言模型（LM）都需要有一个签名。签名由三个简单元素组成：一个关于语言模型要解决的子任务的简要描述；描述我们将提供给语言模型的一个或多个输入字段；描述我们期望从语言模型获取的一个或多个输出字段。

让我们为基本的问答定义一个简单的签名：

class BasicQA(dspy.Signature):
    """Answer questions with short factoid answers."""

    question = dspy.InputField()
    answer = dspy.OutputField(desc="often between 1 and 5 words")

既然我们有了签名，现在让我们定义并使用一个预测器。预测器是一个知道如何使用语言模型来实现签名的模块。重要的是，预测器可以学习以适应任务的行为！

generate_answer = dspy.Predict(BasicQA)

# Call the predictor on a particular input.
pred = generate_answer(question=dev_example.question)

# Print the input and the prediction.
print(f"Question: {dev_example.question}")
print(f"Predicted Answer: {pred.answer}")
print(f"label Answer: {dev_example.answer}")

# 查看直接的推理历史记录
turbo.inspect_history(n=1)

使用 CoT 思维链

加入 CoT 思维链使得模型重新回答，并输出思考过程。

# Define the predictor. Notice we're just changing the class. The signature BasicQA is unchanged.
generate_answer_with_chain_of_thought = dspy.ChainOfThought(BasicQA)

# Call the predictor on the same input.
pred = generate_answer_with_chain_of_thought(question=dev_example.question)

# Print the input, the chain of thought, and the prediction.
print(f"Question: {dev_example.question}")
print(f"Thought: {pred.rationale}")
print(f"Predicted Answer: {pred.answer}")

使用检索模型

检索模型就是根据向量匹配度从之前定义好的数据库中检索到背景知识，用于后续和 prompt 一起送进大模型。使用检索器非常简单。模块 dspy.Retrieve(k) 将搜索与给定查询最匹配的前 k 个段落。

retrieve = dspy.Retrieve(k=3)

print(retrieve(dev_example.question))
topK_passages = retrieve(dev_example.question).passages

print(f"Top {retrieve.k} passages for question: {dev_example.question} \n", '-' * 30, '\n')

for idx, passage in enumerate(topK_passages):
    print(f'{idx+1}]', passage, '\n')

基础检索增强生成（RAG）

我们将构建一个检索增强的答案生成管道。对于一个问题，我们将在训练数据集中搜索前 3 个相关段落，然后将它们作为上下文传递给生成模块以生成答案。

首先定义这个签名：context, question --> answer

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")

下面我们需要定义一个用于 RAG 的执行程序，它需要两个方法：

• __init__ 方法将声明它需要的子模块：dspy.Retrieve 和 dspy.ChainOfThought。后者用于实现我们的 GenerateAnswer 签名。
• forward 方法将描述使用这些模块来回答问题的控制流程

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)

定义了这个程序之后，现在让我们编译它。编译程序将更新每个模块中存储的参数。在我们的设置中，这主要是通过收集和选择好的示例来用在 Prompt 中。

设置指标和优化器

编译依赖于三件事：

• 一个训练集。我们将使用上面提到的部分问答示例的 trainset
• 一个验证指标。我们将定义一个快速的 validate_context_and_answer 函数，检查预测的答案是否正确。
• 一个特定的提示优化器。DSPy 编译器包括许多提示优化器，可以优化你的程序。

不同的提示优化器在优化成本与质量等方面提供了不同的折中方案。在教程中，笔者将使用一个简单的默认提示优化器 BootstrapFewShot。

同时定义评估模型 validate_context_and_answer 用于检查是否问题准确并且检索到准确的背景。

from dspy.teleprompt import BootstrapFewShot

# Validation logic: check that the predicted answer is correct.
# Also check that the retrieved context does actually contain that answer.
def validate_context_and_answer(example, pred, trace=None):
    answer_EM = dspy.evaluate.answer_exact_match(example, pred)
    answer_PM = dspy.evaluate.answer_passage_match(example, pred)
    return answer_EM and answer_PM

# Set up a basic teleprompter, which will compile our RAG program.
teleprompter = BootstrapFewShot(metric=validate_context_and_answer)

# Compile!
compiled_rag = teleprompter.compile(RAG(), trainset=trainset_q)

现在可以通过一些提问来进行验证：

# Ask any question you like to this simple RAG program.
my_question = "What castle did David Gregory inherit?"

# Get the prediction. This contains `pred.context` and `pred.answer`.
pred = compiled_rag(my_question)

# Print the contexts and the answer.
print(f"Question: {my_question}")
print(f"Predicted Answer: {pred.answer}")
print(f"Retrieved Contexts (truncated): {[c[:200] + '...' for c in pred.context]}")

也可以通过 turbo.inspect_history(n=1) 来看到底模型中发生了什么内容。

评估

首先，让我们评估预测答案的准确性（精确匹配）。用于后续结果的进一步优化。

from dspy.evaluate.evaluate import Evaluate

# Set up the `evaluate_on_hotpotqa` function. We'll use this many times below.
evaluate_on_hotpotqa = Evaluate(devset=devset, num_threads=1, display_progress=True, display_table=5)

# Evaluate the `compiled_rag` program with the `answer_exact_match` metric.
metric = dspy.evaluate.answer_exact_match
evaluate_on_hotpotqa(compiled_rag, metric=metric)

研究检索的准确性也可能具有指导意义。通常，我们可以检查检索到的段落是否包含答案。

三、总结与展望

通过以上的介绍，基本可以掌握使用 DSPy 框架上实现自己数据和模型的推理。至于解决目前这种弱检索问题，使得结果更加精确，可以继续深入官方的教程学习后续更高级搜索行为的代码。

相信您已经掌握了如何在三十分钟内快速上手 DSPy 框架。无论是自动化 Prompt 生成还是优化，DSPy 都能为您的自然语言处理项目提供极大的帮助。希望您在今后的工作中能够充分利用 DSPy，提高开发效率。

四、最佳实践与注意事项

在实际生产环境中使用 DSPy 时，建议关注以下几点以提升稳定性和性能：

1. 数据质量优先：DSPy 的效果高度依赖于训练数据的质量。确保标注数据准确且覆盖多样化的场景，有助于优化器找到更鲁棒的解决方案。
2. 监控与日志：在生产部署前，务必启用详细的日志记录。利用 inspect_history 分析模型推理路径，识别常见的失败模式（如幻觉、检索不相关等）。
3. 成本控制：虽然 DSPy 旨在减少手动调优成本，但频繁的 API 调用仍会产生费用。建议在生产环境中缓存常用查询结果，并合理设置 num_passages 等参数以平衡检索精度与开销。
4. 模型选择：根据任务复杂度选择合适的基座模型。对于简单任务，小参数量的本地模型可能已足够；对于复杂推理，则建议使用 GPT-4 等大模型以获得最佳效果。
5. 持续迭代：LLM 技术更新迅速，定期重新运行优化器以适应新版本的模型或新增的数据分布，是保持系统长期有效性的关键。

通过遵循上述实践，开发者可以更高效地利用 DSPy 构建企业级的大语言模型应用。