news 2026/9/29 8:05:48

Agentic RAG实战:基于LangGraph与Elasticsearch构建智能检索增强生成系统

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
Agentic RAG实战:基于LangGraph与Elasticsearch构建智能检索增强生成系统

1. 从“检索增强”到“智能体驱动”:Agentic RAG 到底在解决什么问题

1.1 传统 RAG 的天花板在哪里

做过 RAG 项目的人大概都有过这种体验:用户问一个问题,系统先去向量库捞 Top-K 文档,拼进 Prompt,然后让 LLM 生成答案。流程跑通了,Demo 看着也不错,但一上真实业务就开始露馅。

最典型的问题有三个。第一,检索质量不可控。用户问“上季度华东区退货率异常的原因”,向量检索可能召回一堆讲退货政策的文档,因为“退货”这个词的语义相似度最高,但真正有用的可能是某张数据表里的区域指标,或者某次会议纪要里提到的物流延误。第二,单轮检索不够用。复杂问题往往需要多跳推理,比如先查“哪些 SKU 退货率超过阈值”,再查“这些 SKU 的供应商是谁”,最后查“该供应商近期有没有交付延迟记录”。传统 RAG 一次性检索,根本覆盖不了这种链路。第三,没有自我纠错能力。检索回来的内容不相关,LLM 要么硬编,要么说“根据已知信息无法回答”,但它不会主动说“我换个关键词再搜一次”或者“这个问题的答案可能需要查另一个数据源”。

这三个问题的本质是一样的:传统 RAG 把“检索”和“生成”当成了两个独立的、一次性的步骤,中间没有决策、没有反馈、没有迭代。它像一个只会按固定流程办事的流水线工人,而不是一个会思考、会调整策略的研究助理。

1.2 Agentic RAG 的核心思路:让检索变成“智能体行为”

Agentic RAG 的破局点,就是把 LLM 从“被动生成器”升级为“主动决策者”。整个系统不再是一条直线,而是一个由智能体驱动的循环:理解问题 → 规划策略 → 选择工具 → 执行检索 → 评估结果 → 决定下一步。

这里的关键变化在于,检索不再是唯一动作,而是智能体可以调用的众多工具之一。它可以查向量库,也可以查 Elasticsearch 做关键词匹配,还可以调用 API 查实时数据,甚至可以直接问用户澄清需求。更重要的是,它能在每一轮检索后判断“够不够”“对不对”“要不要换方向”,然后决定是继续深挖、换工具重试,还是直接生成答案。

LangGraph 在这套架构里扮演的角色,就是把这个循环“画”出来并跑起来。它用图结构定义节点和边,节点可以是 LLM 调用、工具执行、条件判断,边则控制流转逻辑。相比 LangChain 的链式结构,LangGraph 支持循环、分支和状态持久化,天然适合实现 Agentic RAG 这种需要多轮决策的场景。

1.3 谁适合参考这套方案

如果你正在做企业知识库、智能客服、数据分析助手、合规审查这类需要“查资料再回答”的项目,并且已经感受到传统 RAG 的局限,那 Agentic RAG 就是下一步该走的路。它不适合那种问题极其简单、答案固定、检索一次就够的场景,因为引入智能体会增加延迟和成本。但只要你的业务里存在“复杂查询”“多源数据”“需要推理”的成分,这套架构带来的准确率提升是值得的。

下面我会从架构设计、核心组件、实操落地、问题排查几个层面,把我在实际项目中踩过的坑和总结的经验完整拆开讲。

2. 架构拆解:Agentic RAG 的四个核心模块与选型逻辑

2.1 规划模块:让 LLM 先想清楚再动手

传统 RAG 拿到问题直接检索,Agentic RAG 则要求先做规划。规划模块的任务是:把用户输入的自然语言问题,拆解成可执行的检索步骤或工具调用序列。

我常用的做法是让 LLM 输出一个结构化的计划,比如 JSON 格式的步骤列表,每一步包含“目标”“工具”“参数”。举个例子,用户问“对比一下我们和竞品在售后响应时间上的差距”,规划模块可能输出:

