教程：预处理不同类型的文件

_{最后更新：2025 年 7 月 22 日}

级别：入门
完成时间：15 分钟
目标：完成本教程后，您将学会如何构建一个索引 Pipeline，该 Pipeline 将使用 FileTypeRouter 根据文件的类型对其进行预处理。

💡 (可选)：在本教程中创建索引 Pipeline 后，有一个可选部分将展示如何在你刚刚创建的 Document Store 上创建 RAG Pipeline。此部分需要您提供一个 Hugging Face API 密钥。

使用的组件

FileTypeRouter：此组件将帮助您根据文件的 MIME 类型将其路由到不同的组件。
MarkdownToDocument：此组件将帮助您将 Markdown 文件转换为 Haystack Documents。
PyPDFToDocument：此组件将帮助您将 PDF 文件转换为 Haystack Documents。
TextFileToDocument：此组件将帮助您将文本文件转换为 Haystack Documents。
DocumentJoiner：此组件将帮助您连接来自 Pipeline 不同分支的 Documents。
DocumentCleaner (可选)：此组件将帮助您通过删除多余的空格等使 Documents 更易读。
DocumentSplitter：此组件将帮助您将 Document 分割成块。
SentenceTransformersDocumentEmbedder：此组件将帮助您为 Documents 创建 embeddings。
DocumentWriter：此组件将帮助您将 Documents 写入 DocumentStore。

概述

在本教程中，您将构建一个索引 Pipeline，该 Pipeline 预处理不同类型的文件（markdown、txt 和 pdf）。每个文件将拥有自己的 FileConverter。其余的索引 Pipeline 则相对标准——将文档分割成块、修剪空格、创建 embeddings 并将它们写入 Document Store。

可选地，您可以继续查看如何将这些 Documents 用于查询 Pipeline。

准备 Colab 环境

安装依赖项

%%bash
pip install haystack-ai
pip install "sentence-transformers>=4.1.0" "huggingface_hub>=0.23.0"
pip install markdown-it-py mdit_plain pypdf
pip install gdown

下载所有文件

本教程中使用的文件存储在 GDrive 文件夹中。您可以直接从 GDrive 文件夹下载文件，或者运行下面的代码。如果您在 Colab 上运行此教程，则会在左侧“文件”选项卡的“/recipe_files”文件夹下找到下载的文件。

与大多数现实生活中的数据一样，这些文件也是不同类型的混合体。

import gdown

url = "https://drive.google.com/drive/folders/1n9yqq5Gl_HWfND5bTlrCwAOycMDt5EMj"
output_dir = "recipe_files"

gdown.download_folder(url, quiet=True, output=output_dir)

创建用于索引文档的 Pipeline

接下来，您将创建一个用于索引文档的 Pipeline。为保持简单，您将使用 InMemoryDocumentStore，但此方法也适用于任何其他 DocumentStore。

对于我们数据源中的每种文件类型（在此案例中为 .pdf、.txt 和 .md），您都需要一个不同的文件转换器类。我们的 FileTypeRouter 将每种文件类型连接到正确的转换器。

一旦所有文件都转换为 Haystack Documents，我们就可以使用 DocumentJoiner 组件将它们合并成一个文档列表，然后一起输入到索引 Pipeline 的其余部分。

from haystack.components.writers import DocumentWriter
from haystack.components.converters import MarkdownToDocument, PyPDFToDocument, TextFileToDocument
from haystack.components.preprocessors import DocumentSplitter, DocumentCleaner
from haystack.components.routers import FileTypeRouter
from haystack.components.joiners import DocumentJoiner
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack import Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()
file_type_router = FileTypeRouter(mime_types=["text/plain", "application/pdf", "text/markdown"])
text_file_converter = TextFileToDocument()
markdown_converter = MarkdownToDocument()
pdf_converter = PyPDFToDocument()
document_joiner = DocumentJoiner()

从那里开始，此索引 Pipeline 的步骤会更标准化。DocumentCleaner 用于删除空格。然后 DocumentSplitter 将它们分割成 150 个单词的块，并留有少量重叠以避免丢失上下文。

document_cleaner = DocumentCleaner()
document_splitter = DocumentSplitter(split_by="word", split_length=150, split_overlap=50)

现在您将添加一个 SentenceTransformersDocumentEmbedder 来为文档创建 embeddings。作为此 Pipeline 的最后一步，DocumentWriter 将它们写入 InMemoryDocumentStore。

document_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
document_writer = DocumentWriter(document_store)

创建完所有组件后，将它们添加到索引 Pipeline 中。

preprocessing_pipeline = Pipeline()
preprocessing_pipeline.add_component(instance=file_type_router, name="file_type_router")
preprocessing_pipeline.add_component(instance=text_file_converter, name="text_file_converter")
preprocessing_pipeline.add_component(instance=markdown_converter, name="markdown_converter")
preprocessing_pipeline.add_component(instance=pdf_converter, name="pypdf_converter")
preprocessing_pipeline.add_component(instance=document_joiner, name="document_joiner")
preprocessing_pipeline.add_component(instance=document_cleaner, name="document_cleaner")
preprocessing_pipeline.add_component(instance=document_splitter, name="document_splitter")
preprocessing_pipeline.add_component(instance=document_embedder, name="document_embedder")
preprocessing_pipeline.add_component(instance=document_writer, name="document_writer")

接下来，连接它们 👇

