Software 2.0

DSPy+Milvus=RAG优化

APPLICATION Jan 17, 2025

DSPy 是由斯坦福大学 NLP 研究组引入的一个开创性的程序化框架，旨在优化语言模型中的提示和权重，特别是在将大型语言模型 (LLMs) 跨多个管道阶段集成的场景中。与依赖手动构建和调整的传统提示工程技术不同，DSPy 采用基于学习的方法。通过吸收查询-答案示例，DSPy 动态生成优化的提示，这些提示针对特定任务进行定制。这种创新方法使整个管道的重新组装成为可能，消除了对持续手动提示调整的需要。DSPy 的 Pythonic 语法提供了各种可组合和声明式的模块，简化了对 LLM 的指令。

使用 DSPy 的好处：

  • 编程方法：DSPy 为 LM 管道开发提供了一种系统化的编程方法，将管道抽象为文本转换图，而不是仅仅提示 LLM。其声明式模块使结构化设计和优化成为可能，取代了传统提示模板的试错方法。
  • 性能提升：DSPy 在现有方法上展示了显著的性能提升。通过案例研究，它在标准提示和专家创建的演示中表现更佳，展示了其即使编译到较小的 LM 模型中的 versatility 和 effectiveness。
  • 模块化抽象：DSPy 有效地抽象了 LM 管道开发中的复杂方面，如分解、微调和模型选择。借助 DSPy，一个简洁的程序可以无缝地转化为对各种模型（如 GPT-4、Llama2-13b 或 T5-base）的指令，简化开发并提高性能。

为什么在 DSPy 中使用 Milvus：

  • DSPy 是一个强大的编程框架，可以提升 RAG（检索增强生成）应用程序的性能。这类应用程序需要检索有用的信息来提高答案质量，这需要向量数据库。
  • Milvus 是一个知名的开源向量数据库，可提高性能和可扩展性。通过 MilvusRM（Milvus 检索器模块），在 DSPy 中集成 Milvus 变得非常简单。
  • 现在，开发人员可以轻松地定义和优化使用 DSPy 的 RAG 程序，并利用 Milvus 强大的向量搜索功能。这种合作使得 RAG 应用程序更高效、更可扩展，结合了 DSPy 的编程能力与 Milvus 的搜索功能。

现在，我们通过一个快速示例来演示如何在 DSPy 中使用 Milvus 优化 RAG 应用程序。

1、前提条件

在构建 RAG 应用程序之前，安装 DSPy 和 PyMilvus。

$ pip install "dspy-ai[milvus]"
$ pip install -U pymilvus

如果你使用的是 Google Colab，为了启用新安装的依赖项，可能需要重启运行时（点击顶部的“运行时”菜单，然后从下拉菜单中选择“重新启动会话”）。

2、加载数据集

在此示例中，我们使用 HotPotQA，一个包含复杂问题-答案对的数据集，作为训练数据集。我们可以通过 HotPotQA 类加载它们。

from dspy.datasets import HotPotQA

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

# 告诉 DSPy，“question”字段是输入。任何其他字段都是标签和/或元数据。
trainset = [x.with_inputs("question") for x in dataset.train]
devset = [x.with_inputs("question") for x in dataset.dev]

3、将数据摄入 Milvus 向量数据库

将上下文信息摄入 Milvus 集合以进行向量检索。该集合应具有嵌入字段和文本字段。在此示例中，我们使用 OpenAI 的 text-embedding-3-small 模型作为默认查询嵌入函数。

import requests
import os

os.environ["OPENAI_API_KEY"] = "<YOUR_OPENAI_API_KEY>"
MILVUS_URI = "example.db"
MILVUS_TOKEN = ""

from pymilvus import MilvusClient, DataType, Collection
from dspy.retrieve.milvus_rm import openai_embedding_function

client = MilvusClient(uri=MILVUS_URI, token=MILVUS_TOKEN)

if "dspy_example" not in client.list_collections():
    client.create_collection(
        collection_name="dspy_example",
        overwrite=True,
        dimension=1536,
        primary_field_name="id",
        vector_field_name="embedding",
        id_type="int",
        metric_type="IP",
        max_length=65535,
        enable_dynamic=True,
    )
text = requests.get(
    "https://raw.githubusercontent.com/wxywb/dspy_dataset_sample/master/sample_data.txt"
).text

for idx, passage in enumerate(text.split("\n")):
    if len(passage) == 0:
        continue
    client.insert(
        collection_name="dspy_example",
        data=[
            {
                "id": idx,
                "embedding": openai_embedding_function(passage)[0],
                "text": passage,
            }
        ],
    )

4、定义 MilvusRM

现在，你需要定义 MilvusRM。

from dspy.retrieve.milvus_rm import MilvusRM
import dspy

retriever_model = MilvusRM(
    collection_name="dspy_example",
    uri=MILVUS_URI,
    token=MILVUS_TOKEN,  # 如果 Milvus 连接不需要令牌，则忽略此字段
    embedding_function=openai_embedding_function,
)
turbo = dspy.OpenAI(model="gpt-3.5-turbo")
dspy.settings.configure(lm=turbo)

5、构建签名

现在我们已经加载了数据，开始定义管道子任务的签名。我们可以识别我们的简单输入问题和输出答案，但由于我们正在构建 RAG 管道，我们将从 Milvus 检索上下文信息。因此，我们定义我们的签名如下：

class GenerateAnswer(dspy.Signature):
    """用简短的事实答案回答问题。"""

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

6、构建管道

现在，定义 RAG 管道。

class RAG(dspy.Module):
    def __init__(self, rm):
        super().__init__()
        self.retrieve = rm

        # 此签名指示 COT 模块的任务。
        self.generate_answer = dspy.ChainOfThought(GenerateAnswer)

    def forward(self, question):
        # 使用 milvus_rm 检索问题的上下文。
        context = self.retrieve(question).passages
        # COT 模块接受“上下文、查询”并输出“答案”。
        prediction = self.generate_answer(context=context, question=question)
        return dspy.Prediction(
            context=[item.long_text for item in context], answer=prediction.answer
        )

7、执行管道并获取结果

现在，我们已经构建了这个 RAG 管道。让我们试一下并获取结果。

rag = RAG(retriever_model)
print(rag("who write At My Window").answer)

输出：

Townes Van Zandt

我们可以在数据集上评估定量结果。

from dspy.evaluate.evaluate import Evaluate
from dspy.datasets import HotPotQA

evaluate_on_hotpotqa = Evaluate(
    devset=devset, num_threads=1, display_progress=False, display_table=5
)

metric = dspy.evaluate.answer_exact_match
score = evaluate_on_hotpotqa(rag, metric=metric)
print("rag:", score)

8、优化管道

在定义了这个程序之后，下一步是编译。此过程更新每个模块内的参数以提高性能。编译过程依赖于三个关键因素：

  • 训练集：我们将使用训练数据集中的 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)

# 现在 compiled_rag 已经优化并准备好回答你的新问题！
# 现在，让我们评估编译后的 RAG 程序。
score = evaluate_on_hotpotqa(compiled_rag, metric=metric)
print("compile_rag:", score)

Ragas 的分数从之前的 50.0 增加到 52.0，表明答案质量有所提高。

9、结束语

DSPy 通过其可编程接口，在语言模型交互中取得了飞跃，使模型提示和权重的算法化和自动化优化成为可能。通过在 DSPy 中实现 RAG，使其适应不同的语言模型或数据集变得轻而易举，大大减少了繁琐的手动干预需求。

原文链接：Integrate Milvus with DSPy

汇智网翻译整理，转载请标明出处

Tags

admin

Recommended for you