生产级RAG实战：Haystack混合检索与LangGraph工具合约设计

1. 从"能跑通"到"敢上线"：生产级 RAG 的分水岭在哪里

很多人第一次用 Haystack 或 LangGraph 搭 RAG，跑通一个"上传 PDF 然后问答"的 Demo 只花了半小时，于是觉得这事成了。等到真正要接入业务、面对真实用户的时候，问题才一个接一个冒出来：检索结果时好时坏、工具调用偶尔死循环、上下文塞满了却答非所问、同一个问题两次回答不一致。这些不是"代码写错了"，而是 Demo 和生产系统之间那条看不见的分水岭。

这篇是系列的第二部分，重点不再是"怎么把 RAG 跑起来"，而是"怎么把它做成一个敢让真实用户用的系统"。核心围绕三件事展开：生产级 RAG 的检索质量工程、工具合约（Tool Contract）的设计与约束、以及上下文工程（Context Engineering）的落地方法。技术栈依然是 Haystack 负责检索与流水编排、LangGraph 负责有状态的 Agent 流程控制。

适合谁看？如果你已经写过最基础的 RAG 链路，知道 embedding、向量库、prompt 大概是怎么回事，但一到"检索命中率上不去""Agent 老是乱调工具""上下文窗口不够用"这些具体问题上就卡住，那这篇就是写给你的。我会尽量把每个设计决策背后的"为什么"讲清楚，而不是甩一段代码让你自己猜。

先说一个我在实际项目里反复验证过的判断：RAG 系统的质量瓶颈，八成不在模型，而在检索和上下文组织。模型换大一号，效果可能提升 5%；但把检索策略和上下文结构做对，效果能翻倍。这也是为什么这个系列要花大量篇幅讲 Haystack 的检索流水和 LangGraph 的状态管理——它们才是决定上限的地方。

2. Haystack 检索流水：把"召回"当成一门工程来做

2.1 为什么单一向量检索在生产环境必然翻车

Demo 阶段大家通常只用一路 dense retrieval（稠密向量检索），把 query 编码成向量，去向量库里找最近邻。这在语义相近的场景下确实好用，但生产环境的 query 是"脏"的：有缩写、有错别字、有专有名词、有精确的编号或型号。这时候纯向量检索的短板就暴露了——它对精确匹配天然不敏感。

举个我踩过的真实例子：用户问"XX-2000 型号的额定功率是多少"，向量检索可能召回一堆"功率相关"的段落，但就是漏掉了那个真正写着"XX-2000"的表格。因为向量空间里"XX-2000"和"额定功率"的语义距离，未必比"XX-3000 的功率"更近。

解决办法是混合检索（Hybrid Retrieval）：dense 负责语义召回，sparse（如 BM25）负责关键词精确召回，两路结果用融合算法合并。Haystack 里做这件事非常自然，它把 retriever 抽象成了可组合的组件。

from haystack import Pipeline from haystack.components.retrievers.in_memory import InMemoryBM25Retriever, InMemoryEmbeddingRetriever from haystack.components.joiners import DocumentJoiner pipeline = Pipeline() pipeline.add_component("bm25", InMemoryBM25Retriever(document_store=store, top_k=10)) pipeline.add_component("embed", InMemoryEmbeddingRetriever(document_store=store, top_k=10)) pipeline.add_component("joiner", DocumentJoiner(join_mode="reciprocal_rank_fusion")) pipeline.connect("bm25.documents", "joiner.documents") pipeline.connect("embed.documents", "joiner.documents")

这里的关键选择是join_mode。常见的有两种：concatenate简单拼接去重，reciprocal_rank_fusion（RRF）按排名倒数加权融合。生产环境我强烈建议用 RRF，因为它不依赖两路分数的绝对量纲——BM25 的分数和向量相似度分数根本不在一个尺度上，直接比大小是没有意义的，而 RRF 只看排名，天然规避了这个问题。

2.2 top_k 不是拍脑袋定的，要算一笔账

很多人 top_k 随手写个 5 或者 10。这个数字其实应该由下游决定。假设你的上下文窗口是 8k token，每个 chunk 平均 300 token，那么理论上最多能塞 20 多个 chunk。但你不能全塞满，因为还要留给系统提示、对话历史、以及模型生成的空间。

