十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

基于LangGraph的最小RAG实战:FastAPI+LangChain+pgvector构建知识库

基于LangGraph的最小RAG实战:FastAPI+LangChain+pgvector构建知识库 最近不少人来问 Agentic RAG 怎么做我一般会先反问一句你现在的基础 RAG 跑明白了吗结果发现大多数情况是把 Agent 框架先堆上去了检索质量却一塌糊涂。所以我今天想直接把话说明白别急着上 Agentic RAG先用 LangGraph 把一个最小 RAG 跑通跑出稳定的效果再谈复杂编排。这篇文章不讲花活就讲你用 FastAPI LangChain LangGraph pgvector 能把 RAG 做的最小可用的那种状态以及我在这条路上踩过的坑。1. 为什么我会劝你先做最小 RAG而不是直接梭哈 Agentic RAG1.1 Agentic RAG 没解决你之前没解决的检索问题很多人对 Agentic RAG 的理解是让大模型自己决定什么时候检索、检索哪个库、要不要追问。听起来很美但有个前提你没注意到——它默认你检索这件事本身是可靠的。如果你的分块不合理、向量库召回率低、文档里全是噪音Agent 再怎么调度拿到的上下文也是脏的。它只会把脏数据包装得特别自信。我见过一个真实案例某团队把 LangGraph 的 ReAct Agent 套在知识库上Agent 一顿操作调了三次检索最后生成出来的答案还是过时的原因是原始文档里根本没有对应版本的信息或者文档被分块后把关键字段切碎了怎么检索都唤不起来。这种问题加多少层 Agent 逻辑都救不了反而会因为多个跳转让错误更难排查。所以我的判断是如果你的目标是一个知识库问答系统第一优先级永远是给定一个明确的问题能在正确的地方把正确的片段捞出来。这个基本功没到 80 分Agentic 只是给错误信息加了层华丽外壳。1.2 最小 RAG的验收标准跑通、可测、能改我定义的最小 RAG 不是能在本地打印一段结果那种 Demo而是要达到三个标准跑通从加载文档到最终生成回答全链路走完处理异常时不会整个服务挂掉。可测给你一个测试集能批量跑问答能算出命中率、正确率而不是靠肉眼抽查两三条。能改修改分块大小、嵌入模型、检索 TopK、Prompt 模板改完能快速对比效果差异。能做到这三点你再考虑引入 Agentic 才有意义。否则你只会陷入加了 Agent 后输出变好还是变坏完全靠运气的玄学状态。2. 技术选型LangGraph、LangChain、FastAPI、pgvector 这一套是怎么配的2.1 LangGraph 和 LangChain 的区别一句话讲清楚LangChain 是一堆组件库里面有各种 LLM 封装、Prompt 模板、检索器、Memory。LangGraph 是一个状态编排框架它把你的业务流程定义成一张图每个节点是一段逻辑边是节点之间的流转条件。很多人问LangGraph 是不是要取代 LangChain这个说法其实不太准。LangGraph 并不排斥 LangChain 的组件它俩更像是流程控制器 工具库的关系。你可以只用 LangGraph 写流程然后从 LangChain 里拿文档加载器、拆分器、向量存储封装也可以完全不用 LangChain 的组件流程里自己写 HTTP 请求去调模型LangGraph 照样能跑。在最小 RAG 场景里我的选型组合是FastAPI对外提供 HTTP 接口简单直接。LangChain负责文档加载、文本切分、向量存储封装。LangGraph把检索和生成定义成节点管理状态流转。pgvector作为向量存储存文本嵌入向量也顺带存原始文本。Ollama 或 OpenAI 接口提供嵌入模型和生成模型。2.2 为什么我把向量库选成 pgvector 而不是 Milvus知识库初期的数据量往往不设上万个文档不到百万级向量用专门的向量数据库Milvus、Qdrant反而增加运维负担。pgvector 方案的好处是把结构化数据、非结构化向量数据放在同一个 PostgreSQL 实例里你不需要额外搭建服务备份、权限、SQL 查询都是你熟悉的。等到数据量真实涨上来再迁移到 Milvus 也来得及因为 LangChain 的向量存储接口是兼容的替换成本不高。我通常建议团队里本来就熟悉 PostgreSQL 且数据量在百万级向量以下优先用 pgvector如果是纯做大规模私有化知识库、对检索性能极敏感、且已经有微服务部署能力再考虑 Milvus。不要为了炫技把一个简单项目复杂化。3. 最小 RAG 的最小实现数据准备与检索链3.1 分块和嵌入这里决定了 RAG 的天花板很多人一上来就去调 Prompt、调 LangGraph 节点结果检索出来的文档片段压根不对。我可以负责任地说RAG 项目 90% 的效果问题出在分块和嵌入这两个环节后面只是把这层天花板兑现出来。分块的核心指标不是每块多少个 token而是每一块是否是一个语义上自洽的单元。我踩过的坑是一段技术文档里有一个表格表格前后是两段叙述如果简单按 512 字符切分表格被切得七零八落检索时语义被严重污染。后来我把自定义分隔符按\n\n、\n、。、的优先级来做递归切分LangChain 里就是RecursiveCharacterTextSplitter并且把表格单独提取出来作为一个块。这才是分块该有的思路而不是无脑切固定长度。嵌入模型的选择更关键。如果你用的是本地部署的 Ollama推荐用bge-m3或nomic-embed-text这类中文表现还不错的模型文本向量维度通常在 1024 左右。如果用的是 OpenAI 接口text-embedding-3-small的性价比就够用。不管用哪个你必须保证两个一致性同一批文档、同一个查询必须用同一个嵌入模型和同一套参数向量存储里保存的 model 名称、维度要在元数据里记好否则后面换模型会导致新旧向量不能混查。3.2 用 FastAPI 暴露 RAG 服务先别急着做对话模板最小阶段你的后端只需两个接口POST /ingest接收文档内容分块、嵌入、写入 pgvector。POST /query接收用户问题检索上下文调 LLM 生成回答。这里我建议先不要做流式输出也不要加多轮对话记忆用最朴素的请求-响应就能把核心链路验证明白。等核心链路稳定了再在 /query 里面逐步扩展。ingest接口的伪代码逻辑大概是from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import PGVector from langchain_ollama import OllamaEmbeddings embeddings OllamaEmbeddings(modelbge-m3) splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap80, separators[\n\n, \n, 。, , , , ] ) docs splitter.create_documents([raw_text]) vector_store PGVector.from_documents( embeddingembeddings, documentsdocs, collection_namemy_knowledge_base, connection_stringpostgresqlpsycopg://user:passlocalhost:5432/ragdb )这段代码你跑通时基本就能确认文本能不能进库、向量维度对不对、collection_name 是否正确。很多问题在第一步就能暴露出来而不是等到 query 阶段才发现检索不出来任何东西。4. 关键一步用 LangGraph 把流程变成状态机4.1 状态设计Question → Context → Answer用 LangGraph 写 RAG本质上是在定义一张有向图图中的节点就是函数这些函数共享一个状态对象。我习惯把状态定义成这样from typing import TypedDict, List, Optional class RAGState(TypedDict): question: str context: List[str] # 检索到的文档片段列表 relevant_scores: List[float] # 每个片段的相似度分数 answer: Optional[str] # 最终生成的回答 steps: List[str] # 记录执行过的节点方便排查这里的steps字段是我后来加上的。一开始没加后来发现调试时很难判断某次请求究竟走了哪个分支是检索失败了还是生成超时了。加上这一步日志后LangGraph 可视化调试时能直观看完整路径问题定位快得多。4.2 把检索和生成拆成节点并接上条件边最小 RAG 只需要两个核心节点retrieve_node和generate_node。from langgraph.graph import StateGraph, START, END def retrieve_node(state: RAGState) - dict: # 这里调用 pgvector 向量检索 documents vector_store.similarity_search_with_score( state[question], k4 ) context [doc.page_content for doc, _ in documents] scores [score for _, score in documents] return { context: context, relevant_scores: scores, steps: [*state[steps], retrieve] } def generate_node(state: RAGState) - dict: context_text \n\n.join(state[context]) prompt f基于以下资料回答问题\n\n{context_text}\n\n问题{state[question]} # 假设调用聊天模型 answer chat_model.invoke(prompt) return { answer: answer, steps: [*state[steps], generate] } graph StateGraph(RAGState) graph.add_node(retrieve_node, retrieve_node) graph.add_node(generate_node, generate_node) graph.add_edge(START, retrieve_node) graph.add_edge(retrieve_node, generate_node) graph.add_edge(generate_node, END) app graph.compile()跑一次调用result app.invoke({ question: 数据库连接超时应该怎么排查, context: [], relevant_scores: [], answer: None, steps: [] }) print(result[answer])这样写到的 LangGraph 好像挺简单甚至有人觉得这不是一样吗我直接用 if else 也能写。但区别在于当后面你需要加判断检索质量是否够好、判断是否需要多轮检索、判断是否应该反问用户澄清意图这些逻辑时你在图上加节点、加条件边就行而不会把函数内部改成一坨纠缠不清的分支。LangGraph 的价值不在第一版而在你开始迭代到第 5 版、第 10 版的时候。这也是我强调先跑最小闭环的原因——只有你把基础的二节点图掌握了后续加条件边才是顺理成章而不是一上来就把图画成蜘蛛网。5. 跑通最小版之后再谈Agentic怎么加5.1 Agentic RAG 本质是给流程加决策点有了上面的基础图Agentic RAG 的Agentic体现在哪里我的理解是它是在检索与生成之间插入了若干路由判断和自反思逻辑。比如检索结果的相关分数整体都很低说明知识库里压根没有相关内容这时不再硬生成而是返回我找不到相关依据。不同的问题类型走不同的检索路径比如操作类问题走故障排查知识库名词解释类问题走产品文档库这时要加一个路由节点。生成出来的答案让模型自己重新审视一遍有没有用到检索到的内容有没有胡说如果有问题重新检索后再生成一遍。这些都是 Agentic 的表现但它们的实现完全可以在 LangGraph 里通过加节点和条件边来逐步叠加。比如加上检索质量判断后状态流转就变成def check_context_node(state: RAGState) - dict: if state[relevant_scores] and state[relevant_scores][0] 0.6: return {next_node: fallback_node} else: return {next_node: generate_node} graph.add_node(check_context_node, check_context_node) graph.add_conditional_edges( retrieve_node, lambda state: state.get(next_node, generate_node), {generate_node: generate_node, fallback_node: fallback_node} )这个设计的意义在于你不需要把知识库命中率不高这种情况写死在某一个大函数里而是把它变成了图上的一条可见分支。后续要加再检索一次的逻辑只要在条件边里增加一个retry_node不用动其他节点。5.2 升级到 Agentic 之前先确认你手里有这三样东西如果你真的打算从最小 RAG 升级到 Agentic RAG我建议你先确认自己已有三个资产否则升级就是空转一套评估集至少 30-50 条真实问题覆盖好问题、模糊问题、无答案问题。每次升级后跑同一套题对比回答质量。一个可复现的检索基线记录当前分块策略、TopK、嵌入模型下的平均相似度分数。否则你根本不知道新增的 Agent 层是帮了忙还是帮倒忙。一条可追踪的链路LangGraph 每一步的输入输出都在状态里留痕出了问题能回放。没有这三个资产你会陷入一种很尴尬的处境Agent 加了功能多了但项目到底变好还是变差说不清楚。这也是为什么我在题目里强调先把最小 RAG 跑明白——因为跑明白意味着你已经有了评估集和基线Agentic 只是在基线上的增量试验。6. 我在这个项目里踩过的几个具体的坑6.1 分块参数不匹配检索出来一堆废话我第一次做的知识库分块设成了chunk_size1024, chunk_overlap50结果检索出来的 Top4 上下文中有两块是同一个内容的不同切片还有一块几乎全是干巴巴的代码注释。核心问题在于chunk_size用的 token 数还是字符数没有统一。LangChain 的RecursiveCharacterTextSplitter默认是字符数不是 token 数。如果你的模型是按 token 计费或上下文窗口有限建议用len()token的方式重写长度函数或者干脆在切分后做一个超过指定 token 数再切一刀的后处理。我当时是在切分后用tiktoken统计每个块的实际 token 数超过上限就强制二次切分。6.2 pgvector 索引没建好数据一多查询就慢小数据量时 pgvector 查询看起来很快但文档到几千片后全表扫描会明显变慢。一开始我忽略了这个问题因为测试集只有几十条。后来到上万片时单次查询要几百毫秒体验很不舒服。解决办法是给向量列建立 HNSW 索引CREATE INDEX ON my_knowledge_base_docstore USING hnsw (embedding vector_cosine_ops);注意一点如果你用的距离类型是内积或欧几里得距离索引类型也要对应改。建立索引前先确认向量列的数据类型是vector(n)否则无法建索引。6.3 本地 Ollama 嵌入不稳定用 Ollama 跑bge-m3嵌入时我遇到过一种情况同一个问题连续查询两次返回的相似度分数波动很大而且第一次索引的数据和后续查询用的模型不一致因为我中途调整过 Ollama 模型版本。这个问题的根因是模型版本漂移。Ollama 拉新模型后旧嵌入还是旧模型生成的导致向量空间不一致。解决思路是在 collection 元数据里记录用到的嵌入模型名和版本号升级模型后强制全量重建索引或者至少做增量重算不要新旧混用。最小阶段最省心的做法就是固定模型版本不随便升级。6.4 LangGraph 状态里的不可序列化对象还有一次踩坑是往 LangGraph state 里塞了数据库 connection 对象。LangGraph 在节点间传递状态时默认需要对状态做序列化某些对象直接塞进去会在执行时报错或者导致 checkpoint 功能失效。经验是状态里只放基础类型、字符串列表、分数、字典这类可序列化数据。数据库连接、HTTP 客户端、嵌入模型实例这些资源放在节点函数内部初始化或者用依赖注入的方式装配不要让它们进入状态对象。这个习惯在后期接 LangGraph 的持久化和断点续跑功能时特别重要。7. 从能跑到能用还有一段路我上面写的这套最小 RAG看代码量并不大FastAPI 两个接口LangGraph 两个节点pgvector 一张表。但它能真正用在整个项目里还需要补一些工程细节接口层加鉴权和限流、文档更新时同步清洗对应向量、记录每次 query 的耗时和相关分数用于后续分析。这些都不是 Agentic 的范畴但却是知识库系统能不能被业务团队接受的关键。我的强烈建议是先把这套最小 RAG 部署到内网让真实用户问一周真实问题把检索不准的 case 收集起来。你会发现大多数问题集中在文档没覆盖、分块切坏、同义词没扩召回这三类。把这些问题处理完你再认真考虑 Agentic RAG 的引入那时你会更清楚每一个 Agent 节点要解决的问题是什么。我记得有几个项目团队用了两三天就把 LangGraph 的最小 RAG 搭起来了然后花了两周在调分块、滤掉垃圾上下文。后来他们评估下来说幸亏没有直接上 Agent否则排查范围会成倍扩大完全不知道该从模型还是从流程上开刀。最小 RAG 和 Agentic RAG 之间差的不是代码量是对检索链路的掌控力。先把底座打稳再谈自动化决策这是我这两年做知识库项目最大的体感。
返回列表