Agentic RAG这个概念最近在技术圈刷屏大家讨论的焦点已经从“RAG怎么做”变成了“RAG怎么不做得像个呆子”。我自己的体感很明显静态RAG的套路化太严重了——用户问个稍微带点弯的问题它就只会埋头检索一段相似文本然后原样吐出来既不会判断问题里到底需要几份证据也不会在找不到资料的时候换个思路继续找。这种体验在Demo里还能看一旦放到生产环境面对真实用户的各种问法立刻就会露馅。所以当我看到基于FastAPI、LangChain、LangGraph和pgvector组合出来的Agentic RAG方案时第一反应是这条路走对了它把RAG从“查资料”这个单一动作扩展成了“理解、推理、决策”的完整闭环。这篇文章不是讲概念ppt我就直接从实战出发把从静态RAG迁移到Agentic RAG过程中踩过的坑、验证过的方案、以及最终跑通的架构设计完完整整写出来。如果你是正在做RAG应用开发的工程师或者准备给团队现有检索系统做智能化升级这篇内容应该能让你少走不少弯路。1. 传统RAG的瓶颈为什么静态方案撑不住真实场景1.1 静态RAG的典型工作流和它的“死穴”静态RAG的工作流大家都熟用户提问先做向量化然后在向量库里做相似度检索取回Top-K条片段拼到Prompt里丢给大模型生成答案。整个过程像一条流水线每个环节都是预定好的没有分支没有反馈没有自主判断。这个设计在文档库单一、问题模式固定的场景下是够用的比如FAQ问答、政策条款查询准度其实还不错。但一旦问题的复杂度上来静态RAG就撑不住了。我举一个自己项目中真实遇到的例子用户问“我们公司的报销制度里发票丢失的情况下还能不能走线下纸质审批跟电子发票的流程有什么区别”这个问题里至少藏着三个检索点——发票丢失的处理条款、线下纸质审批的适用条件、电子发票流程的说明。静态RAG的做法是拿整个问句去做向量最近邻搜索结果检索回来的Top-K片段大概率只覆盖了其中某一两个点剩下没有覆盖到的点大模型就只能靠自己的“想象力”去编造幻觉问题就是这么来的。另一个死穴是静态RAG没有“反思”能力。它检索完一轮就直接生成答案哪怕检索结果明显跑偏甚至根本是空结果它也不会触发二次检索或者换一种检索策略。我见过不少线上案例用户的问题明明在知识库里有明确答案但因为搜索时的表述方式和文档原文差异较大静态RAG召回失败系统就给出一个“抱歉我没找到相关信息”的扛不住式回答这在用户体验上是不可接受的。1.2 Agentic RAG到底改了什么三大核心能力Agentic RAG本质上不是颠覆RAG的检索和生成链路而是在外围加了一个“大脑”来调度这两个环节。具体来说它补齐了静态RAG缺失的三个能力。第一个是理解能力。Agent不再把用户的原话直接拿去检索而是先做一轮query分析——拆解用户意图识别问题里涉及的条件、实体、目标文档类型甚至把多跳问题拆成子查询。这一步做得好不好直接决定后续检索的质量。第二个是推理能力。Agent会结合检索回来的结果判断证据是否充分。如果发现检索到的内容互相矛盾或者缺失某个关键信息它会自动决定是换一种检索方式还是拆解成更细的查询还是把已有结果先汇总起来再继续下一步。这个“边检索边推理”的过程是Agentic RAG最核心的不同点。第三个是决策能力。Agent手里不只有向量检索这一个工具它可以调用全文检索、SQL查询、Web搜索、甚至调用外部API来补充信息。在每一步Agent都要根据当前的状态选择合适的工具和策略。这个决策过程不是静态规则写死的而是根据用户问题的上下文动态调整的。一句话总结静态RAG是“一段Prompt 一次检索”Agentic RAG是“一个会思考的工作流 多轮动态检索”。2. 技术选型思路FastAPI、LangGraph和pgvector为什么是黄金组合2.1 LangGraph提供了Agent循环的脚手架做Agentic RAG最怕的事情是自己从零造一个Agent引擎。你不仅要处理状态管理、循环控制、工具注册还要考虑各种边界情况比如死循环、上下文爆炸、工具调用超时。这些问题的复杂度一旦上来纯手写根本维护不住。LangGraph在这块的价值体现在它对图结构的原生支持上。Agent的执行流程可以被建模为一个有向图节点表示“执行计划”“调用检索工具”“分析结果”“生成回答”这些动作边表示节点之间的跳转条件。这个模型和Agentic RAG的自然流程高度匹配而且LangGraph内置了状态管理和条件边机制让我可以轻松实现“如果检索结果不足则重新规划”这类逻辑不需要额外造一套流程引擎。另外一个让我选择LangGraph的原因是它对历史会话的处理。Agentic RAG的检索决定有时候依赖前面轮次的对话上下文LangGraph的状态机制天然支持在图中维护一个状态对象每一步Agent决策都会读取和更新这个状态这就解决了多轮对话中的上下文传递问题。2.2 FastAPI守住服务入口pgvector当向量主库FastAPI在这套架构里的角色是支撑入口层。我的Agentic服务对外不是直接暴露LangGraph的内部状态而是通过一层REST API来封装。为什么选FastAPI除了它异步性能好、自带OpenAPI文档之外更重要的是它的生命周期管理能力——LangGraph的Agent实例可以作为应用级单例在FastAPI启动时初始化每个请求进来时复用同一个工作流实例只是在状态里隔离各自的会话数据。这个模式在并发压力下实测表现非常稳。pgvector则是向量检索层的承载主体。我没有选择专门的向量数据库原因很简单——我们的业务数据本身已经存在PostgreSQL里如果为了向量检索单独引入一套新的存储就需要维护两套元数据的一致性这在生产和运维层面是很大的负担。pgvector作为PostgreSQL的扩展直接把向量检索的能力内嵌进了原库包括metadata过滤、SQL条件筛选这些能力都能在向量检索时直接使用大大降低了架构复杂度。配合HNSW索引在百万级向量规模下也能保持几十毫秒级别的召回速度日常业务完全拉得开。开了归一化的embedding、调好HNSW参数之后检索延迟和准确率表现都在可接受范围内这部分后面我会单独展开。3. 核心架构设计Agentic RAG的执行引擎3.1 查询理解层把用户问题拆成可执行的检索计划整个Agentic RAG引擎的第一道关卡是查询理解层。我把它拆成两个环节意图分类和查询改写。意图分类的作用是判断用户的问题究竟属于哪一种检索任务。我定义了五类意图单点查询、多跳查询、比较查询、总结查询和无效查询。单点查询最常见比如“离职流程是什么”多跳查询需要多个子查询串联比如“北京分公司的报销流程和要求”比较查询需要抽取多个实体的属性做对比总结查询需要在多个文档中聚合信息无效查询则是和知识库内容无关的问题。这个分类我一开始试图用规则来做比如通过关键词匹配但效果很差。后来换成用大模型做分类在每个Agent执行循环开始前先让LLM根据用户的完整问题输出一个结构化的意图标签和检索计划。这样做的好处是LLM有一种“全局视角”能够给出比规则更准确的判断。查询改写模块负责把复杂问题拆解成多个子查询。这里的关键是不要丢信息。我的做法是让模型同时输出子查询列表和每个子查询的目的说明然后用这些子查询分别触发一轮独立的检索。拆解后的子查询之间可能是并列关系也可能是串联关系——串联关系就需要前一个子查询的结果作为后一个搜索的条件这个依赖关系我也会作为结构化信息传入后续的Agent循环。3.2 路由与工具注册让Agent手里有“牌”可打查询理解完成之后任务进入Agent执行循环。这个循环的每一步Agent都需要决定“下一步该做什么”。为了让它有得选我给它注册了一组工具包括向量检索工具、SQL查询工具、文档重读工具、以及一个用于信息不完全时的扩展搜索工具。向量检索工具是主力负责在pgvector里做相似度召回SQL查询工具用于那些可以从结构化数据库中直接取数的场景比如“近三个月各部门的报销单数量”文档重读工具用来对已经定位到的高价值文档做更细粒度的段落定位本质上是在第一次粗派检索后根据结果再做一次更精确的向量recall扩展搜索工具则是在判定知识库覆盖不足时允许Agent接入外部搜索API做背书。工具注册这块在LangGraph里的实现非常灵活。每个工具就是一个Python函数同时注明它的名称、描述和参数结构。LLM Agent在推理时会基于用户问题、历史状态和工具描述来选择调用哪个工具并填充对应的参数。这一层做得好不好依赖一个点工具描述必须清晰、没有歧义否则Agent很容易选错工具。3.3 自我纠正机制从“只检索一次”到“边想边查”这是Agentic RAG比静态RAG最突出的一点。自我纠正机制让生成不再是单向流水线而是变成了一个多轮决策过程。我在LangGraph里设计了一个条件循环Agent每执行一轮“检索-评估”都要判断当前累积的证据是否足够支撑最终答案。如果不足就继续规划下一轮检索如果资料相互矛盾就触发冲突消解如果连检索多次都无法获取有效信息Agent会明确告诉用户“知识库中没有找到相关信息建议补充资料”。这个自纠正过程里最麻烦的是防止死循环。我在状态里维护了一个step计数器和最大检索步数限制默认是5轮。一旦超过这个轮数无论证据充分与否Agent都会强制结束检索并输出一个基于现有信息的答案同时标注信息完整度。这个设计在线上运行时非常重要因为我见过太多Agent项目因为循环失控导致单次请求耗时十几秒甚至几十秒用户体验完全没法接受。4. 从零搭建Agentic RAG完整实操过程4.1 环境准备与依赖安装我这里以Python 3.10环境为例先把核心依赖列出来pip install fastapi uvicorn langchain langgraph pgvector sentence-transformers pydantic需要额外安装PostgreSQL的pgvector扩展。如果你用的是Docker可以直接拉带pgvector的镜像docker run --name pgvector-demo -e POSTGRES_PASSWORDpassword -p 5432:5432 -d pgvector/pgvector:pg16启动后创建扩展CREATE EXTENSION IF NOT EXISTS vector;然后建一张最基础的向量表CREATE TABLE IF NOT EXISTS documents ( id SERIAL PRIMARY KEY, content TEXT NOT NULL, metadata JSONB, embedding vector(768) );这里vector的维度要和你的Embedding模型输出维度一致。我用的是bge-large-zh-v1.5输出维度是1024所以上面768只是示例实际需要按模型调整。如果模型维度写错了插入时会直接报维度不匹配的错踩坑过一次。4.2 搭建pgvector检索工具层先写一个负责向量检索的封装模块这个模块会被LangGraph里的Agent当作工具来调用。import os import pgvector from pgvector.psycopg2 import register_vector import psycopg2 class VectorRetriever: def __init__(self, conn_string: str, embedding_modelNone): self.conn psycopg2.connect(conn_string) register_vector(self.conn) self.embedding_model embedding_model def get_embedding(self, text: str): return self.embedding_model.encode(text).tolist() def query(self, query_text: str, top_k: int 5, metadata_filter: dict None): query_embedding self.get_embedding(query_text) sql SELECT content, metadata, 1 - (embedding %s::vector) AS similarity FROM documents ORDER BY embedding %s::vector LIMIT %s params [query_embedding, query_embedding, top_k] if metadata_filter: # 这里可以拼接 metadata 的 JSONB 过滤条件 pass cur self.conn.cursor() cur.execute(sql, params) rows cur.fetchall() return [{content: r[0], metadata: r[1], score: r[2]} for r in rows]这里特别要说明的是相似度计算方式。pgvector里我用的是操作符这是余弦距离距离越小表示越相似所以取结果时的score是1 - distance。如果你想要和内积#做对比可以根据你的embedding模型特性来选择因为某些模型针对内积做过优化效果会更好。检索层还有一个容易被忽略的点就是metadata过滤。在真实的RAG应用里文档往往带有业务属性比如部门、日期、文档类型。如果在检索时能先用metadata缩小范围再算向量相似度不仅准确率更高性能也能明显提升。pgvector支持这种“过滤后排序”的方式SQL里带上WHERE条件就能实现。4.3 用LangGraph定义Agent工作流接下来是核心环节——用LangGraph把Agentic RAG的执行流程定义出来。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] query: str sub_queries: list retrieved_chunks: list current_step: int max_steps: int final_answer: str def analyze_query(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f 你是一个检索规划助手。请分析用户问题并输出检索计划。 用户问题: {state[query]} 请输出JSON格式: {{ intent: single|multi|comparison|summary|invalid, sub_queries: [..., ...], needs_sql: false }} response llm.invoke(prompt) return {sub_queries: response.get(sub_queries, [state[query]])} def retrieve(state: AgentState): planner state.get(sub_queries, [state[query]]) results [] for q in planner: retrieved vector_retriever.query(q, top_k3) results.extend(retrieved) return {retrieved_chunks: results} def generate_answer(state: AgentState): chunks state[retrieved_chunks] context \n\n.join([c[content] for c in chunks[:10]]) llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f基于以下资料回答问题。如果资料不足请明确回答“资料不足”。\n\n资料:\n{context}\n\n问题:\n{state[query]}\n\n答案: answer llm.invoke(prompt) return {final_answer: answer.content} def evaluate_retrieval(state: AgentState): # 判断检索结果是否充分 if not state[retrieved_chunks] and state[current_step] state[max_steps]: return retrieve # 没有结果继续检索 return generate # 有足够结果生成答案然后把这些节点和图结构串起来graph StateGraph(AgentState) graph.add_node(analyze, analyze_query) graph.add_node(retrieve, retrieve) graph.add_node(generate, generate_answer) graph.set_entry_point(analyze) graph.add_edge(analyze, retrieve) graph.add_conditional_edges( retrieve, evaluate_retrieval, {retrieve: retrieve, generate: generate} ) graph.add_edge(generate, END) app graph.compile()这段代码已经搭出了基线Agent循环。其中最关键的是evaluate_retrieval这个条件边函数它决定了Agent是继续检索还是进入生成阶段。在真实项目中我会在这个函数里引入更多判断逻辑比如检索结果的相关性得分、覆盖的子查询比例、是否出现矛盾信息等而不只是判断结果是否为空。4.4 把Agent工作流封装成FastAPI服务有了编译好的LangGraph工作流接下来把它封装成API。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAgentic RAG Service) class QueryRequest(BaseModel): query: str session_id: str class QueryResponse(BaseModel): answer: str retrieved_count: int steps_used: int app.post(/query, response_modelQueryResponse) async def handle_query(request: QueryRequest): result await app.async_run_agent( request.query, session_idrequest.session_id ) return QueryResponse( answerresult[final_answer], retrieved_countlen(result[retrieved_chunks]), steps_usedresult[current_step] )这里有个细节值得注意LangGraph的StateGraph在同步模式下是直接调用app.invoke()但FastAPI是异步框架如果在线程池里跑同步LLM调用虽然也能工作但会阻塞部分资源。更优雅的方式是用asyncio.create_task或者把LangGraph放到线程池里跑。如果LLM客户端本身支持异步可以逐步把工具里的调用改成异步版这样整体吞吐能再提升一截。4.5 一条真实查询的执行过程演示我把一个用户问题丢进去跑了一遍把过程写出来给大家直观体会一下Agentic RAG的执行链路。用户输入“电子发票丢失了用纸质发票流程补报销最高能报多少”第一步查询分析节点输出意图识别结果多跳查询子查询为“电子发票丢失如何处理”和“纸质发票报销限额”。第二步Agent先执行第一个子查询的向量检索库里有《电子发票管理规定》和《报销管理细则2024版》两个文档返回结果。Agent在评估阶段发现“纸质发票流程补报销”这个信息点在第一个子查询的返回片段中关联度不足于是主动触发了一次文档重读用“纸质发票 补报销 限额”作为新的检索式重新在《报销管理细则2024版》里定位。第三步两次检索取回的内容拼在一起Agent判定证据完备进入生成阶段。最终答案里既提到了电子发票丢失后允许转为纸质流程申请也提到了年度报销额度的上限符合业务规则。大家注意第三步里的这个判定逻辑如果静态RAG第一次检索的Top-K可能已经丢失了“限额”相关信息大模型要么编一个数字要么回答不完整。Agentic RAG通过自我评估发现问题、主动补检供电质量完全不可同日而语。5. 常见问题与排查技巧实录5.1 检索质量差Agent再聪明也白搭我遇到最多的问题是Agent循环跑得很欢每步都在检索但最终答案依然不行。这时候症状往往不在Agent而在向量检索层。你换个角度想想如果工具本身召回的Top-K片段本身相关性就低Agent的自我纠正最多也就是把一堆错误结果再组织一遍结果当然是错的。排查的要点是先看召回结果。我在调试阶段加了一层日志中间件把每一轮子查询的检索结果都记录到本地文件看具体哪句query的召回出了问题。最常见的原因是query改写阶段拆解出来的子查询语义偏差比如“哪些报销单可以走线下审核”应该拆成“线下审核流程”和“线下审核条件”结果模型拆成了“报销单”和“线下审核”前一个子查询太宽泛导致召回了一堆无关片段。解决方案是在子查询生成后加入一条“查询凝练”步骤让模型把每个子查询拆得更细节合并同义去掉停用词。这一步只增加了一次LLM调用但对召回精准度的提升非常明显。5.2 Agent循环“空转”检索结果一模一样另一个典型问题是Agent进入空转状态。比如第一次检索“报销流程”返回了文档A的前500字第二次同样用“报销流程”检索还是返回文档A的前500字Agent拿不到新信息又判断证据不足就陷入一个“检索-重试-再检索”的循环。我的限流器虽然能兜底强制结束但中间的几次重复调用的成本已经白白消耗了。解决办法有两个。第一检索工具内部做一个去重机制对已经出现在历史状态里的chunk做排除不让同一个片段在后续轮次里再次返回。具体可以在retrieve节点里读取state中已有的concatenated chunk ID列表然后用WHERE id NOT IN (...)进行过滤。第二让评估函数感知“本轮有没有新增信息”如果两轮检索的结果完全相同直接判定为“无法获得进一步信息”强制进入生成阶段而不是再尝试一轮。这种处理方式让系统的无效循环明显减少了单请求平均耗时降了约30%。5.3 pgvector配置和查询性能瓶颈pgvector的坑很大程度集中在索引和数据规模上。初期数据量小的时候几十万条向量全表扫描也能扛得住但数据量到几百万以后就会发现每次查询的时间飙升。HNSW索引不是等数据量大了再加减而是建表后数据量达到一定规模之前就要规划好否则建索引的过程会非常痛苦。我用的建索引SQLCREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops) WITH (m 16, ef_construction 64);这里的m控制每个节点的最大连接数ef_construction控制构建索引时的动态列表长度。这两个值不是越大越好m太大会让索引变得很大插入速度也会下降ef_construction太大同样会增加构建时间。我的经验是对于百万级数据量m16、ef_construction64是一个相对均衡的配置查询阶段可以根据实时需求调ef_search参数比如SET hnsw.ef_search 100;来平衡召回准确率和速度。另外还有一个小细节pgvector对向量列的数据类型要求是vector(n)n必须和你模型输出的维度完全一致。如果你换了Embedding模型维度变了现有的表结构就得改或者重建所以换模型之前一定要先想清楚。5.4 工具调用出错Agent会连环踩坑工具本身就是外部系统就免不了出错。比如SQL查询工具连不上数据库或者扩展搜索API返回超时这些错误如果只是简单抛异常Agent可能连续多次重试同一个工具不仅耗时间和tokens还有可能把异常信息带进后续推理导致最终答案完全跑偏。我的处理方式是在工具层做了一个统一的错误捕获机制。每个工具调用都会有try-except当异常发生时工具返回的不是异常对象而是标准化的错误上下文内容格式大概是{tool: sql_query, error: connection_timeout, message: 数据库连接超时无法执行查询}Agent拿到这个结果后会把这个“失败信息”当作一次有效的工具返回结果来处理从而触发它的决策逻辑要么换一个工具试试要么告诉用户某个数据项暂时不可用而不是傻傻地重试同一个错误工具。这种设计让系统的稳定性实打实地上了一个台阶。6. 几个给你省时间的实操心得6.1 从静态RAG迁移到Agentic RAG不是推翻重写很多小伙伴听到“Agentic”就觉得要把原来所有代码都换掉这其实是个误区。我做完整个项目后发现原来的文档切分、向量化、存储这部分基础建设完全保留真正新增的部分是查询分析、Agent循环、工具调度这三个层面。如果你现在已经有静态RAG的服务完全可以在不改动底层向量库的前提下在API层之上叠一个LangGraph的Agent过程编排就能逐步感受到Agentic RAG的效果。有一个坑在这里要提醒静态RAG时代你搭建的评估方式并不完全适用于Agentic RAG。静态RAG判定“这次检索对不对”只需要看召回内容的相关性但Agentic RAG还要评估“Agent有没有做出正确的决策”——比如它该不该触发二次检索、该不该换工具、该不该停在当前轮次。所以迁移期间我们团队花了大量时间构建一套新的、带过程指标的评估集每一条测试不只是标注最终答案的正误还要标注过程节点的合理性。6.2 搜索日志和链路追踪必须从第一天就接入Agentic RAG的调试复杂度和静态RAG完全不是一个量级。静态RAG出了错看一遍输入输出基本就能定位问题Agentic RAG的错可能发生在查询拆解、检索、评估、工具调用、答案生成任何一个环节如果没有完整的链路追踪排查起来会让人崩溃。我在项目起步时就让每个请求都有一个trace_idLangGraph每执行一个节点就把运行信息绑定到这个trace_id上包括节点名称、输入输出摘要、耗时、token消耗。前端调试页面里可以按照trace_id查看整条执行链路每一步用了什么工具、检索了哪些片段、评估结果是什么都一目了然。这个投入在初期会拖慢一些开发进度但对于Agent型应用的后期迭代来说是值得的。6.3 评估方式要跟着升级指标不能还盯着Top-K最后一件事也是我觉得Agentic RAG项目中容易走偏的地方——评估策略。很多人延续静态RAG的套路只关注准确率和召回率这两个指标在衡量最终答案方面的作用有限因为你很难判断“答案是错的”到底是因为检索环节出了问题还是Agent决策环节出了问题。我给团队的评估体系分成了三层。第一层是任务成功率由人工或强模型打分判断最终回答是否符合业务标准第二层是检索质量独立评估每一轮检索召回的chunk是否和对应子查询相关第三层是决策质量评估Agent在每一轮的状态跳转是否合理比如“在已有足够证据的情况下是否多做了无用检索”、“在证据不足的时候是否识别到了缺什么”。三层的指标分离开以后返工定位问题的效率确实高了很多。把评估体系做细是我在Agentic RAG落地中觉得最有价值、也最容易被忽略的一件事。Agentic RAG带来的不只是技术上的提升更多是一种设计视角的转变。过去我们问“怎么让检索更准”现在应该问“怎么让系统在不确定中自己找到通往准确答案的路径”。这种转变意味着更大的工程复杂度但也意味着系统真正的智能化潜力。整个过程做下来的体会是值得投入的路走的时候没有捷径。
从静态RAG到Agentic RAG：基于LangGraph与pgvector的工程实践
Agentic RAG这个概念最近在技术圈刷屏大家讨论的焦点已经从“RAG怎么做”变成了“RAG怎么不做得像个呆子”。我自己的体感很明显静态RAG的套路化太严重了——用户问个稍微带点弯的问题它就只会埋头检索一段相似文本然后原样吐出来既不会判断问题里到底需要几份证据也不会在找不到资料的时候换个思路继续找。这种体验在Demo里还能看一旦放到生产环境面对真实用户的各种问法立刻就会露馅。所以当我看到基于FastAPI、LangChain、LangGraph和pgvector组合出来的Agentic RAG方案时第一反应是这条路走对了它把RAG从“查资料”这个单一动作扩展成了“理解、推理、决策”的完整闭环。这篇文章不是讲概念ppt我就直接从实战出发把从静态RAG迁移到Agentic RAG过程中踩过的坑、验证过的方案、以及最终跑通的架构设计完完整整写出来。如果你是正在做RAG应用开发的工程师或者准备给团队现有检索系统做智能化升级这篇内容应该能让你少走不少弯路。1. 传统RAG的瓶颈为什么静态方案撑不住真实场景1.1 静态RAG的典型工作流和它的“死穴”静态RAG的工作流大家都熟用户提问先做向量化然后在向量库里做相似度检索取回Top-K条片段拼到Prompt里丢给大模型生成答案。整个过程像一条流水线每个环节都是预定好的没有分支没有反馈没有自主判断。这个设计在文档库单一、问题模式固定的场景下是够用的比如FAQ问答、政策条款查询准度其实还不错。但一旦问题的复杂度上来静态RAG就撑不住了。我举一个自己项目中真实遇到的例子用户问“我们公司的报销制度里发票丢失的情况下还能不能走线下纸质审批跟电子发票的流程有什么区别”这个问题里至少藏着三个检索点——发票丢失的处理条款、线下纸质审批的适用条件、电子发票流程的说明。静态RAG的做法是拿整个问句去做向量最近邻搜索结果检索回来的Top-K片段大概率只覆盖了其中某一两个点剩下没有覆盖到的点大模型就只能靠自己的“想象力”去编造幻觉问题就是这么来的。另一个死穴是静态RAG没有“反思”能力。它检索完一轮就直接生成答案哪怕检索结果明显跑偏甚至根本是空结果它也不会触发二次检索或者换一种检索策略。我见过不少线上案例用户的问题明明在知识库里有明确答案但因为搜索时的表述方式和文档原文差异较大静态RAG召回失败系统就给出一个“抱歉我没找到相关信息”的扛不住式回答这在用户体验上是不可接受的。1.2 Agentic RAG到底改了什么三大核心能力Agentic RAG本质上不是颠覆RAG的检索和生成链路而是在外围加了一个“大脑”来调度这两个环节。具体来说它补齐了静态RAG缺失的三个能力。第一个是理解能力。Agent不再把用户的原话直接拿去检索而是先做一轮query分析——拆解用户意图识别问题里涉及的条件、实体、目标文档类型甚至把多跳问题拆成子查询。这一步做得好不好直接决定后续检索的质量。第二个是推理能力。Agent会结合检索回来的结果判断证据是否充分。如果发现检索到的内容互相矛盾或者缺失某个关键信息它会自动决定是换一种检索方式还是拆解成更细的查询还是把已有结果先汇总起来再继续下一步。这个“边检索边推理”的过程是Agentic RAG最核心的不同点。第三个是决策能力。Agent手里不只有向量检索这一个工具它可以调用全文检索、SQL查询、Web搜索、甚至调用外部API来补充信息。在每一步Agent都要根据当前的状态选择合适的工具和策略。这个决策过程不是静态规则写死的而是根据用户问题的上下文动态调整的。一句话总结静态RAG是“一段Prompt 一次检索”Agentic RAG是“一个会思考的工作流 多轮动态检索”。2. 技术选型思路FastAPI、LangGraph和pgvector为什么是黄金组合2.1 LangGraph提供了Agent循环的脚手架做Agentic RAG最怕的事情是自己从零造一个Agent引擎。你不仅要处理状态管理、循环控制、工具注册还要考虑各种边界情况比如死循环、上下文爆炸、工具调用超时。这些问题的复杂度一旦上来纯手写根本维护不住。LangGraph在这块的价值体现在它对图结构的原生支持上。Agent的执行流程可以被建模为一个有向图节点表示“执行计划”“调用检索工具”“分析结果”“生成回答”这些动作边表示节点之间的跳转条件。这个模型和Agentic RAG的自然流程高度匹配而且LangGraph内置了状态管理和条件边机制让我可以轻松实现“如果检索结果不足则重新规划”这类逻辑不需要额外造一套流程引擎。另外一个让我选择LangGraph的原因是它对历史会话的处理。Agentic RAG的检索决定有时候依赖前面轮次的对话上下文LangGraph的状态机制天然支持在图中维护一个状态对象每一步Agent决策都会读取和更新这个状态这就解决了多轮对话中的上下文传递问题。2.2 FastAPI守住服务入口pgvector当向量主库FastAPI在这套架构里的角色是支撑入口层。我的Agentic服务对外不是直接暴露LangGraph的内部状态而是通过一层REST API来封装。为什么选FastAPI除了它异步性能好、自带OpenAPI文档之外更重要的是它的生命周期管理能力——LangGraph的Agent实例可以作为应用级单例在FastAPI启动时初始化每个请求进来时复用同一个工作流实例只是在状态里隔离各自的会话数据。这个模式在并发压力下实测表现非常稳。pgvector则是向量检索层的承载主体。我没有选择专门的向量数据库原因很简单——我们的业务数据本身已经存在PostgreSQL里如果为了向量检索单独引入一套新的存储就需要维护两套元数据的一致性这在生产和运维层面是很大的负担。pgvector作为PostgreSQL的扩展直接把向量检索的能力内嵌进了原库包括metadata过滤、SQL条件筛选这些能力都能在向量检索时直接使用大大降低了架构复杂度。配合HNSW索引在百万级向量规模下也能保持几十毫秒级别的召回速度日常业务完全拉得开。开了归一化的embedding、调好HNSW参数之后检索延迟和准确率表现都在可接受范围内这部分后面我会单独展开。3. 核心架构设计Agentic RAG的执行引擎3.1 查询理解层把用户问题拆成可执行的检索计划整个Agentic RAG引擎的第一道关卡是查询理解层。我把它拆成两个环节意图分类和查询改写。意图分类的作用是判断用户的问题究竟属于哪一种检索任务。我定义了五类意图单点查询、多跳查询、比较查询、总结查询和无效查询。单点查询最常见比如“离职流程是什么”多跳查询需要多个子查询串联比如“北京分公司的报销流程和要求”比较查询需要抽取多个实体的属性做对比总结查询需要在多个文档中聚合信息无效查询则是和知识库内容无关的问题。这个分类我一开始试图用规则来做比如通过关键词匹配但效果很差。后来换成用大模型做分类在每个Agent执行循环开始前先让LLM根据用户的完整问题输出一个结构化的意图标签和检索计划。这样做的好处是LLM有一种“全局视角”能够给出比规则更准确的判断。查询改写模块负责把复杂问题拆解成多个子查询。这里的关键是不要丢信息。我的做法是让模型同时输出子查询列表和每个子查询的目的说明然后用这些子查询分别触发一轮独立的检索。拆解后的子查询之间可能是并列关系也可能是串联关系——串联关系就需要前一个子查询的结果作为后一个搜索的条件这个依赖关系我也会作为结构化信息传入后续的Agent循环。3.2 路由与工具注册让Agent手里有“牌”可打查询理解完成之后任务进入Agent执行循环。这个循环的每一步Agent都需要决定“下一步该做什么”。为了让它有得选我给它注册了一组工具包括向量检索工具、SQL查询工具、文档重读工具、以及一个用于信息不完全时的扩展搜索工具。向量检索工具是主力负责在pgvector里做相似度召回SQL查询工具用于那些可以从结构化数据库中直接取数的场景比如“近三个月各部门的报销单数量”文档重读工具用来对已经定位到的高价值文档做更细粒度的段落定位本质上是在第一次粗派检索后根据结果再做一次更精确的向量recall扩展搜索工具则是在判定知识库覆盖不足时允许Agent接入外部搜索API做背书。工具注册这块在LangGraph里的实现非常灵活。每个工具就是一个Python函数同时注明它的名称、描述和参数结构。LLM Agent在推理时会基于用户问题、历史状态和工具描述来选择调用哪个工具并填充对应的参数。这一层做得好不好依赖一个点工具描述必须清晰、没有歧义否则Agent很容易选错工具。3.3 自我纠正机制从“只检索一次”到“边想边查”这是Agentic RAG比静态RAG最突出的一点。自我纠正机制让生成不再是单向流水线而是变成了一个多轮决策过程。我在LangGraph里设计了一个条件循环Agent每执行一轮“检索-评估”都要判断当前累积的证据是否足够支撑最终答案。如果不足就继续规划下一轮检索如果资料相互矛盾就触发冲突消解如果连检索多次都无法获取有效信息Agent会明确告诉用户“知识库中没有找到相关信息建议补充资料”。这个自纠正过程里最麻烦的是防止死循环。我在状态里维护了一个step计数器和最大检索步数限制默认是5轮。一旦超过这个轮数无论证据充分与否Agent都会强制结束检索并输出一个基于现有信息的答案同时标注信息完整度。这个设计在线上运行时非常重要因为我见过太多Agent项目因为循环失控导致单次请求耗时十几秒甚至几十秒用户体验完全没法接受。4. 从零搭建Agentic RAG完整实操过程4.1 环境准备与依赖安装我这里以Python 3.10环境为例先把核心依赖列出来pip install fastapi uvicorn langchain langgraph pgvector sentence-transformers pydantic需要额外安装PostgreSQL的pgvector扩展。如果你用的是Docker可以直接拉带pgvector的镜像docker run --name pgvector-demo -e POSTGRES_PASSWORDpassword -p 5432:5432 -d pgvector/pgvector:pg16启动后创建扩展CREATE EXTENSION IF NOT EXISTS vector;然后建一张最基础的向量表CREATE TABLE IF NOT EXISTS documents ( id SERIAL PRIMARY KEY, content TEXT NOT NULL, metadata JSONB, embedding vector(768) );这里vector的维度要和你的Embedding模型输出维度一致。我用的是bge-large-zh-v1.5输出维度是1024所以上面768只是示例实际需要按模型调整。如果模型维度写错了插入时会直接报维度不匹配的错踩坑过一次。4.2 搭建pgvector检索工具层先写一个负责向量检索的封装模块这个模块会被LangGraph里的Agent当作工具来调用。import os import pgvector from pgvector.psycopg2 import register_vector import psycopg2 class VectorRetriever: def __init__(self, conn_string: str, embedding_modelNone): self.conn psycopg2.connect(conn_string) register_vector(self.conn) self.embedding_model embedding_model def get_embedding(self, text: str): return self.embedding_model.encode(text).tolist() def query(self, query_text: str, top_k: int 5, metadata_filter: dict None): query_embedding self.get_embedding(query_text) sql SELECT content, metadata, 1 - (embedding %s::vector) AS similarity FROM documents ORDER BY embedding %s::vector LIMIT %s params [query_embedding, query_embedding, top_k] if metadata_filter: # 这里可以拼接 metadata 的 JSONB 过滤条件 pass cur self.conn.cursor() cur.execute(sql, params) rows cur.fetchall() return [{content: r[0], metadata: r[1], score: r[2]} for r in rows]这里特别要说明的是相似度计算方式。pgvector里我用的是操作符这是余弦距离距离越小表示越相似所以取结果时的score是1 - distance。如果你想要和内积#做对比可以根据你的embedding模型特性来选择因为某些模型针对内积做过优化效果会更好。检索层还有一个容易被忽略的点就是metadata过滤。在真实的RAG应用里文档往往带有业务属性比如部门、日期、文档类型。如果在检索时能先用metadata缩小范围再算向量相似度不仅准确率更高性能也能明显提升。pgvector支持这种“过滤后排序”的方式SQL里带上WHERE条件就能实现。4.3 用LangGraph定义Agent工作流接下来是核心环节——用LangGraph把Agentic RAG的执行流程定义出来。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] query: str sub_queries: list retrieved_chunks: list current_step: int max_steps: int final_answer: str def analyze_query(state: AgentState): llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f 你是一个检索规划助手。请分析用户问题并输出检索计划。 用户问题: {state[query]} 请输出JSON格式: {{ intent: single|multi|comparison|summary|invalid, sub_queries: [..., ...], needs_sql: false }} response llm.invoke(prompt) return {sub_queries: response.get(sub_queries, [state[query]])} def retrieve(state: AgentState): planner state.get(sub_queries, [state[query]]) results [] for q in planner: retrieved vector_retriever.query(q, top_k3) results.extend(retrieved) return {retrieved_chunks: results} def generate_answer(state: AgentState): chunks state[retrieved_chunks] context \n\n.join([c[content] for c in chunks[:10]]) llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f基于以下资料回答问题。如果资料不足请明确回答“资料不足”。\n\n资料:\n{context}\n\n问题:\n{state[query]}\n\n答案: answer llm.invoke(prompt) return {final_answer: answer.content} def evaluate_retrieval(state: AgentState): # 判断检索结果是否充分 if not state[retrieved_chunks] and state[current_step] state[max_steps]: return retrieve # 没有结果继续检索 return generate # 有足够结果生成答案然后把这些节点和图结构串起来graph StateGraph(AgentState) graph.add_node(analyze, analyze_query) graph.add_node(retrieve, retrieve) graph.add_node(generate, generate_answer) graph.set_entry_point(analyze) graph.add_edge(analyze, retrieve) graph.add_conditional_edges( retrieve, evaluate_retrieval, {retrieve: retrieve, generate: generate} ) graph.add_edge(generate, END) app graph.compile()这段代码已经搭出了基线Agent循环。其中最关键的是evaluate_retrieval这个条件边函数它决定了Agent是继续检索还是进入生成阶段。在真实项目中我会在这个函数里引入更多判断逻辑比如检索结果的相关性得分、覆盖的子查询比例、是否出现矛盾信息等而不只是判断结果是否为空。4.4 把Agent工作流封装成FastAPI服务有了编译好的LangGraph工作流接下来把它封装成API。from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleAgentic RAG Service) class QueryRequest(BaseModel): query: str session_id: str class QueryResponse(BaseModel): answer: str retrieved_count: int steps_used: int app.post(/query, response_modelQueryResponse) async def handle_query(request: QueryRequest): result await app.async_run_agent( request.query, session_idrequest.session_id ) return QueryResponse( answerresult[final_answer], retrieved_countlen(result[retrieved_chunks]), steps_usedresult[current_step] )这里有个细节值得注意LangGraph的StateGraph在同步模式下是直接调用app.invoke()但FastAPI是异步框架如果在线程池里跑同步LLM调用虽然也能工作但会阻塞部分资源。更优雅的方式是用asyncio.create_task或者把LangGraph放到线程池里跑。如果LLM客户端本身支持异步可以逐步把工具里的调用改成异步版这样整体吞吐能再提升一截。4.5 一条真实查询的执行过程演示我把一个用户问题丢进去跑了一遍把过程写出来给大家直观体会一下Agentic RAG的执行链路。用户输入“电子发票丢失了用纸质发票流程补报销最高能报多少”第一步查询分析节点输出意图识别结果多跳查询子查询为“电子发票丢失如何处理”和“纸质发票报销限额”。第二步Agent先执行第一个子查询的向量检索库里有《电子发票管理规定》和《报销管理细则2024版》两个文档返回结果。Agent在评估阶段发现“纸质发票流程补报销”这个信息点在第一个子查询的返回片段中关联度不足于是主动触发了一次文档重读用“纸质发票 补报销 限额”作为新的检索式重新在《报销管理细则2024版》里定位。第三步两次检索取回的内容拼在一起Agent判定证据完备进入生成阶段。最终答案里既提到了电子发票丢失后允许转为纸质流程申请也提到了年度报销额度的上限符合业务规则。大家注意第三步里的这个判定逻辑如果静态RAG第一次检索的Top-K可能已经丢失了“限额”相关信息大模型要么编一个数字要么回答不完整。Agentic RAG通过自我评估发现问题、主动补检供电质量完全不可同日而语。5. 常见问题与排查技巧实录5.1 检索质量差Agent再聪明也白搭我遇到最多的问题是Agent循环跑得很欢每步都在检索但最终答案依然不行。这时候症状往往不在Agent而在向量检索层。你换个角度想想如果工具本身召回的Top-K片段本身相关性就低Agent的自我纠正最多也就是把一堆错误结果再组织一遍结果当然是错的。排查的要点是先看召回结果。我在调试阶段加了一层日志中间件把每一轮子查询的检索结果都记录到本地文件看具体哪句query的召回出了问题。最常见的原因是query改写阶段拆解出来的子查询语义偏差比如“哪些报销单可以走线下审核”应该拆成“线下审核流程”和“线下审核条件”结果模型拆成了“报销单”和“线下审核”前一个子查询太宽泛导致召回了一堆无关片段。解决方案是在子查询生成后加入一条“查询凝练”步骤让模型把每个子查询拆得更细节合并同义去掉停用词。这一步只增加了一次LLM调用但对召回精准度的提升非常明显。5.2 Agent循环“空转”检索结果一模一样另一个典型问题是Agent进入空转状态。比如第一次检索“报销流程”返回了文档A的前500字第二次同样用“报销流程”检索还是返回文档A的前500字Agent拿不到新信息又判断证据不足就陷入一个“检索-重试-再检索”的循环。我的限流器虽然能兜底强制结束但中间的几次重复调用的成本已经白白消耗了。解决办法有两个。第一检索工具内部做一个去重机制对已经出现在历史状态里的chunk做排除不让同一个片段在后续轮次里再次返回。具体可以在retrieve节点里读取state中已有的concatenated chunk ID列表然后用WHERE id NOT IN (...)进行过滤。第二让评估函数感知“本轮有没有新增信息”如果两轮检索的结果完全相同直接判定为“无法获得进一步信息”强制进入生成阶段而不是再尝试一轮。这种处理方式让系统的无效循环明显减少了单请求平均耗时降了约30%。5.3 pgvector配置和查询性能瓶颈pgvector的坑很大程度集中在索引和数据规模上。初期数据量小的时候几十万条向量全表扫描也能扛得住但数据量到几百万以后就会发现每次查询的时间飙升。HNSW索引不是等数据量大了再加减而是建表后数据量达到一定规模之前就要规划好否则建索引的过程会非常痛苦。我用的建索引SQLCREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops) WITH (m 16, ef_construction 64);这里的m控制每个节点的最大连接数ef_construction控制构建索引时的动态列表长度。这两个值不是越大越好m太大会让索引变得很大插入速度也会下降ef_construction太大同样会增加构建时间。我的经验是对于百万级数据量m16、ef_construction64是一个相对均衡的配置查询阶段可以根据实时需求调ef_search参数比如SET hnsw.ef_search 100;来平衡召回准确率和速度。另外还有一个小细节pgvector对向量列的数据类型要求是vector(n)n必须和你模型输出的维度完全一致。如果你换了Embedding模型维度变了现有的表结构就得改或者重建所以换模型之前一定要先想清楚。5.4 工具调用出错Agent会连环踩坑工具本身就是外部系统就免不了出错。比如SQL查询工具连不上数据库或者扩展搜索API返回超时这些错误如果只是简单抛异常Agent可能连续多次重试同一个工具不仅耗时间和tokens还有可能把异常信息带进后续推理导致最终答案完全跑偏。我的处理方式是在工具层做了一个统一的错误捕获机制。每个工具调用都会有try-except当异常发生时工具返回的不是异常对象而是标准化的错误上下文内容格式大概是{tool: sql_query, error: connection_timeout, message: 数据库连接超时无法执行查询}Agent拿到这个结果后会把这个“失败信息”当作一次有效的工具返回结果来处理从而触发它的决策逻辑要么换一个工具试试要么告诉用户某个数据项暂时不可用而不是傻傻地重试同一个错误工具。这种设计让系统的稳定性实打实地上了一个台阶。6. 几个给你省时间的实操心得6.1 从静态RAG迁移到Agentic RAG不是推翻重写很多小伙伴听到“Agentic”就觉得要把原来所有代码都换掉这其实是个误区。我做完整个项目后发现原来的文档切分、向量化、存储这部分基础建设完全保留真正新增的部分是查询分析、Agent循环、工具调度这三个层面。如果你现在已经有静态RAG的服务完全可以在不改动底层向量库的前提下在API层之上叠一个LangGraph的Agent过程编排就能逐步感受到Agentic RAG的效果。有一个坑在这里要提醒静态RAG时代你搭建的评估方式并不完全适用于Agentic RAG。静态RAG判定“这次检索对不对”只需要看召回内容的相关性但Agentic RAG还要评估“Agent有没有做出正确的决策”——比如它该不该触发二次检索、该不该换工具、该不该停在当前轮次。所以迁移期间我们团队花了大量时间构建一套新的、带过程指标的评估集每一条测试不只是标注最终答案的正误还要标注过程节点的合理性。6.2 搜索日志和链路追踪必须从第一天就接入Agentic RAG的调试复杂度和静态RAG完全不是一个量级。静态RAG出了错看一遍输入输出基本就能定位问题Agentic RAG的错可能发生在查询拆解、检索、评估、工具调用、答案生成任何一个环节如果没有完整的链路追踪排查起来会让人崩溃。我在项目起步时就让每个请求都有一个trace_idLangGraph每执行一个节点就把运行信息绑定到这个trace_id上包括节点名称、输入输出摘要、耗时、token消耗。前端调试页面里可以按照trace_id查看整条执行链路每一步用了什么工具、检索了哪些片段、评估结果是什么都一目了然。这个投入在初期会拖慢一些开发进度但对于Agent型应用的后期迭代来说是值得的。6.3 评估方式要跟着升级指标不能还盯着Top-K最后一件事也是我觉得Agentic RAG项目中容易走偏的地方——评估策略。很多人延续静态RAG的套路只关注准确率和召回率这两个指标在衡量最终答案方面的作用有限因为你很难判断“答案是错的”到底是因为检索环节出了问题还是Agent决策环节出了问题。我给团队的评估体系分成了三层。第一层是任务成功率由人工或强模型打分判断最终回答是否符合业务标准第二层是检索质量独立评估每一轮检索召回的chunk是否和对应子查询相关第三层是决策质量评估Agent在每一轮的状态跳转是否合理比如“在已有足够证据的情况下是否多做了无用检索”、“在证据不足的时候是否识别到了缺什么”。三层的指标分离开以后返工定位问题的效率确实高了很多。把评估体系做细是我在Agentic RAG落地中觉得最有价值、也最容易被忽略的一件事。Agentic RAG带来的不只是技术上的提升更多是一种设计视角的转变。过去我们问“怎么让检索更准”现在应该问“怎么让系统在不确定中自己找到通往准确答案的路径”。这种转变意味着更大的工程复杂度但也意味着系统真正的智能化潜力。整个过程做下来的体会是值得投入的路走的时候没有捷径。