我的经验公式是：

可用 chunk 数 = (上下文窗口 - 系统提示 - 历史预留 - 生成预留) / 平均 chunk 长度

按 8k 窗口算：系统提示 500、历史预留 1500、生成预留 1000，剩 5000 token 给检索内容，除以 300 约等于 16 个 chunk。但这是上限，实际召回后还要经过 rerank 和过滤，所以召回阶段 top_k 可以设大一点（比如 20-30），精排后再砍到 5-8 个。召回要"宁滥勿缺"，精排才负责"去伪存真"。

2.3 Rerank 是性价比最高的一步优化

如果只能做一件事来提升 RAG 质量，我会选加 reranker。原因很简单：召回模型（bi-encoder）为了速度，把 query 和 document 分别编码，两者之间没有交互；而 reranker（cross-encoder）把 query 和 document 拼在一起过模型，能捕捉细粒度的相关性。代价是慢，所以只能用在少量候选上。

Haystack 里接一个 reranker 组件就行：

from haystack.components.rankers import TransformersSimilarityRanker ranker = TransformersSimilarityRanker(model="BAAI/bge-reranker-base", top_k=6) pipeline.add_component("ranker", ranker) pipeline.connect("joiner.documents", "ranker.documents")

实测下来，在中文知识库场景里，加 reranker 后 top-3 命中率通常能从 60% 出头提到 80% 以上。这个提升幅度，比换更大的生成模型划算得多。选型上，bge-reranker系列对中文友好，如果追求更轻量可以用bge-reranker-v2-m3，多语言场景表现稳定。

注意：reranker 的 top_k 和召回阶段的 top_k 是两个概念。召回 top_k 是"给 reranker 多少候选"，reranker 的 top_k 是"最终留给生成模型多少"。别把这两个搞混，否则要么 reranker 没料可排，要么生成模型被塞爆。

2.4 分块策略：别再用固定长度硬切了

chunk 切分是 RAG 里最容易被忽视、却影响巨大的一环。固定 512 字符硬切的问题在于，它会把一个完整的语义单元拦腰截断——比如一个表格被切成两半，一段论证被切掉结论。

我在项目里常用的策略是递归分块 + 语义边界优先：优先按段落切，段落太长再按句子切，句子还长才按字符切。Haystack 的DocumentSplitter支持按split_by="sentence"或split_by="word"并配合split_length和split_overlap。

