不少人上来就问 Agentic RAG 怎么做,LangGraph 的 Graph 怎么画,Agent 怎么编排。我自己也经历过这个阶段,追着新概念跑,回头才发现,基础的最简 RAG 没跑熟,后面全是坑。这篇就聊聊用 LangGraph 把最小 RAG 跑明白这件事,给想入坑 RAG、LangGraph 的同学一条更稳的路径。
先说清楚这篇文章的定位:不聊 Agentic RAG 的花活,不做天花乱坠的架构设计,就讲怎么用 LangGraph 从零搭一个能用的最小 RAG。你会理解 LangGraph 的核心概念,看到完整的代码实现,最后我会把实际操作中踩过的坑、排查思路一并整理出来。适合刚接触 RAG、对 LangGraph 只听过名字,或者被各种高级玩法劝退的开发者。
1. 为什么我劝你先别碰 Agentic RAG
1.1 Agentic RAG 的火爆与认知误区
Agentic RAG 这个概念在 2024 年下半年开始刷屏,各大技术社区都在聊。它的思路不难理解:传统 RAG 是固定的"检索-生成"管道,Agentic RAG 则是让一个大模型 Agent 去决定什么时候检索、检索什么、要不要重新检索、多个工具怎么调用。听起来很香,对吧?搜索结果能自我修正,回答质量能提升,用户问"今天天气怎么样"这种跟知识库无关的问题,Agent 也知道不检索直接回答。
但问题也出在这。很多初学者一上来就照着 Agentic RAG 的架构图画流程,什么规划器、路由器、推理循环、工具调用,整得比微服务还复杂。然后呢?检索质量没做好,分块策略一塌糊涂,向量检索的结果根本不相关,Agent 再怎么"智能",拿到垃圾上下文也只能输出垃圾回答。这不是 Agent 的锅,是地基没打牢。
我见过太多的案例,花了两周搭 Agentic RAG,最后排查问题的时候发现:召回率低是因为没做 query 改写,回答差是因为上下文塞了太多噪音,甚至还有 embedding 模型选错的。这些问题,在最小 RAG 阶段就该暴露出来。换句话说,Agentic RAG 不会帮你解决检索质量问题,它只会让检索质量问题变得更难排查——因为决策逻辑变复杂了,你分不清是 Agent 决策错了,还是底层检索就错了。
1.2 最小 RAG 的边界到底在哪里
那什么是最小 RAG?我给它画一条清晰边界:只包含四个环节的闭环——
- 文档加载与分块;
- 向量化与存储;
- 检索召回;
- 拼接上下文生成回答。
没有任何额外的决策逻辑,没有路由,没有多轮改写,没有工具调用,就是一条直线流程:用户提问进来,系统检索,系统回答,结束。
我见过不少同学对这个边界不以为然,觉得"这也太简单了"。但恰恰是这个最简单的闭环,藏着 RAG 系统最核心、最影响最终效果的那几个变量:分块粒度、embedding 模型选择、检索 TopK 设置、提示词上下文拼装方式。这几个变量,你不在最小系统里搞清楚,后面加多少 Agent 逻辑都是空中楼阁。
还有一个现实层面的理由:最小 RAG 的问题域足够小,小到你能控制变量。检索结果不对,你只需要检查分块、嵌入、检索参数这三个环节,不需要同时考虑 Agent 是不是选错了工具、是不是少调用了一次检索、是不是上下文被其他信息污染了。这种可调试性,在技术选型初期比什么都值钱。
2. LangGraph 到底解决了什么问题
2.1 LangGraph 的核心概念:State、Node、Edge
用 LangGraph 写最小 RAG 之前,得先把它的三个核心抽象搞明白——State(状态)、Node(节点)、Edge(边)。
State 是贯穿整个图的数据载体。你可以把它理解成一条流水线上的传递带,每个节点从这个传递带上读数据,处理完再写回去。LangGraph 的 State 本质上是一个 TypedDict,你定义了它的结构,所有节点都能读写里面的字段。
Node 是处理单元。一个节点就是一个 Python 函数,输入是当前的 State,输出是一个字典,字典里的字段会更新到 State 上。最小 RAG 里我们只需要两个节点:检索节点和生成节点,分别负责从向量库里召回文档和调用大模型生成回答。
Edge 决定执行顺序。普通边表示"上一个节点跑完,下一个节点接着跑",条件边则表示"根据当前状态决定下一步走哪里"。最小 RAG 只需要普通边就够了,从检索节点连到生成节点,一条线走到底。
这个设计的精妙之处在于,它把"流程的控制逻辑"和"业务的处理逻辑"彻底解耦了。你不需要在业务代码里写 if else 判断下一步做什么,图的执行顺序由 LangGraph 引擎统一调度。这为以后加分支、加循环、加 Agent 决策留好了口子,但当下你只需要关注节点内部的处理逻辑。
2.2 LangGraph 与 LangChain 的关系,别再混为一谈
LangChain 和 LangGraph 是两代不同的东西,很多人分不清,我也曾经被这俩名字绕晕过。一句话总结:LangChain 的核心价值是提供了一堆封装好的组件——文档加载器、文本分割器、向量存储封装、模型调用封装、提示词模板——它是 RAG 的"零件库";LangGraph 的核心价值是编排这些零件的执行流程——它是"流水线控制系统"。
打个比方,LangChain 是工具箱里的各种扳手、螺丝刀,LangGraph 是指导你怎么按顺序使用这些工具的工作手册。你完全可以用纯 LangChain 写一个最小 RAG:加载文档、分割、入库、检索、拼接 Prompt、调用模型,这些都是 LangChain 封装好的 API。但你会发现,流程是写死在业务代码里的,想加个判断逻辑、加个重试循环,就得自己用 Python 代码硬拼。
LangGraph 给我的感觉是,它把流程控制这件事从业务代码里彻底抽离出来了。你可以直观地看到"先检索、再生成"这个流程,也可以轻松地在中间插入一个新节点,或者加一条条件边。而且 LangGraph 自带状态管理,不用手动维护变量传递,这在大一点的系统里省心得多。
有个问题很多人会问:LangGraph 是不是要替代 LangChain?从官方定位来看不是替代关系,是互补关系。LangGraph 的节点里跑的还是 LangChain 的组件。所以在最小 RAG 实操里,我会两者混用:用 LangChain 做文档加载、文本分割、向量存储和模型调用,用 LangGraph 把整个流程串起来。
2.3 为什么最小 RAG 也值得上 LangGraph
有人会说:最小 RAG 用 LangChain 的 RetrievalQA 链就能跑,为什么非要用 LangGraph?
这个问题的答案,取决于你看的是"当前的最小系统"还是"未来的系统演进"。如果只是跑个 Demo 验证一下 RAG 效果,LangChain 的 LCEL 确实够用。但如果你确定后面要往 Agentic RAG 演进——这是大多数做了知识库问答的人都会走的方向——那我建议一开始就上 LangGraph。
原因很简单:流程图的骨架是不变的。你后面加 query 改写节点、加相关性判断节点、加多轮对话管理节点,都是在检索和生成这两个基础节点之间做文章。如果一开始就用 LangGraph 搭好了骨架,后面的演进就是在现有的图上加节点、加边,而不是推倒重来。
还有一个很实际的原因,LangGraph 的图结构天然可观测。你可以一步步打印出每个节点处理完之后的 State,清晰地看到检索结果是什么、最终生成用的上下文是什么,这对调试 RAG 系统的检索质量简直太方便了。我后面排查问题全靠这个能力。
3. 最小 RAG 实操:用 LangGraph 把闭环跑起来
3.1 环境准备与依赖安装
先准备环境,Python 3.10 以上版本,建议用虚拟环境隔离。
pip install langgraph langchain langchain-community langchain-openai chromadb这里我用的向量库是 Chroma,因为它轻量、本地运行、拿来写示例最省事。生产环境你可能换 Milvus、pgvector、Weaviate,但核心逻辑是一样的。模型方面用 OpenAI 的嵌入模型和对话模型做示例,但接口是通用的,换成 Ollama 或者其他本地模型的成本很低。
注意:LangChain 的社区包一直在更新,有些 API 会变动。安装时最好固定一个版本,或者说跑不通的时候先看看是不是版本问题。我在 1.0 系列版本上测试过下面的代码,但如果你用的是 0.x 版本,个别 API 可能不一样。
3.2 定义 RAG 的 State
State 定义是整个 LangGraph 应用的地基,想清楚 State 里放什么字段,等于想清楚了系统的数据流。最小 RAG 只需要三个字段。
from typing import TypedDict, List class RAGState(TypedDict): question: str # 用户提问 context: List[str] # 检索到的文档片段 answer: str # 最终生成的回答这个定义很简单,但设计思路值得说两句。context 字段就是检索节点和生成节点之间的"接力棒":检索节点往里面写入文档片段,生成节点从里面读取并拼接到提示词里。如果你后面想做检索结果的重排,就在检索节点和生成节点之间再加一个节点,处理完之后再更新 context 字段——图结构的变化成本非常低,这正是 LangGraph 灵活性的体现。State 字段的类型注解不要随便省,直觉上它只是一个辅助,但 LangGraph 内部在更新 State 时会用到类型信息,写对了能减少很多莫名其妙的问题。
有人可能会问:为什么 context 用 List[str] 而不是直接用拼接好的字符串?我的建议是尽量保持字段的语义化。List[str] 记录的是"检索到的每一段原文",拼接的操作放到生成节点里做。这样中间无论插入重排、过滤还是去重节点,操作的对象都是结构化的数据,而不是一个已经拼好的字符串——处理起来会灵活得多。
3.3 文档加载、分块与向量化入库
在检索节点能工作之前,得先把知识库准备好。这一步在 LangGraph 的图外完成,属于"前置准备流程",但它直接决定了检索质量的底线。
from langchain_community.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma # 1. 加载文档 loader = TextLoader("data/faq.txt") documents = loader.load() # 2. 分块 text_splitter = RecursiveCharacterTextSplitter( chunk_size=400, chunk_overlap=40, separators=["\n\n", "\n", "。", "!", "?", ". ", " ", ""], ) chunks = text_splitter.split_documents(documents) # 3. 向量化入库 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./data/chroma_db", )分块参数是最值得花时间调的部分,没有之一。我见过太多案例,检索效果差不是因为模型不好,而是分块策略不对。chunk_size 设到 400 到 800 之间是一个对多数场景都能用的合理范围,但具体要看你的文档类型。
文档里全是短段落问答?可以减小 chunk_size,让每个块只包含一对问答,检索命中更精准。文档是长文章、需要保留完整逻辑?那就要加大 chunk_size,否则一段逻辑被切碎了,检索到一半上下文,回答就会没头没尾。chunk_overlap 的作用是保留相邻块之间的边界信息,防止重要内容被拦腰截断。经验值是 chunk_size 的 10% 到 20%,太小了没效果,太大了会产生大量重复内容、浪费向量库空间。
注意:分块不是"设个参数跑一遍"就完事的事。强烈建议第一次跑通后,故意对几个典型问题做检索测试,直接看检索到的块跟问题相关不相关。这一步调试的时间,比你后面调 Agent 的时间值钱得多。
3.4 定义检索节点与生成节点
前置准备做完了,现在进入 LangGraph 的核心环节:定义节点。每个节点就是一个普通 Python 函数。
from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate # 检索节点:从向量库召回相关文档 def retrieve_node(state: RAGState) -> dict: question = state["question"] docs = vectorstore.similarity_search(question, k=4) return {"context": [doc.page_content for doc in docs]} # 生成节点:拼接上下文,调用大模型生成回答 def generate_node(state: RAGState) -> dict: context = "\n\n".join(state["context"]) prompt = PromptTemplate.from_template( "你是知识库问答助手,请根据以下资料回答问题。\n" "若资料中不包含答案,请如实说明,不要编造。\n\n" "资料:\n{context}\n\n" "问题:{question}\n\n" "回答:" ) llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) chain = prompt | llm response = chain.invoke({"context": context, "question": state["question"]}) return {"answer": response.content}先说 retrieve_node。similarity_search 是向量检索里最基础的相似度搜索,默认用余弦相似度。k=4 的意思是召回最相似的 4 个文档块,这个数字建议先设小一点,4 到 6 是常见起点。你想,用户的问题通常只需要一两个信息点就能回答,召回太多块,上下文里塞满了不相关的噪音,大模型反倒被干扰。等跑通了,你可以对比 k=4 和 k=8 的效果差异,再决定最合适的值。
再说 generate_node。这里我做了两件很重要的事:一是明确告诉模型"资料中不包含答案就老实说",这是对抗大模型幻觉最基础的一招。RAG 系统的回答质量问题,很大一部分不是模型能力不行,而是模型在资料不足时硬要编一个答案。二是在提示词里把"资料"和"问题"明确分开,让模型清楚知道什么信息是事实依据、什么信息是需要回答的提问。
3.5 构建图并执行
节点定义完之后,图的构建就水到渠成了。
from langgraph.graph import StateGraph, START, END # 1. 创建图 graph = StateGraph(RAGState) # 2. 添加节点 graph.add_node("retrieve", retrieve_node) graph.add_node("generate", generate_node) # 3. 添加边 graph.add_edge(START, "retrieve") graph.add_edge("retrieve", "generate") graph.add_edge("generate", END) # 4. 编译图 app = graph.compile() # 5. 执行 result = app.invoke({"question": "如何重置密码?"}) print(result["answer"])这里我要解释一下 START 和 END 这两个特殊标记。START 是图的入口节点,所有执行流程都从它开始;END 是图的出口节点,执行到这里意味着整个流程结束。把检索节点放在 START 后面,生成节点放在 END 前面,意思就是一进场就检索,检索完就生成,生成完就结束——这就是最小 RAG 的全部流程。
执行的时候,app.invoke 传入一个字典,字典的 key 必须跟 State 定义的字段对应。这里你只需要传入 question,context 和 answer 会在流程中被检索节点和生成节点依次填充。你可以打印 result 看看,里面会有完整的三个字段值,这就是一次完整的 RAG 闭环。
如果你想更直观地看到每一步的状态变化,可以把 invoke 换成 stream:
for chunk in app.stream({"question": "如何重置密码?"}, stream_mode="updates"): print(chunk)stream 会逐节点打印输出,瞬间就能看清楚每个节点处理完之后 State 变成了什么样。我调试时几乎离不开它——检索节点返回了什么、生成节点拿到了什么上下文、最终回答了什么问题,一目了然。
3.6 结合 FastAPI 做一个最小接口
跑通脚本之后,下一步顺理成章:把最小 RAG 封装成一个服务。热词里提到了 FastAPI,这也是我实际项目里最常用的方式。
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI(title="Minimal RAG API") class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): answer: str @app.post("/query", response_model=QueryResponse) def query(req: QueryRequest): result = app.invoke({"question": req.question}) return QueryResponse(answer=result["answer"])这个接口很简单,但它是从"脚本能跑"到"系统能用"的关键一步。有了 HTTP 接口,前端、企业微信机器人、Slack 机器人、工单系统都能接进来。注意这里有个细节:在 FastAPI 里调用 LangGraph 应用,建议启动时就把 app 编译好,放到全局变量里,而不是每次请求都重新编译。LangGraph 的 compile 有开销,每次走一遍会拖慢响应。
4. 常见问题与排查技巧实录
4.1 检索结果不相关,先不要怀疑模型
最大、最常见、最让人抓狂的问题就是:检索回来的文档跟问题完全没关系,模型再怎么聪明也只能胡说八道。我踩过这个坑之后总结了一条铁律:回答质量出了问题,百分之八十是检索召回的问题,不是生成的问题。
排查思路建议按顺序来:
- 先看召回内容。直接打印 state["context"],看看检索节点到底找回了什么文档。如果内容毫不相关,问题在检索侧,不在生成侧。
- 再看分块是否合理。如果召回的是某一段很长的文本中间的一截,内容被切得支离破碎,那就是分块的 chunk_size 设置不合理,重要信息在切割时被截断了。
- 然后看 embedding 是否匹配。检索时用的 embedding 模型必须跟入库时用的完全一致。换了 embedding 模型,向量空间都变了,检索结果必然崩掉。这个错误我在工程里见过不止一次。
如果是 query 本身比较复杂——比如多个意图混在一起、缩略语很多——最小 RAG 阶段很难完美处理,这是正常的。先记录问题,后面再考虑在检索前加一个 query 改写节点。
4.2 上下文里全是重复内容
chunk_overlap 设得太大,文件里本身有大量重复段落,或者同一个内容被多个文档重复收录,都会导致召回结果里出现大量重复片段。后果是 context 被冗余信息堆满,模型生成的回答可能啰嗦,或者被重复信息带偏。
我习惯在检索节点后面加一个简单的去重处理:
def retrieve_node(state: RAGState) -> dict: docs = vectorstore.similarity_search(state["question"], k=6) seen = set() unique_docs = [] for doc in docs: content = doc.page_content.strip() if content not in seen: seen.add(content) unique_docs.append(content) return {"context": unique_docs[:4]}先取 6 条,去重后再留 4 条,既保证了足够的候选集,又剔除了冗余信息。这个套路简单粗暴,但很实用。
4.3 需要追加新文档,但不想全部重建索引
最小 RAG 跑起来了,业务同学问:新文档怎么加?如果你用的是 Chroma,本地持久化后可以用 add_documents 追加:
new_chunks = text_splitter.split_documents(new_docs) vectorstore.add_documents(new_chunks) vectorstore.persist()增量入库的关键前提是:分割逻辑和 embedding 配置必须跟首次建库时保持一致,否则后加的内容检索风格不一致,召回质量会受影响。这也是为什么我建议把所有配置集中放到一个配置文件或环境变量里,不要散落在各个脚本中。
生成的 Persist 目录要定期备份,Chroma 的本地文件如果损坏,整个向量库就废了。我习惯把向量库目录纳入 Git LFS 管理,或者定期同步到对象存储。
4.4 构建图中容易踩的三个坑
最小 RAG 的图结构虽然只有两个节点,但我在帮读者看代码时,发现不少人会犯三个低级错误:
一是给 State 字段设置了默认值但类型声明不规范,导致 LangGraph 更新字段时出现类型不匹配。这个在前期定义 State 时要严格写清楚 List[str] 这种泛型。
二是在节点里修改了 State 的字段但返回值写错了 key,比如拼写错误,导致字段没更新成功,生成节点拿不到 context。debug 这种问题最快的方式是上面提到的 stream 逐节点打印。
三是忘记在编译前添加 START 到第一个节点的边。LangGraph 对图结构的完整性有一定要求,缺了入口边会直接报错或运行异常。解决方案很简单——按我给的完整代码来,别自己精简掉 add_edge(START, "retrieve") 这一步。
5. 从最小 RAG 到 Agentic RAG 的演进建议
5.1 最小系统是后续所有复杂度的试验田
最小 RAG 跑通之后,你手里其实多了一个非常趁手的试验田。所有后续的优化,都可以在这个基础上做 A/B 对比:
- 想优化检索?换一个 embedding 模型,对比同样问题集的召回效果;
- 想优化分块?改 chunk_size 参数,看看回答质量是变好还是变差;
- 想加 reranker?在检索节点和生成节点之间插入一个重排节点;
- 想减小幻觉?把提示词改得更严格,或者加一个"资料不含答案时直接拒绝回答"的判断。
这些验证,在一个只有两个节点的图里做,成本最低、干扰最少。等你把每个环节都调明白了,再引入 Agentic RAG 的各种决策逻辑,每一步的收益和成本都会非常清楚。
我举个具体例子。当初我在最小 RAG 上验证了 query 改写对检索质量的提升效果——用户问"帮我看看上个月的数据"这种模糊问题时,直接检索效果很差,但把它改写成"上个月销售数据汇总报告"之后,召回质量明显提升。于是后面做 Agent 决策逻辑时,"什么时候需要改写、什么时候不需要"就有了可靠的实践依据,而不是拍脑袋定规则。
5.2 Agentic RAG 最值得加的三种能力
如果最小 RAG 已经满足了上面的要求,我再建议考虑进入 Agentic 阶段。以我的经验,最有价值、最值得先做的三种能力是:
第一种是检索决策。在检索之前加一个轻量的判断节点:这个问题需要检索吗?如果需要,是检索知识库还是直接用模型能力回答?这能有效减少无关问题对检索结果的干扰。
第二种是多轮改写。用户的问题往往是基于上文语境的,原文检索命中率很低。用一个节点把用户当前问题和历史会话信息合成一个更完整的独立问题,再交给检索节点,这是提升企业知识库问答效果的关键一步。
第三种是结果验证。生成回答之后,加一个验证节点,让模型判断生成的答案是否基于给定的资料,如果发现"跑偏"了,就回到检索节点重新检索。这就是最基础的 Agentic 循环。
这三种能力,每一种都对应 LangGraph 里的一个节点或一条条件边。你会发现,当你把最小 RAG 跑明白之后,这些复杂度的增加是有序的、可控的。与其一开始就设计一个大而全的智能体,不如从最小闭环开始,一步步加能力,每加一步都能验证、都能回退。
我个人在实际操作中的体会是:RAG 系统的效果天花板,八成取决于最小闭环里的基础设置——分块策略、检索精度、上下文组织方式,只有两成取决于上层决策逻辑。先把最小 RAG 跑明白,不光是技术路径上的选择,更是一种能够贯穿整个项目生命周期的做事方式。最后再分享一个小技巧:把调试时的典型问题整理成一个回归测试集,每次改参数、改代码之后都跑一遍。这个习惯帮我挡住了无数次"这次改好了别的地方又崩了"的尴尬,强烈建议你也试试。