使用Haystack和Hypster实现“模块化检索和生成（RAG）”

由 Midjourney AI 生成的图片，使用了作者提供的提示词

跟上人工智能的最新进展可能是一项挑战，尤其是在像检索增强生成（RAG）这样的快速变化领域。面对众多不同的解决方案和实现方法，很容易让人感到不知所措。

我为此挣扎了很长一段时间，试图理解每一篇新文章或技巧，以让RAG系统变得更好。每一篇新的论文、教程或博客文章都让我觉得新鲜，记住所有这些新方法的缩写变得越来越难跟上，这些缩写听起来就像是宝可梦里的角色名。

然后我找到了高某某等人（2024）的这篇论文“模块化RAG：将RAG系统变为积木式的可重构框架点击链接查看原文。"

论文中显示的构建RAG解决方案所用组件的主要图表。来源：模块化RAG

本文提供了一种结构化的途径，将RAG系统拆分为一个统一的框架，该框架可以涵盖各种解决方案和方法论。他们提出了六大主要组成部分：

• 索引: 整理您的数据以便高效搜索。
• 预检索: 处理用户的查询以准备搜索。
• 检索: 找到最相关的内容。
• 后检索: 优化检索到的信息。
• 生成: 使用大型语言模型生成响应。
• 编排: 控制系统流程。

本文的关键见解在于，现有的各种RAG解决方案可以用这些组件以类似乐高积木的方式进行描述。这种模块化提供了一个理解和设计RAG系统的框架，并且使构建过程更加灵活和清晰，具有更大的灵活性和清晰度的特点。

论文中，作者们展示了这样做的可能性，通过将现有RAG解决方案的例子用相同的构建模块来表达。比如：

自适应RAG流程中的-这里的决策者决定是否使用检索。来源：模块化RAG

FLARE - 前瞻性的主动检索 (Active RE 检索)，每个句子都能触发一次检索。来源：Modular RAG

我非常推荐阅读这篇论文以及该论文作者高云帆的一系列博客文章：模块化RAG和RAG流系列：（第一部分，第二部分）：第一部分，第二部分。

个人来说，这个框架对我很有帮助，让我理解了不同的RAG方法是如何相互关联的，现在我可以轻松了解新的论文和实现。

那么，我们该怎么真正实现这个“模块化RAG”框架呢？

由于它更像是一个元框架（meta-framework）——这在实际操作中意味着什么？那是否意味着我们需要实现所有可能的组件组合方式？还是我们只需构建各个独立的组件，让开发人员自己决定如何组合这些组件？

我认为，在大多数实际情况中，并不一定要尝试覆盖所有可能的RAG配置，而是要根据每个项目的需求和限制来缩小相关配置的范围。

在这次教程里，我将通过一个具体的例子来展示如何利用一小组选项来构建可配置系统。这将帮助你获得正确的视角和工具，让你能够创建你自己的模块化RAG版本，包含适合你特定需求的相关配置。

我们来探索一下我们将要使用的两个主要工具。

haystack 是一个开源框架，用于构建生产级的 LLM 应用、检索增强的生成管道和前沿的搜索系统，这些系统可以高效地处理大量文档。

好的地方：

• 优秀的组件设计
• 管道非常灵活，支持动态配置
• 文档极其（！）详尽
• 框架内置了众多现有实现，并与生成式 AI 提供商进行了集成。

不足：

• 管道接口可能有点啰嗦
• 在管道外部使用组件不太方便

我试用过几个不同的生成式AI框架，其中这个Haystack框架是我最容易理解、使用和定制的，最简单易懂。

**hypster** 是一个轻量级的、Python 风格的配置系统，适用于 AI 和机器学习项目的配置需求。它提供简洁且直观的 Python 风格语法，支持层次结构和可替换的配置。

Hypster 是一个我开发的新开源项目，旨在为 AI 和机器学习的工作流程提供一种新的编程范式——一种超越单一解决方案，迈向“工作流的叠加”或“超工作流”的范式。

Hypster 允许您定义一系列可能的配置选项，并轻松在它们之间切换，以进行实验和优化配置。这使得添加和自定义自己的配置空间变得容易，用不同的设置实例化这些配置空间，并最终选择适合您生产环境的最佳配置。

注: Hypster 正在积极开发中。暂时不建议在生产环境中使用。

