☰
Agentic RAG实战:基于LangGraph与Elasticsearch构建可迭代检索增强生成系统
2026/9/30 16:38:56 网站建设 项目流程

1. 为什么传统 RAG 开始不够用了

1.1 从“检索增强”到“检索即思考”的转变

RAG 这个词,做 LLM 应用的人都不陌生。核心逻辑就一句话:用户问问题,系统先去知识库里捞相关文档,再把文档塞进大模型的上下文,让模型基于这些材料生成答案。这套范式在过去两年里几乎是所有“企业知识库问答”的标配方案,LangChain 里几十行代码就能跑通一个 demo。

但真正在生产环境里跑过 RAG 项目的人都知道,传统 RAG 的天花板非常明显。用户问“我们公司去年 Q3 在华东区的销售额同比变化,以及主要驱动因素是什么”,朴素 RAG 的做法是拿这句话去做向量检索,捞回来一堆可能包含“华东区”“销售额”字样的文档片段,然后让模型硬答。结果往往是:数据不全、口径不对、时间范围搞错,甚至模型直接编一个看起来合理的数字。

问题出在哪?传统 RAG 把“检索”当成了一次性的、被动的动作,检索完就生成,中间没有任何判断、验证、补检索的环节。它不会问自己:“我捞到的这些材料够不够回答问题?”“有没有遗漏关键维度?”“这个数字和那个数字矛盾了怎么办?”

Agentic RAG 要解决的就是这个问题。它的核心思想是:把检索从“一次性动作”变成“一个由智能体驱动的、可迭代的决策过程”。智能体会规划需要查什么、判断查到的够不够、不够就换个查询词再查、发现矛盾就去交叉验证、最后才组织答案。这中间的每一步,都是 LLM 在做决策,而不是靠固定的代码流程。

1.2 一个生活化的类比:从“自动售货机”到“私人助理”

我用一个类比来解释这个差异。传统 RAG 像自动售货机:你投币、选商品、掉出来,流程固定,选错了只能重新投币。Agentic RAG 像私人助理:你说“帮我准备一份竞品分析”,助理会先想需要哪些维度的信息,然后去查资料,查完发现某个竞品的数据缺失,会主动再去找,最后整理成一份有逻辑的报告交给你。

这个差异落到技术实现上,就是控制流的归属问题。传统 RAG 的控制流是开发者写死的:检索 → 拼接 → 生成。Agentic RAG 的控制流是 LLM 动态决定的:它可能检索三次、可能先检索再计算再检索、可能发现某个工具更适合当前任务而放弃检索。LangGraph 这类框架之所以在 Agentic RAG 的讨论里频繁出现,就是因为它提供了图结构的控制流编排能力,让“循环”“条件分支”“多智能体协作”这些传统链式框架很难表达的模式变得自然。

1.3 谁适合往下读

这篇内容适合三类人:一是已经跑通过基础 RAG demo、但发现效果在生产环境不达标的开发者;二是正在做企业知识库、智能客服、文档问答类项目,需要提升回答准确率和覆盖度的工程师;三是对 LangGraph、Agent 编排感兴趣,想找一个具体落地场景来练手的技术人。我会从架构设计讲到代码实现,再讲到踩坑经验,尽量让不同基础的人都能拿走能用的东西。

2. Agentic RAG 的核心架构拆解

2.1 四个关键角色:规划者、检索者、评估者、生成者

Agentic RAG 不是单一模块的升级,而是整个流程的重构。我把它拆成四个核心角色,每个角色可以由同一个 LLM 扮演,也可以拆成多个专职智能体。

规划者(Planner)负责把用户的原始问题拆解成可执行的检索步骤。比如“对比 A 产品和 B 产品在定价、功能、用户口碑三个维度的差异”,规划者会输出一个结构化的查询计划:先查 A 的定价、再查 B 的定价、再查 A 的功能列表……这一步的价值在于,它把模糊的自然语言问题转化成了明确的检索目标,避免了“拿整句话去检索”导致的语义稀释。

