★ FEATURED ARTICLE 写这个系列之前我一直觉得 LangChain 做 RAG 就是三板斧加载文档、切片、向量化检索最后把上下文拼进 Prompt 丢给大模型。直到需求开始变复杂比如用户会多轮追问、检索结果不好时需要自动重写问题、生成完答案后还要卡一道质量检查我才发现普通 Chain 那种线性编排已经撑不住了。这篇是“RAG 在 LangChain 中的实现”的第五篇核心只有一件事利用 LangGraph 把 RAG 应用重新编排成一张真正可控的有向图让检索、判断、生成、纠错不再是一条直线跑到底而是按我们定义的规则在节点之间流转。适合已经写过基础 RAG、但正在被分支逻辑和对话记忆折腾的开发者也适合想直接拿一个可运行骨架再往上加功能的朋友。1. 为什么RAG需要LangGraph从链式调用到状态机1.1 链式RAG卡在哪基础 RAG 用 LCELLangChain Expression Language写起来很顺手检索、组装 Prompt、调用模型三步结束。但一旦流程中出现“如果……就……”或者“不行就再来一次”这种逻辑链式表达就开始别扭了。举个例子假设你发现向量检索出来的文档质量和问题压根不相关你想加一个判断如果文档不相关就换一个关键词重新检索。这个逻辑如果用 Chain 实现要么写成嵌套的RunnableLambda要么让模型通过工具调用自己决定要不要重新检索。前者写起来绕后者的流程会变成黑盒——模型到底调了几次工具、每次检索到什么你很难从外部观察到。换句话说RAG 本质上是一个需要分支、循环和中间判定的流程型应用而 Chain 是一种线性流水线。流水线上想增加一个“质检不合格就回流重新加工”的环节除非拆掉整条线重铺否则只能在外围打补丁。LangGraph 解决的就是这个“流程控制”问题。1.2 LangGraph的四个核心概念LangGraph 从名字就能看出来它是把应用定义成一张图。图里最核心的四个概念搞懂它们RAG 的编排就通了。State状态一个类型化的字典是节点之间传递的公共数据。在 RAG 里它一般包含原始问题、检索到的文档、最终答案可能还有对话历史。Node节点一个普通的 Python 函数输入是整个 State输出是一个部分更新的字典。你告诉它要更新哪些 key框架负责把这些更新合并回状态里。Edge边决定执行顺序。普通边是“做完 A 一定去做 B”条件边是“做完 A 之后根据当前状态判断去 B 还是去 C”。Checkpointer检查点保存每一步执行完的状态快照。它有两个实用场景一是程序中断后可以从最近的快照恢复二是多轮对话时用同一个会话 ID 把历史状态取回来。这四个概念用做菜来类比很直观。State 是案板上所有的食材和半成品Node 是“切菜”“炒菜”“装盘”这些动作Edge 是“味道淡了就加盐”这种判断Checkpointer 则是每完成一步就拍一张照片中途接个电话回来看一眼照片就能接着干。1.3 用状态图重写RAG的收益把 RAG 从 Chain 换到 LangGraph最直接的收益不是性能而是控制力。检索、判断、重新检索、生成、校验这些环节原本散落在代码逻辑或者 Prompt 指令里现在变成了图上看得见的节点和边。你可以随时打印任意节点的输入输出看每一步到底发生了什么你可以给“检索质量差”单独建一个路由分支让流程自动重写问题后再次检索你还可以把“检查点”打开给每个会话存一份状态快照实现真正的多轮记忆。这些能力用 Chain 不是做不了而是做起来会越来越费劲最后变成一团只有自己能看懂的胶水代码。投资一点学习成本换来的是整个流程的可观测性和可干预性这是我认为 LangGraph 最值得迁移的点。2. 从零搭建LangGraph RAG环境与基础结构2.1 环境准备与最小数据集先搭一个最简环境。安装依赖的时候建议把 LangGraph 版本控制在较新的稳定版因为新旧版本的 API 差异不小具体差异后面我会单独讲。pip install -U langchain langgraph langchain-community chromadb向量库我用 Chroma理由很简单轻量、本地优先、接口清晰适合演示也适合中小规模项目。如果你已经有 Elasticsearch 或者 PGVector 之类的存储替换起来也不难因为 LangChain 的向量库接口是统一的。大模型部分我用兼容 OpenAI 接口的服务来演示API Key 和地址通过环境变量注入避免写死在代码里。import os os.environ[LLM_API_KEY] sk-your-key os.environ[LLM_API_BASE] https://api.example.com/v1接着准备一份最小数据集。不用贪多拿两三个产品说明文档或者 FAQ 就行。这里用DirectoryLoader加载一个目录里所有 TXT 文件再做切片和向量化。from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings loader DirectoryLoader(./docs, glob*.txt, loader_clsTextLoader) docs loader.load() splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) chunks splitter.split_documents(docs) embeddings OpenAIEmbeddings(modelyour-embedding-model) vectorstore Chroma.from_documents(chunks, embeddings, persist_directory./chroma_db)切片参数chunk_size500, chunk_overlap50是演示用的经验值。500 个字符足够覆盖一个完整的知识点50 的重叠能让相邻切片之间保留上下文衔接。实际项目中这个值要根据你的文档类型和模型上下文窗口来调不是越大越好。2.2 定义状态与检索、生成节点图应用的第一步是定义状态。用TypedDict声明字段相当于给整个流程图定了数据契约。from typing import TypedDict, List from langchain_core.documents import Document class RagState(TypedDict): question: str documents: List[Document] answer: str然后定义检索节点。这个节点做的事情很纯粹拿用户问题去向量库做相似度检索把 TopK 文档写回 State。def retrieve_node(state: RagState): query state[question] docs vectorstore.similarity_search(query, k4) return {documents: docs}这里有个新手容易困惑的点节点为什么返回一个新字典而不是直接修改传入的state因为 LangGraph 的节点设计是函数式的它只声明“我想更新这些 key”框架负责把这些返回值合并回公共状态。这种设计的好处是每个节点可以被单独测试也方便 Checkpointer 记录每一步的变化。生成节点的职责是组装上下文并调用大模型。这里核心是 Prompt 里明确要求模型只能基于给定文档回答减少幻觉。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI llm ChatOpenAI( modelyour-llm-model, temperature0, api_keyos.environ[LLM_API_KEY], base_urlos.environ[LLM_API_BASE], ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严谨的问答助手只能根据下面的文档内容回答用户问题不要编造答案。\n\n文档\n{context}), (human, 问题{question}), ]) def generate_node(state: RagState): context \n\n.join([doc.page_content for doc in state[documents]]) messages prompt.format_messages(contextcontext, questionstate[question]) answer llm.invoke(messages) return {answer: answer.content}temperature0是为了让回答更稳定对 RAG 场景来说稳定性比“创造性”重要得多。文档拼接顺序、分隔符样式这些细节也会影响模型对材料的注意力建议在项目初期就固定下来。2.3 组装状态图并调用节点定义好了接下来就是画图把节点放进图里连上边指定入口最后编译。from langgraph.graph import StateGraph, START, END graph StateGraph(RagState) graph.add_node(retrieve, retrieve_node) graph.add_node(generate, generate_node) graph.add_edge(START, retrieve) graph.add_edge(retrieve, generate) graph.add_edge(generate, END) app graph.compile()运行起来很简单result app.invoke({question: 这个产品的试用版有期限吗}) print(result[answer])invoke的入参是初始状态LangGraph 会从START开始执行依次经过retrieve、generate最后到END。最终返回的是完整的状态字典里面既有答案也有中间检索到的文档。这个“中间结果也能拿到”的能力是普通 Chain 给不了的。注意新版 LangGraph 用START、END这两个常量来标记入口和出口老版本里常见的是graph.set_entry_point(retrieve)或者add_edge(generate, __end__)这种字符串写法。建议新项目统一用常量老项目保持一种写法别混用混用会踩版本坑。3. 进阶编排多轮记忆、条件路由与自查纠错3.1 带对话记忆的多轮RAG基础版只能处理单轮问答。用户第二句问“那它的价格呢”这里的“它”指代上一轮聊的东西如果没有历史上下文检索环节根本不知道用户想问什么。解决思路是先改状态把对话历史塞进去然后在检索之前加一个节点把当前问题和历史合并成真正可检索的查询。class RagState(TypedDict): question: str search_query: str history: List[dict] documents: List[Document] answer: str改写节点负责把历史拼进系统消息让模型生成一个带指代消解的检索词。def rewrite_with_history(state: RagState): if not state[history]: return {search_query: state[question]} history_text \n.join( f用户{turn[question]}\n助手{turn[answer]} for turn in state[history][-4:] ) msg f根据对话历史把用户最新问题改写成一个可独立检索的查询。\n\n历史\n{history_text}\n\n最新问题{state[question]} rewritten llm.invoke(msg) return {search_query: rewritten.content.strip()}检索节点和生成节点都要改成读search_query。生成节点还要负责把历史放进 Prompt让模型知道它是在延续一段对话。真正让多轮记忆生效的步骤是给compile传入checkpointer。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory) thread_config {configurable: {thread_id: session-001}} app.invoke({question: 这个产品有试用版吗}, configthread_config) app.invoke({question: 那它的有效期是多长}, configthread_config)两次调用使用同一个thread_idLangGraph 会把第一轮的状态快照恢复出来作为第二轮的历史输入。这里的“历史”不是我们自己存变量而是 Checkpointer 自动维护的。它还能在断点处停下等人工审核后再继续跑这对需要人在环上的业务场景非常有用。3.2 检索质量评估与条件路由向量检索的 TopK 结果里经常混着不相关的文本。如果直接把这些不相关内容拼进 Prompt模型可能被带偏输出看起来有依据、实际上来自错误材料。所以要在“检索”和“生成”之间加一道质检闸门。从工程角度判断文档相关性有两种做法一种是简单的规则比如文档数量、关键词覆盖率省 token 但不够通用另一种是让一个小模型做二分类判断“这篇文档能不能支撑回答这个问题”。这里演示比较实用的 LLM 判断方案def grade_node(state: RagState): question state[question] if not state[documents]: return {retry_count: state.get(retry_count, 0) 1, filtered_docs: []} filtered [] for doc in state[documents]: msg f判断这篇文档是否包含回答该问题所需的信息。只输出相关 或 不相关。\n\n问题{question}\n\n文档{doc.page_content[:500]} verdict llm.invoke(msg).content.strip() if 相关 in verdict: filtered.append(doc) return {filtered_docs: filtered}路由函数根据质检结果决定下一步def route_after_grade(state: RagState): if state.get(retry_count, 0) 2: return generate if not state[filtered_docs]: return rewrite return generate在图上加一条条件边graph.add_node(grade, grade_node) graph.add_node(rewrite, rewrite_node) graph.add_edge(retrieve, grade) graph.add_conditional_edges(grade, route_after_grade, { rewrite: rewrite, generate: generate, }) graph.add_edge(rewrite, retrieve)这样流程就变成检索 → 质检 → 不通过就改写问题再检索 → 通过就生成答案。rewrite_node负责让模型换个关键词或者换个表达方式重写问题。retry_count是必须加的安全阀否则检索结果一直不合格时流程会无限循环。3.3 生成质量校验与修正循环有了检索端检查还可以再加一个生成端检查。很多 RAG 项目最后发现答案有幻觉问题不只出在检索还出在模型生成时没有严格遵守上下文。增加一个校验节点专门判断“答案是否完整回答了问题”以及“答案内容是否都能从文档中找到出处”。def validate_node(state: RagState): instruction ( 检查以下回答是否符合要求\n 1. 是否直接回答了用户问题\n 2. 是否完全基于给定文档没有编造内容。\n 只输出 PASS 或 FAIL。\n\n f问题{state[question]}\n f文档{ .join(d.page_content[:200] for d in state.get(filtered_docs, state[documents]))}\n f回答{state[answer]} ) verdict llm.invoke(instruction).content.strip() return {validation: verdict}路由逻辑可以设计成验证不通过并且重试次数没超限就回到改写节点重写问题后再走一遍完整流程。def route_after_validate(state: RagState): if state.get(validation) FAIL and state.get(retry_count, 0) 2: return rewrite return generate这个循环看起来简单但它把“你要好好回答哦”这种不可靠的 Prompt 约束变成了代码层面的硬控制。实际效果是最终交付给用户的答案至少经过了一道自动质检哪怕校验模型能力有限也能拦截掉相当一部分明显的幻觉输出。4. 常见问题与排查技巧实录4.1 状态不更新与键名不一致我见过最多的问题是节点之间状态字段对不上。比如retrieve_node里写的是{documents: docs}但generate_node里读的是state[docs]结果生成节点拿到的永远是空的。排查这种问题先别急着看框架文档。最简单有效的办法是在每个节点入口打印一下state.keys()一眼就能看出哪个字段缺失。def retrieve_node(state: RagState): print( retrieve invoked, keys:, state.keys()) ...还有一种情况是节点被条件边跳过了。你以为是 A→B→C但条件路由把 B 跳过去了C 自然拿不到 B 产生的字段。这时候把路由函数的输出打出来看看实际走的是哪条边问题基本就清楚了。4.2 版本差异带来的兼容问题LangGraph 版本更新频繁API 变化不小。老版本里入口节点用set_entry_point出口用字符串__end__新版本改用START和END常量。如果你从网上复制了一段代码跑起来报ValueError: Node __end__ not in graph大概率就是版本不匹配。我的建议是固定版本不要每次都用最新版。项目启动时在requirements.txt里锁住大版本比如langgraph0.2,0.3。升级的时候单独抽一个分支做回归测试跑一遍所有典型流程再合入。4.3 并行分支的状态合并多路召回是 RAG 的常见场景比如同时检索 HR 文档和技术文档。你可能会写两个检索节点都从START引出最后汇合到同一个生成节点。这里有一个坑如果两个节点都返回{documents: [...]}LangGraph 默认状态下后执行的节点会覆盖先执行的节点而不是把两边结果合并起来。想要合并要么用不同的 key 承载各自的结果再在合并节点里统一处理要么给状态字段配置 reducer。后者更高级但也更容易误用from typing import Annotated from langgraph.graph import add class RagState(TypedDict): documents: Annotated[List[Document], add]operator.add会把多个节点返回的 list 拼接起来听起来很好用但你要注意如果后续某个节点也返回documents历史值会被继续累加不是替换。状态流里的数据会膨胀得很奇怪。我的建议是并行分支尽量用不同 key合并逻辑显式写在专门的节点里可读性更高也更好调试。4.4 成本与延迟控制加质检节点、校验节点之后一次完整 RAG 请求可能从原来的一次 LLM 调用变成三到五次。单看每次调用不多但线上量一大账单和时延都会肉眼可见地涨。我的控制策略有以下几个质检和校验这类“判断题”尽量用较小、较快的模型不要和主生成的模型用同一个规格。给所有 LLM 调用设置超时和重试上限避免某个节点卡死拖垮整个流程。把“是否开启校验”做成配置项线上可以先关掉灰度验证时再逐步打开。检索结果在短时间内稳定的场景可以用缓存把重复的向量查询和质检结果缓存起来。成本控制的核心思路是不是每个环节都需要最强模型。判断题交给轻量模型生成题交给主力模型最后验收时再抽查一部分结果性价比是最高的。下面是几个高频问题的速查表我平时排查时就靠它现象可能原因处理方法下游节点拿不到字段返回的 key 写错或没定义在 State 中打印各节点state.keys()对比节点没被执行条件边路由跳过了该节点打印路由函数返回值END 相关报错版本混用或代码来自新版/旧版统一使用 START/END 常量锁定版本并行分支结果互相覆盖多个节点写同一个 key改用独立 key 或使用 reducer 合并答案质量不稳定校验模型过弱或未开启校验单独配置一个轻量判断模型流程异常耗时LLM 调用次数过多增加超时、缓存、关闭部分节点5. 从Chain换到LangGraph后的几点实操体会用 LangGraph 重写 RAG 之后我最大的感受是终于能看见流程的每一步了。之前用 Chain出了问题只能看最终结果猜是哪一步跑偏了。现在每个节点都能打印输入输出、每个路由判断都有迹可循定位问题的效率高了不少。一个很实用的技巧写代码之前在纸上把图先画出来。每个节点想清楚三件事输入什么、更新什么、要不要从这里出分支。画清楚了再动手代码基本一遍过。另一个技巧是get_graph()编译完的图对象可以直接打印结构快速核对节点和边的连接是否符合预期不用反复看框架内部逻辑。print(app.get_graph().draw_ascii())最后提醒一句不是每个操作都值得拆成节点。只有需要单独观察、复用或者分支判断的环节才值得成为图里的节点。节点拆得太多调试成本会反超收益。至于后续想往哪里扩展我个人的下一步计划是接入一个人工审核节点——让自动化的 RAG 先跑跑完不确定的答案转给人工看LangGraph 的 Checkpointer 和断点机制正好能承接这部分需求。
阅读完成 · 觉得有帮助？