这是一个高级的教程。它假设你已经熟悉RAG的主要部分。

我来分解代码库的主要部分并逐一解释，在解释过程中分享我的看法。

完整的并且更新了的代码位于以下仓库。别忘了给这个仓库点个⭐️。

我们从大型语言模型的配置空间定义开始吧。

# 从hypster导入config和HP模块
from hypster import config, HP

    @config  
    def llm_config(hp: HP):  
      anthropic_models = {"haiku": "claude-3-haiku-20240307",   
                          "sonnet": "claude-3-5-sonnet-20240620"}  
      openai_models = {"gpt-4o-mini": "gpt-4o-mini",   
                       "gpt-4o": "gpt-4o",   
                       "gpt-4o-latest": "gpt-4o-2024-08-06"}  

      model_options = {**anthropic_models, **openai_models}  
      model = hp.select(model_options, default="gpt-4o-mini")  
      temperature = hp.number_input(0.0)  

      if model in openai_models.values():  
        from haystack.components.generators import OpenAIGenerator  

        llm = OpenAIGenerator(model=model,   
                              generation_kwargs={"temperature": temperature})  
      else: #anthropic  
        from haystack_integrations.components.generators.anthropic import AnthropicGenerator  

        llm = AnthropicGenerator(model=model,   
                                 generation_kwargs={"temperature": temperature})

这段代码片段展示了 Hypster 和 Haystack 的基本示例。我们使用 @config 装饰器定义了一个名为 llm_config 的函数，该函数定义了我们 LLM 的配置空间。配置空间包括选择不同 LLM 提供商（如 Anthropic 或 OpenAI）及其相应模型的选项，以及一个用于调节温度的参数。

在 llm_config 函数中，我们使用条件逻辑根据选择的模型创建合适的Haystack组件。这让我们能够在不同的大模型之间轻松切换，无需改动代码结构。

比如说，要创建一个使用“‘Haiku’模型”的Anthropic生成器模型，我们可以将温度设置为0.5，按照以下方式创建配置：

    result = llm_config(final_vars="最终参数"["llm"],   
                        selections="选择"{"模型" : "俳句"},   
                        overrides="覆盖设置"{"温度" : 0.5})

(Note: The direct translation of function names and parameters like "final_vars", "selections", and "overrides" into Chinese might not be appropriate in technical contexts, so they are left untranslated in this case.)

Corrected version ensuring technical term consistency and appropriate translation:

    result = llm_config(final_vars=["llm"],   
                        selections={"模型" : "俳句"},   
                        overrides={"温度" : 0.5})

让我们继续构建索引管道，并定义如何处理我们的输入文件，也就是PDF文件。

@config  
// 索引配置函数，接受配置参数作为输入  
def 索引配置(hp: 配置参数):  
    from haystack import 管道  
    from haystack.components.converters import PDF转文档组件  
    管道 = 管道()  
    管道.add_component("loader", PDF转文档组件())

来丰富文档，我们将添加一个可选功能，即根据文档的前1000个字符通过LLM生成摘要。接下来，这是基于文档的前1000个字符进行的。

这里有一个很酷的技巧，我们使用文档的前n个字符（包括标点和空格），然后当我们把文档分成几部分时，每个部分都会带上这些信息，用于生成嵌入和响应。

      使用LLM丰富文档 = hp.select([True, False], default=True)  
      if 使用LLM丰富文档:  
        from textwrap import dedent  
        from haystack.components.builders import PromptBuilder  
        from src.haystack_utils import AddLLMMetadata  

        template = dedent("""  
            请用不超过15个词的一句话总结文档的主要主题。  
            然后列出3-5个关键词或缩写，用于搜索。  
            上下文：  
            {{ documents[0].content[:1000] }}  

            ============================  

            输出格式如下：  
            摘要：  
            关键词：  
        """)  

        llm = hp.propagate("configs/llm.py")  
        pipeline.add_component("prompt_builder", PromptBuilder(template=template))  
        pipeline.add_component("llm", llm["llm"])  
        pipeline.add_component("document_enricher", AddLLMMetadata())  

        pipeline.connect("loader", "prompt_builder")  
        pipeline.connect("prompt_builder", "llm")  
        pipeline.connect("llm", "document_enricher")  
        pipeline.connect("loader", "document_enricher")  
        分割器来源 = "document_enricher"  
      else:  
        分割器来源 = "loader"  

      分割方式 = hp.select(["sentence", "word", "passage", "page"],   
                           default="sentence")  
      分割器 = DocumentSplitter(split_by=分割方式,   
                                分割长度=hp.int_input(10),   
                                重叠部分=hp.int_input(2))  
      pipeline.add_component("splitter", 分割器)  
      pipeline.connect(分割器来源, "splitter")