检索者(Retriever)执行具体的检索动作。这里可以是向量检索、关键词检索、混合检索,也可以是调用外部 API 或数据库查询。在 Agentic 架构里,检索者通常被封装成工具(Tool),由智能体按需调用。

评估者(Evaluator)是 Agentic RAG 区别于传统 RAG 的关键角色。它负责判断检索结果的相关性、完整性和一致性。常见做法是用 LLM 对每个检索片段打分,或者用专门的 rerank 模型做精排。如果评估发现材料不足,会触发重新检索或调整查询策略。

生成者(Generator)最后基于通过评估的材料组织答案。和传统 RAG 不同的是,生成者拿到的材料是经过筛选和验证的,而且可能附带结构化的推理链。

这四个角色的协作方式,决定了 Agentic RAG 的实际效果。我见过不少项目只加了“评估”环节,效果就有明显提升,因为大部分 RAG 错误其实来自“检索到了不相关的内容但模型还是硬答”。

2.2 为什么选 LangGraph 而不是 LangChain 的 Chain

这是被问得最多的问题之一。LangChain 的 Chain 是线性的、有向无环的,适合“步骤固定、顺序执行”的场景。但 Agentic RAG 需要循环(检索不够就再检索)、需要条件分支(评估通过就生成,不通过就重查)、需要状态管理(多轮检索的结果要累积)。这些用 Chain 表达非常别扭,你得写一堆回调或者自定义逻辑。

LangGraph 把整个流程建模成一张图,节点是执行单元,边是控制流。循环就是一条指回前面节点的边,条件分支就是带条件的边。状态通过一个共享的 State 对象在节点间传递。这种模型和 Agentic RAG 的需求天然契合。

我个人的经验是:如果你的 RAG 流程里出现了“如果……就再查一次”这种逻辑,就该考虑 LangGraph 了。用 Chain 硬写也能实现,但可维护性和可观测性会差很多。

2.3 检索层的选型:Elasticsearch 在 Agentic RAG 里的位置

热词里 Elasticsearch 出现频率很高,这不是偶然。Agentic RAG 对检索层的要求比传统 RAG 更高,因为智能体会发起多次、多种类型的检索请求。纯向量检索在语义匹配上强,但在精确匹配(比如产品型号、错误码、人名)上弱;纯关键词检索反过来。混合检索(Hybrid Search)是更稳妥的选择,而 Elasticsearch 从 8.x 版本开始原生支持向量检索和混合检索,加上它本身在全文检索、聚合分析、过滤条件上的成熟度,成了很多团队的首选。

另一个考虑是运维成本。很多团队已经有 ES 集群在跑日志或搜索业务,复用它来做 RAG 的检索层,比新引入一个专用向量数据库要省事。当然,如果你的场景是纯语义问答、数据量在百万级以下,用轻量级的向量库也完全够用。选型没有绝对优劣,关键看你的检索需求复杂度和现有基础设施。

3. 核心环节的实操实现

3.1 环境准备与依赖安装

先把基础环境搭起来。我假设你已经有一个可用的 LLM 接口(本地部署或 API 都行),Python 环境 3.10 以上。

pip install langgraph langchain langchain-community elasticsearch pip install sentence-transformers # 用于本地 embedding

Elasticsearch 的启动,Windows 用户直接下载压缩包解压后运行bin\elasticsearch.bat即可,注意默认需要 2GB 以上堆内存,在config\jvm.options里可以调。Linux 用户如果用 Docker,一条命令:

docker run -d --name es -p 9200:9200 -e "discovery.type=single-node" -e "ES_JAVA_OPTS=-Xms2g -Xmx2g" elasticsearch:8.13.0

注意:ES 8.x 默认开启安全认证,本地开发可以临时关闭xpack.security.enabled=false,但生产环境务必配置好证书和账号密码。

3.2 定义共享状态:Agentic RAG 的“记忆”

LangGraph 的核心概念之一是 State。它是在节点之间传递的数据结构,相当于整个流程的“共享内存”。Agentic RAG 的 State 通常包含这些字段:

