企业级RAG系统实战：从LangChain链式调用到LangGraph图编排

先说明一句，这类从“文档进、答案出”的知识库系统，真正难的不是 LLM 调用，而是数据接入、分块策略、检索质量、流程编排这几层怎么串起来。早期我用 LangChain 的链式调用做 RAG，流程一复杂就到处打补丁：要重试得加回调，要做多轮得改状态，要分支判断得写 if else。后来换成 LangGraph 的状态图编排，才感觉这个架构跟企业级 RAG 的复杂度是匹配的。

这篇笔记基于我最近做的一个模拟项目——某设备制造商的内部技术知识库，数据源包括产品手册 PDF、售后记录 Excel、研发规范 Word 文档，目标是让售后工程师通过对话快速定位维修方案。全程不涉及具体商业产品，代码思路可直接迁移。适合已经跑通过简单 RAG Demo、想把系统做成能上生产环境的朋友参考。

1. 为什么从“链式调用”转向 LangGraph 图编排

先说结论：企业级 RAG 的流程不是一条直线，而是多条支线、条件分支、循环重试交织在一起的状态机。LangChain 的 LCEL 链适合三步五步的固定流程，一旦遇到以下场景，链式写法就开始吃力。

1.1 链式 RAG 的典型困境

普通 RAG 链路是“检索 → 拼接 Prompt → 生成”。但在企业场景里，真实需求远不止这三步：

• 用户提问“你们那个贴片机报 E102 错误怎么办”，系统先要判断这是产品咨询还是故障报修，走不同的处理分支。
• 检索结果太差时，需要自动改写查询词，重新检索一次，甚至循环多次。
• 售后手册里没有直接答案，可能要跨文档检索出厂检验记录、维修工单。
• 每次对话都要携带历史上下文，还要能回滚、能重试。

这些靠链式调用写出来，代码里全是if branch_a: return chain_b之类的硬编码，调试起来非常痛苦。

1.2 LangGraph 解决问题的切入点

LangGraph 的核心抽象是“图”：节点是处理逻辑，边是流转关系，所有数据通过一个共享的 State 对象传递。它的几个特性刚好命中企业 RAG 的痛点：

• 支持条件路由，节点可以根据状态决定走哪条边，典型场景就是“查询改写后判断是否需要二次检索”。
• 原生支持循环，也就是“检索 -> 判断 -> 再检索 -> 再判断”这种循环依赖，在链式框架里要实现得很绕。
• 通过 Checkpointer 机制持久化状态，天然支持多轮对话和断点恢复。
• 可视化调试体验好，编译出来的图可以直接画出执行路径，定位问题比看一堆日志高效得多。

说句实在话，如果你的项目只是给个人博客做个问答机器人，LangGraph 确实有点重。但一旦涉及多源数据、权限过滤、带确认流程的复杂交互，用图编排是值得的。我做这个项目时发现，到了后期加需求，图结构的扩展成本明显比链式低。

1.3 选型时的边界条件

我选 LangGraph 还有一层考虑：团队里其他系统已经基于 LangChain 生态，LangGraph 可以复用已有的组件和工具封装，不用从零搭一套。如果你的项目已经深度依赖别的编排框架，倒也不必为了同样功能强行迁移，先评估流程复杂度再决定。

2. 数据读取层设计：多格式文档的加载与分块

标题里“从数据读取”这几个字，恰恰是大部分 RAG 项目翻车的起点。我见过不少团队把精力全放在模型调参上，最后发现回答不准的根因是索引里的文档被切碎了、表格内容丢了一半。

2.1 多格式统一加载策略

制造企业的文档来源复杂：PDF 有扫描版也有文字版，Word 里嵌套表格，Excel 里是多 Sheet 的参数清单。我当时做了一件事：建立一层“统一文档加载器”，每个格式一个适配器，输出统一的数据结构。

def load_document(file_path: str) -> dict: ext = pathlib.Path(file_path).suffix.lower() if ext == ".pdf": return load_pdf(file_path) elif ext == ".docx": return load_docx(file_path) elif ext == ".xlsx": return load_excel(file_path) elif ext in {".md", ".txt"}: return load_plain_text(file_path) else: raise ValueError(f"不支持的文档类型: {ext}")

统一结构至少包含三个字段：text（文本内容）、metadata（来源文件、页码、章节标题、更新时间）、page_info（分页或分区信息，后续引用的关键）。扫描版 PDF 必须提前走 OCR，否则后面的分块和检索全是无源之水。

关键经验：表格类的 Excel 不要直接整表转文本，而是按 Sheet + 表头语义组织。比如一张设备参数表，转成“设备型号: XK-2000, 气压范围: 0.4-0.6MPa, 重复精度: ±0.02mm”这类扁平文本，检索效果比硬切表格单元格好很多。这是我在项目中试了四五种方案之后确定的格式。

