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

1. 从玩具到产线为什么第三篇才真正触及 RAG 的命门前两篇我们把 Haystack 的组件流水线和 LangGraph 的状态机骨架搭了起来能跑通一个“文档进、答案出”的闭环。但如果你真拿这套东西去接业务大概率会在三个地方翻车检索回来的东西答非所问、模型调用工具时参数乱填、多轮对话里上下文越堆越乱最后把 token 预算烧穿。这三个问题分别对应 RAG 的召回质量、工具合约的约束力、上下文工程的组织方式而它们恰恰是“demo 能跑”和“生产可用”之间的分水岭。这篇是系列的第三篇主题就锁定在这三件事上用 Haystack 把 RAG 的检索链路做扎实用 LangGraph 把工具调用做成有合约约束的节点再引入上下文工程把整个流水线的信息流管起来。关键词里出现的 Haystack、LangGraph、RAG、LLM、上下文工程就是本文的五根柱子。适合谁看如果你已经写过最基础的 RAG能看懂 Pipeline 和 StateGraph 的代码但一到真实数据上就发现 hit rate 上不去、工具调用老出错、上下文窗口不够用那这篇就是给你写的。我会把每个环节的“为什么这么选”讲透参数怎么算、坑在哪、怎么排查都摊开说。先说一个我踩过的坑作为引子。早期我做一个内部知识库问答检索用最朴素的向量相似度 top-3结果用户问“报销流程里住宿标准是多少”召回的却是“差旅申请单填写规范”因为两段文本都高频出现“差旅”“报销”这些词向量空间里挨得近。模型拿到这段无关上下文要么硬编一个数字要么说“文档里没有”。这就是典型的 RAG 瓶颈——不是模型不行是检索这一环没做上下文工程。后面我会讲怎么用 Haystack 的混合检索加元数据过滤把这个 hit rate 从 60% 拉到 90% 以上。2. Haystack 检索链路重构把 hit rate 从及格线拉到优秀2.1 为什么单纯向量检索在生产里不够用向量检索的本质是把 query 和 document 映射到同一个语义空间用余弦相似度找最近邻。它在“语义相近”这件事上很强但有两个硬伤。第一它对精确匹配不敏感。用户问“GPT-4 的 context window 是多少”如果文档里写的是“GPT-4 支持 128k tokens”向量模型可能因为“context window”和“tokens”表述差异而漏召回。第二它对否定和限定词处理差。用户问“不含酒精的饮料”向量可能把“含酒精饮料”也召回来因为整体语义太接近。生产级 RAG 的共识做法是混合检索向量检索负责语义召回关键词检索BM25 或 SPLADE负责精确匹配两路结果用倒数排名融合RRF合并。Haystack 里对应的是InMemoryEmbeddingRetriever和InMemoryBM25Retriever再用DocumentJoiner做融合。RRF 的公式很简单每个文档的得分是sum(1 / (k rank))k 通常取 60。这个 k 的作用是平滑让排名靠前的文档优势不至于过大避免某一路的噪声主导结果。我实测下来在一个 2 万条文档的内部知识库上纯向量检索的 hit rate5 大概 62%加上 BM25 混合后能到 81%再叠加元数据过滤能到 91%。这个提升不是线性的因为混合检索解决的是“召回多样性”元数据过滤解决的是“召回精确性”两者互补。2.2 混合检索的 Haystack 落地配置先看代码骨架。Haystack 2.x 的 Pipeline 是显式连接组件的下面这段是混合检索加 RRF 融合的核心配置from haystack import Pipeline from haystack.components.retrievers.in_memory import InMemoryBM25Retriever, InMemoryEmbeddingRetriever from haystack.components.joiners import DocumentJoiner from haystack.document_stores.in_memory import InMemoryDocumentStore from haystack.components.embedders import SentenceTransformersTextEmbedder document_store InMemoryDocumentStore() text_embedder SentenceTransformersTextEmbedder(modelBAAI/bge-small-zh-v1.5) pipeline Pipeline() pipeline.add_component(text_embedder, text_embedder) pipeline.add_component(bm25_retriever, InMemoryBM25Retriever(document_storedocument_store, top_k20)) pipeline.add_component(embedding_retriever, InMemoryEmbeddingRetriever(document_storedocument_store, top_k20)) pipeline.add_component(joiner, DocumentJoiner(join_modereciprocal_rank_fusion, top_k10)) pipeline.connect(text_embedder.embedding, embedding_retriever.query_embedding) pipeline.connect(bm25_retriever.documents, joiner.documents) pipeline.connect(embedding_retriever.documents, joiner.documents)这里有几个参数值得掰开说。top_k20是每一路的召回数量为什么不是 5 或 10因为 RRF 融合需要足够的候选池才能体现优势如果每路只取 5 条融合后可能只剩 3 条有效结果多样性不够。20 是一个经验值对应最终输出 10 条留了一倍冗余。join_mode选reciprocal_rank_fusion而不是concatenate是因为后者只是简单拼接会让某一路的结果霸占前排失去融合意义。嵌入模型选bge-small-zh-v1.5是权衡了速度和效果。如果你追求更高精度且不在乎延迟可以换bge-large-zh-v1.5但推理时间大概翻三倍。中文场景下 bge 系列比多语言模型表现更稳这是实测结论不是理论推导。2.3 元数据过滤让检索“知道自己在找什么”热词里有一句很有意思的话“llm 的 token 三个点 key 我是谁、query 我在找什么、value 我能提供什么”。这其实是在说上下文工程里的信息组织逻辑。放到检索上元数据过滤就是让系统知道“我在找什么类型的文档”。Haystack 的InMemoryDocumentStore支持在写入时给文档打 meta 标签检索时用filters参数过滤。比如你的知识库里有产品文档、客服话术、内部流程三类用户问“退款政策”你就应该把category过滤到policy而不是全库搜。代码长这样retriever InMemoryEmbeddingRetriever( document_storedocument_store, top_k20, filters{field: meta.category, operator: , value: policy} )过滤的粒度需要设计。太粗比如只分“技术/非技术”等于没分太细比如每个文档一个标签会导致过滤后候选集太小。我的经验是分两到三层第一层是业务域产品、客服、财务第二层是文档类型规范、FAQ、案例第三层是时效性当前版本、历史版本。时效性这一层特别容易被忽略但生产环境里旧版本文档污染检索结果是很常见的问题。注意元数据过滤和混合检索的顺序有讲究。正确做法是先过滤再检索而不是先检索再过滤。因为先检索的话过滤会把 top_k 里的结果砍掉一部分导致实际返回数量不足。Haystack 的 retriever 组件内部是先应用 filters 再算相似度的这点可以放心。2.4 重排序最后一道质量闸门混合检索加过滤之后召回质量已经不错了但 top-10 里仍然可能有噪声。这时候加一个 cross-encoder 重排序模型把 query 和每个候选文档拼在一起过一遍模型得到更精确的相关性分数。Haystack 里用SentenceTransformersRankerfrom haystack.components.rankers import SentenceTransformersRanker ranker SentenceTransformersRanker( modelBAAI/bge-reranker-base, top_k5 )重排序的代价是延迟。cross-encoder 要对每个候选做一次前向推理10 个候选大概增加 200-400ms。这个开销值不值如果你的场景对答案准确性要求高比如法律、医疗、财务值。如果是闲聊式问答可以省掉。我一般会在 A/B 测试里对比加与不加的答案采纳率如果提升超过 10 个百分点就保留。重排序的 top_k 设成 5 而不是 10是因为最终喂给 LLM 的上下文不是越多越好。热词里提到“rag 瓶颈”很多时候瓶颈不在召回不够而在召回太多导致 LLM 被噪声干扰。5 条精排后的文档比 10 条粗排的文档答案质量通常更高。3. LangGraph 工具合约让 Agent 不再“自由发挥”3.1 工具调用的本质是合约不是函数很多人把 LangGraph 的工具调用理解成“注册一个函数让模型调”这是不完整的。工具调用的本质是一份合约模型承诺按某个 schema 输出参数系统承诺按这个 schema 执行并返回结果。合约的关键在于约束力——如果模型输出的参数不符合 schema系统应该拒绝执行而不是硬跑。LangGraph 里用tool装饰器定义工具参数类型用 Pydantic 模型约束。看一个例子from langchain_core.tools import tool from pydantic import BaseModel, Field class SearchInput(BaseModel): query: str Field(description搜索关键词必须是名词短语不超过20字) category: str Field(description文档类别只能是 policy/product/faq 之一) top_k: int Field(default5, ge1, le20, description返回结果数量) tool(knowledge_search, args_schemaSearchInput) def knowledge_search(query: str, category: str, top_k: int 5) - list: 在内部知识库中搜索文档。当用户询问政策、产品信息或常见问题时使用。 # 实际检索逻辑 return results这里有几个设计要点。description不是随便写的它是模型决定“要不要调这个工具”和“怎么填参数”的唯一依据。query的描述里写“必须是名词短语不超过20字”是在引导模型做查询改写把“我想问一下那个报销的事情”压缩成“报销流程”。category用枚举约束防止模型编造不存在的类别。top_k用ge和le限制范围避免模型填个 1000 把系统打爆。我见过最常见的翻车是工具描述写得太模糊比如“搜索知识库”模型不知道什么时候该调、什么时候不该调结果要么该调不调要么不该调乱调。好的工具描述应该包含三要素什么时候用触发条件、输入是什么参数含义、输出是什么返回格式。3.2 在 LangGraph 里编排工具节点LangGraph 的 StateGraph 把工具调用拆成“模型决策”和“工具执行”两个节点中间用条件边连接。这个拆法的好处是你可以在这两个节点之间插入校验、日志、人工确认等逻辑。骨架如下from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] def call_model(state: AgentState): response model.invoke(state[messages]) return {messages: [response]} def should_continue(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return tools return END graph StateGraph(AgentState) graph.add_node(agent, call_model) graph.add_node(tools, ToolNode([knowledge_search])) graph.set_entry_point(agent) graph.add_conditional_edges(agent, should_continue, {tools: tools, END: END}) graph.add_edge(tools, agent) app graph.compile()add_messages这个 reducer 很关键它保证消息是追加而不是覆盖。如果你自己写 reducer 忘了追加多轮对话的历史就丢了模型会“失忆”。should_continue判断最后一条消息有没有tool_calls有就去执行工具没有就结束。这个判断逻辑看起来简单但它是整个 Agent 循环的开关。ToolNode是 LangGraph 预置的工具执行节点它会自动解析模型输出的tool_calls按名字找到对应工具用参数调用然后把结果包装成ToolMessage塞回消息列表。如果你需要在这中间加校验比如检查参数是否越界可以不用ToolNode自己写一个执行节点。3.3 工具合约的校验与降级策略模型不是每次都听话。它可能填错参数类型、编造不存在的枚举值、或者该调工具的时候直接编答案。所以工具执行节点必须做校验。我的做法是在ToolNode外面包一层def safe_tool_node(state: AgentState): last_message state[messages][-1] results [] for tool_call in last_message.tool_calls: try: tool tool_map[tool_call[name]] validated_args tool.args_schema(**tool_call[args]) result tool.invoke(validated_args.dict()) results.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) except ValidationError as e: results.append(ToolMessage( contentf参数校验失败{e}。请检查参数格式后重试。, tool_call_idtool_call[id] )) return {messages: results}校验失败时不要把异常直接抛出去而是把错误信息作为ToolMessage返回给模型让它自己修正。这比直接报错中断流程要好因为模型看到“参数校验失败category 必须是 policy/product/faq 之一”之后下一轮大概率会填对。这就是所谓的“自修正循环”。降级策略也要有。如果工具连续调用失败三次应该停止循环返回一个兜底回复而不是无限重试。LangGraph 里可以用状态里的计数器实现class AgentState(TypedDict): messages: Annotated[list, add_messages] tool_failures: int在should_continue里判断tool_failures 3就强制结束。这个阈值我一般设 3因为模型连续三次犯同一个错的概率很低如果真发生了说明是工具描述或 schema 设计有问题继续重试也是浪费 token。实操心得工具描述里的“什么时候用”比“怎么用”更重要。我踩过的坑是花大量篇幅写参数说明结果模型在不该调工具的时候乱调。后来在描述开头加一句“仅当用户明确询问内部政策或产品信息时使用”误调用率下降了一大半。4. 上下文工程把 token 花在刀刃上4.1 上下文工程不是提示词工程的升级版热词里同时出现了“大模型提示词工程与上下文工程”和“llm ontology”这两个词经常被混用但它们的关注点不同。提示词工程关注的是“怎么问”上下文工程关注的是“给模型看什么”。在 RAG 和 Agent 场景里上下文工程的重要性远高于提示词工程因为模型看到的上下文里检索文档和工具返回结果占了 80% 以上的 token提示词只占很小一部分。上下文工程的核心问题是在有限的 token 预算里怎么组织信息让模型做出正确决策。这包括四个子问题检索结果怎么排序和截断、工具返回结果怎么压缩、多轮对话历史怎么摘要、系统提示怎么分层。每一个都直接影响答案质量和成本。我做过一个测算一个中等复杂度的 RAG 问答系统提示 500 token检索文档 5 篇每篇 800 token 共 4000 token对话历史 2000 token工具返回 1000 token加起来 7500 token。如果模型是 128k 上下文的看起来绰绰有余但实际推理成本是按 token 算的7500 token 的输入比 3000 token 贵一倍多。而且上下文越长模型注意力越分散答案质量反而可能下降。所以上下文工程的目标不是“塞满”而是“精准”。4.2 检索结果的上下文组织策略检索回来 5 篇文档怎么放进 prompt 里最朴素的做法是直接拼接但这有三个问题。第一文档之间没有边界模型可能混淆信息。第二长文档里只有一两句话相关其余是噪声。第三文档顺序影响模型注意力放前面的文档被引用的概率更高。我的做法是三步处理。第一步给每篇文档加编号和来源标记格式是[文档1 | 来源报销制度v2.3]这样模型引用时能说清楚出处。第二步对每篇文档做句子级切分只保留和 query 语义相似度最高的 2-3 句而不是整篇塞进去。这一步可以用嵌入模型算句子和 query 的相似度也可以用简单的关键词重叠度。第三步按相关性分数降序排列但把最相关的那篇放在最前面和最后面各一次如果 token 预算允许利用“首因效应”和“近因效应”提升模型对关键信息的注意力。句子级切分的代码大概是这样def compress_document(doc: str, query: str, max_sentences: int 3) - str: sentences split_sentences(doc) query_embedding embed(query) scored [(sent, cosine_sim(embed(sent), query_embedding)) for sent in sentences] scored.sort(keylambda x: x[1], reverseTrue) top_sentences [s[0] for s in scored[:max_sentences]] return .join(top_sentences)这个压缩能把每篇文档从 800 token 压到 200 token 左右5 篇文档从 4000 token 降到 1000 token省了 75% 的预算而关键信息基本没丢。代价是多了一次嵌入计算大概增加 50-100ms 延迟但省下的 token 成本远大于这个延迟。4.3 多轮对话历史的摘要与裁剪多轮对话是上下文膨胀的重灾区。用户问了 10 轮历史消息可能有 5000 token但其中大部分是寒暄和重复确认真正有用的信息可能只有几百 token。全量保留既贵又干扰模型。我的策略是分层处理。最近 3 轮对话保留原文因为这是模型最需要关注的即时上下文。更早的对话做摘要用一个小模型比如 7B 级别的把历史压缩成一段 200 字以内的摘要包含用户的核心诉求、已确认的信息、未解决的问题。摘要的 prompt 大概是“把以下对话压缩成一段话保留用户身份、已确认事实、待办事项去掉寒暄和重复内容。”在 LangGraph 里这个摘要逻辑可以做成一个独立节点在消息数超过阈值时触发def summarize_history(state: AgentState): if len(state[messages]) 10: return {} old_messages state[messages][:-6] summary summarizer.invoke(old_messages) return {messages: [SystemMessage(contentf历史对话摘要{summary})] state[messages][-6:]}注意这里保留了最近 6 条消息3 轮把更早的替换成一条摘要。这样既控制了 token又保留了即时上下文。摘要节点的触发条件用消息数量而不是 token 数量是因为 token 计算需要额外调用而消息数量是现成的。4.4 系统提示的分层设计系统提示不是一大段文字而应该分层。我的分层是角色层、能力层、约束层、格式层。角色层定义“你是谁”能力层定义“你能做什么”约束层定义“你不能做什么”格式层定义“输出长什么样”。角色层一句话就够“你是一个内部知识库助手基于检索到的文档回答员工问题。”能力层说明工具用法“你可以调用 knowledge_search 工具检索文档当文档中没有答案时明确告知用户。”约束层是重点“不要编造文档中没有的信息不要回答与内部政策无关的问题如果检索结果与问题无关直接说‘未找到相关信息’。”格式层定义输出结构“回答时先给结论再给依据引用文档时标注来源编号。”分层的好处是每一层可以独立调整。比如发现模型爱编造就加强约束层发现输出太啰嗦就调整格式层。如果全混在一起改一处可能影响其他行为。注意约束层的措辞要用“不要”而不是“尽量避免”。“尽量避免编造”这种表述模型会理解为“可以偶尔编造”而“不要编造”是硬约束。这个差别在实际测试里很明显。5. 常见问题与排查技巧实录5.1 检索质量问题的排查路径检索出问题第一步不是改代码而是看数据。把 query 和召回的文档对打印出来人工判断相关性。如果 top-5 里只有 1-2 篇相关说明召回有问题如果 top-5 都相关但答案还是错说明是上下文组织或模型的问题。召回问题的排查顺序是先看嵌入模型是否适合你的语言和领域中文场景用 bge 系列英文用 e5 或 gte再看分块策略块太大导致语义稀释块太小导致上下文断裂一般 256-512 token 一块比较稳最后看混合检索的权重BM25 和向量的 top_k 比例可以调精确查询多的场景加大 BM25 权重。我整理了一个速查表现象可能原因排查方法解决方向召回文档完全不相关嵌入模型不匹配换模型对比 hit rate中文用 bge英文用 e5相关文档排在第 8 位相似度区分度不够看分数分布加 BM25 混合或重排序精确术语查不到向量对精确匹配弱测试 BM25 单独效果提高 BM25 权重旧版本文档被召回元数据缺失检查文档 meta加时效性过滤答案引用错误来源文档编号混乱检查 prompt 格式加来源标记和编号5.2 工具调用失败的典型场景工具调用失败最常见的是参数格式错误。模型可能把top_k填成字符串5而不是整数5或者把category填成政策而不是policy。Pydantic 的校验会拦住这些但错误信息要设计好让模型能看懂怎么改。第二个常见场景是工具选择错误。用户问“今天天气怎么样”模型调了knowledge_search因为工具描述里没写清楚适用范围。解决办法是在工具描述里加否定条件“不要用于天气、时间、闲聊等非知识库问题。”第三个场景是循环调用。模型调工具拿到结果后觉得不满意又调一次反复几次耗尽 token。解决办法是在状态里记录工具调用次数超过阈值强制结束并在系统提示里加一句“同一问题最多检索两次第二次仍无结果就告知用户未找到”。5.3 上下文超限的应急处理即使做了摘要和压缩偶尔还是会超限尤其是用户粘贴一大段文本进来的时候。应急处理是在调用模型前做一次 token 计数如果超过预算就按优先级裁剪先裁对话历史再裁检索文档的句子数最后裁工具返回结果。系统提示和当前 query 永远不裁。Haystack 和 LangGraph 都没有内置的 token 计数裁剪组件需要自己写。用 tiktoken 库可以精确计数import tiktoken def count_tokens(text: str, model: str gpt-4) - int: encoding tiktoken.encoding_for_model(model) return len(encoding.encode(text)) def trim_context(messages: list, max_tokens: int 6000) - list: total sum(count_tokens(m[content]) for m in messages) while total max_tokens and len(messages) 2: removed messages.pop(1) # 保留 system 和最后一条 total - count_tokens(removed[content]) return messages这个裁剪逻辑是“从中间删”保留开头的系统提示和结尾的当前 query删掉中间的历史。虽然粗暴但应急够用。更好的做法是按优先级删但实现复杂度高看你的场景需不需要。5.4 性能与成本的平衡技巧RAG 系统的成本主要在嵌入计算和 LLM 调用。嵌入计算可以缓存相同的 query 不用重复算。LLM 调用可以通过上下文压缩减少输入 token。还有一个容易被忽略的点是批量处理如果多个用户同时查询把嵌入请求批量发送比逐个发送快很多Haystack 的 embedder 支持批量输入。延迟方面混合检索的两路可以并行执行LangGraph 支持并行节点。重排序是串行的但可以只对 top-5 做重排而不是 top-20减少计算量。工具调用如果涉及外部 API要设超时避免一个慢请求拖垮整个流程。我个人的经验是一个设计良好的生产级 RAGP95 延迟控制在 3 秒以内是可行的成本控制在每次查询 0.01 元以内也是可行的。关键是把上下文工程做扎实不要用“多召回、长上下文”来掩盖检索质量问题。6. 把三篇串起来一个可复用的生产级骨架走到这里Haystack 负责的检索层和 LangGraph 负责的编排层已经能咬合上了。检索层输出的是经过混合召回、元数据过滤、重排序、句子压缩的高质量上下文编排层负责的是工具合约校验、状态管理、循环控制、上下文裁剪。两层之间的接口就是“一组带来源标记的文档片段”这个接口设计得越干净两层的耦合越低后续替换组件越容易。热词里提到的“agentic rag”和“rag 智能体”本质上就是这两层的结合RAG 提供事实依据Agent 提供决策和行动能力。但要注意Agent 不是越多越好。我见过一些项目把简单问答也做成多轮工具调用结果延迟翻倍、成本翻倍答案质量却没提升。判断标准很简单如果一个问题用单次检索加一次生成就能答好就不要上 Agent 循环。Agent 的价值在于处理需要多步推理、多源信息整合、或者需要执行动作的场景。最后分享一个我在实际项目里用的检查清单每次上线前过一遍检索层混合检索是否开启、元数据过滤是否配置、重排序是否启用、句子压缩是否生效工具层每个工具的 description 是否包含触发条件和否定条件、参数 schema 是否有范围约束、校验失败是否有自修正机制、是否有循环次数上限上下文层系统提示是否分层、对话历史是否有摘要、token 预算是否设了硬上限、超限裁剪是否有优先级监控层检索 hit rate 是否可观测、工具调用成功率是否可观测、token 消耗是否可观测、P95 延迟是否可观测这个清单不复杂但能挡住 80% 的上线事故。剩下的 20% 靠的是对业务数据的理解——知道你的用户会怎么问、你的文档长什么样、你的模型在什么情况下会犯什么错。这些没有通用答案只能靠一次次排查积累。