中级

用 Haystack 2.x 搭一套可调试的 RAG 检索增强流水线

2026.09.16· 6 个步骤 · 18 分钟阅读· 🔍 Haystack

想给 Agent 接上「会查资料」的能力，最稳的做法是先把 RAG（检索增强）跑成一条可调试的流水线。Haystack 2.x 用声明式的 Pipeline 把文档切分、向量化、向量库检索、Prompt 拼装、大模型生成串成可复用组件，每个环节都能单独打印结果排查问题。本教程从零搭一条 Chroma 版 RAG 流水线。

🔍 本教程适合：做知识库问答/RAG 的开发者。需要基础 Python 与一点向量检索概念。

先搞懂：2.x 与旧版的关键区别

Haystack 2.0 起移除了旧的 Pipeline / PromptNode / InMemoryDocumentStore 等类，改为 @component 装饰器 + 现代 Pipeline。网上不少老教程会直接抛 ImportError，本教程全部用 2.x 现行写法。

环节	2.x 组件
文档嵌入	SentenceTransformersDocumentEmbedder
向量库	ChromaDocumentStore（集成包）
检索	ChromaEmbeddingRetriever
生成	OpenAIChatGenerator + ChatPromptBuilder

Step 1：安装依赖

1 装对版本

pip install "haystack-ai>=2.0.0" chromadb sentence-transformers openai

版本陷阱：旧组合会拉到 pre-2.0 包导致 ImportError。务必确认 haystack-ai 是 2.x，且 Chroma 走官方集成包 haystack_integrations，而非老的 langchain 桥接。

Step 2：准备文档与 Document Store

2 建向量库

用 Chroma 集成包建一个 Document Store（内存或持久化），后面索引与查询共用同一实例。

from haystack.dataclasses import Document
from haystack_integrations.document_stores.chroma import ChromaDocumentStore

document_store = ChromaDocumentStore(
    collection_name="my_rag",
    persistence_path="./chroma_data",   # 持久化到磁盘
)
docs = [
    Document(content="我们的退款周期是 7 个工作日。", meta={"title":"退款政策"}),
    Document(content="标准配送需要 3 到 5 天。", meta={"title":"配送说明"}),
]

💡 真实场景用 TextFileToDocument / PyPDFToDocument 等转换器读文件，再用 DocumentSplitter 切分，比手工塞字符串稳得多。

Step 3：建索引流水线

3 把文档变成向量

索引流水线负责：文档 → 嵌入 → 写库。两个组件用 connect 串起来。

from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.components.writers import DocumentWriter

indexing = Pipeline()
indexing.add_component("embedder", SentenceTransformersDocumentEmbedder())
indexing.add_component("writer", DocumentWriter(document_store))
indexing.connect("embedder.documents", "writer.documents")
indexing.run({"embedder": {"documents": docs}})

Step 4：建查询流水线（RAG 核心）

4 检索 + 拼 Prompt + 生成

查询流水线四件套：查询嵌入 → Chroma 检索 → 拼装上下文 → 大模型生成。连接关系要一一对应。

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack_integrations.components.retrievers.chroma import ChromaEmbeddingRetriever
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses.chat_message import ChatMessage

retriever = ChromaEmbeddingRetriever(document_store, top_k=2)
prompt = [
  ChatMessage.from_user(
    "根据上下文回答。若没有答案就说「未找到」。\n"
    "上下文：\n{% for d in documents %}{{ d.content }}\n{% endfor %}\n"
    "问题：{{ query }}"
  )
]
querying = Pipeline()
querying.add_component("q_embed", SentenceTransformersTextEmbedder())
querying.add_component("retriever", retriever)
querying.add_component("prompt", ChatPromptBuilder(template=prompt))
querying.add_component("llm", OpenAIChatGenerator())
querying.connect("q_embed.embedding", "retriever.query_embedding")
querying.connect("retriever.documents", "prompt.documents")
querying.connect("prompt", "llm")

必填变量：ChatPromptBuilder 的模板变量（documents / query）建议显式声明为必填，否则多分支时可能静默不执行。模板里的循环用 Jinja 语法。

Step 5：跑通并调试检索质量

5 看中间结果

先单独打印检索到的文档与分数，确认「找得对」，再信生成的答案。

res = querying.run({"q_embed":{"text":"退款要多久"},
                    "prompt":{"query":"退款要多久"}})
for d in res["retriever"]["documents"]:
    print(d.score, d.meta, d.content[:50])
print(res["llm"]["replies"][0].text)

🔧 检索质量差先调这三处：切分粒度、嵌入模型、top_k。生成胡说八道多半是检索没命中，别急着怪模型。

Step 6：包成 Agent 工具与评估

6 接入 Agent 并量化

把上面的查询流水线封装成一个工具函数，Agent 需要查资料时调用；并用 Ragas / DeepEval 跑自动评估，盯住「检索命中率」和「答案忠实度」。

def rag_tool(question: str) -> str:
    r = querying.run({"q_embed":{"text":question},
                      "prompt":{"query":question}})
    return r["llm"]["replies"][0].text

# 评估（以官方文档 API 为准）
# from ragas import evaluate ...  # 关注 faithfulness / answer_relevancy

成本与选型：嵌入模型决定检索质量与费用，生成模型决定答案质量与费用，两者独立调优；评估指标要落到可观测看板，别只在笔记本里跑一次。具体评估库版本以官方文档为准。

常见问题速查

现象	原因与解决
ImportError	装了 pre-2.0 老包，重装 haystack-ai 2.x
连接报 key 错	OpenAIChatGenerator 需要 OPENAI_API_KEY 环境变量
检索为空	索引流水线没跑 / collection 名不一致
答案不引用资料	top_k 太小或切分把关键信息拆散

📚 更多教程 📰 智能体资讯 🔍 RAG 📚 向量库 🧩 Haystack ⚙️ 可调试