使用Haystack框架构建QA-RAG代理

想象一下，您正在构建一个需要回答有关产品问题的客户支持 AI 代理。有时，它需要从您的文档中提取信息，有时则需要在网上搜索最新更新。在这类复杂的人工智能应用中，RAG 代理系统就派上了用场。将它们视为智能研究助手，它们不仅了解您的内部文档，还能决定何时去搜索网络。在本指南中，我们将介绍使用 Haystack 框架构建 QA RAG 代理系统的过程。

• 了解什么是 LLM 代理，并理解它与 RAG 系统有何不同。
• 熟悉用于 LLM 代理应用的 Haystack 框架。
• 了解从模板构建提示的过程，学习如何将不同的提示连接在一起。
• 学习如何在 Haystack 中使用 ChromaDB 创建嵌入。
• 学习如何建立从嵌入到生成的完整本地开发系统。

什么是LLM代理？

LLM 代理是一种人工智能系统，它可以根据对任务的理解自主做出决策并采取行动。与主要生成文本回复的传统 LLM 不同，LLM 代理可以做更多事情。它可以思考、计划和行动，只需极少的人工输入。它可以评估自己的知识，识别何时需要更多信息或外部工具。LLM 代理不依赖静态数据或索引知识，而是决定信任哪些来源，以及如何收集最佳见解。

这类系统还能为工作挑选合适的工具。它可以决定何时需要检索文档、运行计算或自动执行任务。它们的与众不同之处在于能够将复杂问题分解为多个步骤并独立执行，这使其在研究、分析和工作流程自动化方面具有重要价值。

RAG与RAG代理

传统的 RAG 系统遵循线性流程。当收到查询时，系统首先会识别请求中的关键要素。然后搜索知识库，扫描相关信息，以帮助设计准确的响应。一旦检索到相关信息或数据，系统就会对其进行处理，生成有意义且与上下文相关的回复。

您可以通过下图轻松了解这些流程。

Source: Author

现在，RAG 代理系统通过以下方式加强了这一过程：

• 评估查询要求
• 在多个知识源之间做出决定
• 可能将不同来源的信息结合起来
• 自主决定响应策略
• 提供源归属响应

关键区别在于系统能够就如何处理查询做出智能决策，而不是遵循固定的检索生成模式。

了解Haystack框架组件

Haystack 是一个开源框架，用于构建生产就绪的人工智能、LLM 应用程序、RAG 管道和搜索系统。它为构建 LLM 应用程序提供了一个强大而灵活的框架。它允许您集成来自 Huggingface、OpenAI、CoHere、Mistral 和 Local Ollama 等不同平台的模型。你还可以在 AWS SageMaker、BedRock、Azure 和 GCP 等云服务上部署模型。

Haystack 为高效数据管理提供了强大的文档存储。它还配备了一套全面的评估、监控和数据集成工具，可确保应用程序各层性能的平稳运行。它还拥有强大的社区协作能力，可以定期从不同的服务提供商那里集成新的服务。

Source: Haystack

使用Haystack可以构建什么？

• 使用强大的检索和生成技术，在您的数据上简单推进 RAG。
• 使用 GPT-4 、Llama3.2、Deepseek-R1 等最新 GenAI 模型的聊天机器人和代理。
• 在混合类型（图像、文本、音频和表格）知识库上生成多模态问题解答系统。
• 从文档中提取信息或构建知识图谱。

Haystack构件

Haystack 有两个构建全功能 GenAI LLM 系统的主要概念–组件和管道。让我们通过一个关于日本动漫人物的 RAG 的简单例子来了解它们。

组件

组件是 Haystack 的核心构件。它们可以执行文档存储、文档检索、文本生成和嵌入等任务。Haystack 有许多组件，你可以在安装后直接使用，它还提供 API，让你通过编写 Python 类来制作自己的组件。

它还收集了来自合作伙伴公司和社区的集成组件。

安装库并设置 Ollama

$ pip install haystack-ai ollama-haystack
# On you system download Ollama and install LLM
ollama pull llama3.2:3b
ollama pull nomic-embed-text
# And then start ollama server
ollama serve

导入一些组件

from haystack import Document, Pipeline
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack_integrations.components.generators.ollama import OllamaGenerator