from typing import TypedDict, List, Annotated from operator import add class RAGState(TypedDict): question: str # 原始问题 query_plan: List[str] # 规划出的查询列表 retrieved_docs: Annotated[List[dict], add] # 累积的检索结果 evaluation: str # 评估结论 retry_count: int # 重试次数 final_answer: str # 最终答案

这里有个细节值得说:retrieved_docs用了Annotated[List[dict], add],意思是每次节点返回新文档时,会追加到已有列表而不是覆盖。这个设计让多轮检索的结果能自然累积,不用手动合并。retry_count是用来防止无限循环的,后面会讲。

3.3 规划节点:把大问题拆成小查询

规划节点的职责是让 LLM 输出一个查询列表。Prompt 的设计很关键,我一般会这样写:

PLANNER_PROMPT = """你是一个检索规划助手。用户提出了一个问题,你需要把它拆解成若干个独立的检索查询。 要求: 1. 每个查询应该聚焦一个具体的信息点 2. 查询之间尽量不要语义重叠 3. 最多输出 5 个查询 4. 直接输出查询列表,每行一个,不要编号,不要解释 用户问题:{question} """

实测下来,限制查询数量上限非常重要。不限制的话,LLM 有时候会输出十几个查询,导致检索成本飙升,而且很多是冗余的。5 个是我在多个项目里试出来的比较平衡的值,覆盖度够,成本可控。

规划节点的代码:

def plan_node(state: RAGState): response = llm.invoke(PLANNER_PROMPT.format(question=state["question"])) queries = [q.strip() for q in response.content.strip().split("\n") if q.strip()] return {"query_plan": queries[:5], "retry_count": 0}

3.4 检索节点:混合检索的落地

检索节点遍历查询列表,对每个查询执行混合检索。Elasticsearch 的混合检索可以用knn加query组合实现:

def retrieve_node(state: RAGState): docs = [] for query in state["query_plan"]: query_vector = embed_model.encode(query).tolist() resp = es.search( index="knowledge_base", body={ "knn": { "field": "embedding", "query_vector": query_vector, "k": 5, "num_candidates": 50 }, "query": { "match": {"content": query} }, "size": 5 } ) for hit in resp["hits"]["hits"]: docs.append({ "content": hit["_source"]["content"], "score": hit["_score"], "source": hit["_source"].get("source", "unknown") }) return {"retrieved_docs": docs}

这里k和num_candidates的取值有讲究。k是最终返回数量,num_candidates是每个分片预选的候选数量,一般设为k的 5 到 10 倍。设太小会漏掉好结果,设太大影响性能。我一般从k=5, num_candidates=50起步,根据召回评估再调。

3.5 评估节点:决定“够不够”的关键判断

评估节点是 Agentic RAG 的灵魂。它的输出决定了流程是继续检索还是进入生成。我用的是 LLM 打分加规则判断的组合:

EVAL_PROMPT = """判断以下检索材料是否足以回答用户问题。 用户问题:{question} 检索材料: {docs} 请输出: - 如果材料充足,输出 "SUFFICIENT" - 如果材料不足,输出 "INSUFFICIENT: <缺少什么信息>" """

评估节点还要处理一个边界情况:如果重试次数超过阈值,强制进入生成。否则智能体可能陷入“检索-评估-再检索”的死循环。我一般设max_retry=2,也就是最多额外检索两轮。

def evaluate_node(state: RAGState): docs_text = "\n---\n".join([d["content"] for d in state["retrieved_docs"][-10:]]) response = llm.invoke(EVAL_PROMPT.format( question=state["question"], docs=docs_text )) result = response.content.strip() if result.startswith("SUFFICIENT") or state["retry_count"] >= 2: return {"evaluation": "SUFFICIENT"} return {"evaluation": result, "retry_count": state["retry_count"] + 1}

3.6 生成节点与图的组装

生成节点基于累积的检索材料组织答案,Prompt 里要强调“只基于给定材料回答,材料中没有的信息不要编造”:

