Haystack怎么用？从文档入库到检索管线与RAG问答的入门指南

先分清三个角色

Haystack中的组件负责具体工作，例如转换文件、生成向量或调用模型；Pipeline负责连接组件的输入输出；Document Store保存文档、元数据和向量，供检索器读取。基础RAG通常分为两条管线：索引管线处理资料，查询管线根据问题检索并生成回答。

下面统一使用Haystack 2.x的组件与接口。准备一份允许使用的公开产品手册，将正文保存为manual.txt，以空行分隔段落，并保留章节标题、产品型号和版本信息。示例只处理文本文件；PDF需要替换转换组件，扫描件还需先做OCR。

准备环境与组件

在独立Python环境安装依赖，避免与Haystack 1.x的farm-haystack包混装：

pip install haystack-ai sentence-transformers

生成环节使用OpenAI接口，运行前需通过环境变量配置OPENAI_API_KEY，并确认账户可使用示例模型。以下代码按顺序放入同一个Python文件，各段共享变量：

from haystack import Pipeline

from haystack.document_stores.in_memory import InMemoryDocumentStore

from haystack.components.converters import TextFileToDocument

from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter

from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder

from haystack.components.writers import DocumentWriter

from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever

from haystack.components.builders import PromptBuilder

from haystack.components.generators import OpenAIGenerator

store = InMemoryDocumentStore(embedding_similarity_function="cosine")

embedding_model = "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"

索引管线：从文件到向量

转换器把文件变成Document；清理器处理多余空白等内容；切分器生成便于检索的片段；文档Embedding组件为每段生成向量；写入器将结果保存到store。

indexing = Pipeline()

indexing.add_component("convert", TextFileToDocument())

indexing.add_component("clean", DocumentCleaner())

indexing.add_component("split", DocumentSplitter(split_by="passage", split_length=3, split_overlap=1))

indexing.add_component("embed", SentenceTransformersDocumentEmbedder(model=embedding_model))

indexing.add_component("write", DocumentWriter(document_store=store))

indexing.connect("convert.documents", "clean.documents")

indexing.connect("clean.documents", "split.documents")

indexing.connect("split.documents", "embed.documents")

indexing.connect("embed.documents", "write.documents")

indexing.run({"convert": {"sources": ["manual.txt"], "meta": {"source": "公开产品手册", "file": "manual.txt"}}})

这里按空行分段，每个片段包含三个段落，相邻片段重叠一段。它适合已整理的中文文本；若原文没有空行，应先整理结构。入库前检查清理和切分结果，避免参数表、警告语与适用条件被拆散。

查询管线：检索、组装提示词、生成

文档Embedding处理资料，查询Embedding处理问题。两端应使用同一模型与一致的向量处理配置；只保证维度相同仍不够。更换模型后，应重新生成文档向量并入库。

template = "请仅依据以下资料回答，资料不足时明确说明无法从手册确认。引用片段编号，不要补充猜测。\n{% for doc in documents %}[{{ loop.index }}] {{ doc.content }}\n{% endfor %}\n问题：{{ question }}"

querying = Pipeline()

querying.add_component("embed", SentenceTransformersTextEmbedder(model=embedding_model))

querying.add_component("retrieve", InMemoryEmbeddingRetriever(document_store=store, top_k=3))

querying.add_component("prompt", PromptBuilder(template=template, required_variables=["documents", "question"]))

querying.add_component("generate", OpenAIGenerator(model="gpt-4o-mini"))

querying.connect("embed.embedding", "retrieve.query_embedding")

querying.connect("retrieve.documents", "prompt.documents")

querying.connect("prompt.prompt", "generate.prompt")

question = "这款产品的工作温度范围是多少？"

result = querying.run({"embed": {"text": question}, "prompt": {"question": question}}, include_outputs_from={"retrieve"})

print(result["generate"]["replies"])

print([(i + 1, d.content, d.meta) for i, d in enumerate(result["retrieve"]["documents"])])

问题分别传给向量组件和提示词组件。输出同时保留回答与检索片段，便于把回答中的编号对应到原文和文件信息。首次运行需要下载Embedding模型，生成步骤还需要网络与接口权限。

用两类问题检查依据

• 可回答问题：选择手册明确记载的参数，检查前三条片段是否包含数值、单位、适用型号，以及回答是否准确引用。
• 不可回答问题：询问手册未记载的未来功能或内部采购价，检查模型是否说明资料不足。检索器仍可能返回相似片段，不能把相似度当成答案存在的证明。

若回答有误，先看检索片段，再检查切分与提示词。引用编号也需要人工核对，提示词约束不能保证消除幻觉。

从演示存储走向生产

InMemoryDocumentStore适合单进程原型，本例没有配置持久化，进程结束后不能依赖其中的数据。生产环境应选择支持所需持久化、过滤和并发能力的Document Store集成，并配套相应检索器。上线前还需确定文档唯一标识、更新与删除规则，保存来源和版本，让答案能够追溯到具体资料。