企业官网建设流程全解析

Haystack Transformers 集成组件详解：零样本分类、NER、本地 Chat、抽取式 QA 与路由的完整 API 实战指南

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

本文基于 Haystack 官方参考文档中的 Transformers 集成 API 页面整理与扩充，系统讲解transformers-haystack集成包提供的六大本地模型组件：TransformersZeroShotDocumentClassifier、TransformersNamedEntityExtractor、TransformersChatGenerator、TransformersExtractiveReader、TransformersTextRouter与TransformersZeroShotTextRouter。读完后，你将能够掌握各组件的初始化参数、run调用语义与序列化方式，并能把本地 Hugging Face 模型以 Pipeline 组件形式接入 Haystack 检索-生成工作流。

一、集成包定位：为什么这些组件不在 Haystack 核心里

在深入各个组件之前，先明确一个重要的架构事实：这批基于 Hugging Facetransformers的组件不属于 Haystack 核心仓库，而是迁移到了独立的transformers-haystack集成包中，导入路径统一为haystack_integrations.components.*。

从源码结构看，当前仓库（Haystack 核心，VERSION.txt显示版本为 3.2.0-rc0）的haystack/components/目录下并不包含任何 Transformers 组件。迁移依据可以在仓库中直接验证：

• 迁移指南 明确列出对应关系，例如from haystack.components.routers import TransformersTextRouter迁移后应写为from haystack_integrations.components.routers.transformers import TransformersTextRouter，且HuggingFaceLocalChatGenerator更名为TransformersChatGenerator、ExtractiveReader更名为TransformersExtractiveReader；
• 弃用发布说明 说明TransformersZeroShotDocumentClassifier、TransformersTextRouter、TransformersZeroShotTextRouter、HuggingFaceLocalChatGenerator、NamedEntityExtractor和ExtractiveReader已从核心弃用并计划移除，需通过pip install transformers-haystack安装独立包继续使用。

这意味着本文所有组件的使用前提只有一个：安装transformers-haystack包（及其传递依赖 Hugging Facetransformers）。这些组件与核心仓库提供的Pipeline、Document、InMemoryDocumentStore等原语组合使用，后者在当前仓库中分别位于 pipeline.py、document.py 和 document_store.py。

此外，所有组件共享三条底层约定，后文不再重复：

1. 设备选择：device参数接受ComponentDevice，为None时自动选择默认设备；若在 pipeline kwargs 中显式指定了 device/device map，则会覆盖device参数。
2. 认证令牌：token参数默认值形如Secret.from_env_var(["HF_API_TOKEN", "HF_TOKEN"], strict=False)，即自动读取环境变量HF_API_TOKEN或HF_TOKEN，未设置时不报错，仅公开模型可用。
3. 预热与序列化：warm_up()惰性加载 Hugging Face pipeline（首次run时若未手动预热会自动触发），to_dict()/from_dict()支持 YAML/JSON 序列化整个 Pipeline 配置。

二、TransformersZeroShotDocumentClassifier：给文档打上无监督标签

该组件对文档执行零样本（zero-shot）分类：在初始化时提供 NLI 模型与标签集合，对每个文档预测标签并写入其metadata的classification字段。默认在Document.content上分类，也可通过classification_field指定对某个 metadata 字段分类。文档推荐的可用模型包括valhalla/distilbart-mnli-12-3、cross-encoder/nli-distilroberta-base和cross-encoder/nli-deberta-v3-xsmall。

2.1 初始化参数

__init__( model: str, labels: list[str], multi_label: bool = False, classification_field: str | None = None, device: ComponentDevice | None = None, token: Secret | None = Secret.from_env_var( ["HF_API_TOKEN", "HF_TOKEN"], strict=False ), huggingface_pipeline_kwargs: dict[str, Any] | None = None, ) -> None

2.2 实战示例：检索后按情感分类

以下示例将 BM25 检索器与分类器串联，检索出的文档会依据"positive"/"negative"标签被打上分类元数据。其中InMemoryBM25Retriever属于 Haystack 核心，可在 bm25_retriever.py 中查看其实现：

from haystack import Document from haystack.components.retrievers.in_memory import InMemoryBM25Retriever from haystack.core.pipeline import Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack_integrations.components.classifiers.transformers import TransformersZeroShotDocumentClassifier documents = [Document(id="0", content="Today was a nice day!"), Document(id="1", content="Yesterday was a bad day!")] document_store = InMemoryDocumentStore() retriever = InMemoryBM25Retriever(document_store=document_store) document_classifier = TransformersZeroShotDocumentClassifier( model="cross-encoder/nli-deberta-v3-xsmall", labels=["positive", "negative"], ) document_store.write_documents(documents) pipeline = Pipeline() pipeline.add_component(instance=retriever, name="retriever") pipeline.add_component(instance=document_classifier, name="document_classifier") pipeline.connect("retriever", "document_classifier") queries = ["How was your day today?", "How was your day yesterday?"] expected_predictions = ["positive", "negative"] for idx, query in enumerate(queries): result = pipeline.run({"retriever": {"query": query, "top_k": 1}}) assert result["document_classifier"]["documents"][0].to_dict()["id"] == str(idx) assert (result["document_classifier"]["documents"][0].to_dict()["classification"]["label"] == expected_predictions[idx])