{ "steps": [ {"goal": "获取本公司售后响应时间数据", "tool": "sql_query", "params": {"table": "service_metrics", "metric": "response_time"}}, {"goal": "获取竞品售后响应时间公开数据", "tool": "web_search", "params": {"query": "竞品名称 售后响应时间 报告"}}, {"goal": "对比分析", "tool": "llm_reason", "params": {}} ] }

这里有个经验:规划粒度不要太细。我试过让 LLM 把每一步的检索关键词都定死,结果它经常生成过于具体的查询,反而漏掉关键信息。更好的做法是让规划模块只定“目标和工具”,具体查询语句交给执行模块根据上下文动态生成。

另一个坑是规划模块容易过度设计。有些问题其实一步检索就能解决,但 LLM 会习惯性地拆成三四步,导致延迟翻倍。我的解决办法是在 Prompt 里加一个约束:“如果问题可以直接通过一次检索回答,不要拆解。”同时在评估环节加一个“计划复杂度”指标,监控平均步骤数是否合理。

2.2 工具层:向量检索、关键词检索与 API 调用的分工

Agentic RAG 的工具层通常包含三类检索工具,它们各有适用场景,不能互相替代。

向量检索适合语义匹配,比如“退货率异常的原因”这种表述,关键词可能是“退货”“异常”“原因”,但文档里可能写的是“逆向物流指标波动分析”。向量检索能跨越字面差异找到语义相关的内容。我一般用 Elasticsearch 的 dense_vector 字段配合 kNN 查询,或者用专门的向量库。选 Elasticsearch 的好处是它同时支持 BM25 关键词检索和向量检索,一套系统搞定两种模式,运维成本低。

关键词检索适合精确匹配,比如查订单号、产品型号、人名、特定术语。向量检索在这些场景下反而容易召回不相关的内容,因为语义相似度会把“ABC-123”和“ABC-124”当成相近的。Elasticsearch 的 BM25 在这方面表现稳定,尤其是配合 analyzer 做分词和归一化之后。

API 调用适合实时数据或结构化数据。比如查库存、查汇率、查工单状态,这些数据不在文档库里,但智能体可以通过 HTTP 请求获取。这里要注意的是,API 工具需要定义清晰的输入输出 schema,否则 LLM 容易传错参数。我通常用 Pydantic 模型定义参数结构,然后在工具描述里写清楚每个字段的含义和格式要求。

三类工具的调度逻辑是:规划模块根据问题类型选择工具,执行模块调用工具并返回结果,评估模块判断结果是否满足当前步骤的目标。如果向量检索召回的内容相关性低,评估模块可以触发“换关键词重试”或“改用关键词检索”。

2.3 评估与反思模块:判断“够不够”和“对不对”

这是 Agentic RAG 区别于传统 RAG 最关键的一环。评估模块要做两件事:相关性判断和充分性判断。

相关性判断是看检索回来的内容是否和当前子目标相关。我常用的方法是让 LLM 对每个检索结果打分(0-1),然后过滤掉低分内容。这里有个细节:不要只让 LLM 看单条结果,而是把问题和结果一起给它,让它判断“这条内容能否帮助回答这个问题”。单独看结果容易误判,因为有些内容脱离问题语境看起来相关,实际上没用。

充分性判断是看当前收集到的信息是否足以回答原始问题。如果不够,评估模块要输出“还缺什么”,然后触发下一轮检索或工具调用。这个判断的 Prompt 设计很关键,我一般会要求 LLM 输出一个结构化结果:

{ "sufficient": false, "missing": "竞品的售后响应时间数据尚未获取", "next_action": "调用 web_search 工具查询竞品公开报告" }

反思模块则是在多轮检索后,如果发现方向不对,主动调整策略。比如连续两轮向量检索都召回低分内容,反思模块可以决定“换用关键词检索”或“扩大检索范围”。这个机制能有效避免智能体在错误方向上死磕。

2.4 状态管理与循环控制:LangGraph 的图结构怎么设计

LangGraph 的核心概念是 StateGraph,它维护一个全局状态对象,所有节点读写这个状态。在 Agentic RAG 里,状态通常包含:原始问题、当前计划、已执行步骤、检索结果列表、评估结论、循环次数。

图的结构大致是:入口节点接收问题 → 规划节点生成计划 → 执行节点调用工具 → 评估节点判断结果 → 条件边决定下一步。条件边根据评估结果选择:如果充分,走生成节点输出答案;如果不充分且未超最大轮次,回到执行节点继续;如果超过最大轮次,走兜底节点返回已有信息并说明局限。

这里有个实操要点:一定要设最大循环次数。我见过没有限制的智能体陷入死循环,反复检索同一个方向,烧掉大量 token。一般设 3-5 轮比较合理,具体看业务复杂度。另外,状态里要记录每一轮的检索结果,避免重复检索相同内容。可以在执行节点加一个去重逻辑,如果新检索结果和已有结果高度重叠,直接跳过。

3. 实操落地:从零搭一套可运行的 Agentic RAG

3.1 环境准备与 Elasticsearch 关键配置

先说一下基础环境。我用的技术栈是 Python 3.10+、LangGraph、Elasticsearch 8.x、以及一个本地或 API 方式的 LLM。Elasticsearch 的安装方式看你的部署环境,Windows 下直接下载压缩包解压运行bin/elasticsearch.bat即可,Linux 下建议用 Docker 或 K8s 部署。KubeSphere 部署 Elasticsearch 的话,注意给足内存,默认的 JVM heap 可能不够用,建议至少 4GB。

索引 mapping 的设计直接影响检索效果。我一般会建一个包含以下字段的索引:

{ "mappings": { "properties": { "content": {"type": "text", "analyzer": "ik_max_word"}, "content_vector": {"type": "dense_vector", "dims": 1024, "index": true, "similarity": "cosine"}, "source": {"type": "keyword"}, "doc_type": {"type": "keyword"}, "timestamp": {"type": "date"} } } }

content字段用 IK 分词器做中文关键词检索,content_vector存 embedding 做向量检索。source和doc_type用于过滤,比如只检索某个数据源或某类文档。这里注意,dense_vector 的 dims 要和你的 embedding 模型输出维度一致,用之前确认清楚。

写入性能方面,如果发现写入慢,先看几个指标:indexing_pressure是否过高、refresh_interval是否设得太短、bulk 请求的 batch size 是否合理。我一般把refresh_interval设为 30s,bulk 每批 500-1000 条,配合wait_for_active_shards: 1。如果磁盘 IO 是瓶颈,考虑加 SSD 或调整translog的 durability 设置。

3.2 用 LangGraph 定义智能体工作流

下面是一个简化的 LangGraph 工作流定义,展示核心节点和条件边:

from langgraph.graph import StateGraph, END from typing import TypedDict, List class RAGState(TypedDict): question: str plan: List[dict] current_step: int retrieved_docs: List[dict] evaluation: dict loop_count: int def plan_node(state: RAGState): # 调用 LLM 生成检索计划 plan = llm_plan(state["question"]) return {"plan": plan, "current_step": 0, "loop_count": 0} def execute_node(state: RAGState): # 执行当前步骤的工具调用 step = state["plan"][state["current_step"]] results = call_tool(step["tool"], step["params"]) return {"retrieved_docs": state["retrieved_docs"] + results} def evaluate_node(state: RAGState): # 评估检索结果是否充分 evaluation = llm_evaluate(state["question"], state["retrieved_docs"]) return {"evaluation": evaluation, "loop_count": state["loop_count"] + 1} def should_continue(state: RAGState): if state["evaluation"]["sufficient"]: return "generate" if state["loop_count"] >= 5: return "fallback" return "execute" graph = StateGraph(RAGState) graph.add_node("plan", plan_node) graph.add_node("execute", execute_node) graph.add_node("evaluate", evaluate_node) graph.add_node("generate", generate_node) graph.add_node("fallback", fallback_node) graph.set_entry_point("plan") graph.add_edge("plan", "execute") graph.add_edge("execute", "evaluate") graph.add_conditional_edges("evaluate", should_continue, { "generate": "generate", "execute": "execute", "fallback": "fallback" }) graph.add_edge("generate", END) graph.add_edge("fallback", END) app = graph.compile()

这段代码的关键在于should_continue这个条件函数,它根据评估结果和循环次数决定下一步走向。实际项目中,评估节点的 Prompt 需要反复调优,我一般会加入 few-shot 示例,让 LLM 更准确地判断“充分性”。

3.3 检索工具的具体实现与参数调优

向量检索的实现,我通常用 Elasticsearch 的 kNN 查询:

def vector_search(query, top_k=5, source_filter=None): query_vector = embed(query) knn = { "field": "content_vector", "query_vector": query_vector, "k": top_k, "num_candidates": top_k * 10 } if source_filter: knn["filter"] = {"term": {"source": source_filter}} response = es.search(index="knowledge_base", knn=knn, size=top_k) return [hit["_source"] for hit in response["hits"]["hits"]]

num_candidates设成top_k的 10 倍是个经验值,太小会影响召回率,太大影响性能。如果发现检索命中率低,先检查 embedding 模型是否适合你的领域数据。通用 embedding 模型在专业领域(比如医疗、法律、工业)表现可能不好,考虑用领域数据微调或换用领域专用模型。

关键词检索用 multi_match 查询:

def keyword_search(query, top_k=5): body = { "query": { "multi_match": { "query": query, "fields": ["content^2", "title^3"], "type": "best_fields" } }, "size": top_k } response = es.search(index="knowledge_base", body=body) return [hit["_source"] for hit in response["hits"]["hits"]]

title字段加权更高,因为标题通常概括了文档核心内容。如果业务里有明确的字段优先级,都可以通过 boost 调整。

3.4 评估 Prompt 的设计与迭代

评估 Prompt 我改过十几版,最终稳定下来的结构是这样的:

你是一个检索质量评估助手。给定一个问题和一组检索结果,请判断:

  1. 每条结果与问题的相关性(0-1 分)
  2. 当前结果是否足以回答问题(是/否)
  3. 如果不足,还缺什么信息

输出 JSON 格式:{"relevance_scores": [...], "sufficient": bool, "missing": "..."}

这里有个技巧:让 LLM 先逐条打分,再综合判断充分性。如果直接问“够不够”,LLM 容易过度乐观。先打分再判断,准确率明显更高。另外,missing字段的描述要具体,比如“缺少竞品 B 在 2024 年 Q1 的响应时间数据”,而不是“信息不足”。具体的缺失描述能指导下一次检索。

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

4.1 检索命中率低:从 embedding 到分词的逐层排查

检索命中率低是最常见的问题。我的排查顺序是:先看 embedding 质量,再看分词效果,最后看查询构造。

Embedding 质量怎么判断?拿几个典型问题,人工标注哪些文档应该被召回,然后看向量检索的实际召回结果。如果召回率低于 60%,大概率是 embedding 模型不适合。我试过用通用模型处理工业设备故障描述,效果很差,换成领域微调模型后召回率提升了 30% 以上。

分词问题在中文场景特别常见。Elasticsearch 默认的 standard 分词器对中文是按字切分,效果很差。一定要装 IK 分词器,并且根据业务词典补充自定义词。比如“退货率”如果被切成“退货”和“率”,检索“退货率异常”时可能匹配不到。在 IK 的 custom 词典里加上“退货率”“响应时间”这类业务术语,效果立竿见影。

查询构造方面,如果用户问题很长,直接拿整个问题去检索效果往往不好。我一般会先用 LLM 把问题压缩成几个关键词,再用这些关键词去检索。比如“上季度华东区退货率异常的原因是什么”压缩成“华东区 退货率 异常 原因”,检索精度会高很多。

4.2 智能体循环不收敛:原因分析与强制终止策略

循环不收敛的表现是:评估模块一直说“不充分”,执行模块反复检索但结果没有实质增加。原因通常有三个:一是评估 Prompt 过于严格,二是检索工具确实找不到相关信息,三是状态管理有问题导致重复检索。