在这里我们可以看到Haystack的管道在起作用。如果用户选择enrich_doc_w_llm==True（即启用文档增强功能），我们将继续添加实现这一增强所需的组件和连接。我们这里使用的是PromptBuilder → LLM → AddLLMMetadata这三种组件。

看看这个——它非常灵活，可以根据条件逻辑即时构建。这非常厉害。

现在有几种方法可以创建或初始化配置对象，例如：

    results = indexing_config(selections={"enrich_doc_w_llm": False,   
                                          "split_by" : "page"},   
                              overrides={"split_length" : 1})

这里我们有一个简单的管道流程，包括加载器和拆分器，以及所选的拆分器设置。

否则，我们可以选择用大语言模型的摘要内容来增强文档内容。

    results = indexing_config(selections={"enrich_doc_w_llm": True}) # 结果 = 配置索引(selections={"enrich_doc_w_llm": True}) # 指使用大模型丰富文档: enrich_doc_w_llm = True

需要注意的是，Hypster 使用每个参数默认设定的值，因此每次使用时不需要指定所有参数选项。这里展示生成管道的一个例子。

（如果没有具体示例图，则无需翻译此部分）

注意我们是如何通过hp.propagate("configs/llm_config.py")轻松地将llm_config插入到索引管道中的。这种传播能力使我们能够层次化地创建嵌套配置。我们可以通过点号选择和覆盖嵌套的llm_config中的参数。例如：

    results = indexing_config(selections={"llm.model": "gpt-4o-latest"})

这里，indexing_config 函数用于配置索引，参数 selections 包含一个键值对，表示选择的模型为最新的 "gpt-4o-latest"。

这将导致实例化一个带有LLM增强任务的索引管道，使用OpenAI的gpt-4o-2024—08模型。

截至目前，我们已经为许多潜在的索引流程构建了一个精简的架构。

为了简洁起见，我将略过嵌入配置的细节，在那里我整合了fastembed和jina嵌入。如果您感兴趣，请查看完整实现。

让我们继续来看检索流程。

Haystack自带内存文档存储，方便快速实验。它内置了嵌入检索器和BM25检索器。接下来我们将构建一个配置空间，可以使用BM25检索器、嵌入检索器或两者结合。

    @config  
    def in_memory_retrieval(hp: HP):  
      from haystack import Pipeline  
      from haystack.document_stores.in_memory import InMemoryDocumentStore  
      from src.haystack_utils import PassThroughDocuments, PassThroughText  

      pipeline = Pipeline()  
      # utility components for the first and last parts of the pipeline    
      pipeline.add_component("query", PassThroughText())  
      pipeline.add_component("retrieved_documents", PassThroughDocuments())  

      retrieval_types = hp.multi_select(["bm25", "embeddings"],   
                                        default=["bm25", "embeddings"])  
      if len(retrieval_types) == 0:  
          raise ValueError("至少需要选择一种检索方式。")  

      document_store = InMemoryDocumentStore()  

      if "embedding" in retrieval_types:  
        from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever  
        embedding_similarity_function = hp.select(["cosine", "dot_product"], default="cosine")  
        document_store.embedding_similarity_function = embedding_similarity_function  
        pipeline.add_component("embedding_retriever", InMemoryEmbeddingRetriever(document_store=document_store))  

      if "bm25" in retrieval_types:  
        from haystack.components.retrievers.in_memory import InMemoryBM25Retriever  
        bm25_algorithm = hp.select(["BM25Okapi", "BM25L", "BM25Plus"], default="BM25L")  
        document_store.bm25_algorithm = bm25_algorithm  
        pipeline.add_component("bm25_retriever", InMemoryBM25Retriever(document_store=document_store))  
        pipeline.connect("query", "bm25_retriever")  

      if len(retrieval_types) == 2:  # 同时包含 bm25 和 嵌入式检索。  
        from haystack.components.joiners.document_joiner import DocumentJoiner  

        bm25_weight = hp.number_input(0.5)  
        join_mode = hp.select(["distribution_based_rank_fusion",   
                              "concatenate", "merge",   
                              "reciprocal_rank_fusion"],  
                              default="distribution_based_rank_fusion")  
        joiner = DocumentJoiner(join_mode=join_mode, top_k=hp.int_input(10),  
                                weights=[bm25_weight, 1-bm25_weight])  

        pipeline.add_component("document_joiner", joiner)  
        pipeline.connect("bm25_retriever", "document_joiner")  
        pipeline.connect("embedding_retriever", "document_joiner")  
        pipeline.connect("document_joiner", "retrieved_documents")  
      elif "embeddings" in retrieval_types: # 只包含嵌入式检索  
        pipeline.connect("embedding_retriever", "retrieved_documents")  
      else:  # 只包含 BM25 检索  
        pipeline.connect("bm25_retriever", "retrieved_documents")

