DSPy和RAG: Retrieval-Augmented Generation 🧠

在当今的自然语言处理领域，检索增强生成（RAG）是一个引人注目的方法，它允许大语言模型（LLMs）利用来自大型知识库的丰富信息，通过查询知识存储来寻找相关段落和内容，从而生成经过深思熟虑的响应。RAG的魅力在于它能够动态利用实时知识，即使在某些主题上并未经过专门训练的情况下，也能提供有深度的回答。

但RAG的复杂性也随之而来，尤其是在构建精细的RAG管道时。为了解决这些复杂性，我们可以借助DSPy，它提供了一种无缝的方式来设置提示管道。

配置语言模型（LM）和检索模型（RM） ⚙️

首先，我们需要设置语言模型（LM）和检索模型（RM），而DSPy通过多种LM和RM API以及本地模型托管来支持这一过程。

在本教程中，我们将使用GPT-3.5（gpt-3.5-turbo）和ColBERTv2检索器（一个免费服务器，托管了一个包含2017年维基百科“摘要”搜索索引的数据库，该数据库包含每篇文章的第一段内容）。我们将在DSPy中配置LM和RM，从而使DSPy能够在需要生成或检索时内部调用相应的模块。

import dspy

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)

加载数据集 📚

本教程中，我们使用HotPotQA数据集，这是一个复杂的问题-答案对集合，通常以多跳的方式进行回答。我们可以通过HotPotQA类加载这个由DSPy提供的数据集：

from dspy.datasets import HotPotQA

# 加载数据集。
dataset = HotPotQA(train_seed=1, train_size=20, eval_seed=2023, dev_size=50, test_size=0)

# 告诉DSPy将“问题”字段视为输入。其他字段是标签和/或元数据。
trainset = [x.with_inputs('question') for x in dataset.train]
devset = [x.with_inputs('question') for x in dataset.dev]

len(trainset), len(devset)

输出:

(20, 50)

构建签名 ✍️

在加载完数据后，我们现在可以开始定义RAG管道的子任务的签名。

我们可以识别简单的输入question和输出answer，但由于我们正在构建RAG管道，我们希望利用来自ColBERT语料库的一些上下文信息。因此，让我们定义我们的签名：context, question --> answer。

class GenerateAnswer(dspy.Signature):
    """回答短小的事实性问题。"""

    context = dspy.InputField(desc="可能包含相关事实")
    question = dspy.InputField()
    answer = dspy.OutputField(desc="通常在1到5个单词之间")

我们为context和answer字段添加了小描述，以定义模型将接收和应该生成的内容的更强指引。

构建管道 🚀

我们将把RAG管道构建为一个DSPy模块，这将需要两个方法：

__init__方法将简单地声明它所需的子模块：dspy.Retrieve和dspy.ChainOfThought。后者被定义为实现我们的GenerateAnswer签名。
forward方法将描述使用我们拥有的模块回答问题的控制流：给定一个问题，我们将搜索前3个相关段落，然后将它们作为上下文提供给答案生成。

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)

优化管道 🔧

编译RAG程序

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

编译依赖于三件事：

训练集。 我们将仅使用上面提到的20个问题-答案示例。
验证指标。 我们将定义一个简单的validate_context_and_answer函数，检查预测的答案是否正确，并且检索到的上下文确实包含该答案。
特定的提示器。 DSPy编译器包括许多提示器，可以优化您的程序。

from dspy.teleprompt import BootstrapFewShot

# 验证逻辑：检查预测答案是否正确。
# 同时检查检索的上下文是否确实包含该答案。
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

# 设置一个基本的提示器，将编译我们的RAG程序。
teleprompter = BootstrapFewShot(metric=validate_context_and_answer)

# 编译！
compiled_rag = teleprompter.compile(RAG(), trainset=trainset)

:::info
提示器： 提示器是强大的优化器，可以将任何程序进行引导，学习如何自生成并选择有效的模块提示。因此，它的名字意味着“远程提示”。

不同的提示器在优化成本与质量等方面提供了各种权衡。在上述示例中，我们使用了一个简单的默认BootstrapFewShot。