2.3 run 的输出语义

run(documents: list[Document], batch_size: int = 1) -> dict[str, Any]

• documents：待分类的文档列表；
• batch_size：批量处理每个文档内容时的批大小。

返回值包含documents键：分类结果写入每个文档metadata["classification"]字典中，其中包含命中的label；当multi_label=True时，每个候选标签的得分还能在classification字典的details键下取到。除run外，组件还提供标准的warm_up()、to_dict()与from_dict(data)序列化方法。

三、TransformersNamedEntityExtractor：把实体标注存进 metadata

该组件对文档集合做命名实体识别（NER），支持 Hugging Face Hub 上任意 token classification 模型（文档示例使用dslim/bert-base-NER）。标注结果作为 metadata 存储回文档中，并提供静态方法get_stored_annotations取出。

3.1 数据结构与使用示例

from haystack import Document from haystack_integrations.components.extractors.transformers import TransformersNamedEntityExtractor documents = [ Document(content="I'm Merlin, the happy pig!"), Document(content="My name is Clara and I live in Berkeley, California."), ] extractor = TransformersNamedEntityExtractor(model="dslim/bert-base-NER") results = extractor.run(documents=documents)["documents"] annotations = [TransformersNamedEntityExtractor.get_stored_annotations(doc) for doc in results] print(annotations)

标注结果由NamedEntityAnnotation数据类描述，字段包括：entity（实体标签，str）、start（实体在文档中的起始索引，int）、end（结束索引，int）、score（模型分数，float | None）。

3.2 API 一览

__init__( *, model: str, pipeline_kwargs: dict[str, Any] | None = None, device: ComponentDevice | None = None, token: Secret | None = Secret.from_env_var( ["HF_API_TOKEN", "HF_TOKEN"], strict=False ) ) -> None

注意model是关键字必填参数（*之后）。pipeline_kwargs透传给 HF pipeline；warm_up()初始化失败或run处理失败均抛出ComponentError。

run(documents: list[Document], batch_size: int = 1) -> dict[str, Any] get_stored_annotations(document: Document) -> list[NamedEntityAnnotation] | None initialized: bool # 提取器是否已就绪

同样提供to_dict()/from_dict(data)完成序列化往返。

四、TransformersChatGenerator：本地跑的 Chat 生成器

这是本集成中功能最重的组件（即原核心组件HuggingFaceLocalChatGenerator更名而来），用于在本地运行 Chat 模型，如Qwen/Qwen3-0.6B或meta-llama/Llama-2-7b-chat-hf——本地模型对硬件有一定要求。最小用法：

from haystack.dataclasses import ChatMessage from haystack_integrations.components.generators.transformers import TransformersChatGenerator generator = TransformersChatGenerator(model="Qwen/Qwen3-0.6B") messages = [ChatMessage.from_user("What's Natural Language Processing? Be brief.")] print(generator.run(messages))

返回{"replies": [ChatMessage, ...]}，ChatMessage携带finish_reason、index、model、usage（含prompt_tokens/completion_tokens/total_tokens）等 meta 信息。

4.1 初始化参数全表

__init__( model: str = "Qwen/Qwen3-0.6B", task: Literal["text-generation", "image-text-to-text"] | None = None, device: ComponentDevice | None = None, token: Secret | None = Secret.from_env_var( ["HF_API_TOKEN", "HF_TOKEN"], strict=False ), chat_template: str | None = None, generation_kwargs: dict[str, Any] | None = None, huggingface_pipeline_kwargs: dict[str, Any] | None = None, stop_words: list[str] | None = None, streaming_callback: StreamingCallbackT | None = None, tools: ToolsType | None = None, tool_parsing_function: Callable[[str], list[ToolCall] | None] | None = None, async_executor: ThreadPoolExecutor | None = None, *, enable_thinking: bool = False ) -> None

4.2 run / run_async 与工具调用

run( messages: list[ChatMessage] | str, generation_kwargs: dict[str, Any] | None = None, streaming_callback: StreamingCallbackT | None = None, tools: ToolsType | None = None, ) -> dict[str, list[ChatMessage]]

• messages支持直接传字符串，会被转换为一条 user 角色的ChatMessage列表；
• 运行期generation_kwargs与初始化值按 key 合并：本次传入的 key 优先，仅初始化设置的 key 保留；
• 运行期tools若提供则覆盖初始化时设置的tools；
• 返回值仅含replies键，为ChatMessage列表。