创建文档和文档存储

document_store = InMemoryDocumentStore()
documents = [
Document(
content="Naruto Uzumaki is a ninja from the Hidden Leaf Village and aspires to become Hokage."
),
Document(
content="Luffy is the captain of the Straw Hat Pirates and dreams of finding the One Piece."
),
Document(
content="Goku, a Saiyan warrior, has defended Earth from numerous powerful enemies like Frieza and Cell."
),
Document(
content="Light Yagami finds a mysterious Death Note, which allows him to eliminate people by writing their names."
),
Document(
content="Levi Ackerman is humanity’s strongest soldier, fighting against the Titans to protect mankind."
),
]

管道

管道是 Haystack 框架的支柱。它们定义了不同组件之间的数据流。管道本质上是有向无环图（DAG）。一个具有多个输出的组件可以连接到另一个具有多个输入的组件。

您可以通过以下方式定义管道

pipe = Pipeline()
pipe.add_component("retriever", InMemoryBM25Retriever(document_store=document_store))
pipe.add_component("prompt_builder", PromptBuilder(template=template))
pipe.add_component(
"llm", OllamaGenerator(model="llama3.2:1b", url="http://localhost:11434")
)
pipe.connect("retriever", "prompt_builder.documents")
pipe.connect("prompt_builder", "llm")

您可以将管道可视化

image_param = {
"format": "img",
"type": "png",
"theme": "forest",
"bgColor": "f2f3f4",
}
pipe.show(params=image_param)

该管道提供

• 模块化工作流程管理
• 灵活的组件安排
• 易于调试和监控
• 可扩展的处理架构

节点

节点是管道中可连接的基本处理单元，这些节点是执行特定任务的组件。

上述流水线中的节点示例

pipe.add_component("retriever", InMemoryBM25Retriever(document_store=document_store))
pipe.add_component("prompt_builder", PromptBuilder(template=template))
pipe.add_component(
"llm", OllamaGenerator(model="llama3.2:1b", url="http://localhost:11434")
)

连接图

连接图定义了组件的交互方式。

从上面的管道中可以直观地看到连接图。

image_param = {
"format": "img",
"type": "png",
"theme": "forest",
"bgColor": "f2f3f4",
}
pipe.show(params=image_param)

动漫管道的连接图

这种图形结构

• 定义组件之间的数据流
• 管理输入/输出关系
• 尽可能实现并行处理
• 创建灵活的处理路径。

现在，我们可以使用提示来查询动漫知识库。

创建提示模板

template = """
Given only the following information, answer the question.
Ignore your own knowledge.
Context:
{% for document in documents %}
{{ document.content }}
{% endfor %}
Question: {{ query }}?
"""

该提示将根据文档库中的信息提供答案。

使用提示和检索器进行查询

query = "How Goku eliminate people?"
response = pipe.run({"prompt_builder": {"query": query}, "retriever": {"query": query}})
print(response["llm"]["replies"])

响应：

对于新手来说，这个 RAG 很简单，但在概念上却很有价值。现在我们已经了解了 Haystack 框架的大部分概念，可以深入研究我们的主要项目了。如果遇到任何新问题，我都会一一解释。

高中物理问答RAG项目

我们将为高中生创建一个基于 NCERT 物理书的问题解答 RAG。它将通过从 NCERT 书籍中获取信息来为问题提供答案，如果没有相关信息，它将搜索网络来获取信息：

• 为此，我将使用：本地 Llama3.2:3b 或 Llama3.2:1b
• 用于嵌入存储的 ChromaDB
• 用于本地嵌入的 Nomic 嵌入文本模型
• 用于网络搜索的 DuckDuckGo 搜索或 Tavily 搜索（可选）

我使用的是完全本地化的免费系统。

设置开发者环境

我们将设置一个 Python 3.12 的 conda 环境

$conda create --name agenticlm python=3.12
$conda activate agenticlm

安装必要的软件包

$pip install haystack-ai ollama-haystack pypdf
$pip install chroma-haystack duckduckgo-api-haystack

现在创建一个名为 qagent 的项目目录。

$md qagent # create dir
$cd qagent # change to dir
$ code .   # open folder in vscode

您可以为项目使用纯 Python 文件，也可以为项目使用 Jupyter Notebook，这并不重要。我将使用纯 Python 文件。

在项目根目录下创建一个main.py文件。

导入必要的库

• 系统包
• 核心 haystack 组件
• 用于嵌入组件的 ChromaDB
• 用于本地推断的 Ollama 组件
• 用于网络搜索的 Duckduckgo

# System packages
import os
from pathlib import Path

# Core haystack components
from haystack import Pipeline
from haystack.components.writers import DocumentWriter
from haystack.components.joiners import BranchJoiner
from haystack.document_stores.types import DuplicatePolicy
from haystack.components.converters import PyPDFToDocument
from haystack.components.routers import ConditionalRouter
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter

# ChromaDB integration
from haystack_integrations.document_stores.chroma import ChromaDocumentStore
from haystack_integrations.components.retrievers.chroma import (
ChromaEmbeddingRetriever,
)

# Ollama integration
from haystack_integrations.components.embedders.ollama.document_embedder import (
OllamaDocumentEmbedder,
)
from haystack_integrations.components.embedders.ollama.text_embedder import (
OllamaTextEmbedder,
)
from haystack_integrations.components.generators.ollama import OllamaGenerator

# Duckduckgo search integration
from duckduckgo_api_haystack import DuckduckgoApiWebSearch

创建文档存储

在这里，文档存储是最重要的，我们将存储嵌入内容以便检索，我们使用 ChromaDB 来存储嵌入内容，正如你在前面的示例中看到的，我们使用 InMemoryDocumentStore 来快速检索，因为那时我们的数据很小，但对于一个强大的检索系统来说，我们不能依赖 InMemoryStore，它会占用内存，而且每次启动系统时我们都会创建嵌入内容。

解决办法是使用矢量数据库，如 Pinecode、Weaviate、Postgres Vector DB 或 ChromaDB。我之所以使用 ChromaDB，是因为它免费、开源、易用且强大。

# Chroma DB integration component for document(embedding) store
document_store = ChromaDocumentStore(persist_path="qagent/embeddings")

persistent_path 是您要存储嵌入内容的位置。

PDF 文件路径

HERE = Path(__file__).resolve().parent
file_path = [HERE / "data" / Path(name) for name in os.listdir("QApipeline/data")]

它将从包含 PDF 文件的数据文件夹中创建一个文件列表。

文档预处理组件

我们将使用 Haystack 内置的文档预处理器，如清理器、分割器和文件转换器，然后使用写入器将数据写入存储空间。

Cleaner：它会清除文档中的多余空格、重复行、空行等。

cleaner = DocumentCleaner()

Splitter：它能以单词、句子、段落、页面等多种方式分割文档。

splitter = DocumentSplitter()

File Converter：使用 pypdf 将 pdf 转换为文档。

file_converter = PyPDFToDocument()

Writer：它会将文档存储到你想要存储文档的位置，对于重复的文档，它会覆盖之前的文档。

writer = DocumentWriter(document_store=document_store, policy=DuplicatePolicy.OVERWRITE)

现在设置文档索引的嵌入器。

Embedder：Nomic 嵌入文本

我们将使用在 Huggingface 和 Ollama 中非常有效且免费的 nomic-embed-text 嵌入器。

在运行索引管道之前，打开终端并键入以下内容，从 Ollama 模型库中提取 nomic-embed-text 和 llama3.2:3b 模型

$ ollama pull nomic-embed-text
$ ollama pull llama3.2:3b

在终端中输入 ollama serve 命令启动 Ollama

现在嵌入组件

embedder = OllamaDocumentEmbedder(
model="nomic-embed-text", url="http://localhost:11434"
)

我们使用 OllamaDocumentEmbedder 组件嵌入文档，但如果要嵌入文本字符串，则必须使用 OllamaTextEmbedder。

创建索引管道

与之前的玩具 RAG 示例一样，我们将首先启动 Pipeline 类。

indexing_pipeline = Pipeline()

现在，我们将把组件逐一添加到管道中

indexing_pipeline.add_component("embedder", embedder)
indexing_pipeline.add_component("converter", file_converter)
indexing_pipeline.add_component("cleaner", cleaner)
indexing_pipeline.add_component("splitter", splitter)
indexing_pipeline.add_component("writer", writer)