如果你喜欢类比，可以将其视为你的训练数据、损失函数和标准DNN监督学习设置中的优化器。而SGD是一个基本的优化器，还有更复杂（且更昂贵！）的优化器，如Adam或RMSProp。
:::

执行管道 🎉

现在我们已经编译了RAG程序，让我们试试它。

# 向这个简单的RAG程序询问任何问题。
my_question = "David Gregory继承了哪个城堡？"

# 获取预测。这包含`pred.context`和`pred.answer`。
pred = compiled_rag(my_question)

# 打印上下文和答案。
print(f"问题: {my_question}")
print(f"预测答案: {pred.answer}")
print(f"检索到的上下文（截断）: {[c[:200] + '...' for c in pred.context]}")

非常好。我们来检查最后的提示给LM的内容。

turbo.inspect_history(n=1)

输出:

回答短小的事实性问题。

---

问题: "At My Window"由哪位美国歌手/词曲作者发行？
答案: John Townes Van Zandt

问题: "Everything Has Changed"是哪个唱片公司发行的专辑中的一首歌？
答案: Big Machine Records
...(截断)

即使我们没有写出任何详细的示例，我们也看到DSPy能够为3-shot检索增强生成引导这个3,000个token的提示，并使用Chain-of-Thought推理在一个极其简单的程序中。

这展示了组合和学习的力量。当然，这只是由特定提示器生成的，其在每个设置中可能完美无瑕，也可能并非如此。正如您将在DSPy中看到的那样，您有一个庞大且系统化的选项空间，可以针对程序的质量和成本进行优化和验证。

您还可以轻松检查学习到的对象。

for name, parameter in compiled_rag.named_predictors():
    print(name)
    print(parameter.demos[0])
    print()

评估管道 📊

我们现在可以在开发集上评估我们的compiled_rag程序。当然，这个小集并不意味着是一个可靠的基准，但它将在说明中很有启发性。

让我们评估预测答案的准确性（精确匹配）。

from dspy.evaluate.evaluate import Evaluate

# 设置`evaluate_on_hotpotqa`函数。我们将多次使用它。
evaluate_on_hotpotqa = Evaluate(devset=devset, num_threads=1, display_progress=False, display_table=5)

# 使用`answer_exact_match`指标评估`compiled_rag`程序。
metric = dspy.evaluate.answer_exact_match
evaluate_on_hotpotqa(compiled_rag, metric=metric)

输出:

平均指标: 22 / 50  (44.0): 100%|██████████| 50/50 [00:00<00:00, 116.45it/s]
平均指标: 22 / 50  (44.0%)

44.0

评估检索 🔍

评估检索的准确性也可能是有启发性的。虽然有多种方式可以做到这一点，但我们可以简单地检查检索的段落是否包含答案。

我们可以利用我们的开发集，其中包括应检索的金标题。

def gold_passages_retrieved(example, pred, trace=None):
    gold_titles = set(map(dspy.evaluate.normalize_text, example['gold_titles']))
    found_titles = set(map(dspy.evaluate.normalize_text, [c.split(' | ')[0] for c in pred.context]))

    return gold_titles.issubset(found_titles)

compiled_rag_retrieval_score = evaluate_on_hotpotqa(compiled_rag, metric=gold_passages_retrieved)

输出:

平均指标: 13 / 50  (26.0): 100%|██████████| 50/50 [00:00<00:00, 671.76it/s]平均指标: 13 / 50  (26.0%)

尽管这个简单的compiled_rag程序能够正确回答相当一部分问题（在这个小集上，超过40%），但检索的质量则低得多。

这可能表明LM在回答问题时往往依赖于它在训练期间记住的知识。为了应对这种较弱的检索，我们将探索一个涉及更高级搜索行为的第二个程序。

通过这个详细的示例，我们可以看到如何利用DSPy构建一个RAG管道，配置和优化我们的模型，并评估其性能。这不仅展示了DSPy的强大能力，也为使用RAG技术提供了清晰的方向。