preprocessing_pipeline.connect("file_type_router.text/plain", "text_file_converter.sources")
preprocessing_pipeline.connect("file_type_router.application/pdf", "pypdf_converter.sources")
preprocessing_pipeline.connect("file_type_router.text/markdown", "markdown_converter.sources")
preprocessing_pipeline.connect("text_file_converter", "document_joiner")
preprocessing_pipeline.connect("pypdf_converter", "document_joiner")
preprocessing_pipeline.connect("markdown_converter", "document_joiner")
preprocessing_pipeline.connect("document_joiner", "document_cleaner")
preprocessing_pipeline.connect("document_cleaner", "document_splitter")
preprocessing_pipeline.connect("document_splitter", "document_embedder")
preprocessing_pipeline.connect("document_embedder", "document_writer")

让我们用我写的一些食谱来测试这个 Pipeline。你是不是已经饿了？

from pathlib import Path

preprocessing_pipeline.run({"file_type_router": {"sources": list(Path(output_dir).glob("**/*"))}})

🎉 如果你只想学习如何预处理文档，可以到此为止了！如果你想看一个在 RAG Pipeline 中使用这些文档的示例，请继续阅读。

(可选) 构建用于查询文档的 Pipeline

现在，让我们构建一个 RAG Pipeline，该 Pipeline 根据您在上面部分创建的文档来回答问题。在此步骤中，我们将使用 HuggingFaceAPIChatGenerator，因此必须为此部分提供一个 Hugging Face API 密钥。我们将使用 Qwen/Qwen2.5-7B-Instruct 模型。

import os
from getpass import getpass

if "HF_API_TOKEN" not in os.environ:
    os.environ["HF_API_TOKEN"] = getpass("Enter Hugging Face token:")

在此步骤中，您将构建一个查询 Pipeline 来回答有关文档的问题。

此 Pipeline 接收提示，在 Document Store 中搜索相关文档，然后将这些文档传递给 LLM 以生成答案。

⚠️ 请注意，我们之前使用了 sentence-transformers/all-MiniLM-L6-v2 为我们的文档创建 embeddings。这就是为什么我们也会使用相同的模型来嵌入传入的问题。

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import HuggingFaceAPIChatGenerator

template = [
    ChatMessage.from_user(
        """
Answer the questions based on the given context.

Context:
{% for document in documents %}
    {{ document.content }}
{% endfor %}

Question: {{ question }}
Answer:
"""
    )
]
pipe = Pipeline()
pipe.add_component("embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
pipe.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store))
pipe.add_component("chat_prompt_builder", ChatPromptBuilder(template=template))
pipe.add_component(
    "llm",
    HuggingFaceAPIChatGenerator(
        api_type="serverless_inference_api",
        api_params={"model": "Qwen/Qwen2.5-7B-Instruct", "provider": "together"}
    ),
)

pipe.connect("embedder.embedding", "retriever.query_embedding")
pipe.connect("retriever", "chat_prompt_builder.documents")
pipe.connect("chat_prompt_builder.prompt", "llm.messages")

尝试运行下面的代码自己动手试试。如果一切顺利，您应该会从所有食谱来源获得一份完整的购物清单。🧂🥥🧄

question = (
    "What ingredients would I need to make vegan keto eggplant lasagna, vegan persimmon flan, and vegan hemp cheese?"
)

pipe.run({"embedder": {"text": question}, "chat_prompt_builder": {"question": question}})

{’llm’: {‘replies’: [ChatMessage(_role=<ChatRole.ASSISTANT: ‘assistant’>, _content=[TextContent(text=‘要制作素食酮生酮茄子千层面、素食柿子布丁和素食大麻奶酪，您需要以下配料：\n\n### 素食酮生酮茄子千层面\n- 2 个大茄子\n- 盐（超多盐）\n- 1/2 杯商店购买的素食马苏里拉奶酪用于顶层\n- 菠菜豆腐里科塔（14 盎司硬豆腐或特硬豆腐、10 盎司菠菜、1 个柠檬汁、适量蒜粉、适量盐）\n- 澳洲坚果奶酪（1 杯澳洲坚果、1/4 杯营养酵母、1/4 杯橄榄油、1 份素食香蒜酱、1 份菠菜豆腐里科塔、1 茶匙蒜粉、半个柠檬汁、适量盐）\n\n### 素食柿子布丁\n- 2 个中等大小的富裕柿子，沥干水分\n- 1 汤匙玉米淀粉\n- 1/2 茶匙琼脂粉\n- 1 汤匙龙舌兰花蜜，或按口味调整\n- 2 汤匙白砂糖\n- 1/4 杯椰奶\n- 1/2 杯杏仁奶\n- 1/2 茶匙香草精\n\n### 素食大麻奶酪\n- 1/2 杯葵花籽\n- 1/2 杯火麻仁\n- 1.5 茶匙味噌酱\n- 1 茶匙营养酵母\n- 1/4 杯活化水\n- 1/4 茶匙盐，或按口味调整\n\n### 其他工具和说明\n- 砂锅（9 x 13）\n- 2 个布丁模\n- 搅拌机或食品加工机\n- 平底锅\n- 手持搅拌器（可选）\n- 干净的玻璃碗\n- 橡皮筋\n- 抹布\n- 刀\n- 热水浴法（用于布丁）\n\n这些配料和工具将允许您按照提供的上下文准备所有三种菜肴。’)], _name=None, _meta={‘model’: ‘Qwen/Qwen2.5-7B-Instruct’, ‘finish_reason’: ‘stop’, ‘index’: 0, ‘usage’: {‘prompt_tokens’: 2031, ‘completion_tokens’: 446}})]}}

下一步

恭喜您构建了一个可以预处理不同文件类型的索引 Pipeline。现在您可以开始将所有混乱的真实世界数据注入您的工作流程中。💥

如果您喜欢这个教程，您可能还会喜欢

要及时了解最新的 Haystack 开发动态，您可以订阅我们的新闻通讯。感谢您的阅读！

使用元数据过滤文档

构建工具调用代理