教程：使用 TransformersTextRouter 和 TransformersZeroShotTextRouter 进行查询分类

_{最后更新：2025 年 8 月 25 日}

级别：中级
完成时间：15 分钟
使用的组件: TransformersZeroShotTextRouter, TransformersTextRouter, InMemoryDocumentStore, SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder, InMemoryEmbeddingRetriever, InMemoryBM25Retriever
目标: 完成本教程后，您将了解 TransformersZeroShotTextRouter 和 TransformersTextRouter，以及如何在管道中使用它们。

概述

使用 Haystack 中提供的最先进的 NLP 模型的一个巨大好处是，它允许用户用纯自然语言问题来表达他们的查询：用户无需绞尽脑汁想出正确的关键词来找到他们问题的答案，只需像询问（一位知识渊博的！）人一样提出他们的问题即可。

但是，仅仅因为用户可以用“纯英语”（或“纯德语”等）提问，并不意味着他们总是会这样。例如，用户可能会输入几个关键词而不是一个完整的问题，因为他们不了解管道的全部功能，或者已经习惯了关键词搜索。虽然标准的 Haystack 管道可能以合理的准确性处理此类查询，但出于各种原因，我们仍然希望我们的管道能够区分接收到的查询类型，以便在用户输入一组关键词而不是一个问题时做出不同的响应。因此，Haystack 提供了内置功能来区分文本输入类型，例如关键词查询、疑问查询（问题）和陈述查询。

给定一个文本输入，分类模型会输出一个标签，该标签可以根据模型的训练方式为 LABEL_0、LABEL_1 等。Haystack 的 TransformersTextRouter 组件使用分类模型，然后将文本路由到以标签命名的输出分支。

在本教程中，您将学习如何使用 TransformersTextRouter 和 TransformersZeroShotTextRouter 根据接收到的查询类型来分支您的 Haystack 管道。

关键词与问题/陈述 — 根据查询是完整的问题/陈述还是关键词集合，将其路由到两个分支之一。
问题与陈述 — 根据查询是问题还是陈述，将自然语言查询路由到两个分支之一。

使用 TransformersTextRouter，还可以根据自定义分类模型来路由查询。使用 TransformersZeroShotTextRouter，您甚至可以进行零样本分类，这意味着您可以指定自定义类别，而无需任何自定义模型。

在解释完所有这些之后，让我们开始吧！

准备 Colab 环境

启用 GPU 运行时

安装 Haystack

首先，使用 pip 安装最新版本的 Haystack。

%%bash

pip install --upgrade pip
pip install haystack-ai torch sentencepiece datasets sentence-transformers

试用 TransformersTextRouter

在将 TransformersTextRouter 集成到管道之前，先单独对其进行测试，看看它实际做了什么。首先，初始化一个简单的、开箱即用的关键词与问题/陈述 TransformersTextRouter，它使用了 shahrukhx01/bert-mini-finetune-question-detection 模型。对于此模型，LABEL_0 对应于关键词查询，LABEL_1 对应于非关键词查询。

from haystack.components.routers import TransformersTextRouter

text_router = TransformersTextRouter(model="shahrukhx01/bert-mini-finetune-question-detection")
text_router.warm_up()

现在向此文本路由器输入一些查询。测试一个关键词查询、一个疑问查询和一个陈述查询。请注意，您无需使用任何标点符号（如问号）即可让文本路由器做出正确的决定。

queries = [
    "Arya Stark father",  # Keyword Query
    "Who was the father of Arya Stark",  # Interrogative Query
    "Lord Eddard was the father of Arya Stark",  # Statement Query
]

下面，您可以看到文本路由器如何处理这些查询：它正确地确定“Arya Stark father”是一个关键词查询，并将其发送到分支 LABEL_0。它还正确地将疑问查询“Who was the father of Arya Stark”和陈述查询“Lord Eddard was the father of Arya Stark”分类为非关键词查询，并将它们发送到分支 1。

result = text_router.run(text=queries[0])
next(iter(result))

import pandas as pd

results = {"Query": [], "Output Branch": [], "Class": []}

for query in queries:
    result = text_router.run(text=query)
    results["Query"].append(query)
    results["Output Branch"].append(next(iter(result)))
    results["Class"].append("Keyword Query" if next(iter(result)) == "LABEL_0" else "Question/Statement")

pd.DataFrame.from_dict(results)

接下来，您将使用 shahrukhx01/question-vs-statement-classifier 来演示一个问题与陈述 TransformersTextRouter。对于此任务，您需要使用此分类模型初始化一个新的文本路由器。