def generate_node(state: RAGState): docs_text = "\n---\n".join([d["content"] for d in state["retrieved_docs"]]) response = llm.invoke(GEN_PROMPT.format( question=state["question"], docs=docs_text )) return {"final_answer": response.content}

最后用 LangGraph 把节点组装成图:

from langgraph.graph import StateGraph, END graph = StateGraph(RAGState) graph.add_node("plan", plan_node) graph.add_node("retrieve", retrieve_node) graph.add_node("evaluate", evaluate_node) graph.add_node("generate", generate_node) graph.set_entry_point("plan") graph.add_edge("plan", "retrieve") graph.add_edge("retrieve", "evaluate") graph.add_conditional_edges( "evaluate", lambda s: "generate" if s["evaluation"] == "SUFFICIENT" else "retrieve", {"generate": "generate", "retrieve": "retrieve"} ) graph.add_edge("generate", END) app = graph.compile()

这段代码里,add_conditional_edges就是 Agentic RAG 的“思考”所在:评估通过就去生成,不通过就回到检索节点。整个循环由 LLM 的判断驱动,而不是写死的流程。

4. 常见问题与排查技巧实录

4.1 检索命中率低:先查 embedding 再查查询词

RAG 效果差,八成问题出在检索层。我排查的顺序是:先看 embedding 模型是否适合你的领域。通用 embedding 模型在专业领域(医疗、法律、工业)上表现会明显下降,这时候要么换领域微调的模型,要么在检索前加一层查询改写。

查询改写是个低成本高收益的手段。用户问“这个报错怎么解决”,直接检索效果很差,因为“这个”指代不明。让 LLM 先结合对话历史把问题改写成“Elasticsearch 启动时报 max virtual memory areas vm.max_map_count 错误怎么解决”,检索命中率会大幅提升。

4.2 评估节点误判:阈值和样本都要调

评估节点用 LLM 判断“够不够”,会出现两种误判:一是材料其实够了但判为不足,导致无谓的重试;二是材料不足但判为充足,导致生成阶段编造。前者浪费成本,后者影响准确性。

我的做法是先用一批标注样本校准评估 Prompt。准备 50 条左右的问题-材料对,人工标注“充足/不足”,然后跑评估节点看准确率。如果误判率高,调整 Prompt 里的判断标准,比如明确“材料中必须包含问题涉及的所有关键实体才算充足”。另外,评估时只传最近 N 条材料(我一般传 10 条),传太多会稀释 LLM 的注意力。

4.3 循环失控:重试次数和查询去重双保险

前面提到了max_retry,这是第一道保险。第二道保险是查询去重:如果评估失败后重新规划,LLM 可能输出和上一轮相同的查询,导致检索到相同结果,白白循环。可以在规划节点里把历史查询传进去,要求 LLM 输出不同的查询。

还有一个隐蔽的坑:如果检索层本身有问题(比如索引为空、embedding 维度不匹配),评估会一直失败,重试到上限才停。所以监控重试率很重要,如果重试率异常高,先检查检索层是否正常工作,而不是调 Prompt。

4.4 Elasticsearch 写入慢的排查思路

热词里有人问“ES 怎么判断写入慢是磁盘问题还是其他问题”,这个在 RAG 场景里也会遇到,因为知识库更新时会有批量写入。排查顺序:

指标正常范围异常含义
indexing.index_total增速稳定骤降可能是磁盘瓶颈
indexing.index_time_in_millis与文档大小成正比偏高说明 CPU 或 IO 压力大
os.disk.io.util低于 80%持续高位说明磁盘饱和
thread_pool.write.queue接近 0持续排队说明写入线程不足
jvm.memory.pressure低于 75%偏高会触发频繁 GC

如果disk.io.util高但index_time正常,多半是磁盘 IO 瓶颈;如果index_time高但磁盘 util 不高,可能是 mapping 设计问题(比如字段过多、dynamic mapping 失控)或者分词器太重。我遇到过一次写入慢,最后发现是某个字段用了nested类型且嵌套层级很深,改成flattened后写入速度翻了五倍。

