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_PASSWORD=password -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_model=None): 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(model="gpt-4o-mini", temperature=0) 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_k=3) 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(model="gpt-4o-mini", temperature=0) 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(title="Agentic 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_model=QueryResponse) async def handle_query(request: QueryRequest): result = await app.async_run_agent( request.query, session_id=request.session_id ) return QueryResponse( answer=result["final_answer"], retrieved_count=len(result["retrieved_chunks"]), steps_used=result["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索引不是等数据量大了再加减,而是建表后数据量达到一定规模之前就要规划好,否则建索引的过程会非常痛苦。
我用的建索引SQL:
CREATE INDEX ON documents USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);这里的m控制每个节点的最大连接数,ef_construction控制构建索引时的动态列表长度。这两个值不是越大越好,m太大会让索引变得很大,插入速度也会下降;ef_construction太大同样会增加构建时间。我的经验是,对于百万级数据量,m=16、ef_construction=64是一个相对均衡的配置,查询阶段可以根据实时需求调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_id,LangGraph每执行一个节点就把运行信息绑定到这个trace_id上,包括节点名称、输入输出摘要、耗时、token消耗。前端调试页面里可以按照trace_id查看整条执行链路,每一步用了什么工具、检索了哪些片段、评估结果是什么都一目了然。这个投入在初期会拖慢一些开发进度,但对于Agent型应用的后期迭代来说,是值得的。
6.3 评估方式要跟着升级,指标不能还盯着Top-K
最后一件事,也是我觉得Agentic RAG项目中容易走偏的地方——评估策略。很多人延续静态RAG的套路,只关注准确率和召回率,这两个指标在衡量最终答案方面的作用有限,因为你很难判断“答案是错的”到底是因为检索环节出了问题,还是Agent决策环节出了问题。
我给团队的评估体系分成了三层。第一层是任务成功率,由人工或强模型打分判断最终回答是否符合业务标准;第二层是检索质量,独立评估每一轮检索召回的chunk是否和对应子查询相关;第三层是决策质量,评估Agent在每一轮的状态跳转是否合理,比如“在已有足够证据的情况下是否多做了无用检索”、“在证据不足的时候是否识别到了缺什么”。三层的指标分离开以后,返工定位问题的效率确实高了很多。把评估体系做细,是我在Agentic RAG落地中觉得最有价值、也最容易被忽略的一件事。
Agentic RAG带来的不只是技术上的提升,更多是一种设计视角的转变。过去我们问“怎么让检索更准”,现在应该问“怎么让系统在不确定中自己找到通往准确答案的路径”。这种转变意味着更大的工程复杂度,但也意味着系统真正的智能化潜力。整个过程做下来的体会是,值得投入的路,走的时候没有捷径。