text_router = TransformersTextRouter(model="shahrukhx01/question-vs-statement-classifier")
text_router.warm_up()

queries = [
    "Who was the father of Arya Stark",  # Interrogative Query
    "Lord Eddard was the father of Arya Stark",  # Statement Query
]

results = {"Query": [], "Output Branch": [], "Class": []}

for query in queries:
    result = text_router.run(text=query)
    results["Query"].append(query)
    results["Output Branch"].append(next(iter(result)))
    results["Class"].append("Question" if next(iter(result)) == "LABEL_1" else "Statement")

pd.DataFrame.from_dict(results)

同样，文本路由器会将问题和陈述发送到预期的输出分支。

文本分类的自定义用例

TransformersTextRouter 非常灵活，还支持除区分关键词查询和疑问查询之外的其他文本分类选项。例如，您可能对检测文本的情感或对主题进行分类感兴趣。您可以通过从 Hugging Face Hub 加载自定义分类模型或使用 TransformersZeroShotTextRouter 进行零样本分类来实现这一点。

传统的文本分类模型经过训练，用于预测少数“硬编码”类别之一，并需要专门的训练数据集。在 Hugging Face Hub 中，您可以找到许多预训练模型，可能与您感兴趣的领域相关。
零样本分类非常通用：通过选择合适的 Transformer 基础模型，您可以在没有任何训练数据集的情况下对文本进行分类。您只需提供候选类别。

使用 TransformersTextRouter 进行自定义分类模型

对于此用例，您可以使用 Hugging Face Hub 上提供的公共模型。例如，如果您想对查询的情感进行分类，可以选择合适的模型，例如 cardiffnlp/twitter-roberta-base-sentiment。

text_router = TransformersTextRouter(model="cardiffnlp/twitter-roberta-base-sentiment")
text_router.warm_up()

queries = [
    "What's the answer?",  # neutral query
    "Would you be so lovely to tell me the answer?",  # positive query
    "Can you give me the damn right answer for once??",  # negative query
]

sent_results = {"Query": [], "Output Branch": [], "Class": []}

for query in queries:
    result = text_router.run(text=query)
    sent_results["Query"].append(query)
    sent_results["Output Branch"].append(next(iter(result)))
    sent_results["Class"].append(
        {"LABEL_0": "negative", "LABEL_1": "neutral", "LABEL_2": "positive"}.get(next(iter(result)), "Unknown")
    )

pd.DataFrame.from_dict(sent_results)

使用 TransformersZeroShotTextRouter 进行零样本分类

TransformersZeroShotTextRouter 允许您通过提供合适的 Transformer 基础模型和定义模型应预测的类别来执行零样本分类。

首先，使用一些自定义类别标签初始化一个 TransformersZeroShotTextRouter。默认情况下，它使用基础大小的零样本分类模型 MoritzLaurer/deberta-v3-base-zeroshot-v1.1-all-33。您可以通过使用 TransformersZeroShotTextRouter(model="MoritzLaurer/deberta-v3-large-zeroshot-v2.0") 来切换到更大的模型 MoritzLaurer/deberta-v3-large-zeroshot-v2.0 并获得更好的结果。

让我们看一个例子。您可能想知道用户查询是否与音乐或电影相关。在这种情况下，labels 参数是一个包含候选类别的列表，组件的输出分支相应命名。

from haystack.components.routers import TransformersZeroShotTextRouter

text_router = TransformersZeroShotTextRouter(labels=["music", "cinema"])
text_router.warm_up()

queries = [
    "In which films does John Travolta appear?",  # cinema
    "What is the Rolling Stones first album?",  # music
    "Who was Sergio Leone?",  # cinema
]

sent_results = {"Query": [], "Output Branch": []}

for query in queries:
    result = text_router.run(text=query)
    sent_results["Query"].append(query)
    sent_results["Output Branch"].append(next(iter(result)))

pd.DataFrame.from_dict(sent_results)

与前面的示例类似，我们可以使用零样本文本分类将问题分组到“权力的游戏”、“星球大战”和“指环王”相关问题中。标签的数量由您决定！

from haystack.components.routers import TransformersZeroShotTextRouter

text_router = TransformersZeroShotTextRouter(labels=["Game of Thrones", "Star Wars", "Lord of the Rings"])
text_router.warm_up()

queries = [
    "Who was the father of Arya Stark",  # Game of Thrones
    "Who was the father of Luke Skywalker",  # Star Wars
    "Who was the father of Frodo Baggins",  # Lord of the Rings
]

results = {"Query": [], "Output Branch": []}

for query in queries:
    result = text_router.run(text=query)
    results["Query"].append(query)
    results["Output Branch"].append(next(iter(result)))

pd.DataFrame.from_dict(results)

正如您所见，“Arya Stark”关于“权力的游戏”的问题被发送到“权力的游戏”分支，而“Luke Skywalker”关于“星球大战”的问题被发送到“星球大战”分支，“Frodo Baggins”关于“指环王”的问题被发送到“指环王”。这意味着您可以让您的管道以不同的方式处理关于这些宇宙的问题。

恭喜！🎉 您已经了解了 TransformersZeroShotTextRouter 和 TransformersTextRouter 的工作原理以及如何单独使用这些组件。现在让我们探索如何在管道中使用它们。

关键字与问题/陈述查询分类的管道

现在您将创建一个包含关键词与问题/陈述查询分类的问答（QA）管道，并根据分类结果路由问题。

获取并索引文档

您将通过下载数据并使用嵌入项索引数据到 DocumentStore 来开始创建问答系统。

在本教程中，您将采用一种简单的方法将文档及其嵌入项写入 DocumentStore。有关包含预处理、清理和拆分的完整索引管道，请参阅我们关于预处理不同文件类型的教程。

初始化 DocumentStore

初始化一个 DocumentStore 来索引您的文档。DocumentStore 存储问答系统用于查找问题答案的文档。在本教程中，您将使用 InMemoryDocumentStore。

from haystack.document_stores.in_memory import InMemoryDocumentStore

document_store = InMemoryDocumentStore()

InMemoryDocumentStore 是最容易入门的 DocumentStore。它不需要外部依赖，并且是小型项目和调试的不错选择。但它在大规模文档集合上的扩展性不太好，因此不适合生产系统。要了解 Haystack 支持的各种外部数据库，请参阅 DocumentStore 集成。

DocumentStore 现已准备就绪。现在是时候填充一些文档了。

获取数据

您将使用古代世界七大奇迹的维基百科页面作为文档。我们预处理了数据并将其上传到 Hugging Face Space：Seven Wonders。因此，您无需执行任何额外的清理或拆分。

获取数据并将其转换为 Haystack 文档

from datasets import load_dataset
from haystack import Document

dataset = load_dataset("bilgeyucel/seven-wonders", split="train")
docs = [Document(content=doc["content"], meta=doc["meta"]) for doc in dataset]

初始化文档嵌入器

要将数据及其嵌入项存储在 DocumentStore 中，请使用模型名称初始化一个 SentenceTransformersDocumentEmbedder 并调用 warm_up() 来下载嵌入模型。

如果您愿意，可以使用不同的 Embedder 来处理您的文档。

from haystack.components.embedders import SentenceTransformersDocumentEmbedder

doc_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
doc_embedder.warm_up()

将文档写入 DocumentStore

使用文档运行 doc_embedder。嵌入器将为每个文档创建嵌入项，并将这些嵌入项保存在 Document 对象embedding 字段中。然后，您可以使用 write_documents() 方法将文档写入 DocumentStore。

docs_with_embeddings = doc_embedder.run(docs)
document_store.write_documents(docs_with_embeddings["documents"])

2) 初始化检索器、文本嵌入器和 TransformersTextRouter

您的管道将是一个简单的检索器管道，但检索器的选择将取决于接收到的查询类型：关键词查询将使用稀疏 BM25Retriever，而问题/陈述查询将使用更准确但计算成本更高的 EmbeddingRetriever。

现在，初始化一个 Router、一个 Embedder，以及两个 Retrievers 和一个 Joiner。

from haystack.components.retrievers.in_memory import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.joiners import DocumentJoiner

text_router = TransformersTextRouter(model="shahrukhx01/bert-mini-finetune-question-detection")
text_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
embedding_retriever = InMemoryEmbeddingRetriever(document_store)
bm25_retriever = InMemoryBM25Retriever(document_store)
document_joiner = DocumentJoiner()

3) 定义管道

正如所承诺的，文本路由器的“问题/陈述”分支 LABEL_0 被馈送到 TextEmbedder，然后是 EmbeddingRetriever，而关键词分支 LABEL_1 被馈送到 BM25Retriever。这两个检索器随后都被馈送到我们的 Joiner。因此，我们的管道可以被视为具有某种菱形形状：所有查询都被发送到路由器，路由器将这些查询分成两个不同的检索器，这两个检索器将其输出馈送到同一个 Joiner。

from haystack import Pipeline