2.2 分块参数：chunk_size 与 overlap 的取舍逻辑

分块是 RAG 最容易忽略但影响最大的环节。块太小，上下文不完整，检索到的片段答非所问；块太大，向量相似度被噪声稀释，还浪费上下文窗口。我的经验是先用文档结构分块，再用字符数做兜底限制。

实操中我用的是递归字符分割器，设定 chunk_size=512、chunk_overlap=64，再按标题层级做硬分隔。具体来说：

• 每个二级标题下的内容优先作为一个完整语义块。
• 如果超过 512 字符，按段落边界继续切。
• overlap 的作用是让跨块上下文不丢，尤其适合表格换行和长句跨块的情况。

补充一条：代码块、列表项密集的文档，建议把分隔符优先级调整一下。默认的["\n\n", "\n", " ", ""]在遇到代码缩进时会把语义切得稀碎，我改成["\n\n", "\n", "；", "。 ", " "]之后效果提升很明显。

2.3 元数据设计：企业检索的隐藏变量

很多人做 RAG 只往向量库里塞文本，忽略了元数据。实际上元数据是权限控制、时间过滤、来源追溯的基础，也是提升检索精度的利器。

我的元数据 Schema 大致如下：

• doc_id：文档唯一 ID
• source：原始文件路径或 URL
• title：文档标题
• chapter：所属章节
• last_modified：最后修改时间
• version：版本号
• permission_group：权限组（对应企业里的部门或角色）

有了permission_group，检索阶段就能对向量库做预过滤，只有当前用户有权限的文档才能进入候选集。这个点如果一开始不做架构设计，后期补会非常痛苦。我们项目早期没做权限过滤，上线内测时一个售后工程师能把研发部的未公开规范搜出来，虽然只是内部测试，但已经给团队敲了警钟。

3. 向量化与检索底座：构建高召回率的索引

数据读取完，接下来是 Embedding 与向量库。这一步决定系统“能查到什么”，直接影响最终回答质量的上限。

3.1 Embedding 模型选型：通用模型与领域模型的平衡

Embedding 模型决定文本相似度计算的质量。企业知识库常有大量专业术语，比如“贴片机”和“SMT 设备”语义相近，但通用向量模型未必能理解。我的做法是分两步走：

• 先用通用中文 Embedding 模型跑通流程，验证整体链路没有问题。
• 再收集领域语料做微调训练，或者直接选用在工业垂直领域表现更好的模型。

实操中，我用一个“术语表增强”的小技巧：把常见同义词和缩写整理成映射表，在查询改写阶段把“贴片机”扩展成“贴片机/SMT 设备/表面贴装设备”。这个做法成本低，见效快，比一开始就微调模型务实得多。

3.2 向量库选型：从开发到生产的平滑演进

我经历过从原型到生产的迁移，所以选型时会特意看这三项指标：

开发阶段可以用轻量级向量库，秒级启动、API 简单，几十万向量够用。但到了生产环境，数据集上千万、并发查询上百，就必须用分布式向量库。这个切换过程需要提前在客户端做一层抽象，避免代码跟具体存储绑死。否则生产切换等于重写检索模块。

3.3 混合检索：关键词与向量的互补

纯向量检索有个经典问题：专有名词和型号代码的精确匹配常常失败。“E102 错误”这类短代码，向量相似度不如关键词命中可靠。因此我在项目中做了混合检索——向量检索召回的 TopK 结果和 BM25 关键词检索召回的 TopK 结果做融合，再统一送去做重排序。

融合策略我用的是最简单有效的 RRF（Reciprocal Rank Fusion）：

def rrf_fusion(results_list, k=60): scores = {} for results in results_list: for rank, doc_id in enumerate(results): scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank) return sorted(scores.items(), key=lambda x: x[1], reverse=True)

RRF 不依赖分数绝对值，天然解决不同检索方式分数不可比的问题。实测混合检索比纯向量检索的召回率提升了大约 17%，尤其在型号、故障码这类短查询上提升显著。

4. 检索增强链路：查询改写、重排序与生成优化

召回上来了，不代表回答能直接见人。企业用户问得很随意，检索模块却需要精确的查询语义。这个“用户问题”到“检索意图”的鸿沟，需要几层机制填补。

4.1 查询改写：从口语问题到检索条件

用户说“机器老报警怎么弄”，系统直接拿这句话去检索，很难命中“E102 错误代码处理方法”。查询改写要做的是：把口语化的、缺少上下文的提问，转化为结构化的检索查询。

我在图里加了一个rewrite_node，输入原始问题，输出两到三条改写后的查询语句。改写规则有几种思路：