run_async(...)与run签名、参数、返回值完全一致，可在 async 代码中以await使用。内部的create_message(text, index, tokenizer, prompt, generation_kwargs, parse_tool_calls=False)负责从生成文本构造带 meta 的ChatMessage，parse_tool_calls=True时会对文本做工具调用解析。生命周期方法为warm_up()（初始化组件并预热 tools）与close()（关闭组件自有的 executor），序列化走to_dict()/from_dict(data)。

五、TransformersExtractiveReader：跨文档可比的抽取式问答

该组件（原核心ExtractiveReader更名而来）执行抽取式 QA。其设计要点在于：对每个候选答案 span 独立打分，不做文档内归一化，从而避免了其他实现“按文档独立归一化导致跨文档答案分数难以比较”的通病，便于在多文档检索结果中按分数统一排序。

5.1 使用示例

from haystack import Document from haystack_integrations.components.readers.transformers import TransformersExtractiveReader docs = [ Document(content="Python is a popular programming language"), Document(content="python ist eine beliebte Programmiersprache"), ] reader = TransformersExtractiveReader() question = "What is a popular programming language?" result = reader.run(query=question, documents=docs) assert "Python" in result["answers"][0].data

5.2 初始化参数

__init__( model: Path | str = "deepset/roberta-base-squad2-distilled", device: ComponentDevice | None = None, token: Secret | None = Secret.from_env_var( ["HF_API_TOKEN", "HF_TOKEN"], strict=False ), top_k: int = 20, score_threshold: float | None = None, max_seq_length: int = 384, stride: int = 128, max_batch_size: int | None = None, answers_per_seq: int | None = None, no_answer: bool = True, calibration_factor: float = 0.1, overlap_threshold: float | None = 0.01, model_kwargs: dict[str, Any] | None = None, ) -> None

5.3 run 与去重

run( query: str, documents: list[Document], top_k: int | None = None, score_threshold: float | None = None, max_seq_length: int | None = None, stride: int | None = None, max_batch_size: int | None = None, answers_per_seq: int | None = None, no_answer: bool | None = None, overlap_threshold: float | None = None, ) -> dict[str, Any]

run支持对 init 参数逐项做运行期覆盖，返回值是按得分降序排列的答案列表（含no answer项时的说明见上表）。配套的deduplicate_by_overlap(answers, overlap_threshold)静态方法按 span 重叠度对同一文档内的抽取式答案去重，返回去重后的list[ExtractedAnswer]；warm_up()用于初始化，to_dict()/from_dict(data)完成序列化。

六、TransformersTextRouter：基于分类模型的多语言路由

该组件用文本分类模型把输入路由到不同连接，标签体系由所选模型决定（可在模型的 HF 页面描述中查看标签含义）。典型场景是多语言分流：

from haystack.components.builders import PromptBuilder from haystack.components.generators import HuggingFaceLocalGenerator from haystack.core.pipeline import Pipeline from haystack_integrations.components.routers.transformers import TransformersTextRouter p = Pipeline() p.add_component( instance=TransformersTextRouter(model="papluca/xlm-roberta-base-language-detection"), name="text_router" ) p.add_component( instance=PromptBuilder(template="Answer the question: {{query}}\nAnswer:"), name="english_prompt_builder" ) p.add_component( instance=PromptBuilder(template="Beantworte die Frage: {{query}}\nAntwort:"), name="german_prompt_builder" ) p.add_component( instance=HuggingFaceLocalGenerator(model="DiscoResearch/Llama3-DiscoLeo-Instruct-8B-v0.1"), name="german_llm" ) p.add_component( instance=HuggingFaceLocalGenerator(model="microsoft/Phi-3-mini-4k-instruct"), name="english_llm" ) p.connect("text_router.en", "english_prompt_builder.query") p.connect("text_router.de", "german_prompt_builder.query") p.connect("english_prompt_builder.prompt", "english_llm.prompt") p.connect("german_prompt_builder.prompt", "german_llm.prompt") # English Example print(p.run({"text_router": {"text": "What is the capital of Germany?"}})) # German Example print(p.run({"text_router": {"text": "Was ist die Hauptstadt von Deutschland?"}}))

初始化与 run 语义

__init__( model: str, labels: list[str] | None = None, device: ComponentDevice | None = None, token: Secret | None = Secret.from_env_var( ["HF_API_TOKEN", "HF_TOKEN"], strict=False ), huggingface_pipeline_kwargs: dict[str, Any] | None = None, ) -> None

• model：文本分类的 HF 模型名或路径（必填）；
• labels：可选的标签列表；不传时组件会用transformers.AutoConfig.from_pretrained从 HF Hub 的模型配置中自动拉取标签；
• token：为True语义下读取HF_API_TOKEN/HF_TOKEN环境变量，可用transformers-cli login生成令牌。