query_classification_pipeline = Pipeline()
query_classification_pipeline.add_component("text_router", text_router)
query_classification_pipeline.add_component("text_embedder", text_embedder)
query_classification_pipeline.add_component("embedding_retriever", embedding_retriever)
query_classification_pipeline.add_component("bm25_retriever", bm25_retriever)
query_classification_pipeline.add_component("document_joiner", document_joiner)

query_classification_pipeline.connect("text_router.LABEL_0", "text_embedder")
query_classification_pipeline.connect("text_embedder", "embedding_retriever")
query_classification_pipeline.connect("text_router.LABEL_1", "bm25_retriever")
query_classification_pipeline.connect("bm25_retriever", "document_joiner")
query_classification_pipeline.connect("embedding_retriever", "document_joiner")

4) 运行管道

下面，您可以看到这种选择如何影响分支结构：关键词查询“arya stark father”和问题查询“Who is the father of Arya Stark?”产生了明显不同的结果，这种区别很可能是由于对关键词和问题/陈述查询使用了不同的检索器。

# Useful for framing headers
equal_line = "=" * 30

# Run only the dense retriever on the full sentence query
res_1 = query_classification_pipeline.run({"text_router": {"text": "Who is the father of Arya Stark?"}})
print(f"\n\n{equal_line}\nQUESTION QUERY RESULTS\n{equal_line}")
print(res_1)

# Run only the sparse retriever on a keyword based query
res_2 = query_classification_pipeline.run({"text_router": {"text": "arya stark father"}})
print(f"\n\n{equal_line}\nKEYWORD QUERY RESULTS\n{equal_line}")
print(res_2)

上面您看到了关键词与问题/陈述分类的一个潜在用途：您可能会选择对关键词查询使用资源消耗较低的检索器，而不是对问题/陈述查询使用。但是，问题与陈述分类呢？

问题与陈述查询分类器的管道

为了说明问题与陈述分类的一个潜在用途，您将构建一个如下所示的管道。

管道将从一个检索器开始，该检索器将处理所有查询。
管道将以一个读取器结束，该读取器将只处理问题查询。

换句话说，您的管道将是一个针对陈述查询的仅检索器管道——给定陈述“Arya Stark was the daughter of a Lord”，您只会得到最相关的文档；但它将是一个针对问题查询的检索器-读取器管道。

为了使事情更具体，您的管道将从一个检索器开始，然后将其馈送到一个设置为执行问题与陈述分类的 QueryClassifier。QueryClassifier 的第一个分支（处理问题查询）将发送到 Reader，而第二个分支将不连接到任何其他节点。因此，管道的最后一个节点取决于查询类型：问题会一直通过 Reader，而陈述只会通过 Retriever。

现在，定义管道。请记住，您无需再次将文档写入 DocumentStore，因为它们已经被索引了。

2) 定义管道和组件

from haystack.components.readers import ExtractiveReader

query_classification_pipeline = Pipeline()
query_classification_pipeline.add_component("bm25_retriever_0", InMemoryBM25Retriever(document_store))
query_classification_pipeline.add_component("bm25_retriever_1", InMemoryBM25Retriever(document_store))
query_classification_pipeline.add_component(
    "text_router", TransformersTextRouter(model="shahrukhx01/question-vs-statement-classifier")
)
query_classification_pipeline.add_component("reader", ExtractiveReader())

query_classification_pipeline.connect("text_router.LABEL_0", "bm25_retriever_0")
query_classification_pipeline.connect("bm25_retriever_0", "reader")
query_classification_pipeline.connect("text_router.LABEL_1", "bm25_retriever_1")

2) 运行管道

以下是此管道的结果：对于像“Who is the father of Arya Stark?”这样的问题查询，您会从 Reader 获得答案；对于像“Arya Stark was the daughter of a Lord”这样的陈述查询，您只会从 Retriever 获得文档。

# Useful for framing headers
equal_line = "=" * 30

# Run the retriever + reader on the question query
query = "Who is the father of Arya Stark?"
res_1 = query_classification_pipeline.run({"text_router": {"text": query}, "reader": {"query": query}})
print(f"\n\n{equal_line}\nQUESTION QUERY RESULTS\n{equal_line}")
print(res_1)

# Run only the retriever on the statement query
query = "Arya Stark was the daughter of a Lord"
res_2 = query_classification_pipeline.run({"text_router": {"text": query}, "reader": {"query": query}})
print(f"\n\n{equal_line}\nKEYWORD QUERY RESULTS\n{equal_line}")
print(res_2)

使用 Multiplexer 简化管道输入

使用 Haystack 创建多代理系统