这里，我们使用了几种“技巧”来使它生效。首先，我们使用hp.multi_select来选择多个选项。其次，我们在管道的开始和结束添加了“辅助”组件（PassThroughText，PassThroughDocuments）以确保每个选择都将从query开始，以retrieved_documents结束，其余部分则相对简单。

几个例子是：

    in_memory_retrieval(selections={"retrieval_types": ["bm25"],   # 检索类型，此处为bm25算法
                                    "bm25_algorithm": "BM25Okapi"}) # 特定的BM25算法实现

作者供图

和：

内存检索(selections={"连接模式": "互惠排名融合"})

在完全实现的过程中，我添加了一个Qdrant向量存储、一个可选的重排序步骤（reranking步骤）以及一个最终的生成管道（pipeline）。这些都是为了展示在这些管道中添加和自定义不同组件的可能性的示例。你也可以在完整的代码库中找到这些示例。

最后，我们有个主配置把所有设置都绑在一起了。

    @config  
    def rag_config(hp: HP):  
      indexing = hp.propagate("configs/indexing.py")  
      indexing_pipeline = indexing["pipeline"]  

      embedder_type = hp.select(["fastembed", "jina"], default="fastembed")  
      match embedder_type:  
        case "fastembed":  
          embedder = hp.propagate("configs/fast_embed.py")  
        case "jina":  
          embedder = hp.propagate("configs/jina_embed.py")  

      indexing_pipeline.add_component("doc_embedder", embedder["doc_embedder"])  
      document_store_type = hp.select(["in_memory", "qdrant"], default="in_memory")  
      match document_store_type:  
        case "in_memory":  
          retrieval = hp.propagate("configs/in_memory_retrieval.py")  
        case "qdrant":  
          retrieval = hp.propagate("configs/qdrant_retrieval.py", overrides={"embedding_dim": embedder["embedding_dim"]})  

      from haystack.components.writers import DocumentWriter  
      from haystack.document_stores.types import DuplicatePolicy  

      document_writer = DocumentWriter(retrieval["document_store"], policy=DuplicatePolicy.OVERWRITE)  
      indexing_pipeline.add_component("document_writer", document_writer)  
      indexing_pipeline.connect("splitter", "doc_embedder")  
      indexing_pipeline.connect("doc_embedder", "document_writer")  

      # 检索与生成管道
      pipeline = retrieval["pipeline"]  
      pipeline.add_component("text_embedder", embedder["text_embedder"])  
      pipeline.connect("query", "text_embedder")  
      pipeline.connect("text_embedder", "embedding_retriever.query_embedding")  

      from src.haystack_utils import PassThroughDocuments  
      pipeline.add_component("docs_for_generation", PassThroughDocuments())  

      use_reranker = hp.select([True, False], default=True)  
      if use_reranker:  
          reranker = hp.propagate("configs/reranker.py")  
          pipeline.add_component("reranker", reranker["reranker"])  
          pipeline.connect("retrieved_documents", "reranker")  
          pipeline.connect("reranker", "docs_for_generation")  
          pipeline.connect("query", "reranker")  
      else:  
          pipeline.connect("retrieved_documents", "docs_for_generation")  

      response = hp.propagate("configs/response.py")  
      from haystack.components.builders import PromptBuilder  
      pipeline.add_component("prompt_builder", PromptBuilder(template=response["template"]))  
      pipeline.add_component("llm", response["llm"])  
      pipeline.connect("prompt_builder", "llm")  
      pipeline.connect("query.text", "prompt_builder.query")  
      pipeline.connect("docs_for_generation", "prompt_builder")