向管道中添加组件并不考虑顺序，因此您可以按任何顺序添加组件。

将组件连接到管道图

indexing_pipeline.connect("converter", "cleaner")
indexing_pipeline.connect("cleaner", "splitter")
indexing_pipeline.connect("splitter", "embedder")
indexing_pipeline.connect("embedder", "writer")

在这里，顺序很重要，因为如何连接组件会告诉管道数据如何在管道中流动。这就好比，你从哪里购买水管设备，顺序并不重要，但如何将它们组合在一起，将决定你是否能喝到水。

转换器转换 PDF 文件后，将其发送给清理器进行清理。然后，清理器将清理后的文档发送给分割器进行分块。然后，这些分块将传递给嵌入器进行矢量化，最后嵌入器将把这些嵌入交给编写器进行存储。

明白了吧！好了，让我给你一张索引的可视化图表，以便你能检查数据流。

绘制索引流水线

image_param = {
"format": "img",
"type": "png",
"theme": "forest",
"bgColor": "f2f3f4",
}
indexing_pipeline.draw("indexing_pipeline.png", params=image_param)  # type: ignore

是的，你可以通过 Haystack 管道轻松创建一个漂亮的美人鱼图。

索引管道图

我想你们现在已经完全理解了 Haystack 管道背后的理念。

实现路由器

现在，我们需要创建一个路由器，通过不同的路径路由数据。在这种情况下，我们将使用条件路由器，它将根据特定条件完成路由工作。条件路由器将根据组件输出评估条件。它将引导数据流通过不同的管道分支，从而实现动态决策。它还将具有强大的回退策略。

# Conditions for routing
routes = [
{
"condition": "{{'no_answer' in replies[0]}}",
"output": "{{query}}",
"output_name": "go_to_websearch",
"output_type": str,
},
{
"condition": "{{'no_answer' not in replies[0]}}",
"output": "{{replies[0]}}",
"output_name": "answer",
"output_type": str,
},
]
# router component
router = ConditionalRouter(routes=routes)

当系统从嵌入式存储上下文中获得 no_answer 回复时，它将转而使用网络搜索工具从互联网上收集相关数据。

对于网络搜索，我们将使用 Duckduckgo API 或 Tavily，这里我使用的是 Duckduckgo。

websearch = DuckduckgoApiWebSearch(top_k=5)

好了，大部分繁重的工作已经完成。现在，是时候进行提示工程了

创建提示模板

我们将使用 Haystack PromptBuilder 组件从模板中创建提示信息

首先，我们将创建一个 qa 提示

template_qa = """
Given ONLY the following information, answer the question.
If the answer is not contained within the documents reply with "no_answer.
If the answer is contained within the documents, start the answer with "FROM THE KNOWLEDGE BASE: ".
Context:
{% for document in documents %}
{{ document.content }}
{% endfor %}
Question: {{ query }}?
"""

它会从文件中获取上下文，并尝试回答问题。但如果在文档中找不到相关上下文，系统就会回复 no_answer。

现在，在从 LLM 得到 no_answer 后的第二次提示中，系统将使用网络搜索工具从互联网上收集上下文。

Duckduckgo 提示词模板

template_websearch = """
Answer the following query given the documents retrieved from the web.
Start the answer with "FROM THE WEB: ".
Documents:
{% for document in documents %}
{{ document.content }}
{% endfor %}
Query: {{query}}
"""

这将有助于系统进入网络搜索并尝试回答查询。

使用 Haystack 的 PromptBuilder 创建提示词

prompt_qa = PromptBuilder(template=template_qa)
prompt_builder_websearch = PromptBuilder(template=template_websearch)

我们将使用 Haystack 提示连接器将提示的各个分支连接到一起。

prompt_joiner = BranchJoiner(str)

实施查询管道

查询管道将嵌入查询，从嵌入中收集上下文资源，并使用 LLM 或网络搜索工具回答我们的查询。

它与索引管道类似。

启动管道

query_pipeline = Pipeline()

为查询管道添加组件

