1. 为什么“流水线”才是生产级 RAG 的真正分水岭
很多人第一次接触 RAG，都是从一段几十行的脚本开始的：把文档切一切、丢进向量库、检索 Top-K、拼进 Prompt、调一次模型，跑通了，觉得“RAG 不过如此”。但真正把它放到生产环境里，问题会一个接一个冒出来——检索召回忽高忽低、多轮对话里上下文越堆越乱、工具调用和知识检索互相打架、模型偶尔胡编乱造还没有兜底。这时候你会发现，RAG 的难点从来不是“能不能跑通”，而是“能不能稳定地跑、可观测地跑、可迭代地跑”。
这篇内容我想聊的，就是怎么用Haystack和LangGraph这两套东西，把 RAG 从“脚本”升级成“流水线”。核心关键词会围绕Haystack、LangGraph、RAG、LLM、上下文工程展开。Haystack 负责把检索、排序、生成这些环节做成可插拔的组件，LangGraph 负责把整个流程编排成一张带状态、带分支、带循环的图。两者结合，才能撑起所谓“生产级”的四个字。
适合谁看？如果你已经写过最基础的 RAG demo，但被召回率、上下文管理、工具调用这些问题卡住，那这篇就是写给你的。如果你还没写过 RAG，也没关系，我会把每个环节的“为什么”讲清楚，你照着思路也能搭起来。整篇会分成几个部分：先讲整体设计思路和选型逻辑，再拆核心细节和实操要点，然后是完整的落地流程，最后是我踩过的坑和排查技巧。内容偏长，建议收藏后慢慢看。
2. 整体设计与选型：Haystack 和 LangGraph 到底谁管什么
2.1 先想清楚：生产级 RAG 和 Demo 的差距在哪
Demo 版 RAG 的假设非常理想化：用户问一个问题，系统检索一次，模型答一次，结束。但生产环境里，用户会追问、会跑题、会问需要查多个知识源才能回答的问题，甚至会在一次对话里既想查文档又想让你调用工具算个数。这些场景逼着你必须引入状态管理和流程编排。
我把生产级 RAG 和 Demo 的差距归纳成四点。第一是多轮状态：对话历史、已检索的文档、已调用的工具结果，这些都需要被记住并在后续轮次里复用。第二是条件分支：不是每个问题都需要检索，有些问题直接答就行；有些问题检索一次不够，需要改写 query 再检索。第三是循环与重试：检索结果质量差的时候，要能自动改写问题、重新检索，而不是硬着头皮生成。第四是可观测与可替换：每个环节的输入输出都要能追踪，组件要能单独替换和评测。
这四点，恰好就是 Haystack 和 LangGraph 各自擅长的领域。理解了这个，选型就不会纠结。
2.2 Haystack 的定位：把 RAG 的“零件”标准化
Haystack 的核心价值在于它把 RAG 涉及的每个环节都抽象成了组件（Component），然后用**管道（Pipeline）**把它们串起来。文档转换、文本切分、Embedding、向量检索、重排序、Prompt 构建、LLM 生成，每一个都是独立组件，有明确的输入输出接口。
这种设计的好处非常实际。你可以今天用 A 家的 Embedding 模型，明天换成 B 家，只要接口对得上，管道其他部分不用动。你也可以在检索后面插一个重排序组件，对比加与不加的效果。组件化带来的可替换性，是 RAG 能够持续迭代的前提。我见过太多项目把检索和生成写死在一起，想换个模型要改半个代码库，这种结构根本没法做 A/B 测试。
Haystack 另一个容易被忽略的点是它的管道序列化能力。一个配置好的管道可以导出成 YAML，这意味着你的 RAG 流程本身是可以版本管理的。今天调好的参数组合，明天可以原样复现，这对生产环境太重要了。
2.3 LangGraph 的定位：把流程编排成一张有状态的图
如果说 Haystack 解决的是“每个零件怎么做”，那 LangGraph 解决的就是“这些零件怎么按逻辑连起来，并且记住中间发生了什么”。LangGraph 的核心模型是图（Graph）：节点（Node）代表一个处理步骤，边（Edge）代表步骤之间的流转，而整个图共享一个**状态（State）**对象。
这个状态对象是关键。它可以是对话消息列表、检索到的文档、工具调用记录，或者任何你需要跨步骤传递的数据。每个节点读取状态、处理、再写回状态。这样一来，多轮对话、条件分支、循环重试都变成了图上的自然表达。
举个具体例子。用户问“帮我对比一下这两份文档里的方案”。在 LangGraph 里，你可以设计成这样一条路径：先判断问题类型，如果是需要检索的，进入检索节点；检索后判断结果是否足够，不够就进入 query 改写节点，改写后回到检索节点形成循环；结果够了就进入生成节点。整个过程状态一直在流转，每一步的决策都有据可查。这种带条件、带循环的编排能力，是普通链式调用做不到的。
2.4 两者怎么配合：分工而不是二选一
经常有人问，Haystack 和 LangGraph 是不是重复了，选一个就行？我的看法是，它们解决的是不同层次的问题，配合起来才完整。
我的实践方案是：用 Haystack 构建检索和生成的组件层，用 LangGraph 做上层编排。具体来说，把 Haystack 的检索管道、重排序、生成器封装成 LangGraph 的节点。LangGraph 负责决定什么时候调用检索、什么时候改写 query、什么时候直接生成、什么时候调用工具。Haystack 负责把每个被调用的环节执行好。
这样分工的好处是，检索逻辑的优化和流程逻辑的优化可以分开进行。你想提升召回率，就去调 Haystack 那边的切分策略和重排序；你想优化多轮对话体验，就去调 LangGraph 那边的状态设计和分支条件。两边互不干扰，各自都能独立评测。
提示：不要一上来就把两者都堆上。如果你的场景就是单轮问答，Haystack 的管道足够用，硬套 LangGraph 只会增加复杂度。LangGraph 的价值在多轮、分支、循环这些场景才体现得出来。
3. 核心细节解析：RAG 流水线里那些决定成败的环节
3.1 文档切分：切得好不好，直接决定召回上限
切分（Chunking）是 RAG 里最容易被轻视、却最影响效果的环节。很多人默认按固定字数切，比如每 500 字一块，重叠 50 字。这个策略在结构规整的文档上勉强能用，但遇到带标题层级、带表格、带代码块的文档，就会把语义切碎。
我的经验是，切分策略要跟着文档结构走。对于 Markdown 或带标题的文档，优先按标题层级切，让每个 chunk 尽量是一个完整的语义单元。对于长段落，再按句子边界切，而不是硬按字数切。Haystack 提供了多种切分组件，可以按句子、按段落、按固定长度切，也可以组合使用。
重叠（Overlap）的设置也有讲究。重叠的目的是防止关键信息正好落在切分边界上被割裂。但重叠太大，会导致检索结果里出现大量重复内容，浪费上下文窗口。我一般把重叠设在 chunk 大小的 10% 到 20% 之间，具体看文档的句子平均长度。如果句子普遍较长，重叠就设大一点。
还有一个细节是元数据（Metadata）的保留。每个 chunk 最好带上它来自哪个文档、哪个章节、第几页。这些元数据在检索后可以用来做过滤，也可以在生成时作为引用来源展示给用户。生产环境里，用户对“这个答案从哪来的”非常在意，元数据是建立信任的基础。
3.2 检索策略：从“单路召回”到“多路融合”
最基础的检索是向量相似度检索，把 query 和文档都转成向量，算余弦相似度，取 Top-K。但纯向量检索有个明显短板：它对精确匹配不敏感。用户问一个具体的编号、人名、专有名词，向量检索可能召回一堆语义相近但没答到点上的内容。
解决办法是混合检索（Hybrid Retrieval），把向量检索和关键词检索（比如 BM25）结合起来。向量检索负责语义召回，关键词检索负责精确召回，两路结果融合后再排序。Haystack 支持把多个检索器组合，融合策略可以用加权求和，也可以用倒数排名融合（RRF）。RRF 的好处是不需要调权重，对两路结果的分数尺度不敏感，我一般优先用它。
检索之后还有一步重排序（Reranking）。初次检索为了召回率，通常会取比较多的候选，比如 20 到 50 条。但真正送进 LLM 的只能有几条，因为上下文窗口有限。重排序模型会对候选做更精细的相关性打分，把最相关的排到前面。实测下来，加一个重排序环节，最终答案质量提升非常明显，尤其是候选里混有干扰项的时候。
3.3 上下文工程：不是塞得越多越好
上下文工程（Context Engineering）是这两年越来越被重视的概念。它的核心问题是：在有限的上下文窗口里，放什么、怎么放、放多少。
我见过最常见的错误是把检索到的所有内容一股脑塞进 Prompt，觉得信息越多越好。结果适得其反：无关内容稀释了关键信息，模型注意力被分散，还容易产生幻觉。上下文工程要做的，是精选、排序、压缩。
精选就是通过重排序和阈值过滤，只保留真正相关的内容。排序是把最相关的放在 Prompt 的开头或结尾，因为模型对首尾位置的信息更敏感。压缩则是把长文档摘要成关键句，或者只提取和问题直接相关的段落。Haystack 的生成组件支持自定义 Prompt 模板，你可以在模板里控制上下文的组织方式。
还有一个容易被忽略的点是上下文的格式。给每段检索内容加上来源标记和分隔符，能帮助模型区分不同来源，减少混淆。比如用类似“文档 1：…… 文档 2：……”的结构，比把所有内容拼成一大段效果要好。
3.4 工具合约：让 LLM 知道什么时候该“动手”
工具调用（Tool Calling）是 RAG 从“问答”走向“干活”的关键一步。用户的问题不一定都能靠检索文档回答，有些需要实时计算、查数据库、调外部接口。这时候就需要 LLM 能识别出“这个问题该用工具”，并正确地构造调用参数。
这里有个概念叫工具合约（Tool Contract），指的是每个工具的名称、描述、参数结构要定义得足够清晰，让 LLM 能准确判断何时调用、怎么传参。工具描述写得好不好，直接决定调用准确率。我的经验是，描述里要写清楚三件事：这个工具能做什么、什么时候用、参数是什么含义。含糊的描述会让模型乱调或者该调不调。
LangGraph 在工具调用上的优势是，你可以把工具调用做成图上的一个节点，并且根据调用结果决定下一步走向。比如工具返回了结果，就进入生成节点；工具报错了，就进入重试或降级节点。这种基于结果的动态流转，比单纯的链式调用灵活得多。
3.5 状态设计：多轮对话的“记忆”怎么管
多轮对话里，状态管理是核心难题。用户第三轮的问题，可能依赖第一轮提到的背景。如果每轮都重新检索、重新构建上下文，不仅浪费，还容易丢失上下文连贯性。
LangGraph 的状态对象可以设计成一个结构化的字典，包含对话历史、当前检索到的文档、已调用的工具结果、当前的问题改写版本等。每个节点按需读取和更新。关键是要区分哪些状态需要跨轮保留，哪些每轮重置。对话历史通常要保留，但检索到的文档可能每轮都要重新获取，因为用户的问题变了。
还有一个技巧是对话历史的压缩。轮次多了以后，历史消息会占满上下文窗口。这时候可以用一个摘要节点，把早期对话压缩成摘要，只保留最近几轮的完整内容。这样既保留了上下文，又控制了长度。
4. 实操落地：从零搭一条可运行的 RAG 流水线
4.1 环境准备与依赖安装
先把基础环境搭起来。我习惯用虚拟环境隔离依赖，避免版本冲突。Python 版本建议 3.10 以上，因为 LangGraph 和较新的 Haystack 版本对 Python 版本有要求。
python -m venv rag_env source rag_env/bin/activate # Windows 用 rag_env\Scripts\activate pip install haystack-ai langgraph langchain-openai这里说明一下选型理由。haystack-ai是 Haystack 2.x 的包名，2.x 版本相比 1.x 做了大量重构，组件化和管道能力更强，新项目直接用 2.x。langgraph是 LangGraph 的核心包。langchain-openai用来接 OpenAI 兼容的模型接口，如果你用的是其他模型服务，换成对应的集成包即可。
注意：Haystack 1.x 和 2.x 的 API 差异很大，网上很多老教程是 1.x 的，照抄会报错。认准 2.x 的文档和写法。
4.2 用 Haystack 搭一条基础检索管道
先构建 Haystack 这边的检索能力。假设我们有一批文档，流程是：读取文档、切分、生成 Embedding、存入文档存储、检索。
from haystack import Document, Pipeline from haystack.components.preprocessors import DocumentSplitter from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack.document_stores.in_memory import InMemoryDocumentStore # 1. 准备文档 raw_docs = [Document(content="这里放你的文档内容...")] # 2. 切分：按句子切，chunk 大小 200 词，重叠 30 词 splitter = DocumentSplitter(split_by="sentence", split_length=5, split_overlap=1) split_result = splitter.run(documents=raw_docs) chunks = split_result["documents"] # 3. 生成文档向量并写入存储 doc_store = InMemoryDocumentStore() doc_embedder = SentenceTransformersDocumentEmbedder(model="BAAI/bge-small-zh-v1.5") doc_embedder.warm_up() docs_with_emb = doc_embedder.run(documents=chunks)["documents"] doc_store.write_documents(docs_with_emb) # 4. 组装检索管道 retrieval_pipeline = Pipeline() retrieval_pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder(model="BAAI/bge-small-zh-v1.5")) retrieval_pipeline.add_component("retriever", InMemoryEmbeddingRetriever(document_store=doc_store, top_k=10)) retrieval_pipeline.connect("text_embedder.embedding", "retriever.query_embedding") # 5. 测试检索 result = retrieval_pipeline.run({"text_embedder": {"text": "你的问题"}}) for doc in result["retriever"]["documents"]: print(doc.score, doc.content[:80])这段代码里几个参数值得说明。split_length=5表示每 5 个句子组成一个 chunk，split_overlap=1表示相邻 chunk 重叠 1 个句子。Embedding 模型选了bge-small-zh-v1.5，这是中文场景下性价比很高的选择，体积小、速度快、效果稳。top_k=10是初次召回数量，后面会经过重排序再筛选。
4.3 加入重排序，提升上下文质量
初次检索的 10 条结果里，真正相关的可能只有三四条。加一个重排序组件，把最相关的排到前面，再截取前几条送进生成。
from haystack.components.rankers import SentenceTransformersSimilarityRanker ranker = SentenceTransformersSimilarityRanker(model="BAAI/bge-reranker-base", top_k=4) ranked = ranker.run(query="你的问题", documents=result["retriever"]["documents"]) top_docs = ranked["documents"]top_k=4表示重排序后只保留 4 条。这个数字不是拍脑袋定的，要结合你的上下文窗口和每条 chunk 的长度来算。假设每条 chunk 平均 200 词，4 条就是 800 词，加上 Prompt 模板和对话历史，总长度要控制在模型上下文窗口的安全范围内。留出足够余量，避免超长被截断。
4.4 用 LangGraph 编排带分支和循环的流程
现在把检索能力封装成 LangGraph 的节点，编排一条完整的流程。先定义状态结构。
from typing import TypedDict, List class RAGState(TypedDict): question: str rewritten_question: str documents: List[str] answer: str retry_count: int状态里包含原始问题、改写后的问题、检索到的文档、最终答案、重试次数。重试次数用来控制循环，防止无限改写。
接下来定义节点。检索节点调用 Haystack 管道，评估节点判断检索结果是否足够，改写节点在结果不足时改写问题，生成节点产出最终答案。
from langgraph.graph import StateGraph, END def retrieve_node(state: RAGState): query = state.get("rewritten_question") or state["question"] result = retrieval_pipeline.run({"text_embedder": {"text": query}}) docs = result["retriever"]["documents"] ranked = ranker.run(query=query, documents=docs)["documents"] return {"documents": [d.content for d in ranked]} def evaluate_node(state: RAGState): # 简单策略：文档数量足够且分数达标就通过 if len(state["documents"]) >= 2: return {"retry_count": state.get("retry_count", 0)} return {"retry_count": state.get("retry_count", 0) + 1} def rewrite_node(state: RAGState): # 这里可以调用 LLM 改写问题，简化起见用拼接示意 new_q = state["question"] + " 详细说明" return {"rewritten_question": new_q} def generate_node(state: RAGState): context = "\n\n".join(state["documents"]) # 调用 LLM 生成，示意 answer = f"基于以下内容回答：{context[:200]}..." return {"answer": answer} def route_after_evaluate(state: RAGState): if len(state["documents"]) >= 2 or state.get("retry_count", 0) >= 2: return "generate" return "rewrite"然后把这些节点组装成图。
graph = StateGraph(RAGState) graph.add_node("retrieve", retrieve_node) graph.add_node("evaluate", evaluate_node) graph.add_node("rewrite", rewrite_node) graph.add_node("generate", generate_node) graph.set_entry_point("retrieve") graph.add_edge("retrieve", "evaluate") graph.add_conditional_edges("evaluate", route_after_evaluate, {"generate": "generate", "rewrite": "rewrite"}) graph.add_edge("rewrite", "retrieve") graph.add_edge("generate", END) app = graph.compile() result = app.invoke({"question": "你的问题", "retry_count": 0}) print(result["answer"])这条图里，evaluate之后有条件边：结果够就生成，不够就改写再检索，形成循环。retry_count上限设为 2，防止死循环。这就是 LangGraph 相比普通链式调用的核心优势——流程可以根据中间结果动态调整。
4.5 工具调用的接入方式
工具调用可以做成图上的一个分支。在入口处加一个判断节点，识别问题是否需要工具。如果需要，走工具节点；如果不需要，走检索节点。
def route_question(state: RAGState): q = state["question"] if "计算" in q or "多少" in q: return "tool" return "retrieve" def tool_node(state: RAGState): # 调用具体工具，示意 return {"answer": "工具调用结果"}工具节点的关键是参数校验。LLM 生成的调用参数不一定合法，要在执行前做校验，不合法就返回错误信息让模型重新生成。这一步能挡掉大量运行时异常。
5. 常见问题与排查技巧实录
5.1 检索召回率低，答案总是答非所问
这是最高频的问题。排查顺序我一般是这样：先看切分，再看 Embedding，最后看检索策略。
切分问题最常见。如果 chunk 切得太碎，单条 chunk 信息不完整，检索到了也没用。如果切得太大，一条 chunk 里混了多个主题，向量表示被稀释。我的做法是把切分后的 chunk 抽样打印出来，人工看几条，判断语义是否完整。
Embedding 模型的问题也要排查。中文场景下，用英文为主的模型效果会打折。换成中文优化过的模型，召回率往往有明显提升。另外，query 和文档要用同一个 Embedding 模型，这个低级错误我见过不止一次。
检索策略上，如果纯向量效果不好，就上混合检索。关键词检索能补上向量检索对精确匹配不敏感的短板。
5.2 多轮对话里上下文越来越乱
这个问题通常出在状态设计上。如果每轮都把全部历史消息塞进 Prompt，几轮之后上下文就被占满了，而且早期无关信息会干扰当前回答。
我的处理方式是分层管理对话历史。最近 2 到 3 轮保留完整内容，更早的对话压缩成摘要。摘要节点可以用 LLM 生成，把之前的对话要点提炼成几句话。这样既保留了背景，又控制了长度。
另外，检索到的文档不要跨轮复用。用户这轮的问题和上轮不同，上轮检索的文档大概率不相关。每轮重新检索，只在状态里保留当前轮的文档。
5.3 工具调用该调不调，或者乱调
工具调用的准确率，八成取决于工具描述写得好不好。描述太简单，模型不知道什么时候该用；描述太模糊，模型会滥用。
我的经验是给每个工具写清楚三要素：功能、触发场景、参数说明。比如一个计算器工具，描述里要写明“当用户问题涉及数值计算时使用”，参数说明里写清楚每个参数的类型和含义。描述里最好带一两个使用示例，模型对示例的敏感度很高。
还有一个技巧是在系统 Prompt 里明确工具的使用边界。告诉模型“只有涉及实时数据或计算时才调用工具，纯知识问答直接回答”。这条约束能显著减少不必要的调用。
5.4 生成结果有幻觉，引用了不存在的内容
幻觉的根源通常是上下文里没有答案，但模型被要求必须回答。解决办法有两个方向。
一是在 Prompt 里明确允许“不知道”。告诉模型“如果检索内容不足以回答问题，就明确说明无法回答，不要编造”。这一条能挡掉相当一部分幻觉。
二是在流程里加校验节点。生成答案后，用一个轻量模型或规则检查答案里的关键信息是否能在检索内容里找到出处。找不到就标记为可疑，触发重新检索或降级回复。这个校验节点会增加一点延迟，但对可信度的提升很值。
5.5 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 | 解决思路 |
|---|---|---|---|
| 召回内容不相关 | 切分过碎或过大 | 抽样检查 chunk 语义完整性 | 调整切分粒度和重叠 |
| 精确问题答不准 | 纯向量检索短板 | 检查是否命中关键词 | 引入混合检索和重排序 |
| 多轮上下文混乱 | 历史消息无节制堆积 | 查看 Prompt 长度 | 分层管理，压缩早期历史 |
| 工具该调不调 | 工具描述不清晰 | 检查工具定义 | 补全功能、场景、参数说明 |
| 答案有幻觉 | 上下文不足仍强行生成 | 检查检索是否命中 | 允许“不知道”+ 答案校验 |
| 响应越来越慢 | 上下文过长或循环过多 | 统计 token 和循环次数 | 控制上下文长度，设重试上限 |
5.6 几个我踩过的坑
第一个坑是忽略 Embedding 模型的维度匹配。换了 Embedding 模型但没重建索引，检索结果全是乱的。换模型一定要重新生成所有文档向量。
第二个坑是重排序模型和检索模型不匹配。重排序模型有自己的输入格式要求，query 和文档的拼接方式不对，打分就不准。用之前先看文档，确认输入格式。
第三个坑是LangGraph 状态里放了不可序列化的对象。状态最好只放基本类型和列表字典，放复杂对象在持久化和调试时会出问题。
第四个坑是循环没有退出条件。改写检索的循环一定要有次数上限，否则遇到检索不到的内容会一直转。我一般设 2 到 3 次上限，超过就降级回复。
6. 关于上下文工程和工具合约的一点延伸思考
上下文工程这个概念，很多人理解成“怎么把 Prompt 写好”，但我觉得它的范围更广。它其实是在回答一个系统性问题：在 LLM 的每一次调用里，如何用有限的 token 预算，传递最有效的信息。这包括检索内容的选择、对话历史的压缩、工具结果的呈现、系统指令的组织。每一个环节都在争夺那点宝贵的上下文空间。
工具合约也是类似。它不只是“定义几个函数”，而是在 LLM 和外部能力之间建立一份清晰的契约。这份契约写得好，模型就能可靠地调用；写得含糊，模型就会乱来。我现在的习惯是，每加一个工具，都先想清楚它的边界：什么情况下用、什么情况下不用、参数怎么传、出错怎么反馈。把这些想明白了再写代码，比写完再调要省事得多。
Haystack 和 LangGraph 的组合，本质上是在给这套复杂的上下文和流程管理提供工程化的支撑。Haystack 让每个信息处理环节可替换、可评测，LangGraph 让整个决策流程可编排、可观测。两者配合，RAG 才真正从“能跑”变成“能扛”。
最后分享一个我最近在用的调试技巧：把 LangGraph 每一步的状态变化都打日志，包括进入节点时的状态和离开节点时的状态。跑几条典型问题，把日志拉出来看，流程哪里绕了、哪里状态没更新、哪里检索结果被丢弃了，一目了然。这个习惯帮我定位过好几个隐蔽的流程 bug，比盯着代码看效率高得多。
