1. 为什么“流水线”才是生产级 RAG 的真正分水岭很多人第一次搭 RAG脑子里想的是一条直线文档切块、向量化、存库、检索、拼进 Prompt、丢给模型。跑通 Demo 大概一个下午就够了但一上生产就原形毕露——召回忽高忽低、多轮对话记不住上下文、工具调用和检索结果打架、模型答非所问还一本正经。问题不在于某个组件选错了而在于你把 RAG 当成了一次性的“函数调用”而它本质上是一条需要编排、需要状态、需要分支和回退的流水线。这篇内容我想聊的就是这条流水线怎么搭。核心工具是两个Haystack负责把检索、排序、生成这些“节点”做扎实LangGraph负责把这些节点编排成一张带状态、能循环、能条件跳转的图。再往上是工具合约和上下文工程这两件容易被忽略但决定成败的事。整套东西的目标很明确让 RAG 从“能跑”变成“敢上线”。适合谁看如果你已经写过最基础的 RAG被召回率和幻觉折磨过或者正在纠结 LangChain 和 LangGraph 到底该用哪个、Haystack 值不值得引入那这篇就是写给你的。我会把选型逻辑、状态设计、工具合约的写法、上下文预算的算法都拆开讲参数和代码都给到能直接抄的程度。零基础也能看懂思路有经验的可以直接跳到实操部分。先说清楚一个前提RAG 的瓶颈从来不在“检索”这一个动作上。热词里有个词叫“rag瓶颈”我特别认同。真正的瓶颈是——检索回来的东西怎么和对话历史、工具结果、系统指令一起塞进有限的上下文窗口还不互相污染。这就是上下文工程要解决的问题也是 Haystack LangGraph 这套组合真正发挥价值的地方。2. 整体架构设计与选型逻辑拆解2.1 为什么是 Haystack 加 LangGraph而不是二选一先回答一个被问烂的问题LangChain 和 LangGraph 什么关系Haystack 又插在哪。我的判断是这样的——LangChain 是组件库LangGraph 是编排引擎Haystack 是检索流水线的专业工具箱。三者不是替代关系而是分工。LangGraph 的强项是“图”节点、边、条件边、状态、检查点、循环。它天生适合表达“检索不达标就改写 query 重试”“工具调用失败就回退到检索”“多轮对话需要持久化状态”这类非线性流程。你用它写 Agent 特别顺手因为 Agent 的本质就是一个带循环的图。但 LangGraph 本身不负责“怎么把文档切得好、怎么把向量检索和关键词检索融合、怎么重排序”。这些活儿 Haystack 干得更专业。Haystack 的 Pipeline 抽象、DocumentStore 生态、Retriever 和 Ranker 的成熟实现是经过大量生产验证的。尤其是它的DocumentJoiner、SentenceWindowRetriever、各种 Ranker拿来即用省掉大量造轮子的时间。所以我的架构是Haystack 管“检索质量”LangGraph 管“流程控制”两者通过一个适配层对接。Haystack 的 Pipeline 对外暴露成一个函数或一个 ToolLangGraph 把它当成图里的一个节点来调用。这样职责清晰各自发挥长处。提示不要试图用 LangGraph 重写 Haystack 的检索逻辑也不要用 Haystack 的 Pipeline 硬凑 Agent 循环。各干各的接口对齐是最省心的做法。2.2 生产级 RAG 的四个必备能力Demo 级 RAG 和生产级 RAG 的差距我总结成四个能力缺一个都会在真实场景翻车。第一是混合检索。纯向量检索对语义相似但关键词不匹配的 query 很友好但对精确术语、编号、专有名词就拉胯。生产环境必须向量 BM25 双路召回再融合Haystack 的DocumentJoiner支持reciprocal_rank_fusion这是标配。第二是重排序。召回阶段追求高 recall宁可多召回一些排序阶段追求高 precision把真正相关的顶上来。一个 Cross-Encoder Ranker 能把 hit rate 提升一大截代价是延迟。这个取舍后面细讲。第三是状态管理。多轮对话里“它”“这个”“上面说的”指代什么必须靠状态记住。LangGraph 的 State 和 Checkpointer 就是干这个的而且支持持久化服务重启不丢会话。第四是工具合约。当 RAG 不只是问答还要调 API、查数据库、执行计算时工具的参数校验、错误处理、超时控制就变成刚需。工具合约写不好Agent 会疯狂调用错误工具或者传错参数最后烧一堆 token 还解决不了问题。2.3 上下文工程把有限的窗口花在刀刃上上下文工程这个词最近很火但很多人理解得太窄以为就是“写 Prompt”。我的理解是上下文工程是在有限的 token 预算下决定哪些信息进窗口、以什么顺序进、以什么格式进、冲突时谁优先。一个典型的 RAG 请求候选上下文来源有系统指令、对话历史、检索到的文档、工具返回结果、用户当前 query。这些东西加起来轻松超过模型窗口。你必须做预算分配。我的经验值是系统指令占 10%对话历史占 20%检索文档占 50%工具结果占 15%留 5% 给模型输出。当然这要看具体场景但先定预算再填内容比事后截断要靠谱得多。热词里有个说法我特别喜欢“LLM 的 token 三个点——key 我是谁、query 我在找什么、value 我能提供什么”。这其实是在说每一段塞进上下文的内容都要能回答这三个问题之一。回答不了的就是噪音该砍就砍。这个标准比“看起来相关”严格得多也实用得多。3. 核心细节解析与实操要点3.1 Haystack 检索流水线的关键节点配置Haystack 的 Pipeline 是声明式的你把组件连起来它负责执行。一个生产级的检索流水线我通常这么搭from haystack import Pipeline from haystack.components.retrievers import InMemoryBM25Retriever, InMemoryEmbeddingRetriever from haystack.components.joiners import DocumentJoiner from haystack.components.rankers import TransformersSimilarityRanker pipeline Pipeline() pipeline.add_component(bm25, InMemoryBM25Retriever(document_storestore, top_k20)) pipeline.add_component(embed, InMemoryEmbeddingRetriever(document_storestore, top_k20)) pipeline.add_component(joiner, DocumentJoiner(join_modereciprocal_rank_fusion)) pipeline.add_component(ranker, TransformersSimilarityRanker(modelBAAI/bge-reranker-base, top_k5)) pipeline.connect(bm25.documents, joiner.documents) pipeline.connect(embed.documents, joiner.documents) pipeline.connect(joiner.documents, ranker.documents)几个参数值得说清楚。top_k20是召回阶段的两路各 20融合后可能 30 多篇再交给 Ranker 砍到 5 篇。为什么召回要这么多因为 RRF 融合后排名会变多召回一些给 Ranker 更多选择。Ranker 的top_k5是最终进上下文的这个数字取决于你的文档块大小和窗口预算块小可以多给几篇块大就少给。reciprocal_rank_fusion这个融合算法值得展开。它的核心思想是一篇文档如果在多个召回列表里都排得靠前那它大概率真的相关。公式是score Σ 1/(k rank)k 通常取 60。这个算法不需要归一化分数对异构检索器特别友好是我最推荐的融合方式。注意BM25 对中文需要先分词。Haystack 默认的 BM25 用的是英文分词中文场景要么换jieba分词器要么直接用支持中文的 DocumentStore。这个坑我踩过检索结果全是乱的。3.2 LangGraph 状态设计别把 State 当垃圾桶LangGraph 的 State 是整个图的共享内存所有节点读写它。新手最容易犯的错是把什么都往 State 里塞最后 State 变成一个巨大的、没人说得清里面有什么的字典。我的原则是State 只放“跨节点需要传递的最小必要信息”。一个 RAG Agent 的 State我通常这么定义from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class RAGState(TypedDict): messages: Annotated[list, add_messages] query: str retrieved_docs: list tool_results: list retry_count: int final_answer: strmessages用add_messages注解这是 LangGraph 的内置 reducer负责把新消息追加而不是覆盖。retry_count用来控制重试循环防止无限循环烧钱。retrieved_docs和tool_results分开存因为它们的生命周期和优先级不同。这里有个关键设计query 和 messages 分开。messages 是完整对话历史query 是当前这一轮经过改写、用于检索的实际查询。为什么要分开因为多轮对话里用户这轮说的可能是“那它的价格呢”直接拿这句去检索必然拉胯需要先用 LLM 结合历史改写成“XX 产品的价格是多少”。改写后的 query 存进 State检索用它展示给用户还是用原始 messages。3.3 工具合约让 Agent 调用工具不再靠猜工具合约这个词听起来玄其实就是把工具的输入输出定义清楚包括类型、必填项、取值范围、错误码。LangGraph 里工具用tool装饰器定义但光写个函数签名不够你得把 docstring 写到位因为 LLM 是靠 docstring 来决定调不调、怎么调的。from langchain_core.tools import tool tool def search_knowledge_base(query: str, top_k: int 5) - str: 在内部知识库中检索信息。 Args: query: 检索关键词应该是完整的自然语言问题不要只给关键词。 top_k: 返回文档数量范围 1-10默认 5。数量越多延迟越高。 Returns: 检索到的文档内容多篇之间用 --- 分隔。 Raises: 检索失败时返回错误说明字符串不会抛异常。 ...注意几个细节。第一query的描述里明确说“不要只给关键词”因为 LLM 经常偷懒只传几个词导致检索质量差。第二top_k给了范围防止 LLM 传个 1000 进来把上下文撑爆。第三明确说“不抛异常返回错误字符串”这样 Agent 拿到错误能自己决定重试还是换工具而不是整个图崩掉。工具合约的另一个重点是幂等性和副作用。查询类工具随便调但写操作类工具下单、发邮件、改数据必须做幂等否则 Agent 重试一次就重复执行了。我的做法是给这类工具加一个idempotency_key参数由调用方生成服务端去重。4. 实操过程与核心环节实现4.1 从零搭一条可运行的检索图先搭骨架。LangGraph 的图用StateGraph构建节点是函数边是流转规则。from langgraph.graph import StateGraph, END def rewrite_query(state: RAGState) - dict: # 结合历史把用户输入改写成独立可检索的 query ... return {query: rewritten} def retrieve(state: RAGState) - dict: docs haystack_pipeline.run({query: state[query]})[ranker][documents] return {retrieved_docs: docs} def grade_docs(state: RAGState) - str: # 判断检索质量决定是否重试 if not state[retrieved_docs] or state[retry_count] 2: return generate return generate if is_relevant(state) else rewrite def generate(state: RAGState) - dict: context build_context(state) answer llm.invoke(context) return {final_answer: answer, messages: [answer]} graph StateGraph(RAGState) graph.add_node(rewrite, rewrite_query) graph.add_node(retrieve, retrieve) graph.add_node(generate, generate) graph.set_entry_point(rewrite) graph.add_edge(rewrite, retrieve) graph.add_conditional_edges(retrieve, grade_docs, {rewrite: rewrite, generate: generate}) graph.add_edge(generate, END) app graph.compile(checkpointermemory)这个图里最关键的是grade_docs这个条件边。它做的是检索质量评估——检索回来的文档到底能不能回答问题。评估可以用 LLM 打分也可以用轻量模型甚至可以用规则比如关键词覆盖率。如果评估不通过且重试次数没超就回到rewrite改写 query 重试。这个循环是生产级 RAG 和 Demo 的核心区别之一。checkpointermemory是持久化状态的关键。用MemorySaver是内存版生产要换成数据库版比如 Postgres这样服务重启会话不丢。每个会话用thread_id区分调用时传config{configurable: {thread_id: user_123}}。4.2 上下文预算的分配与拼装build_context这个函数是上下文工程的核心。我把它拆成三步分配预算、按优先级填充、冲突消解。def build_context(state: RAGState, max_tokens: int 8000) - str: budget { system: int(max_tokens * 0.10), history: int(max_tokens * 0.20), docs: int(max_tokens * 0.50), tools: int(max_tokens * 0.15), } # 按预算截断各部分 system_part truncate(SYSTEM_PROMPT, budget[system]) history_part truncate_history(state[messages], budget[history]) docs_part pack_docs(state[retrieved_docs], budget[docs]) tools_part pack_tools(state[tool_results], budget[tools]) return assemble(system_part, history_part, docs_part, tools_part)pack_docs不是简单拼接而是按相关性排序后贪心填充每篇文档前面加上来源标记比如[文档1]这样模型引用时能对上号。如果一篇文档太长就截断但保留开头和结尾因为关键信息往往在这两处。冲突消解是个容易被忽略的点。如果检索文档和工具结果矛盾怎么办我的规则是工具结果优先级高于检索文档因为工具是实时查询文档可能是旧的。在拼装时给工具结果加一句“以下为实时查询结果优先级最高”模型会倾向于采信。提示上下文里每段内容都加一个明确的分隔标记和来源说明模型对“这段是什么”越清楚用起来越准。别指望模型自己猜。4.3 工具调用的接入与错误处理把工具接进图里用 LangGraph 的ToolNode最省事。它自动处理工具调用的解析、执行、结果回填。from langgraph.prebuilt import ToolNode tools [search_knowledge_base, query_database, calculate] tool_node ToolNode(tools) graph.add_node(tools, tool_node)但ToolNode默认遇到工具报错会抛异常整个图就挂了。生产环境必须包一层错误处理。我的做法是给每个工具内部 try/except把异常转成结构化的错误信息返回让 Agent 自己决定下一步。比如数据库超时返回{error: timeout, retryable: true}Agent 看到retryable就知道可以重试。工具调用的另一个坑是并行调用。LLM 可能一次返回多个工具调用ToolNode会并行执行。如果这些工具有依赖关系比如第二个工具要用第一个的结果就会出错。解决办法是在工具描述里写清楚依赖或者干脆拆成多轮让模型一次只调一个。我倾向于后者虽然慢一点但可控。4.4 一个完整的请求生命周期把上面这些串起来一个用户请求的完整流程是这样的用户输入“帮我查下上季度华东区的销售数据和去年同期比怎么样”。第一步rewrite_query结合历史改写成“查询上季度华东区销售数据并与去年同期对比”。第二步retrieve走 Haystack 混合检索召回相关文档。第三步grade_docs评估发现文档里有销售数据但没有对比分析判定需要工具。第四步进入工具节点调用query_database拿实时数据。第五步generate把检索文档和工具结果一起拼进上下文生成对比分析。整个过程 State 里记录了 query、docs、tool_resultscheckpointer 存了会话下次用户问“那华南区呢”历史还在。这个流程里每一步的输入输出都是可观测的。我在生产环境会给每个节点加日志和耗时统计这样出问题能快速定位是检索慢、工具慢还是生成慢。5. 常见问题与排查技巧实录5.1 检索质量问题的排查顺序检索不准是最常见的问题但原因可能出在好几个地方。我整理了一个排查顺序从便宜到贵排查项检查方法常见问题分块策略看 chunk 大小和重叠块太大语义稀释太小上下文断裂分词器中文是否用了 jieba默认英文分词导致 BM25 失效Embedding 模型是否匹配语言和领域通用模型在专业领域表现差融合权重RRF 的 k 值是否合理k 太小导致排名波动大Ranker是否启用、模型是否合适没启用 Ranker 导致精度低Query 改写多轮是否做了改写指代词直接检索必然失败我的经验是80% 的检索问题出在分块和 query 改写上而不是模型选型。分块我一般用 512 token 加 50 token 重叠专业文档可以更小。Query 改写一定要做尤其是多轮对话场景。5.2 Agent 循环失控与 token 烧钱Agent 最危险的行为是陷入循环反复调用工具或反复检索token 哗哗地烧。防护措施有三层第一层是硬性重试上限。State 里放retry_count超过阈值强制走生成节点。我一般设 2 到 3 次。第二层是单次请求 token 预算。在图的入口记录初始 token 数每个节点执行后累加超过预算就中断并返回已有结果。LangGraph 可以在节点里检查 State 里的累计值。第三层是工具调用去重。同一个工具用相同参数连续调用两次第二次直接返回缓存结果。这个在ToolNode外面包一层缓存就能实现。注意循环失控往往不是模型笨而是工具返回的结果让模型觉得“没解决问题”。检查工具返回格式确保模型能看懂结果比调 Prompt 更有效。5.3 上下文污染与幻觉幻觉的根源经常是上下文里塞了矛盾或无关的信息。排查时我会做一件事把实际拼装好的上下文打印出来逐段问“这段对回答当前问题有用吗”。没用的删掉矛盾的标优先级。很多时候删掉一半上下文回答质量反而上升。另一个技巧是强制引用。在 Prompt 里要求模型每个事实性陈述后面标注来源[文档1]或[工具结果]没有来源的不许说。这不能完全消除幻觉但能让你快速发现哪些是编的。5.4 性能与延迟优化生产环境延迟是硬指标。RAG 的延迟大头在检索和生成。优化手段检索并行BM25 和向量检索并行跑别串行。Ranker 按需启用召回结果少且分数高时跳过 Ranker。流式生成LangGraph 支持stream模式首 token 延迟大幅降低。缓存相同 query 的检索结果缓存相同文档的 embedding 缓存。我实测下来一个优化到位的 RAG 请求P95 延迟能控制在 2 秒以内其中检索 300ms、Ranker 400ms、生成 1.2s。Ranker 是延迟大头如果对精度要求没那么极致可以换成更轻量的模型或者干脆跳过。5.5 常见问题速查表现象可能原因解决方向答非所问Query 未改写、检索不准加 query 改写检查分块反复调同一工具工具返回格式模型看不懂结构化返回加错误说明多轮丢上下文没用 checkpointer配置持久化传 thread_id中文检索差分词器不匹配换 jieba 或中文优化模型延迟高Ranker 太重、串行执行并行检索按需 Ranker幻觉多上下文矛盾、无引用要求清理上下文强制引用循环烧钱无重试上限加 retry_count 和 token 预算这套东西搭下来RAG 才算真正能上生产。Haystack 保证检索质量LangGraph 保证流程可控工具合约保证调用可靠上下文工程保证窗口不浪费。四个环节缺一不可任何一个偷懒都会在真实流量下暴露。我个人在实际操作中的体会是别追求一步到位。先把检索和生成跑通再加 Ranker再加工具再加循环和评估。每加一层都测一遍看指标是升是降。RAG 调优是个迭代过程没有银弹但有一套可复现的方法论就能少走很多弯路。
生产级RAG流水线：Haystack与LangGraph实战指南
1. 为什么“流水线”才是生产级 RAG 的真正分水岭很多人第一次搭 RAG脑子里想的是一条直线文档切块、向量化、存库、检索、拼进 Prompt、丢给模型。跑通 Demo 大概一个下午就够了但一上生产就原形毕露——召回忽高忽低、多轮对话记不住上下文、工具调用和检索结果打架、模型答非所问还一本正经。问题不在于某个组件选错了而在于你把 RAG 当成了一次性的“函数调用”而它本质上是一条需要编排、需要状态、需要分支和回退的流水线。这篇内容我想聊的就是这条流水线怎么搭。核心工具是两个Haystack负责把检索、排序、生成这些“节点”做扎实LangGraph负责把这些节点编排成一张带状态、能循环、能条件跳转的图。再往上是工具合约和上下文工程这两件容易被忽略但决定成败的事。整套东西的目标很明确让 RAG 从“能跑”变成“敢上线”。适合谁看如果你已经写过最基础的 RAG被召回率和幻觉折磨过或者正在纠结 LangChain 和 LangGraph 到底该用哪个、Haystack 值不值得引入那这篇就是写给你的。我会把选型逻辑、状态设计、工具合约的写法、上下文预算的算法都拆开讲参数和代码都给到能直接抄的程度。零基础也能看懂思路有经验的可以直接跳到实操部分。先说清楚一个前提RAG 的瓶颈从来不在“检索”这一个动作上。热词里有个词叫“rag瓶颈”我特别认同。真正的瓶颈是——检索回来的东西怎么和对话历史、工具结果、系统指令一起塞进有限的上下文窗口还不互相污染。这就是上下文工程要解决的问题也是 Haystack LangGraph 这套组合真正发挥价值的地方。2. 整体架构设计与选型逻辑拆解2.1 为什么是 Haystack 加 LangGraph而不是二选一先回答一个被问烂的问题LangChain 和 LangGraph 什么关系Haystack 又插在哪。我的判断是这样的——LangChain 是组件库LangGraph 是编排引擎Haystack 是检索流水线的专业工具箱。三者不是替代关系而是分工。LangGraph 的强项是“图”节点、边、条件边、状态、检查点、循环。它天生适合表达“检索不达标就改写 query 重试”“工具调用失败就回退到检索”“多轮对话需要持久化状态”这类非线性流程。你用它写 Agent 特别顺手因为 Agent 的本质就是一个带循环的图。但 LangGraph 本身不负责“怎么把文档切得好、怎么把向量检索和关键词检索融合、怎么重排序”。这些活儿 Haystack 干得更专业。Haystack 的 Pipeline 抽象、DocumentStore 生态、Retriever 和 Ranker 的成熟实现是经过大量生产验证的。尤其是它的DocumentJoiner、SentenceWindowRetriever、各种 Ranker拿来即用省掉大量造轮子的时间。所以我的架构是Haystack 管“检索质量”LangGraph 管“流程控制”两者通过一个适配层对接。Haystack 的 Pipeline 对外暴露成一个函数或一个 ToolLangGraph 把它当成图里的一个节点来调用。这样职责清晰各自发挥长处。提示不要试图用 LangGraph 重写 Haystack 的检索逻辑也不要用 Haystack 的 Pipeline 硬凑 Agent 循环。各干各的接口对齐是最省心的做法。2.2 生产级 RAG 的四个必备能力Demo 级 RAG 和生产级 RAG 的差距我总结成四个能力缺一个都会在真实场景翻车。第一是混合检索。纯向量检索对语义相似但关键词不匹配的 query 很友好但对精确术语、编号、专有名词就拉胯。生产环境必须向量 BM25 双路召回再融合Haystack 的DocumentJoiner支持reciprocal_rank_fusion这是标配。第二是重排序。召回阶段追求高 recall宁可多召回一些排序阶段追求高 precision把真正相关的顶上来。一个 Cross-Encoder Ranker 能把 hit rate 提升一大截代价是延迟。这个取舍后面细讲。第三是状态管理。多轮对话里“它”“这个”“上面说的”指代什么必须靠状态记住。LangGraph 的 State 和 Checkpointer 就是干这个的而且支持持久化服务重启不丢会话。第四是工具合约。当 RAG 不只是问答还要调 API、查数据库、执行计算时工具的参数校验、错误处理、超时控制就变成刚需。工具合约写不好Agent 会疯狂调用错误工具或者传错参数最后烧一堆 token 还解决不了问题。2.3 上下文工程把有限的窗口花在刀刃上上下文工程这个词最近很火但很多人理解得太窄以为就是“写 Prompt”。我的理解是上下文工程是在有限的 token 预算下决定哪些信息进窗口、以什么顺序进、以什么格式进、冲突时谁优先。一个典型的 RAG 请求候选上下文来源有系统指令、对话历史、检索到的文档、工具返回结果、用户当前 query。这些东西加起来轻松超过模型窗口。你必须做预算分配。我的经验值是系统指令占 10%对话历史占 20%检索文档占 50%工具结果占 15%留 5% 给模型输出。当然这要看具体场景但先定预算再填内容比事后截断要靠谱得多。热词里有个说法我特别喜欢“LLM 的 token 三个点——key 我是谁、query 我在找什么、value 我能提供什么”。这其实是在说每一段塞进上下文的内容都要能回答这三个问题之一。回答不了的就是噪音该砍就砍。这个标准比“看起来相关”严格得多也实用得多。3. 核心细节解析与实操要点3.1 Haystack 检索流水线的关键节点配置Haystack 的 Pipeline 是声明式的你把组件连起来它负责执行。一个生产级的检索流水线我通常这么搭from haystack import Pipeline from haystack.components.retrievers import InMemoryBM25Retriever, InMemoryEmbeddingRetriever from haystack.components.joiners import DocumentJoiner from haystack.components.rankers import TransformersSimilarityRanker pipeline Pipeline() pipeline.add_component(bm25, InMemoryBM25Retriever(document_storestore, top_k20)) pipeline.add_component(embed, InMemoryEmbeddingRetriever(document_storestore, top_k20)) pipeline.add_component(joiner, DocumentJoiner(join_modereciprocal_rank_fusion)) pipeline.add_component(ranker, TransformersSimilarityRanker(modelBAAI/bge-reranker-base, top_k5)) pipeline.connect(bm25.documents, joiner.documents) pipeline.connect(embed.documents, joiner.documents) pipeline.connect(joiner.documents, ranker.documents)几个参数值得说清楚。top_k20是召回阶段的两路各 20融合后可能 30 多篇再交给 Ranker 砍到 5 篇。为什么召回要这么多因为 RRF 融合后排名会变多召回一些给 Ranker 更多选择。Ranker 的top_k5是最终进上下文的这个数字取决于你的文档块大小和窗口预算块小可以多给几篇块大就少给。reciprocal_rank_fusion这个融合算法值得展开。它的核心思想是一篇文档如果在多个召回列表里都排得靠前那它大概率真的相关。公式是score Σ 1/(k rank)k 通常取 60。这个算法不需要归一化分数对异构检索器特别友好是我最推荐的融合方式。注意BM25 对中文需要先分词。Haystack 默认的 BM25 用的是英文分词中文场景要么换jieba分词器要么直接用支持中文的 DocumentStore。这个坑我踩过检索结果全是乱的。3.2 LangGraph 状态设计别把 State 当垃圾桶LangGraph 的 State 是整个图的共享内存所有节点读写它。新手最容易犯的错是把什么都往 State 里塞最后 State 变成一个巨大的、没人说得清里面有什么的字典。我的原则是State 只放“跨节点需要传递的最小必要信息”。一个 RAG Agent 的 State我通常这么定义from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class RAGState(TypedDict): messages: Annotated[list, add_messages] query: str retrieved_docs: list tool_results: list retry_count: int final_answer: strmessages用add_messages注解这是 LangGraph 的内置 reducer负责把新消息追加而不是覆盖。retry_count用来控制重试循环防止无限循环烧钱。retrieved_docs和tool_results分开存因为它们的生命周期和优先级不同。这里有个关键设计query 和 messages 分开。messages 是完整对话历史query 是当前这一轮经过改写、用于检索的实际查询。为什么要分开因为多轮对话里用户这轮说的可能是“那它的价格呢”直接拿这句去检索必然拉胯需要先用 LLM 结合历史改写成“XX 产品的价格是多少”。改写后的 query 存进 State检索用它展示给用户还是用原始 messages。3.3 工具合约让 Agent 调用工具不再靠猜工具合约这个词听起来玄其实就是把工具的输入输出定义清楚包括类型、必填项、取值范围、错误码。LangGraph 里工具用tool装饰器定义但光写个函数签名不够你得把 docstring 写到位因为 LLM 是靠 docstring 来决定调不调、怎么调的。from langchain_core.tools import tool tool def search_knowledge_base(query: str, top_k: int 5) - str: 在内部知识库中检索信息。 Args: query: 检索关键词应该是完整的自然语言问题不要只给关键词。 top_k: 返回文档数量范围 1-10默认 5。数量越多延迟越高。 Returns: 检索到的文档内容多篇之间用 --- 分隔。 Raises: 检索失败时返回错误说明字符串不会抛异常。 ...注意几个细节。第一query的描述里明确说“不要只给关键词”因为 LLM 经常偷懒只传几个词导致检索质量差。第二top_k给了范围防止 LLM 传个 1000 进来把上下文撑爆。第三明确说“不抛异常返回错误字符串”这样 Agent 拿到错误能自己决定重试还是换工具而不是整个图崩掉。工具合约的另一个重点是幂等性和副作用。查询类工具随便调但写操作类工具下单、发邮件、改数据必须做幂等否则 Agent 重试一次就重复执行了。我的做法是给这类工具加一个idempotency_key参数由调用方生成服务端去重。4. 实操过程与核心环节实现4.1 从零搭一条可运行的检索图先搭骨架。LangGraph 的图用StateGraph构建节点是函数边是流转规则。from langgraph.graph import StateGraph, END def rewrite_query(state: RAGState) - dict: # 结合历史把用户输入改写成独立可检索的 query ... return {query: rewritten} def retrieve(state: RAGState) - dict: docs haystack_pipeline.run({query: state[query]})[ranker][documents] return {retrieved_docs: docs} def grade_docs(state: RAGState) - str: # 判断检索质量决定是否重试 if not state[retrieved_docs] or state[retry_count] 2: return generate return generate if is_relevant(state) else rewrite def generate(state: RAGState) - dict: context build_context(state) answer llm.invoke(context) return {final_answer: answer, messages: [answer]} graph StateGraph(RAGState) graph.add_node(rewrite, rewrite_query) graph.add_node(retrieve, retrieve) graph.add_node(generate, generate) graph.set_entry_point(rewrite) graph.add_edge(rewrite, retrieve) graph.add_conditional_edges(retrieve, grade_docs, {rewrite: rewrite, generate: generate}) graph.add_edge(generate, END) app graph.compile(checkpointermemory)这个图里最关键的是grade_docs这个条件边。它做的是检索质量评估——检索回来的文档到底能不能回答问题。评估可以用 LLM 打分也可以用轻量模型甚至可以用规则比如关键词覆盖率。如果评估不通过且重试次数没超就回到rewrite改写 query 重试。这个循环是生产级 RAG 和 Demo 的核心区别之一。checkpointermemory是持久化状态的关键。用MemorySaver是内存版生产要换成数据库版比如 Postgres这样服务重启会话不丢。每个会话用thread_id区分调用时传config{configurable: {thread_id: user_123}}。4.2 上下文预算的分配与拼装build_context这个函数是上下文工程的核心。我把它拆成三步分配预算、按优先级填充、冲突消解。def build_context(state: RAGState, max_tokens: int 8000) - str: budget { system: int(max_tokens * 0.10), history: int(max_tokens * 0.20), docs: int(max_tokens * 0.50), tools: int(max_tokens * 0.15), } # 按预算截断各部分 system_part truncate(SYSTEM_PROMPT, budget[system]) history_part truncate_history(state[messages], budget[history]) docs_part pack_docs(state[retrieved_docs], budget[docs]) tools_part pack_tools(state[tool_results], budget[tools]) return assemble(system_part, history_part, docs_part, tools_part)pack_docs不是简单拼接而是按相关性排序后贪心填充每篇文档前面加上来源标记比如[文档1]这样模型引用时能对上号。如果一篇文档太长就截断但保留开头和结尾因为关键信息往往在这两处。冲突消解是个容易被忽略的点。如果检索文档和工具结果矛盾怎么办我的规则是工具结果优先级高于检索文档因为工具是实时查询文档可能是旧的。在拼装时给工具结果加一句“以下为实时查询结果优先级最高”模型会倾向于采信。提示上下文里每段内容都加一个明确的分隔标记和来源说明模型对“这段是什么”越清楚用起来越准。别指望模型自己猜。4.3 工具调用的接入与错误处理把工具接进图里用 LangGraph 的ToolNode最省事。它自动处理工具调用的解析、执行、结果回填。from langgraph.prebuilt import ToolNode tools [search_knowledge_base, query_database, calculate] tool_node ToolNode(tools) graph.add_node(tools, tool_node)但ToolNode默认遇到工具报错会抛异常整个图就挂了。生产环境必须包一层错误处理。我的做法是给每个工具内部 try/except把异常转成结构化的错误信息返回让 Agent 自己决定下一步。比如数据库超时返回{error: timeout, retryable: true}Agent 看到retryable就知道可以重试。工具调用的另一个坑是并行调用。LLM 可能一次返回多个工具调用ToolNode会并行执行。如果这些工具有依赖关系比如第二个工具要用第一个的结果就会出错。解决办法是在工具描述里写清楚依赖或者干脆拆成多轮让模型一次只调一个。我倾向于后者虽然慢一点但可控。4.4 一个完整的请求生命周期把上面这些串起来一个用户请求的完整流程是这样的用户输入“帮我查下上季度华东区的销售数据和去年同期比怎么样”。第一步rewrite_query结合历史改写成“查询上季度华东区销售数据并与去年同期对比”。第二步retrieve走 Haystack 混合检索召回相关文档。第三步grade_docs评估发现文档里有销售数据但没有对比分析判定需要工具。第四步进入工具节点调用query_database拿实时数据。第五步generate把检索文档和工具结果一起拼进上下文生成对比分析。整个过程 State 里记录了 query、docs、tool_resultscheckpointer 存了会话下次用户问“那华南区呢”历史还在。这个流程里每一步的输入输出都是可观测的。我在生产环境会给每个节点加日志和耗时统计这样出问题能快速定位是检索慢、工具慢还是生成慢。5. 常见问题与排查技巧实录5.1 检索质量问题的排查顺序检索不准是最常见的问题但原因可能出在好几个地方。我整理了一个排查顺序从便宜到贵排查项检查方法常见问题分块策略看 chunk 大小和重叠块太大语义稀释太小上下文断裂分词器中文是否用了 jieba默认英文分词导致 BM25 失效Embedding 模型是否匹配语言和领域通用模型在专业领域表现差融合权重RRF 的 k 值是否合理k 太小导致排名波动大Ranker是否启用、模型是否合适没启用 Ranker 导致精度低Query 改写多轮是否做了改写指代词直接检索必然失败我的经验是80% 的检索问题出在分块和 query 改写上而不是模型选型。分块我一般用 512 token 加 50 token 重叠专业文档可以更小。Query 改写一定要做尤其是多轮对话场景。5.2 Agent 循环失控与 token 烧钱Agent 最危险的行为是陷入循环反复调用工具或反复检索token 哗哗地烧。防护措施有三层第一层是硬性重试上限。State 里放retry_count超过阈值强制走生成节点。我一般设 2 到 3 次。第二层是单次请求 token 预算。在图的入口记录初始 token 数每个节点执行后累加超过预算就中断并返回已有结果。LangGraph 可以在节点里检查 State 里的累计值。第三层是工具调用去重。同一个工具用相同参数连续调用两次第二次直接返回缓存结果。这个在ToolNode外面包一层缓存就能实现。注意循环失控往往不是模型笨而是工具返回的结果让模型觉得“没解决问题”。检查工具返回格式确保模型能看懂结果比调 Prompt 更有效。5.3 上下文污染与幻觉幻觉的根源经常是上下文里塞了矛盾或无关的信息。排查时我会做一件事把实际拼装好的上下文打印出来逐段问“这段对回答当前问题有用吗”。没用的删掉矛盾的标优先级。很多时候删掉一半上下文回答质量反而上升。另一个技巧是强制引用。在 Prompt 里要求模型每个事实性陈述后面标注来源[文档1]或[工具结果]没有来源的不许说。这不能完全消除幻觉但能让你快速发现哪些是编的。5.4 性能与延迟优化生产环境延迟是硬指标。RAG 的延迟大头在检索和生成。优化手段检索并行BM25 和向量检索并行跑别串行。Ranker 按需启用召回结果少且分数高时跳过 Ranker。流式生成LangGraph 支持stream模式首 token 延迟大幅降低。缓存相同 query 的检索结果缓存相同文档的 embedding 缓存。我实测下来一个优化到位的 RAG 请求P95 延迟能控制在 2 秒以内其中检索 300ms、Ranker 400ms、生成 1.2s。Ranker 是延迟大头如果对精度要求没那么极致可以换成更轻量的模型或者干脆跳过。5.5 常见问题速查表现象可能原因解决方向答非所问Query 未改写、检索不准加 query 改写检查分块反复调同一工具工具返回格式模型看不懂结构化返回加错误说明多轮丢上下文没用 checkpointer配置持久化传 thread_id中文检索差分词器不匹配换 jieba 或中文优化模型延迟高Ranker 太重、串行执行并行检索按需 Ranker幻觉多上下文矛盾、无引用要求清理上下文强制引用循环烧钱无重试上限加 retry_count 和 token 预算这套东西搭下来RAG 才算真正能上生产。Haystack 保证检索质量LangGraph 保证流程可控工具合约保证调用可靠上下文工程保证窗口不浪费。四个环节缺一不可任何一个偷懒都会在真实流量下暴露。我个人在实际操作中的体会是别追求一步到位。先把检索和生成跑通再加 Ranker再加工具再加循环和评估。每加一层都测一遍看指标是升是降。RAG 调优是个迭代过程没有银弹但有一套可复现的方法论就能少走很多弯路。