query_pipeline.add_component("text_embedder", OllamaTextEmbedder())
query_pipeline.add_component(
"retriever", ChromaEmbeddingRetriever(document_store=document_store)
)
query_pipeline.add_component("prompt_builder", prompt_qa)
query_pipeline.add_component("prompt_joiner", prompt_joiner)
query_pipeline.add_component(
"llm",
OllamaGenerator(model="llama3.2:3b", timeout=500, url="http://localhost:11434"),
)
query_pipeline.add_component("router", router)
query_pipeline.add_component("websearch", websearch)
query_pipeline.add_component("prompt_builder_websearch", prompt_builder_websearch)

在这里，我们使用 OllamaGenerator 组件来生成 LLM，该组件可使用 Llama3.2:3b 或 1b 或任何你喜欢的 LLM 工具来生成答案。

将所有组件连接在一起，用于查询流和答案生成

query_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
query_pipeline.connect("retriever", "prompt_builder.documents")
query_pipeline.connect("prompt_builder", "prompt_joiner")
query_pipeline.connect("prompt_joiner", "llm")
query_pipeline.connect("llm.replies", "router.replies")
query_pipeline.connect("router.go_to_websearch", "websearch.query")
query_pipeline.connect("router.go_to_websearch", "prompt_builder_websearch.query")
query_pipeline.connect("websearch.documents", "prompt_builder_websearch.documents")
query_pipeline.connect("prompt_builder_websearch", "prompt_joiner")

总结上述连接：

1. 从 text_embedder 发送的嵌入到 retriever 的查询嵌入。
2. Retriever 向 prompt_builder 的文档发送数据。
3. 提示生成器转到提示连接器，与其他提示连接。
4. 提示生成器将数据传递给 LLM 进行生成。
5. LLM 的回复会转到路由器，检查回复是否有no_answer 。如果没有 ，则转到网络搜索模块。
6. 网络搜索将数据作为查询发送到网络搜索提示。
7. 网络搜索文档向网络搜索文档发送数据。
8. 网络搜索提示将数据发送给提示连接器。
9. 提示连接器将数据发送到 LLM，以便生成答案。

为什么不亲眼看看呢？

绘制查询管道图

query_pipeline.draw("agentic_qa_pipeline.png", params=image_param)  # type: ignore

查询图形

我知道这张图很大，但它能让你清楚地看到野兽腹下发生了什么。

现在是享受我们辛勤工作成果的时候了。

创建一个便于查询的函数。

def get_answer(query: str):
response = query_pipeline.run(
{
"text_embedder": {"text": query},
"prompt_builder": {"query": query},
"router": {"query": query},
}
)
return response["router"]["answer"]

这是一个用于生成答案的简单函数。

现在运行主脚本，为 NCERT 物理书编制索引

indexing_pipeline.run({"converter": {"sources": file_path}})

这是一项一次性工作，索引完成后必须注释这一行，否则它将开始重新索引图书。

在文件底部，我们写入查询的驱动代码

if __name__ == "__main__":
query = "Give me 5 MCQ on resistivity?"
print(get_answer(query))

从书本知识中了解电阻率的 MCQ

书中没有提到的另一个问题

if __name__ == "__main__":
query = "What is Photosynthesis?"
print(get_answer(query))

输出

让我们再试一个问题。

if __name__ == "__main__":
query = (
"Tell me what is DRIFT OF ELECTRONS AND THE ORIGIN OF RESISTIVITY from the book"
)
print(get_answer(query))

因此，它是可行的！我们可以使用更多的数据、书籍或 PDF 文件进行嵌入，从而生成更多的上下文感知答案。此外，GPT-4o、Anthropic’s Claude 或其他云 LLM 也能更好地完成工作。

结论

我们的代理 RAG 系统展示了 Haystack 框架的灵活性和健壮性，以及其组合组件和管道的能力。通过部署到网络服务平台，以及使用 OpenAI 和 nthropic 等更好的付费 LLM，该 RAG 可随时投入生产。您可以使用 Streamlit 或基于 React 的 Web SPA 构建用户界面，以获得更好的用户体验。

你可以在这里找到文章中使用的所有代码。

• 与传统 RAG 相比，RAG 代理系统能提供更智能、更灵活的响应。
• Haystack 的管道架构可实现复杂的模块化工作流。
• 路由器可在生成响应时进行动态决策。
• 连接图提供灵活、可维护的组件交互。
• 整合多个知识源可提高响应质量。