• 补全上下文：承接多轮对话中提到的设备型号、故障现象。
• 术语归一化：把“报警”映射到故障代码体系里的“错误代码 E102”。
• 拆解复合问题：“这台机器怎么调压力，还有耗材多久换”，拆成两个独立查询。

这个步骤看似是 LLM 调用，其实背后的规则和领域词表才是关键。团队整理的术语表越细，改写质量越高。

4.2 重排序：粗粒度召回之后的精排

混合检索召回的候选集通常是 Top20 甚至 Top50，如果直接拼进 Prompt，既浪费 token，又引入大量无关信息。必须经过重排序这一步，用更精细的语义匹配模型把最相关的几个文档块找出来。

重排序模型和 Embedding 模型是两个层次的模型。Embedding 追求“快速筛掉大部分无关内容”，Rerank 追求“在较少的候选中精确排序”。项目里我设定召回 Top20 进 Rerank，最终只保留 Top4 给生成模块。从效果看，答案的相关性、引用准确性都有明显提升。

4.3 生成 Prompt 优化：让模型学会“不知道就承认”

答案生成阶段，我的核心经验是两点：引用落地、明确边界。

• 引用落地：要求模型在回答中引用具体知识块编号（比如“根据《维修手册》第 12 页”），我使用结构化 Prompt 强制模型把证据来源放进输出字段。
• 明确边界：企业场景里，模型不懂装懂是最危险的。Prompt 里明确写“如果知识库中没有依据，直接说不知道，不要推测”。

还有一个关键技巧：把问题类型从 Prompt 里显式传入。我用的生成模板里包含question_type字段，系统先判断用户问题属于“故障排查”“参数查询”“操作指导”哪个类型，不同类型采用不同的回答格式。故障排查类需要步骤化输出，参数查询类则直接给表格。这个设计让输出稳定很多，用户体验也好了不少。

5. LangGraph 图编排实战：状态、节点与条件边

这部分是标题里的重头戏。用一个实际例子走一遍 LangGraph 的图设计：多轮对话、检索失败重试、条件路由全部包括。

5.1 Graph 的核心概念：State、Node、Edge

先明确三个概念：

• State（状态）：整个图的全局共享数据，由 TypedDict 或 Pydantic 模型定义。
• Node（节点）：一个接收 State 并返回 State 更新的函数。
• Edge（边）：确定节点之间的流转关系，分为普通边和条件边。

下图是模拟项目的核心架构（用文字描述，不依赖图形工具）：

入口节点router_node判断意图，走rewrite_node或直接走retrieve_node；检索结果经judge_node判断质量，不够好则循环回rewrite_node重新检索，最多两轮；通过则进rerank_node、generate_node，最后输出。

整个图的定义代码如下：

from typing import TypedDict from langgraph.graph import StateGraph, END class RAGState(TypedDict): question: str history: list rewritten_question: str docs: list retry_count: int answer: str sources: list def router_node(state: RAGState) -> dict: # 示例逻辑：判断是否需要查询改写 if needs_rewrite(state["question"]): return {"rewritten_question": ""} return {"rewritten_question": state["question"]} def rewrite_node(state: RAGState) -> dict: written = rewrite_query(state["question"], state["history"]) return {"rewritten_question": written} def retrieve_node(state: RAGState) -> dict: docs = hybrid_search(state["rewritten_question"], top_k=20) return {"docs": docs} def judge_node(state: RAGState) -> dict: if not state["docs"] and state["retry_count"] < 2: return {"retry_count": state["retry_count"] + 1} return {} def rerank_node(state: RAGState) -> dict: top = rerank_docs(state["rewritten_question"], state["docs"], top_k=4) return {"docs": top} def generate_node(state: RAGState) -> dict: answer, sources = generate_with_context(state["question"], state["docs"]) return {"answer": answer, "sources": sources} # 搭建图 graph = StateGraph(RAGState) graph.add_node("router", router_node) graph.add_node("rewrite", rewrite_node) graph.add_node("retrieve", retrieve_node) graph.add_node("judge", judge_node) graph.add_node("rerank", rerank_node) graph.add_node("generate", generate_node) graph.set_entry_point("router") graph.add_edge("router", "rewrite") graph.add_edge("rewrite", "retrieve") graph.add_edge("retrieve", "judge") # 条件边：检索结果不佳时循环重试 graph.add_conditional_edges( "judge", lambda state: "retry" if state.get("retry_count", 0) > 0 else "continue", {"retry": "rewrite", "continue": "rerank"}, ) graph.add_edge("rerank", "generate") graph.add_edge("generate", END) app = graph.compile()

5.2 条件路由和循环的思考

这个图最关键的决策在judge_node到rewrite_node的重试回路。为什么不用简单 if else？因为重试不是一步，它需要重新改写、再检索、再判断，相当于在一个图内部做子循环。用状态图表达这个流程，每个环节的数据流和状态变更都很清晰，后期要加“重试次数上限”或者“多轮重写”都很容易。