从这里，我们几乎可以在任何子组件中定义任何我们想要的东西。例如：

results = rag_config(selections={"索引.enrich_doc_w_llm": True,  
                                 "索引.llm模型": "gpt-4o-mini",  
                                 "文档存储": "qdrant",  
                                 "嵌入类型": "fastembed",  
                                 "重排序器.model": "tiny-bert-v2",  
                                 "响应.llm模型": "sonnet"},  
    覆盖设置={"索引.splitter拆分长度": 6,  
              "重排序器.top_k": 3})

我们已经建立了一套具体的工作流程，

我们现在可以一个个地来执行它们了：

    indexing_pipeline = results["indexing_pipeline"]  
    indexing_pipeline.warm_up()  

    file_paths = ["data/raw/modular_rag.pdf", "data/raw/enhancing_rag.pdf"]  
    for file_path in file_paths:  # 在这些文件路径中，对于每一个文件路径，这可以被并行化处理  
        indexing_pipeline.run({"loader": {"sources": [file_path]}})  

    query = "模块化RAG框架的六个主要模块是什么？"  

    pipeline = results["pipeline"]  
    pipeline.warm_up()  
    response = pipeline.run({"query": {"text": query}})  

    print("响应：", response["llm"]["replies"][0])

    回答：该模块化RAG框架的六个主要部分是索引、预取、检索、后处理、生成和编排。

    以下引用来自文档1：“根据RAG目前的发展阶段，我们建立了六个主要部分：索引、预取、检索、后处理、生成和编排。”

真是太好了！👏

概要

对于你们中的一些人来说，一下子接受这么多信息可能有点难以消化。你们可能刚刚开始接触Haystack，而这可能也是你们第一次见到Hypster。这完全可以理解。

代码很复杂，但我觉得这来源于构建这样一个模块化系统本身所固有的复杂性。此外，定义工作流的具体流程是一个视觉任务，有时候通过文本阅读起来更费劲。

说到底，这是我第一次看到一个完全可配置的模块化RAG系统。这对我来说真的挺激动人心的，也希望你跟我一样激动！

我认为这代表了AI/ML项目的根本不同方法。我们不是为了单一解决方案而构建代码库，而是构建了一个可以容纳多种工作流程的代码库——一种“多重工作流程的叠加”或“超工作流”。

一旦你开始做这种编程，你就会马上发现许多令人难以置信的好处。

1. 超参数优化 很容易实现（更多内容将在未来的帖子中介绍）
2. 根据不同场景使用不同的配置。例如，类型为X的查询可以使用一个将BM25检索器的权重设置得较高的RAG系统，而类型为Y的查询则主要集中在密集嵌入技术上。
3. 代理工具的使用 - 可以相对轻松地将其封装成一个工具，可以在不同的场景中实例化并使用，这意味着… 是的！我们可以将这变成一个AI代理使用的工具。想想这能带来哪些可能性。
4. 生产环境中的A/B测试 - 我们可以将这个RAG超参数空间部署到生产环境中，并通过为每个单独的API请求指定配置来进行A/B测试。

那，你觉得怎么样？

让我能够分享这些知识非常重要，因此你的反馈很有价值。如果你对这种实现或整体方法上的任何问题或评论，请随时在本文中写下你的意见。

我也为寻求利用最先进的生成式人工智能和机器学习工具来解决业务难题的公司提供顾问服务和自由职业，采用结构化、务实的方法。

Feel free to contact me via 电子邮件地址, LinkedIn, 或我的网站

• 高, Y., 熊, Y., 王, M., & 王, H. (2024). 模块化RAG：将RAG系统变成乐高式的可重构框架. arXiv预印论文 arXiv:2407.21059.

• 海斯特的文档页 | DeepLearning.ai 课程 | GitHub 仓库
• Hypster的简介 | 文档页 | GitHub 仓库
• 模块化-RAG的GitHub 仓库

• 未加注释的图片均由作者创作。
• 我与Deepset/Haystack没有任何关系。