4.5 多轮检索结果冲突:让生成节点做仲裁

Agentic RAG 多轮检索后,可能捞到互相矛盾的材料(比如两个文档对同一个参数的描述不一致)。这时候不要指望评估节点解决,它的职责是判断“够不够”,不是“对不对”。我的做法是在生成节点的 Prompt 里加一条指令:“如果材料之间存在矛盾,请指出矛盾并说明各自来源,不要强行选择一个。”这样至少用户能看到冲突,而不是被一个看似确定的错误答案误导。

5. 从能跑到好用:几个提升效果的经验

5.1 给检索结果加元数据过滤

纯语义检索有个问题:它不理解“最新”“去年”“华东区”这类限定条件。解决办法是在检索时加元数据过滤。ES 的查询里可以加filter子句,比如时间范围、部门、文档类型。这些过滤条件可以由规划节点在生成查询时一并输出,作为结构化字段传给检索节点。

我做过一个对比:同一个知识库,不加过滤的 RAG 在“2024 年政策”这类问题上的准确率是 61%,加了时间过滤后提升到 89%。元数据过滤是性价比最高的优化手段之一,前提是你的文档在入库时打好了标签。

5.2 用 rerank 模型做二次精排

混合检索召回的结果,相关性排序未必最优。加一个 rerank 模型(比如 BGE-reranker 系列)对召回结果重新打分,能明显提升 top-k 的精度。代价是增加一次模型推理,延迟会上升。我的经验是:如果召回数量在 20 条以内,rerank 的收益很明显;如果召回上百条,先做粗筛再 rerank,否则延迟吃不消。

5.3 把“不知道”作为合法输出

很多 RAG 项目效果差的根源,是模型被训练成“必须回答”。材料里没有的信息,它也要编一个。在生成 Prompt 里明确写“如果材料中没有相关信息,直接回答‘根据现有资料无法回答该问题’”,并且在实际评估时把这类回答也纳入正确率统计。允许说不知道,反而提升了用户对系统的信任,因为用户能区分“系统知道”和“系统不知道”,而不是被错误答案误导。

5.4 监控指标:别只看最终答案

Agentic RAG 的链路比传统 RAG 长,出问题的环节更多。我建议至少监控这几个指标:检索召回率(人工抽检 top-5 里有多少相关)、评估节点通过率、平均重试次数、端到端延迟、生成答案的引用准确率。其中评估节点通过率最能反映系统健康度,如果通过率突然下降,多半是检索层或知识库出了问题。

我在实际项目里踩过最深的坑,是早期只盯着最终答案的准确率,忽略了中间环节。结果有一次知识库索引重建后字段名变了,检索节点静默返回空结果,评估节点因为“没有材料”判为不足,重试到上限后生成节点基于空材料编了一个答案,最终准确率只掉了几个百分点,但实际上是整个检索链路已经断了。后来加了检索结果数量的监控,这类问题就能第一时间发现。

5.5 关于 Agentic RAG 的边界

最后说点实在的。Agentic RAG 不是银弹,它解决的是“复杂问题需要多步检索和判断”的场景。如果你的场景是简单的 FAQ 问答,用户问题短、答案在单篇文档里,传统 RAG 加个好点的 rerank 就够了,上 Agentic 架构反而增加延迟和成本。判断标准很简单:如果你的 RAG 错误里,有相当一部分是“检索到了相关内容但模型没用对”或者“需要组合多个文档才能回答”,那 Agentic RAG 值得投入;如果错误主要是“知识库里根本没有这个信息”,那先补知识库,架构再花哨也没用。

这个方向后续可以扩展的点还很多,比如把工具调用加进来(让智能体在检索之外还能查数据库、调 API)、多智能体分工(规划、检索、验证各用一个专职 agent)、以及和 GraphRAG 结合做实体关系推理。但这些都是在你把基础的单智能体循环跑稳、指标可观测之后才值得做的事。先把检索质量、评估准确率、循环控制这三件事做扎实,比堆架构重要得多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询