1. 为什么要把 Haystack 和 LangGraph 放在一起用1.1 单框架做生产级 RAG 的真实瓶颈我最早做 RAG 项目的时候用的是最朴素的组合向量库加一个检索接口拼一段提示词丢给 LLM能跑通就上线。Demo 阶段确实很爽但一旦进入真实业务问题就集中爆发了。最典型的三类第一检索回来的内容质量不稳定同一个问题换个问法召回结果天差地别第二多轮对话里上下文管理混乱用户追问“那第二个呢”系统根本不知道“第二个”指的是什么第三需要调用外部工具的时候整个流程就变成了一堆 if-else 硬编码改一处牵动全身。这些问题的本质是检索、编排、生成这三件事被揉在了一个函数里。Haystack 擅长的是检索管线的工程化——文档预处理、切分、嵌入、索引、检索、重排它有一套成熟的 Pipeline 抽象每个节点职责清晰可以单独替换和调试。但 Haystack 在复杂的状态流转、条件分支、循环重试这些“控制流”层面并不是它的强项。而 LangGraph 恰恰相反它把 LLM 应用建模成一张有状态图节点是计算单元边是流转逻辑支持条件边、循环、中断恢复、人工介入天然适合表达“检索不达标就改写查询再检索”这类带反馈的流程。所以把两者放在一起不是赶时髦而是各取所长Haystack 负责把知识库这条线做扎实LangGraph 负责把整个决策流程编排清楚。你可以理解为Haystack 是后厨负责把食材处理好、按需取用LangGraph 是前厅经理决定什么时候去后厨取什么菜、取几次、取不到怎么办。1.2 这套组合到底解决了什么问题具体来说这套架构解决的是**从“能回答”到“可靠地回答”**之间的鸿沟。我把它拆成三个层面检索层用 Haystack 的 DocumentStore 和 Retriever 做混合检索向量加关键词再用 Ranker 做重排保证召回的相关性。这一层的关键是可评估每次改动都能用命中率和 MRR 量化。编排层用 LangGraph 的 StateGraph 定义节点和边把“理解意图→检索→判断是否够用→不够就改写→再检索→生成”这条链路显式画出来。每个节点是一个纯函数输入输出都是状态字典的一部分调试的时候可以逐节点打印。上下文层这是最容易被忽视但最影响效果的部分。检索回来的文档怎么裁剪、怎么排序、怎么和对话历史融合、怎么控制 token 预算这些都属于上下文工程的范畴。LangGraph 的状态机制让上下文的流转变得可追踪你随时知道当前提示词里到底塞了什么。提示不要一上来就追求全自动的 Agent。先把“检索→生成”这条最短路径用 LangGraph 跑通再逐步加条件分支和工具调用。我见过太多项目一上来就设计五六个 Agent 互相调用最后连问题出在哪个环节都定位不了。1.3 适合谁来参考这套方案这套内容适合三类人一是已经用 LangChain 或 Haystack 做过 RAG Demo但上线后效果不稳定的开发者二是需要构建带工具调用能力的问答系统比如查数据库、调 API、做多步推理的场景三是对 LangGraph 的状态图模型感兴趣想找一个完整落地案例来学习的工程师。前提是你对 Python 比较熟知道什么是向量嵌入大致了解 LLM 的调用方式。如果这些还不清楚建议先补一下基础不然直接看编排部分会有点吃力。2. 核心组件选型与架构设计思路2.1 Haystack 在检索管线里的定位Haystack 2.x 的 Pipeline 设计比 1.x 清晰很多核心概念就几个Component 是节点Pipeline 把节点连起来每个 Component 的 run 方法接收和返回字典。做 RAG 检索最常用的组件组合是这样的from haystack import Pipeline from haystack.components.embedders import SentenceTransformersDocumentEmbedder, SentenceTransformersTextEmbedder from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever from haystack.document_stores.in_memory import InMemoryDocumentStore document_store InMemoryDocumentStore() doc_embedder SentenceTransformersDocumentEmbedder(modelBAAI/bge-base-zh-v1.5) doc_embedder.warm_up() text_embedder SentenceTransformersTextEmbedder(modelBAAI/bge-base-zh-v1.5) retriever InMemoryEmbeddingRetriever(document_storedocument_store, top_k10) pipeline Pipeline() pipeline.add_component(doc_embedder, doc_embedder) pipeline.add_component(text_embedder, text_embedder) pipeline.add_component(retriever, retriever) pipeline.connect(text_embedder.embedding, retriever.query_embedding)这里有几个选型上的考量。嵌入模型我选的是 BGE 系列的中文模型原因是它在中文语义相似度任务上表现稳定而且模型体积适中本地部署没有压力。如果你做的是英文为主的知识库可以换成all-MiniLM-L6-v2或者text-embedding-3-small。DocumentStore在开发阶段用内存版就够了生产环境换成 Elasticsearch 或 QdrantHaystack 对这些都有官方集成切换成本很低。为什么检索要单独抽出来而不是直接在 LangGraph 节点里调向量库因为 Haystack 的 Pipeline 支持序列化和反序列化你可以把整条检索管线存成 YAML部署的时候直接加载不用改代码。而且它的评估组件如DocumentMRREvaluator、DocumentRecallEvaluator能直接接在管线上跑指标这对持续优化检索质量非常关键。2.2 LangGraph 的状态图模型怎么理解LangGraph 的核心是StateGraph你可以把它想象成一张流程图每个节点是一个处理函数边定义了节点之间的跳转关系。和普通流程图不同的是它有一个共享状态在节点之间传递每个节点读取状态、修改状态、返回更新。from langgraph.graph import StateGraph, END from typing import TypedDict, List class RAGState(TypedDict): question: str rewritten_query: str documents: List[str] answer: str retry_count: int def rewrite_query(state: RAGState) - RAGState: # 用 LLM 改写查询 return {rewritten_query: new_query, retry_count: state.get(retry_count, 0) 1} def retrieve(state: RAGState) - RAGState: # 调 Haystack 检索 return {documents: docs} def generate(state: RAGState) - RAGState: # 生成答案 return {answer: answer} def should_retry(state: RAGState) - str: if len(state[documents]) 3 and state[retry_count] 2: return rewrite return generate 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, should_retry, {rewrite: rewrite, generate: generate}) graph.add_edge(generate, END) app graph.compile()这段代码里最关键的是should_retry这个条件边函数。它决定了检索结果不够好的时候是回到改写节点重试还是直接进入生成。重试次数上限一定要设不然遇到检索不到的内容会死循环。我一般设 2 次超过就直接生成并在答案里注明“根据现有资料可能不完整”。2.3 两者集成的边界在哪里集成的时候最容易犯的错是把 Haystack 的 Pipeline 整个塞进一个 LangGraph 节点里然后所有检索逻辑都在里面跑。这样做的后果是你失去了 LangGraph 对流程的细粒度控制检索过程中的中间状态比如改写后的查询、每个文档的得分都拿不到。我的做法是把 Haystack 的检索拆成两个节点一个负责查询改写和嵌入一个负责实际检索和重排。这样 LangGraph 可以在中间插入判断逻辑比如“如果改写后的查询和原查询语义差异太大就放弃这次检索”。同时Haystack 的 DocumentStore 和 Retriever 作为工具函数被调用而不是作为黑盒 Pipeline。def retrieve_node(state: RAGState) - RAGState: query state[rewritten_query] # 直接调 Haystack 组件而不是跑整个 Pipeline embedding text_embedder.run(textquery)[embedding] docs retriever.run(query_embeddingembedding)[documents] # 重排 reranked reranker.run(documentsdocs, queryquery)[documents] return {documents: [d.content for d in reranked[:5]]}这种拆法的好处是每个环节的输入输出都暴露在状态里调试的时候打印状态就能看到全貌。而且重排模型可以单独替换不影响其他节点。3. 上下文工程的关键细节与实操3.1 文档切分策略对检索质量的影响文档切分是 RAG 里最容易被低估的环节。我见过太多项目直接用固定长度切分结果把一段完整的逻辑切得七零八落检索回来的片段缺少上下文LLM 根本没法用。切分的核心原则是每个 chunk 要能独立表达一个完整的意思。具体怎么做我的经验是分三层处理按结构切如果文档有明确的标题层级Markdown、HTML优先按标题切每个小节作为一个候选 chunk。这样能保证语义完整性。按语义切对于没有明显结构的纯文本用句子边界检测加滑动窗口。窗口大小 256 到 512 token重叠 50 到 100 token。重叠是为了防止关键信息刚好落在边界上被切断。后处理切完之后给每个 chunk 加上它所属的章节标题作为前缀。这一步很关键因为检索的时候标题里的关键词能显著提升召回率。from haystack.components.preprocessors import DocumentSplitter splitter DocumentSplitter( split_bysentence, split_length5, # 每 5 个句子一个 chunk split_overlap1, # 重叠 1 个句子 split_threshold2 # 少于 2 个句子不单独成 chunk )注意split_length和split_overlap这两个参数没有万能值。中文文档句子短可以设大一点英文文档句子长要设小一点。我一般会拿一批真实问题跑一遍检索看命中率再调。3.2 上下文窗口的预算分配LLM 的上下文窗口是有限资源怎么分配直接决定回答质量。我的分配策略是这样的系统提示词占 10%对话历史占 20%检索文档占 50%当前问题占 5%剩余 15% 留给生成。这个比例不是死的但核心思想是检索文档必须是最大头因为它是事实依据的来源。对话历史的管理有个技巧不要把所有历史都塞进去而是只保留最近 N 轮并且对更早的历史做摘要。LangGraph 的状态里可以存一个conversation_summary字段每轮对话结束后更新它。def build_prompt(state: RAGState) - str: docs_text \n\n.join([f[文档{i1}] {d} for i, d in enumerate(state[documents])]) prompt f基于以下资料回答问题。如果资料中没有相关信息请明确说明。 资料 {docs_text} 对话摘要{state.get(conversation_summary, 无)} 问题{state[question]} return prompttoken 计数一定要做不能凭感觉。用tiktoken或者模型自带的 tokenizer 算一下超了就截断文档优先保留得分高的。我一般会留 200 token 的缓冲防止不同模型 tokenizer 差异导致超限。3.3 检索结果的重排与过滤向量检索召回 top_k 之后直接丢给 LLM 是很浪费的。因为向量相似度高不代表真的相关尤其是当知识库里有大量相似表述的时候。重排模型的作用就是用一个更精细的模型通常是交叉编码器对候选文档重新打分。from haystack.components.rankers import SentenceTransformersRanker ranker SentenceTransformersRanker(modelBAAI/bge-reranker-base, top_k5)重排之后我还会加一层相关性阈值过滤。如果最高分的文档低于某个阈值说明知识库里可能根本没有相关内容这时候应该触发“我不知道”的回复而不是硬编一个答案。这个阈值怎么定拿一批已知无答案的问题跑一遍看分数分布取一个能过滤掉大部分无答案问题的值。过滤策略优点缺点适用场景固定阈值实现简单不同查询分布不同难统一知识库主题单一相对阈值自适应查询需要至少一个高分文档知识库主题多样分类器判断准确率高需要标注数据训练对准确性要求极高我一般先用相对阈值最高分文档的分数乘以 0.7 作为底线低于这个线的都过滤掉。如果过滤后一篇不剩就走“无答案”分支。4. 工具合约与 LangGraph 工具调用实战4.1 工具合约的定义与校验工具调用是 RAG 系统从“问答”进化到“干活”的关键一步。但工具调用最容易出问题的地方是参数格式。LLM 生成的参数经常缺字段、类型不对、或者编造不存在的值。所以工具合约必须严格定义并且在调用前做校验。我用 Pydantic 来定义工具的参数模型from pydantic import BaseModel, Field, field_validator class SearchOrderInput(BaseModel): order_id: str Field(description订单编号格式为 ORD- 开头加 8 位数字) fields: list[str] Field(default[status, amount], description需要返回的字段) field_validator(order_id) def validate_order_id(cls, v): if not v.startswith(ORD-) or len(v) ! 12: raise ValueError(订单编号格式不正确) return v然后在 LangGraph 节点里先让 LLM 生成参数再用 Pydantic 校验。校验失败就把错误信息返回给 LLM让它重新生成。这个重试逻辑可以放在条件边里。def call_tool(state: RAGState) - RAGState: try: params SearchOrderInput(**state[tool_params]) result search_order_api(params.order_id, params.fields) return {tool_result: result, tool_error: None} except ValidationError as e: return {tool_error: str(e), tool_result: None} def should_retry_tool(state: RAGState) - str: if state.get(tool_error) and state.get(tool_retry, 0) 2: return regenerate_params return format_result提示工具描述里的参数说明要写得非常具体包括格式、取值范围、示例。LLM 对模糊描述的理解能力远不如你想象的那么强。我一般会在 description 里直接写“格式为 ORD- 开头加 8 位数字”这种硬约束。4.2 多工具场景下的路由策略当系统有多个工具可用时怎么决定调哪个最直接的方式是让 LLM 根据工具描述自己选。但实测下来工具超过 5 个之后选错的概率明显上升。我的做法是两级路由先用一个轻量分类器可以是小模型或者关键词匹配把问题分到某个大类再在该大类下的工具里让 LLM 选。def route_tool(state: RAGState) - str: question state[question] if any(kw in question for kw in [订单, 物流, 退货]): return order_tools elif any(kw in question for kw in [余额, 账单, 充值]): return account_tools else: return knowledge_retrieval这种硬路由看起来不够“智能”但在生产环境里可靠性比智能更重要。关键词匹配的准确率可以通过维护同义词表来提升而且出错了容易定位。纯靠 LLM 路由一旦选错你连为什么错都很难查。4.3 工具调用结果的回填与生成工具返回的结果不能直接丢给用户需要经过一轮格式化生成。因为 API 返回的通常是 JSON用户看不懂。这一步用 LLM 把结构化数据转成自然语言同时要确保数字和事实不能编造。def format_tool_result(state: RAGState) - RAGState: prompt f根据以下数据回答用户问题不要编造数据中没有的信息。 用户问题{state[question]} 工具返回数据{json.dumps(state[tool_result], ensure_asciiFalse)} 请用自然语言回答 answer llm.invoke(prompt) return {answer: answer}这里有个坑LLM 在格式化的时候可能会“顺手”加一些解释性内容比如“您的订单正在派送中预计明天送达请耐心等待”。如果数据里没有“预计明天送达”这就是幻觉。所以提示词里必须强调“不要编造”并且在评估环节加一条“事实一致性”检查。5. 常见问题排查与避坑经验5.1 检索命中率低的排查思路检索命中率低是最常见的问题排查要按顺序来先看嵌入模型是否匹配语言用中文模型处理英文文档或者反过来效果会差很多。确认模型和文档语言一致。再看切分是否合理把检索回来的 chunk 打印出来看是不是完整的句子或段落。如果经常出现半句话就是切分参数有问题。然后看查询改写是否有效把原始查询和改写后的查询都打印出来对比检索结果。如果改写后反而更差说明改写提示词需要调整。最后看重排是否起了反作用有些重排模型对特定领域不敏感重排后反而把相关文档排到后面。可以对比重排前后的命中率。现象可能原因解决方向召回文档完全不相关嵌入模型不匹配换模型或微调召回文档相关但缺关键信息切分太碎增大 chunk 或加标题前缀同一问题换问法结果差异大查询改写不稳定固定改写模板或加 few-shot重排后效果变差重排模型领域不匹配换模型或降低重排权重5.2 LangGraph 状态更新的常见错误LangGraph 的状态更新是合并式的节点返回的字典会和现有状态合并。这里有两个坑列表字段被覆盖如果状态里有个documents列表节点返回{documents: new_docs}原来的列表会被整个替换。如果想追加要用operator.add注解或者手动合并。状态字段类型不一致不同节点对同一个字段返回不同类型会导致后续节点报错。建议在 TypedDict 里把类型定义清楚并且每个节点返回前做类型检查。import operator from typing import Annotated class RAGState(TypedDict): documents: Annotated[list[str], operator.add] # 追加而不是覆盖 retry_count: int5.3 工具调用参数校验失败的处理参数校验失败是高频问题尤其是当 LLM 生成日期、金额、编号这类格式敏感的参数时。我的处理流程是第一次校验失败把 Pydantic 的错误信息原样返回给 LLM让它重新生成。第二次还失败就换一个更简单的提示词只让 LLM 提取关键字段格式由代码补全。第三次失败直接返回“无法理解您的请求请提供更具体的信息”并记录日志。这个降级策略能保证系统不会因为一个参数问题就完全卡死。日志里要记录原始问题、LLM 生成的参数、校验错误信息方便后续分析是提示词问题还是模型能力问题。5.4 上下文超限的应急处理上下文超限在长对话或检索文档多的时候经常发生。应急处理分三步截断优先截断对话历史保留最近 3 轮。检索文档按得分排序从低分开始删。摘要如果截断后还是超就把对话历史用 LLM 压缩成一段摘要。降级如果还超就减少检索文档数量从 top 5 降到 top 3并在答案里注明“基于部分资料回答”。预防措施是在构建提示词之前就算好 token 数留足缓冲。我一般会在状态里存一个token_budget字段每个节点消耗多少 token 都记录这样能提前发现哪个环节在膨胀。6. 从开发到生产的几个关键决策6.1 评估体系的搭建没有评估的 RAG 系统就是盲人摸象。我建议至少建三个评估集有答案的问题集、无答案的问题集、需要工具调用的问题集。每个集合 50 到 100 条覆盖主要业务场景。评估指标分两层检索层看命中率和 MRR生成层看答案准确率和事实一致性。事实一致性可以用 LLM as judge 来做让一个独立的 LLM 判断生成的答案是否完全基于检索文档。def evaluate_faithfulness(answer: str, documents: list[str]) - float: prompt f判断以下答案是否完全基于给定文档不包含文档外的信息。 文档{documents} 答案{answer} 只返回 0 到 1 之间的分数1 表示完全基于文档。 score float(llm.invoke(prompt)) return score6.2 日志与可观测性生产环境必须记录每个节点的输入输出和耗时。LangGraph 支持 callback可以在每个节点执行前后打点。我一般记录节点名、输入状态摘要、输出状态摘要、耗时、token 消耗。这些数据用来定位性能瓶颈和效果波动。import time def with_logging(node_func): def wrapper(state): start time.time() result node_func(state) elapsed time.time() - start logger.info(fnode{node_func.__name__} elapsed{elapsed:.2f}s) return result return wrapper6.3 版本管理与回滚RAG 系统的效果受很多因素影响嵌入模型版本、切分参数、提示词、重排模型。每次改动都要记录版本并且保留回滚能力。我的做法是把这些配置抽到一个 YAML 文件里用 Git 管理。部署的时候加载对应版本的配置出问题就切回上一个版本。retrieval: embedder: BAAI/bge-base-zh-v1.5 splitter: split_by: sentence split_length: 5 split_overlap: 1 retriever: top_k: 10 ranker: model: BAAI/bge-reranker-base top_k: 5 generation: model: gpt-4o-mini temperature: 0.1 max_tokens: 1024这套配置管理方式看起来简单但在实际运维中能省很多事。尤其是当你要同时维护多个环境开发、测试、生产的时候配置和代码分离是必须的。6.4 成本控制的几个手段LLM 调用成本在生产环境是实打实的支出。控制成本主要靠三招缓存、降级、批处理。缓存是对相同或相似查询直接返回历史结果用语义缓存比如 GPTCache比精确匹配命中率更高。降级是在非关键场景用更小更便宜的模型。批处理是把多个独立请求合并成一次调用适合离线评估和批量处理场景。我在实际项目里语义缓存能挡掉 30% 到 40% 的重复查询成本下降非常明显。但要注意缓存的失效策略知识库更新后要及时清理相关缓存。7. 我踩过的几个印象深刻的坑第一个坑是过度依赖 LLM 做路由。早期我让 LLM 自己决定走检索还是走工具结果它经常在该检索的时候去调工具在该调工具的时候去检索。后来改成规则加 LLM 的两级路由稳定性大幅提升。规则负责粗筛LLM 负责细选各司其职。第二个坑是忽略嵌入模型的维度匹配。有次换了一个嵌入模型忘了同步更新向量库的索引结果检索出来的全是随机结果。排查了半天才发现是维度对不上。现在我的检查清单里第一条就是“嵌入模型和索引维度是否一致”。第三个坑是提示词里塞了太多文档。一开始觉得给 LLM 的资料越多越好后来发现文档超过 8 篇之后LLM 反而抓不住重点答案质量下降。现在控制在 3 到 5 篇并且按相关性排序最重要的放最前面。第四个坑是没有做超时和重试。外部 API 调用偶尔会超时没有重试机制的话整个流程就断了。现在所有外部调用都包了重试装饰器超时时间设 10 秒重试 2 次指数退避。这些坑的共同点是看起来都是小问题但在生产环境里都会被放大。Demo 阶段跑通不代表生产可用中间差的就是这些细节的打磨。