评估 Prompt 过于严格的话,调整阈值或增加“如果连续两轮检索结果相似,判定为充分”的规则。检索工具找不到信息的情况,需要在评估模块加一个判断:“如果已尝试所有可用工具且仍无结果,输出‘无法回答’并终止”。状态管理问题最常见的是没有去重,每次检索都返回相同内容。在执行节点加一个去重逻辑,对比新结果和已有结果的 content 字段,如果重复率超过 80%,直接跳过。

强制终止策略除了最大循环次数,还可以加一个“无进展检测”:如果连续两轮的检索结果数量和质量没有提升,强制终止。这个逻辑可以在条件边里实现。

4.3 生成答案与检索内容不一致:归因与修正

有时候检索内容明明相关,但 LLM 生成的答案却偏离了。这通常是 Prompt 的问题。我见过两种典型情况:一是 LLM 过度依赖自己的先验知识,忽略了检索内容;二是检索内容太长,LLM 只看了开头就下结论。

第一种情况的解决办法是在 Prompt 里加约束:“只使用提供的检索内容回答问题,如果检索内容不足以回答,明确说明。”同时可以加一个“引用来源”的要求,让 LLM 在答案里标注每条信息的出处,这样能强制它关注检索内容。

第二种情况需要控制检索内容的长度。我一般会把每条检索结果截断到 500-800 字,只保留最相关的部分。如果一条文档很长,先用 LLM 做摘要,再把摘要给生成模块。这样既保留了关键信息,又不会超出上下文窗口。

4.4 性能与成本平衡:延迟优化的几个实用手段

Agentic RAG 的延迟通常比传统 RAG 高,因为多了规划、评估和多轮检索。优化手段有几个:一是并行化,如果规划模块生成了多个独立步骤,可以并行执行;二是缓存,对常见问题的检索结果做缓存,避免重复查询;三是模型分级,规划和评估用轻量模型,生成用大模型。

我实测下来,规划和评估用 7B 级别的模型就够用,生成用 70B 或 API 大模型。这样整体延迟能控制在可接受范围内。另外,Elasticsearch 的查询本身很快,瓶颈通常在 LLM 调用上。如果延迟敏感,可以考虑把评估逻辑简化成规则判断,比如“检索结果数量超过 N 条且平均相关性分数超过阈值”就判定为充分,不走 LLM。

4.5 常见问题速查表

问题现象可能原因排查方法解决措施
检索命中率低embedding 模型不匹配人工标注测试集,计算召回率换领域模型或微调
中文检索效果差分词器未配置检查 mapping 的 analyzer安装 IK 分词器,补充自定义词典
智能体循环不收敛评估 Prompt 过严查看每轮评估输出调整阈值,加无进展检测
答案偏离检索内容Prompt 约束不足检查生成 Prompt加“仅使用检索内容”约束,要求引用来源
延迟过高LLM 调用过多统计各节点耗时并行化、缓存、模型分级
写入慢refresh_interval 过短查看 indexing_pressure调大 refresh_interval,优化 bulk
重复检索状态未去重检查 retrieved_docs加去重逻辑,对比 content 字段

5. 进阶方向:Agentic RAG 还能怎么玩

5.1 多智能体协作:让不同角色各司其职

单智能体已经能解决大部分问题,但如果业务复杂度再上一个台阶,可以考虑多智能体架构。比如一个“检索智能体”专门负责找资料,一个“分析智能体”专门负责推理和计算,一个“审核智能体”专门检查答案的合规性和准确性。它们通过共享状态或消息传递协作。

LangGraph 支持这种模式,你可以定义多个智能体节点,用条件边控制它们之间的流转。我做过一个合规审查的项目,检索智能体负责找法规条文,分析智能体负责判断业务行为是否违规,审核智能体负责复核。三个角色分工明确,准确率比单智能体高了 15% 左右。代价是延迟和成本增加,适合对准确性要求极高的场景。

5.2 与知识图谱结合:GraphRAG 的思路

向量检索擅长语义匹配,但对结构化关系无能为力。比如“A 公司的母公司旗下的子公司有哪些”,这种问题需要沿着实体关系遍历。GraphRAG 的思路是把知识图谱和向量检索结合,智能体可以先在图谱里定位实体,再沿着关系边扩展,最后用向量检索补充非结构化信息。

实现上,可以用 Neo4j 或 NebulaGraph 存图谱,智能体通过 Cypher 查询遍历关系。LangGraph 里加一个“图谱查询”工具,规划模块根据问题类型决定是否调用。这个方向适合实体关系密集的领域,比如金融、供应链、医疗。

5.3 从 RAG 到 Agentic RAG 的迁移路径

如果你已经有一套传统 RAG 系统,想迁移到 Agentic RAG,我的建议是分步走。第一步,先加一个评估模块,对现有检索结果做相关性打分,过滤低质量内容。这一步改动小,效果立竿见影。第二步,加规划模块,把单轮检索变成多轮。第三步,引入 LangGraph 管理循环和状态。第四步,逐步增加工具类型,从纯向量检索扩展到关键词检索和 API 调用。

每一步都要有评估指标,比如检索命中率、答案准确率、平均延迟。不要一次性全改,否则出了问题很难定位。我在实际迁移中,第一步就把答案准确率从 65% 提到了 78%,因为过滤掉了大量不相关的内容。后面每一步的提升幅度会递减,但累积效果很可观。

5.4 我踩过的一个典型坑:过度依赖 LLM 做决策

最后分享一个我踩过的坑。刚开始做 Agentic RAG 的时候,我把所有决策都交给 LLM,包括工具选择、参数生成、结果评估。结果发现 LLM 经常做出莫名其妙的决定,比如用向量检索去查订单号,或者评估时把不相关的内容判为相关。

后来我调整了策略:能用规则的地方用规则,不能用规则的地方才用 LLM。比如工具选择,如果问题里包含明确的订单号格式,直接走关键词检索,不问 LLM。评估环节,如果检索结果数量为 0,直接判定不充分,不问 LLM。这样既降低了延迟,又提高了稳定性。LLM 适合处理模糊的、需要语义理解的决策,不适合处理确定性的逻辑判断。这个经验让我在后来的项目中少走了很多弯路。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/9/29 8:01:51

Linux进程间通信(IPC)详解:管道、共享内存、消息队列与信号

Linux应用开发里,进程间通信(IPC)是绕不开的一块硬骨头。不管是写嵌入式后台服务、搞中间件,还是单纯为了面试和笔试,你都得把这几种通信方式吃透。这篇笔记我把管道、消息队列、共享内存、信号量、信号这些常见手段从…

作者头像 李华
网站建设 2026/9/29 8:00:30

高校宿舍楼综合布线实战设计:千兆到床头、十年不返工

简介:本资源是一份面向高校网络工程、通信技术及相关专业学生的《综合布线实训教程》课程设计成果,聚焦学生宿舍楼这一典型场景,系统解决多终端接入、多业务融合(数据/语音/视频/安防)下的结构化布线规划与实施问题。内…

作者头像 李华
网站建设 2026/9/29 8:00:12

乌克兰BlackEnergy病毒攻击SCADA实战复盘与防御

简介:这份PDF文档聚焦2015年乌克兰电力系统遭遇BlackEnergy病毒攻击并导致伊万诺-弗兰科夫斯克州大规模停电的真实事件,面向电力系统安全、工控信息安全方向的研究人员与运维人员,提供病毒机理分析与防御思路的参考。文档通过获取不同版本病毒…

作者头像 李华
网站建设 2026/9/29 8:00:08

Windows提权实用指南:敏感信息搜索与凭据收集全流程

1. 为什么敏感信息搜索是Windows提权的第一站考OSCP的时候我有个特别深的体会:真正让你省力的不是某个0day漏洞,而是系统自己堆在角落里的密码。Windows权限提升这条路,很多人一上来就翻内核漏洞、找exp,结果折腾半天不如先静下心…

作者头像 李华