from haystack.components.preprocessors import DocumentSplitter splitter = DocumentSplitter( split_by="sentence", split_length=4, # 每块约 4 句 split_overlap=1, # 相邻块重叠 1 句，防止边界信息丢失 )

split_overlap这个参数值得多说一句。它让相邻 chunk 有重叠内容，好处是跨块的信息不会因为切分而彻底断裂。代价是存储和检索时会有冗余。我的经验是重叠 10%-20% 比较合适，太少起不到衔接作用，太多则检索结果高度重复、浪费上下文。

3. LangGraph 工具合约：让 Agent 调用工具不再"放飞自我"

3.1 工具调用的本质是一份"合约"

LangGraph 里让 LLM 调用工具，很多人直接丢一个函数描述就完事。但在生产环境，工具调用必须被当成一份**合约（Contract）**来设计：输入是什么、输出是什么、什么情况下允许调用、调用失败怎么办，全都要明确。

为什么强调"合约"？因为 LLM 是概率性的，它可能传错参数、可能在不该调用的时候调用、可能对同一个工具反复调用陷入循环。合约的作用就是用确定性的约束去包裹不确定性的模型行为。

一份好的工具合约至少包含四要素：

3.2 用 Pydantic 把参数 schema 钉死

LangGraph 的工具定义通常基于函数签名或 Pydantic 模型。我强烈建议用 Pydantic，因为它能在运行时做类型校验，模型传错参数会直接报错而不是悄悄传下去。

from pydantic import BaseModel, Field from typing import Literal class SearchInput(BaseModel): query: str = Field(description="要检索的自然语言问题，必须是完整问句") scope: Literal["product", "policy", "faq"] = Field( description="检索范围，只能从这三个里选" ) top_k: int = Field(default=5, ge=1, le=20, description="返回条数，1到20之间") def search_knowledge(inp: SearchInput) -> list[dict]: ...

注意scope用了Literal而不是str。这一步很关键——它把"模型可以自由发挥"变成了"模型只能从枚举里选"。实测下来，用Literal约束后，模型选错范围的概率大幅下降。top_k用ge和le卡住范围，防止模型传个 1000 把上下文撑爆。

3.3 前置条件校验：在工具执行前拦一道

光有参数校验还不够，还要校验状态。比如"退款"工具，只有在订单已支付且未退款的状态下才能调用。这类校验放在工具函数内部最前面：

def refund_order(order_id: str, state: dict) -> dict: order = state.get("orders", {}).get(order_id) if not order: return {"error": "订单不存在", "code": "ORDER_NOT_FOUND"} if order["status"] != "paid": return {"error": f"订单状态为 {order['status']}，不可退款", "code": "INVALID_STATE"} ...

这里有个设计要点：错误要以结构化形式返回，而不是抛异常。因为异常会中断整个图流程，而结构化错误可以回传给模型，让它自己决定下一步（比如换个订单号重试，或者告诉用户原因）。这其实就是"工具合约"里返回契约的部分。

3.4 防死循环：给工具调用加上"刹车"

Agent 最让人头疼的问题之一就是死循环——模型反复调用同一个工具，每次都拿到相似结果，却始终不收敛。LangGraph 提供了几种刹车机制。

第一种是递归限制。编译图的时候可以设置：

app = graph.compile() result = app.invoke(input_state, config={"recursion_limit": 25})

超过 25 步就强制停止。这个数字要结合你的图复杂度来定，太小会误杀正常流程，太大则失去保护意义。

第二种是状态里记录调用历史，在路由函数里判断：

def should_continue(state: dict) -> str: history = state.get("tool_calls", []) # 同一个工具连续调用超过 3 次，强制转人工 if len(history) >= 3 and len(set(h["name"] for h in history[-3:])) == 1: return "fallback" return "continue"

第三种更优雅：在工具返回里加入"是否取得新信息"的标记。如果连续两次调用返回的内容高度相似，说明模型在原地打转，直接触发兜底逻辑。这个思路我在实际项目里用过，比单纯数次数更精准。

4. 上下文工程：把有限的窗口花在刀刃上

4.1 上下文不是越多越好，而是越"对"越好

"上下文工程"这个词这两年很火，但很多人理解成了"往 prompt 里塞更多东西"。恰恰相反，上下文工程的核心是做减法——在有限的窗口里，只放对当前决策真正有用的信息。

有个反直觉的实测结论：给模型塞 20 个 chunk，效果往往不如塞 5 个精选 chunk。原因是长上下文里存在"中间遗忘"现象——模型对开头和结尾的信息记得牢，中间部分容易被忽略。塞得越多，真正有用的信息越可能被淹没在中间。

所以上下文工程的第一原则是：先精排，再组装，能砍就砍。

4.2 上下文的分层组装

我在项目里通常把上下文分成四层，按优先级从高到低排列：

1. 系统指令层：角色定义、输出格式要求、安全边界。这层永远在最前面，且尽量精简。
2. 任务上下文层：当前要解决的问题、已知条件、约束。这层是动态的，随对话推进更新。
3. 检索证据层：rerank 后的 top-k chunk，每个 chunk 带上来源标识。
4. 对话历史层：最近几轮对话，通常只保留最近 3-5 轮，更早的做摘要压缩。

组装顺序上，把最重要的信息放在开头和结尾，中间放次要的。这是利用了模型对首尾位置更敏感的特性。

4.3 对话历史的压缩：摘要 + 关键实体

多轮对话里，历史会迅速吃掉窗口。全量保留不现实，直接丢弃又会丢失上下文。我的做法是滚动摘要 + 关键实体表。

滚动摘要：每积累 N 轮对话，就用一次 LLM 调用把这段历史压缩成一段摘要，替换掉原始对话。关键实体表则是一个结构化的字典，记录对话中出现的重要实体（订单号、产品名、用户诉求等），每轮更新。

state = { "summary": "用户咨询 XX-2000 的功率，已告知为 2000W，用户接着问保修政策", "entities": {"product": "XX-2000", "intent": "保修咨询"}, "recent_turns": [...] # 最近 3 轮原文 }

这样组装 prompt 时，摘要提供背景，实体表提供精确锚点，最近几轮提供即时语境。三者结合，比单纯堆历史高效得多。

4.4 检索证据的"引用标注"

生产级 RAG 有个硬要求：答案要能溯源。所以每个 chunk 塞进上下文时，都要带上编号，并要求模型在回答里标注引用来源。

context = "\n\n".join( f"[{i+1}] 来源：{doc.meta['source']}\n{doc.content}" for i, doc in enumerate(docs) )

然后在系统提示里明确要求："回答时用 [1][2] 标注依据的片段编号。" 这样做有两个好处：一是用户能验证答案可信度，二是当模型答错时，你能快速定位是检索错了还是生成错了。这个可观测性在生产环境里价值极高。

5. 把 Haystack 和 LangGraph 缝在一起：谁负责什么

5.1 职责边界：检索归 Haystack，编排归 LangGraph

这两个框架放一起，最容易犯的错是职责混乱——有人在 LangGraph 节点里手写检索逻辑，也有人试图用 Haystack pipeline 去做复杂的条件分支。正确的分工应该是：

• Haystack 负责"无状态的检索与处理"：文档预处理、embedding、混合检索、rerank、prompt 组装。它是一个纯粹的"输入 query 输出证据"的流水线。
• LangGraph 负责"有状态的决策与编排"：维护对话状态、决定何时检索、何时调工具、何时结束、如何处理错误。它是一个状态机。

把检索封装成一个 LangGraph 节点，节点内部调用 Haystack pipeline：

def retrieve_node(state: dict) -> dict: query = state["current_query"] result = haystack_pipeline.run({"embed": {"query": query}, "bm25": {"query": query}}) docs = result["ranker"]["documents"] return {"retrieved_docs": docs, "context": build_context(docs)}

这样 Haystack 的检索能力被"原子化"成一个可复用的节点，LangGraph 则专注于流程控制。两者解耦，各自升级互不影响。

5.2 状态设计：LangGraph 的 state 是系统的"记忆"

LangGraph 的 state 是整个系统的中枢。设计 state 时要有前瞻性，把可能用到的东西都预留好：

from typing import TypedDict, Annotated from operator import add class RAGState(TypedDict): messages: Annotated[list, add] # 对话消息，用 add 累加 current_query: str # 当前待处理的 query retrieved_docs: list # 检索到的文档 tool_calls: Annotated[list, add] # 工具调用历史 summary: str # 滚动摘要 entities: dict # 关键实体 retry_count: int # 重试计数

Annotated[list, add]这个写法是 LangGraph 的精髓之一——它定义了当多个节点都往这个字段写数据时，如何合并。用add表示追加，这样消息和工具调用历史就能自然累积，不用手动管理。

5.3 条件路由：让流程"会拐弯"

LangGraph 相比普通 chain 的最大优势，就是能根据状态动态决定下一步走哪。这是通过条件边（conditional edge）实现的：

def route_after_retrieval(state: dict) -> str: docs = state.get("retrieved_docs", []) if not docs: return "rewrite_query" # 没检索到，改写 query 重试 if state.get("retry_count", 0) > 2: return "fallback" # 重试太多次，兜底 return "generate" # 正常生成 graph.add_conditional_edges( "retrieve", route_after_retrieval, {"rewrite_query": "rewrite", "generate": "generate", "fallback": "fallback"} )

这个"检索不到就改写 query 重试"的回路，是生产级 RAG 的标配。用户的问题表述往往和文档表述不一致，一次检索不中很正常，给系统一两次自我修正的机会，命中率能明显提升。

6. 上线前必须压测的几件事

6.1 检索质量：用命中率而不是"感觉"来衡量

"感觉检索效果还行"是上线大忌。必须用数据说话。核心指标是Hit Rate@k：在测试集里，正确文档出现在 top-k 结果中的比例。

构造测试集的方法：从真实业务里挑 100-200 个问题，人工标注每个问题的正确答案来源文档。然后跑检索，统计命中率。这个工作枯燥但值得——它是你后续所有优化的基准线。

我一般会同时看三个数：Hit Rate@5、Hit Rate@10、以及 MRR（平均倒数排名）。Hit Rate 看"有没有找到"，MRR 看"排得够不够前"。两个都达标，检索才算过关。

6.2 工具调用的稳定性：跑 100 次看方差

LLM 是概率性的，同一个输入跑两次结果可能不同。工具调用尤其如此。上线前我会做一件事：把同一批测试用例跑 100 次，统计工具调用的成功率、平均调用次数、以及异常率。

如果某个工具的成功率只有 70%，那说明要么描述不清，要么参数约束不够，要么前置条件太严。这个方差数据比单次跑通有价值得多。

6.3 上下文长度：留足余量，别卡着上限

上线前一定要测最坏情况下的上下文长度：最长的 query + 最多的检索结果 + 最长的对话历史。如果这个组合逼近甚至超过窗口上限，线上迟早会出问题。

我的做法是设一个"软上限"，比如窗口的 80%。一旦组装后的上下文超过软上限，就触发裁剪逻辑——优先砍对话历史，其次砍低分的检索 chunk。宁可少给点信息，也不能让请求直接失败。

7. 几个我踩过的坑和对应的解法

第一个坑是reranker 拖慢响应。cross-encoder 是串行计算的，候选一多就慢。解法是控制候选数量（召回 top_k 别超过 30），并且 reranker 用 GPU 或者量化版本。如果延迟实在压不下来，可以考虑用更轻量的模型，或者只在"检索结果分数接近"时才启用 rerank。

第二个坑是工具描述里的示例误导模型。我在工具描述里写了个示例参数，结果模型不管什么情况都照着示例填。后来把示例删掉，只保留参数说明，反而正常了。教训是：工具描述要精确，但别给太具体的示例，否则模型会偷懒照抄。

第三个坑是状态字段命名冲突。LangGraph 的 state 里我一开始用了messages和history两个字段，结果两个节点分别往不同字段写，最后组装时数据对不上。后来统一成一个字段，用Annotated管理合并，问题消失。state 字段宁少勿多，一个概念只用一个字段。

第四个坑是检索到的 chunk 里混入了导航栏、页脚这类噪声。PDF 解析出来的文本经常带这些。解法是在预处理阶段做清洗，用规则过滤掉过短、重复度高、或者明显是模板的段落。这一步做扎实，后面检索质量能省很多事。

8. 关于"上下文工程"的一点个人体会

做了几个 RAG 项目之后，我越来越觉得上下文工程是一门"取舍的艺术"。模型能力再强，窗口也是有限的；检索再全，噪声也是存在的。真正决定系统好不好用的，是你在有限资源下把什么信息放到了模型面前。

我现在的习惯是：每次调 prompt 之前，先把组装好的上下文打印出来，自己读一遍。如果连我自己都觉得"这段信息对回答这个问题没用"，那就该砍。模型不会比你更聪明地筛选信息，筛选是你的活，不是它的活。

还有一点：别迷信"更大的模型能解决一切"。我见过太多团队一遇到效果不好就想着换模型，结果换了之后发现瓶颈根本在检索。先把 Haystack 的检索流水调顺，把 LangGraph 的状态和路由设计清楚，把上下文组装做精，这些做到位了，中等规模的模型也能跑出很好的效果。反过来，检索和上下文一塌糊涂，再大的模型也救不回来。

这套组合我目前用在几个知识库问答和智能客服场景里，稳定性还不错。后续如果要做多模态检索（比如图片、表格），Haystack 的组件化设计也能比较平滑地扩展进去——把新的 retriever 接进 pipeline，LangGraph 那边几乎不用动。这大概就是"检索归 Haystack、编排归 LangGraph"这个分工带来的最大好处：每一层都能独立演进，而不会牵一发动全身。