run(text: str) -> dict[str, str]

返回值以预测标签为 key、原文为 value（如{"en": "What is the capital of Germany?"}），因此 Pipeline 连接时使用text_router.<label>形式。输入非字符串会抛TypeError。同样提供warm_up()、to_dict()/from_dict(data)。

七、TransformersZeroShotTextRouter：标签自定义的零样本路由

与TransformersTextRouter不同，零样本路由器的标签由用户在初始化时指定，默认模型为MoritzLaurer/deberta-v3-base-zeroshot-v1.1-all-33。典型用法是判断输入更像“查询”还是“段落”，从而走不同的 embedding 前缀分支：

from haystack import Document # Requires: pip install sentence-transformers-haystack from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersTextEmbedder from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersDocumentEmbedder from haystack.components.retrievers import InMemoryEmbeddingRetriever from haystack.core.pipeline import Pipeline from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack_integrations.components.routers.transformers import TransformersZeroShotTextRouter document_store = InMemoryDocumentStore() doc_embedder = SentenceTransformersDocumentEmbedder(model="intfloat/e5-base-v2") docs = [ Document( content="Germany, officially the Federal Republic of Germany, is a country in the western region of " "Central Europe. The nation's capital and most populous city is Berlin and its main financial centre " "is Frankfurt; the largest urban area is the Ruhr." ), Document( content="France, officially the French Republic, is a country located primarily in Western Europe. " "France is a unitary semi-presidential republic with its capital in Paris, the country's largest city " "and main cultural and commercial centre; other major urban areas include Marseille, Lyon, Toulouse, " "Lille, Bordeaux, Strasbourg, Nantes and Nice." ) ] docs_with_embeddings = doc_embedder.run(docs) document_store.write_documents(docs_with_embeddings["documents"]) p = Pipeline() p.add_component(instance=TransformersZeroShotTextRouter(labels=["passage", "query"]), name="text_router") p.add_component( instance=SentenceTransformersTextEmbedder(model="intfloat/e5-base-v2", prefix="passage: "), name="passage_embedder" ) p.add_component( instance=SentenceTransformersTextEmbedder(model="intfloat/e5-base-v2", prefix="query: "), name="query_embedder" ) p.add_component( instance=InMemoryEmbeddingRetriever(document_store=document_store), name="query_retriever" ) p.add_component( instance=InMemoryEmbeddingRetriever(document_store=document_store), name="passage_retriever" ) p.connect("text_router.passage", "passage_embedder.text") p.connect("passage_embedder.embedding", "passage_retriever.query_embedding") p.connect("text_router.query", "query_embedder.text") p.connect("query_embedder.embedding", "query_retriever.query_embedding") # Query Example p.run({"text_router": {"text": "What is the capital of Germany?"}}) # Passage Example p.run({ "text_router":{ "text": "The United Kingdom of Great Britain and Northern Ireland, commonly known as the " "United Kingdom (UK) or Britain, is a country in Northwestern Europe, off the north-western coast of " "the continental mainland." } })

初始化与 run 语义

__init__( labels: list[str], multi_label: bool = False, model: str = "MoritzLaurer/deberta-v3-base-zeroshot-v1.1-all-33", device: ComponentDevice | None = None, token: Secret | None = Secret.from_env_var( ["HF_API_TOKEN", "HF_TOKEN"], strict=False ), huggingface_pipeline_kwargs: dict[str, Any] | None = None, ) -> None

• labels（必填）：分类标签集合，可为单个标签、逗号分隔字符串或列表；
• multi_label：False时每个序列的标签得分归一化总和为 1；True时标签独立，按 entailment 与 contradiction 分数做 softmax 归一化；
• run(text: str) -> dict[str, str]：与TransformersTextRouter相同，标签为 key、原文为 value，非字符串输入抛TypeError。

注意示例中SentenceTransformersTextEmbedder等嵌入器来自另一个独立集成包sentence-transformers-haystack（参见 MIGRATION.md 的迁移映射表），使用前需另行安装。

八、组件速查与选型建议

选型上可以这样把握：需要“打标签存库”选文档分类器或 NER 提取器（两者结果都落在Document.metadata，天然适配后续MetadataRouter等核心组件）；需要本地问答选TransformersExtractiveReader（其跨文档可比分数是相对其他实现的关键优势）；需要本地对话选TransformersChatGenerator（注意 ChatML 要求与max_new_tokens默认 512）；需要在 Pipeline 前段做分流则两个 Router 按“模型自带标签 / 自定义标签”二选一。所有组件均可通过to_dict()/from_dict()序列化进 Pipeline YAML，配合核心 serialization.py 的注册机制在集群间搬运模型配置——但模型权重本身仍需部署环境可访问 HF Hub 或本地模型路径。

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack