企业官网建设流程全解析

第一次正经用 LangGraph 搭企业级 RAG 知识库，还是在一次内部项目里被逼出来的。当时我手里的系统还是老一套“文档切块 → 向量化 → 检索 → 拼 Prompt → 让大模型回答”的线性管道，跑 demo 没问题，一上生产就痛苦：检索不准、流程不可控、临时要加一个“判断问题需不需要检索”的分支，就得改整条链路的代码。后来我用 LangGraph 把整套流程重写成了图编排，才真正体会到什么叫“把 RAG 当系统设计，而不是当脚本写”。

这篇笔记不是概念科普，而是完整记录我从数据读取、清洗分块、向量化入库，到用 Graph 图编排实现检索问答全流程的实战过程。里面有我的选型理由、参数设置、踩坑记录，也有适合直接抄作业的代码片段。它适合已经跑通过基础 RAG、准备把系统推向真实业务场景的开发者，也适合正在犹豫要不要把工作流迁到图编排框架的人。我尽量把每个“为什么这么做”都讲清楚，读完你应该能动手复现一套完整的企业级 RAG 底座。

1. 项目背景与整体设计思路

1.1 先说说我为什么弃用线性管道

传统 RAG 链路长这样：文档 → 分块 → 向量化 → 检索 → 生成。每个环节是硬编码的先后调用，看着简单，实际维护起来非常痛苦。举个例子，用户问了一个纯闲聊问题“你是谁开发的”，系统也会老老实实去向量库检索一遍，然后拿着毫不相干的上下文去给大模型，最后模型只能根据一堆文档碎片胡编。我当时的解决方案是在代码里加一个if判断，走分支绕开检索。这个判断本身不难，但业务方随后提出了更多需求：某些问题需要先查数据库拿到实时数据，再结合知识库回答；某些多轮对话场景需要先判断用户是否在追问上一轮的结果，是的话直接用历史，不是再触发新的检索。

这些需求本质上都是“流程分支”和“流程编排”，用线性管道实现，就得不断堆if/else、不断传各种临时变量，代码越来越难读，而且状态分散在各处，很难统一管理和排查。LangGraph 让我眼前一亮的点，是它把流程本身变成了一张显式的图：节点是具体操作，边是流向关系，节点之间通过共享状态传递数据。你可以随时在图中插入新节点，也可以为某条边配置条件，让流程根据输入自己决定走向。我觉得它的定位不是“又一个 RAG 框架”，而是一个通用的事件状态编排内核。

1.2 整体架构：离线索引与在线查询两条线

我在设计这个知识库时，将整个系统分成了两条链路：离线索引链路和在线查询链路。

离线链路负责数据准备：读取原始文档 → 清洗 → 分块 → 向量化 → 写入向量库。在线链路负责用户问答：接收问题 → 判断是否需要检索 → 检索召回 → 重排 → 拼装上下文 → 大模型生成 → 返回答案。离线链路我甚至也用 LangGraph 编排了一版，而不是写成普通脚本。这样做的收益是，当文档处理流程本身也需要分支时——比如扫描版 PDF 要走 OCR，文本版直接解析——我用同一套图编排机制就能表达，而不需要额外维护一套定时任务代码。

在线链路的核心图结构，我用文字画一下：

[开始] → [问题路由节点] │ ├─ 需要检索 → [多路召回节点] → [重排节点] → [生成节点] → [结束] │ └─ 不需要检索 → [生成节点] → [结束]

这只是一个最基础的骨架，真实项目里还可以继续扩展：把“检索”拆成向量检索 + 关键词检索两个并行节点，把“生成”拆成“准备 Prompt + 流式输出”两个节点。每条边、每个节点都可以单独加日志、单独设置超时和重试。这种可插拔的灵活性，是线性代码很难给的。

1.3 图编排的核心设计决策

做图编排前先想清楚几个决策点，想清楚能少填很多坑。

第一个决策是状态模型。LangGraph 的节点之间通过一个共享的 State 对象传递信息，这个对象可以是字典，也可以用 TypedDict 定义强类型结构。我为每条链路单独定义了状态结构，比如查询链路的 State 包含question、need_retrieval、contexts、answer字段，每个节点只关心自己需要的字段，避免节点间隐式依赖。

第二个决策是节点粒度。节点切得太粗，比如一个“retrieve_and_generate”节点把检索和生成揉在一起，就丧失了图编排的可控性；切得太细，比如把每一行 Prompt 拼接都做成节点，又会让图的维护成本暴涨。我的经验是以“业务可独立调整的环节”为粒度，检索、重排、生成、路由各自独立，因为它们后面大概率会被单独替换或扩展。

第三个决策是运行时状态的可恢复性。LangGraph 支持通过 Checkpointer 保存每一步执行的状态，这对我这种经常要调线上问题的人来说太重要了。出问题时直接把某个会话的历史状态拉起来看，甚至能从断点恢复执行，而不需要让用户重新输入一遍问题。

2. 数据读取与预处理：脏活累活全在这

2.1 多格式文档加载

企业知识库的文档格式五花八门，我第一次接数据时就遇到了 PDF、Word、Markdown、HTML、CSV、扫描图片等六种以上格式。不同格式要选择不同加载策略，核心标准是“能不能把结构化信息保真地提取为纯文本”。

文本型 PDF 通常可以直接按页提取文字，但要注意分栏排版、页眉页脚混入正文的问题。我当时的处理策略是，先探测文本布局，检测到分栏就按栏重组文本，页眉页脚通过正则和位置信息过滤；扫描版 PDF 没有文本层，必须接 OCR 识别。这里我踩过的坑是：OCR 输出的文本没有段落结构，经常把标题和正文混在一起，导致后续分块效果很差。后来我加了“版面分析”环节，先把页面划分成标题、段落、表格区域，再分别提取，分块质量才上来。

Word、HTML 这类半结构化文档，隐含的层级信息是很有价值的，比如标题的h1/h2、表格的表头和单元格，这些信息如果全部拍平成纯文本就浪费了。我的做法是将它们转换成带轻量标记的 Markdown，保留标题层级和表格结构，这样后续分块时就能按标题边界和表格边界切分。

2.2 文本分块：chunk_size 和 overlap 怎么定

分块决定了检索的颗粒度，也是 RAG 里对效果影响最直接的一步。我先说结论：不要用固定长度硬切，优先考虑按语义边界切，比如段落、章节标题、列表项。固定长度切分最大的问题是会把一个完整的知识点从中间切断，检索时拿到一半内容，生成时自然缺上下文。

具体参数上，chunk_size 和 chunk_overlap 需要根据你的知识库文档特点来定。如果文档结构清晰、段落较长，比如技术手册，可以把 chunk_size 设在 600～800 token，让每个分块包含足够上下文；如果文档是碎片化的问答对或者产品短描述，300～500 token 更合适。chunk_overlap 是相邻分块之间重叠的部分，一般设为 chunk_size 的 10%～15%。它存在的原因是避免被边界切断的关键词恰好落在两块的边缘，导致检索不到。代价是重复内容会占用存储空间和 token。

我的做法是同时维护两种分块：大块用于生成（feed to LLM），小块用于检索（query to vector store），也就是常说的“父子分块”。查询时先用小块做向量检索拿高精度结果，再通过文档结构映射回它所属的大块，用大块内容去喂给大模型。这样既能精准定位用户关心的小片段，又能保证模型拿到完整上下文。

2.3 数据清洗与元数据设计

清洗的必要性容易被低估。我接手的生产文档里，隐藏字符、全角半角混乱、多余空行、扫描件水印几乎无处不在。这些噪声不仅是质量问题，还会直接影响向量化效果。清洗流程我固定为：转码统一 → 去除控制字符 → 清理页眉页脚水印 → 规范空白字符 → 按需纠正 OCR 错字。

元数据设计是另一个容易被忽略的环节。每个分块入库前我都打上这些标签：文档来源、文档版本、最后更新时间、所属业务线、访问权限级别、作者、大块 ID。这些标签后期会用于过滤检索结果。比如权限级别字段，查询时会把它作为一个硬性过滤条件，保证高敏感数据永远不会被无权限用户检索到。如果没有这个设计，知识库在企业环境里根本没法上线，合规这关就过不了。

3. 向量化与入库：检索效果的地基

3.1 Embedding 模型选型

检索效果三分靠分块，七分靠 Embedding，这话不夸张。选型时我主要看四个维度：领域适配度、文本长度上限、向量维度、部署方式。

领域适配度务必用小样本实测，不要只看公开榜单。我用过通用型 Embedding 模型，它的整体效果不错，但一旦涉及企业内部的专有名词、产品型号缩写，检索相关度就明显下降。后来我在内部语料上做了对比测试，换成了对垂直领域更友好的向量模型，同一个查询的 Recall@5 提升明显。具体操作是：准备一批带标准答案的问答对，用hit_rate或者Recall@K做离线评估，用数据说话，而不是靠感觉。

文本长度上限决定了你和分块策略的配合方式。部分 Embedding 模型只支持 512 token 输入，超过上限会被截断，导致长文档的语义丢失。我选的模型支持更长上下文，配合父子分块策略，效果更稳。

部署方式上，当时我优先考虑了自托管开源模型，主要原因是数据不出内网。千万不要忽略这一点，企业知识库的内容往往有保密要求，直接调用外部向量化接口会有数据合规风险。

3.2 向量库选型与索引参数

向量库选型，我总结成一个三角权衡：性能、功能、运维成本。

纯内存方案，比如一些简单的向量索引实现，启动快、适合 demo，但数据量大以后内存吃紧，且缺少增删改查的完善语义，不适合生产。分布式向量数据库虽然运维有成本，但支持高并发、持久化、复杂过滤条件和权限控制，企业场景建议直接上这类方案。如果团队规模小，也可以用支持向量检索的开源数据库，它是关系数据库和向量检索的结合体，业务数据结构化的时候最省事。

索引参数的调优我有过一段痛苦的经历。HNSW 算法里有两个参数影响巨大：M和efConstruction。M控制每个节点的最大连接数，调大能提高召回率，但会显著增加内存占用和构建时间；efConstruction控制构建时考虑的候选数，调大质量更高，但构建变慢。我当时在 200 万条向量上用M=16, efConstruction=100，内存花了约 2GB，召回效果好；后来数据涨到 1000 万，只能把M调到 12 才勉强控制住资源，代价是召回率略降。建议按业务数据量做多组实验，找到“够用且不过度牺牲性能”的参数组合。

3.3 批量写入、增量更新与权限过滤

离线索引时我一次性写过几十万条分块，初期直接 for 循环单条插入，写了一个下午没写完。后来改成批量写入，每条请求带上 200～500 条向量，用并发池控制速度，几十万条半小时内就写完了。批量写入时要注意向量库的吞吐上限，并发太高会把资源打满，影响线上查询，建议给离线写入单独限流。

增量更新比全量重建复杂得多。我的做法是按文档版本做 diff：文档变更后，只重新处理变更部分，旧分块通过元数据里的文档版本字段批量过期。具体实现是在库里维护一个“文档版本表”，查询前先过滤掉非最新版本的分块。这样避免每次更新都推倒全库重建，省下大量计算和存储成本。

权限过滤建议放在查询链路的第一层。节点拿到用户身份后，先转换成权限标签，然后带着标签去检索，在向量库查询语句里追加过滤条件。这个设计比“全部检索完再在应用层过滤”安全且高效，因为后者会先把高敏感内容召回出来，只是显示端做了遮挡，风险面太大。

4. LangGraph 图编排全流程实现

4.1 核心概念速览

LangGraph 的基本概念可以当成流程图的元素来理解：State 是整个流程共享的数据字典，Node 是执行具体处理的函数，Edge 是两个节点之间的跳转关系，Conditional Edge 是根据 State 当前内容动态选择下一跳的边，Checkpointer 则负责把每一步执行后的状态落盘。

节点函数遵循一个约定：输入一个 State，返回一个包含更新字段的字典。LangGraph 会用返回值去更新 State，而不是替换整个 State。不同的节点的返回值可以只更新各自关心的字段，这降低了节点之间的耦合。

先看一个最基础的状态定义：

from typing import TypedDict, Annotated, List class RAGState(TypedDict): question: str # 用户原始问题 need_retrieval: bool # 是否需要检索 contexts: List[str] # 检索到的上下文片段 answer: str # 最终答案

4.2 定义节点：路由、检索、生成

路由节点判断用户问题是否需要检索。我内部维护了一个小型的判断逻辑，把闲聊、问候、知识库范围外的问题归到“免检索”类别。判断方式可以是一组规则，也可以是一个轻量分类模型，初期用规则最省事。

def route_node(state: RAGState) -> dict: q = state["question"] # 这块可以换成更复杂的意图识别逻辑 if is_greeting(q) or is_chitchat(q): return {"need_retrieval": False} return {"need_retrieval": True}

检索节点做多路召回。我只写核心逻辑，以一个向量检索为示例：

def retrieve_node(state: RAGState) -> dict: if not state["need_retrieval"]: return {"contexts": []} query = state["question"] # 单路向量召回，k 值先设 10，后面靠重排筛成 3 docs = vector_store.similarity_search(query, k=10, filter={"permission": get_current_permission()}) # 截取文本内容，也可以在这里注入元数据 return {"contexts": [d.page_content for d in docs]}

生成节点负责组装 Prompt 并调用大模型。如果上下文为空，就直接回答用户问题，避免把空上下文硬塞给模型：

def generate_node(state: RAGState) -> dict: question = state["question"] contexts = state["contexts"] if not contexts: answer = llm.invoke(question) else: prompt = ( "你是企业知识库助手，请严格基于以下资料回答用户问题。\n" f"资料：\n{chr(10).join(contexts)}\n" f"用户问题：{question}" ) answer = llm.invoke(prompt) return {"answer": answer}

4.3 图构建与条件路由

节点定义好之后，就进入灵魂环节：把流程编排成图。我用StateGraph创建图对象，依次注册节点、连接边、配置条件边：

from langgraph.graph import StateGraph, START, END def should_next(node_state: RAGState) -> str: if node_state["need_retrieval"]: return "retrieve" return "generate" graph = StateGraph(RAGState) graph.add_node("route", route_node) graph.add_node("retrieve", retrieve_node) graph.add_node("generate", generate_node) graph.add_edge(START, "route") graph.add_conditional_edges( "route", should_next, {"retrieve": "retrieve", "generate": "generate"}, ) graph.add_edge("retrieve", "generate") graph.add_edge("generate", END) app = graph.compile()

执行的时候调用：

result = app.invoke({"question": "售后保修政策是什么？", "need_retrieval": False, "contexts": [], "answer": ""})

这里要重点理解add_conditional_edges的作用：它接收一个回调函数should_next，根据 State 的当前内容返回一个键名，然后对照映射表决定跳转到哪个节点。这比我之前用if/else串联所有逻辑来得清晰太多，新增分支时只需要改图和映射表，不动节点内部逻辑。

4.4 检索增强：多路召回与重排

单靠一路向量检索，在生产环境是不够的。典型问题有两个：一是向量检索对精确匹配不敏感，用户搜“型号 ABC-200”这种带连字符和数字的关键词，向量相似度排名反而不如 BM25 关键词检索；二是向量检索召回的 TopK 里噪声太多，直接喂给大模型会稀释有效信息。

我的方案是把检索节点升级成多路召回。一路走向量相似度，一路走全文关键词（BM25），再通过 RRF（Reciprocal Rank Fusion）算法融合两路结果。具体实现是在检索节点内部同时发起两种查询，然后按排名归一化打分融合。

重排节点单独建一个节点，放在检索之后、生成之前。它用小而精的排序模型对召回结果重新打分，去掉与问题无关的片段，把最相关的 3 个片段送回生成节点。这样做的收益是：允许召回阶段多取一些候选（比如 10 个），给召回率留足余量，再把精度交给重排环节去控制。图结构上只加一个节点和一条边，极大降低了扩展成本。

4.5 多轮对话记忆

企业知识库问答很少是单轮的，用户会追问、会纠正、会引用上一轮的答案。早期版本我用全局变量存历史记录，一旦并发上来就乱了。LangGraph 的 Checkpointer 机制解决得很干净：它按会话thread_id保存状态，每个会话共享同一份状态快照。

开启多轮记忆只需在编译时传入 Checkpointer，并在调用时指定thread_id：

from langgraph.checkpoint.memory import InMemorySaver app = graph.compile(checkpointer=InMemorySaver()) result1 = app.invoke({"question": "产品 X 的保修期是多久？"}, config={"configurable": {"thread_id": "session-001"}}) result2 = app.invoke({"question": "那逾期续保怎么收费？"}, config={"configurable": {"thread_id": "session-001"}})

第二次提问时，“那逾期续保”中的“那”字，全靠第一轮上下文才能正确理解。我在生成节点里把历史记录也拼进 Prompt 即可。注意 Checkpointer 的存储介质，跨容器部署时建议用持久化存储，而不是内存实现，否则会话重启就丢了。

5. 企业级增强：流式、可观测与容错

5.1 流式响应

把 LangGraph 应用接到前端时，我更推荐用stream而不是invoke。invoke会等到整张图跑完才返回最终答案，用户在这个过程里只能干等，体验非常差。用stream则能拿到每个节点执行后的状态增量，特别适合做类 ChatGPT 的打字机效果。

LangGraph 的流式接口既可以按节点流式返回状态，也可以按 Token 返回大模型的生成内容。如果用的语言模型底层支持流式输出，我可以在生成节点里把 token 源源不断地推给前端。实现上，我用一个异步生成器读取流事件，前端通过 Server-Sent Events 接收，在大模型开始输出后立刻“打字”，首字延迟从四五秒降到一秒钟以内。

5.2 可观测性设计

图编排系统最难排查的点是“流程跑到哪一步了”。我在每个节点入口和出口各打一条结构化日志，包含：会话 ID、节点名、当前状态摘要、耗时、本轮消耗的 token 数。日志统一打到链路追踪后端里，方便按会话 ID 拉出完整执行链路。

LangGraph 的状态快照对排查特别有用。节点 A 之后、节点 B 之前，一旦 B 输出异常，我可以直接捞出 A 的输出快照，确认是不是上游给了脏数据。这比线性代码里靠 print 大法逐个排查高效得多。我还给关键节点加了耗时阈值告警，比如检索节点超过 2 秒就报警，因为查询慢通常意味着向量库性能劣化或者索引参数需要重调。

5.3 容错与降级

企业系统里大模型和向量库都可能抖动，不能在节点里裸调外部服务。我统一封装了带超时和重试的调用器：连接超时 5 秒、读超时 30 秒，失败后指数退避重试三次。重试仍然失败就进入降级路径。

降级的思路是逐级放弃效果、保住基本体验：先尝试完整图流程，如果重排节点挂了，就跳过重排直接用召回结果生成；如果向量库查询超时，同时有 BM25 可用，就退化成纯关键词检索；如果大模型生成失败，但召回出了有价值的片段，就把检索片段整理后直接返回，至少让用户拿到参考信息。换一种说法是：RAG 系统的核心资产之一是检索结果，不要让生成环节的故障把检索资产也吞掉了。

具体实现上，我在每个节点外层包了一层安全执行函数，捕获异常后根据节点类型决定继续走降级分支还是直接终止流程。LangGraph 的条件边很自然地支持了这种“走后门”路径，我在图里为每个高风险节点都画了一条指向降级节点的备用边。

6. 常见问题与排查技巧实录

6.1 检索结果不相关

这是我用 RAG 以来遇到最多的一个问题，而且原因往往不在检索本身。某次线上用户反馈“查产品故障码，返回的全是无关介绍”，我把召回的前 10 条拉出来看，发现故障码文本被分块切断了，故障码本身落在上一块末尾，上一块内容全是设备型号介绍，导致向量检索时相似度信号错位。

排查顺序建议先看中间产物，不要猜。在检索节点出口打印召回的文本片段和对应的元数据，用眼睛看一下是否切题。如果侦查，再往下游查分块策略、Embedding 模型、查询改写。很多情况下，问题不在模型而在于查询和文档的语言表达不一致，例如用户口语说“坏了修不了”，文档里写的是“设备故障报修流程”，向量空间里这两句话相似度不高，这种情况需要在检索前加一个查询改写节点，把口语转成正式的业务术语。

6.2 图执行顺序异常或节点重复执行

有一次我发现同一个检索节点在一轮调用中被重复执行了三遍，第一反应是图结构连错了。排查后发现问题出在状态字段的合并策略上。LangGraph 里如果某个字段在多个节点里被更新，默认策略是“覆盖”，但如果把字段声明成了可累加类型（用于把多路结果合并成列表），又没控制好边的走向，就可能出现“一个节点追加一次、再经过另一条边又追加一次”的情况。

遇到这类问题：先把图结构画出来，用文本方式标清每一条边和条件分支；然后用app.get_state()查看某个会话最新状态，比对字段内容是否符合预期；最后给节点名加前缀序号，比如1-retrieve、2-rerank，日志里按节点名排序后，执行顺序一目了然。这种命名习惯帮我定位了好几次问题，建议保留。

6.3 Token 消耗虚高

大模型调用费是生产 RAG 的主要成本项，而浪费主要来自三个地方：多轮对话把全部历史记录都塞进 Prompt、检索阶段召回了大量无用片段、系统 Prompt 被反复重复注入。

我的优化有三板斧：历史记录只保留最近两轮完整问答，更早的用一轮摘要替代，长对话再叠加“关键信息抽取”压缩；检索召回从 10 个片段经过重排只留 3 个，且按相关度倒序排列，减少模型处理无关信息的开销；系统 Prompt 精简到必要边界，不把大段背景介绍写进每次请求。实测同样 1000 次问答，优化后 token 消耗降低了接近四成，回答质量没有明显下降。

6.4 一套排查落地的调试流程

我在本地排查问题时，不会直接盯着线上日志，而是写一套最小复现脚本。步骤是：固定一条查询问题 → 以离线方式跑一遍完整图流程 → 把每个节点的输入输出 dump 成 JSON → 对比线上结果。因为图编排的每个节点输入输出都是纯数据，天然方便离线回放，这是它比线性代码更适合生产调试的一个重要原因。

还有一个比较实战的建议：给图里的每个节点加上版本信息，比如retrieve_v2。模型和向量库升级后，新旧节点可以并存在同一张图里，通过配置切换版本。这样出问题的时候，能快速对比新旧逻辑差异，而不是直接回滚整个系统。

对我来说，这次用 LangGraph 重构 RAG 知识库最值的一笔投入，是把原来散落各处的if/else和临时变量收敛成了结构化、可观测、可恢复的图流程。节点即模块、边即路由，这套心智模型让后续每一次业务需求变更都变成“加节点、加边”，而不用牵动整条链路。如果你也在准备做企业级知识库，我的建议是别急着上复杂架构，先用手头的数据跑通最小闭环，再把路由、重排、记忆、降级逐层加进去，每一步都用真实查询验证效果。图编排不是银弹，但它确实把 RAG 的复杂度变成了一张可以看清、摸到、控制住的网。