retry_count的设计也是一个细节经验。很多人把重试逻辑写成死循环，缺乏上限控制。我在状态里加了一个计数器，最大重试两次。原因很现实：重试不是越多越好，超过两轮，检索结果通常已经稳定，再改写反而会把问题带偏。

5.3 多轮对话状态管理：Checkpointer 的接入

企业场景里，多轮对话是刚需。用户可能先说“E102 错误”，下一句问“那压力怎么调”，系统需要把前文的设备型号、故障现象带入当前查询。这就是 State 持久化的用武之地。

开篇说到，LangGraph 通过 Checkpointer 机制保存每个会话的状态。接入方式是在编译时传入一个 checkpoint 持久化后端：

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) # 调用时传入线程 ID，相同 ID 共享历史状态 result = app.invoke( {"question": "那压力怎么调"}, config={"configurable": {"thread_id": "user-123"}} )

thread_id就是会话 ID。第二次提问时，history字段自动包含上一轮的上下文，rewrite_node就能据此补全指代。开发阶段用内存版 Checkpointer 没问题，生产环境一定要换成持久化版本，否则服务重启历史全丢。

写这块代码时有个容易踩的坑：字段更新必须用增量的方式返回，LangGraph 默认对新值做覆盖，对列表字段的追加需要显式设置add_to_state之类的注解。我第一次写时就忘了处理历史列表追加，结果多轮对话的历史一直在被替换。

5.4 图执行的失败重试与降级策略

图编排还有一个容易被忽略的价值：单点故障的可控化。我在retrieve_node和generate_node上加了超时和重试包装，比如向量库超时 3 秒，自动降级为纯 BM25 检索。这样即使向量服务抖动，系统也不会整体不可用，顶多是回答精度下降。

6. 企业级落地清单：评估、权限与成本优化

图编排跑通只是开始。真正上生产，还有三件事必须做，顺序不能乱。

6.1 评估闭环：没有评测集就没有迭代方向

不做评测就去调 Prompt、调参数，就像蒙着眼开车。我做了一个标准流程：建一个质量评测集，包含至少 100 条来自真实用户的问题，每个问题标注标准答案和期望引用的文档 ID。然后每次改动，批量跑一遍评测集，看三项核心指标：

• 答案相关性：回答是否准确命中问题点。
• 引用正确性：回答引用的文档是否真的是问题对应的内容。
• 无根据率：回答中没有知识库依据的占比。

这套评测集帮我发现了不少问题，比如某类问题经常检索到过时文档，就是因为元数据里没做版本过滤——只保留last_modified最近且version最高的文档进入检索范围，效果立竿见影。

6.2 权限控制：企业知识库的隐形边界

这层动作必须在检索阶段完成，不能指望 LLM 判断文档是否可答。我采用的做法是：先根据用户身份获取权限组列表，再在检索时对元数据做强制过滤。

一个常见的坑是：权限过滤在查向量库之前做预过滤，可能因为过滤条件过严导致候选集过少。我的处理是放宽预过滤，在重排序后、生成前再做一次硬过滤，双重保险。避免“查到但是不能用”的尴尬。

6.3 成本与性能优化：缓存优先，重算兜底

企业知识库查询有大量重复模式——几十个工程师问到的可能是同一个问题。针对这点，我做了两层缓存：

• 查询改写结果缓存：改写逻辑相对稳定，相同问题不必每次调用模型。
• 生成结果缓存：完全相同的问题直接返回历史答案，按会话和文档版本做失效控制。

实测加上缓存之后，高峰期的 LLM 调用量下降了约 40%，响应时间从 5 秒降到 1 秒左右。唯一的注意点是缓存一致性：知识库文档更新时，相关缓存必须及时失效。我简单粗暴地按文档版本号做缓存键的一部分，保证旧版本更新后答案不会继续复用。

总结一下项目中的关键取舍

这个项目做下来，我最深的体会是：企业级 RAG 的瓶颈通常不在模型，而在数据工程和流程编排。数据读取得不干净，后面所有环节都白搭；流程编排不灵活，需求稍微一复杂就变成补丁工程。LangGraph 的道理并不复杂——它只是逼你把流程当作有状态、有分支、可循环的系统来设计，而不是一条直线写到底。

最后分享一个实用技巧：图编排上线之前，先在本地把每条节点的输入输出保存成 JSON，跑一批典型用例，人工检查每个节点的流转是否符合预期。我靠这个习惯，上线前拦下了至少三处条件边逻辑错误，省掉了生产环境排查的狼狈。如果你的 RAG 也准备走向生产，我建议从“数据读取规范化”和“图编排状态管理”这两个点入手，先把骨架夯实，后续优化就有依据了。