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 RAG3.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设为 30sbulk 每批 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_k5, source_filterNone): 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(indexknowledge_base, knnknn, sizetop_k) return [hit[_source] for hit in response[hits][hits]]num_candidates设成top_k的 10 倍是个经验值太小会影响召回率太大影响性能。如果发现检索命中率低先检查 embedding 模型是否适合你的领域数据。通用 embedding 模型在专业领域比如医疗、法律、工业表现可能不好考虑用领域数据微调或换用领域专用模型。关键词检索用 multi_match 查询def keyword_search(query, top_k5): body { query: { multi_match: { query: query, fields: [content^2, title^3], type: best_fields } }, size: top_k } response es.search(indexknowledge_base, bodybody) return [hit[_source] for hit in response[hits][hits]]title字段加权更高因为标题通常概括了文档核心内容。如果业务里有明确的字段优先级都可以通过 boost 调整。3.4 评估 Prompt 的设计与迭代评估 Prompt 我改过十几版最终稳定下来的结构是这样的你是一个检索质量评估助手。给定一个问题和一组检索结果请判断每条结果与问题的相关性0-1 分当前结果是否足以回答问题是/否如果不足还缺什么信息输出 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 适合处理模糊的、需要语义理解的决策不适合处理确定性的逻辑判断。这个经验让我在后来的项目中少走了很多弯路。 
长河君 · 2026/9/29 8:06:00 · 阅读 1.2k 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 RAG3.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设为 30sbulk 每批 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_k5, source_filterNone): 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(indexknowledge_base, knnknn, sizetop_k) return [hit[_source] for hit in response[hits][hits]]num_candidates设成top_k的 10 倍是个经验值太小会影响召回率太大影响性能。如果发现检索命中率低先检查 embedding 模型是否适合你的领域数据。通用 embedding 模型在专业领域比如医疗、法律、工业表现可能不好考虑用领域数据微调或换用领域专用模型。关键词检索用 multi_match 查询def keyword_search(query, top_k5): body { query: { multi_match: { query: query, fields: [content^2, title^3], type: best_fields } }, size: top_k } response es.search(indexknowledge_base, bodybody) return [hit[_source] for hit in response[hits][hits]]title字段加权更高因为标题通常概括了文档核心内容。如果业务里有明确的字段优先级都可以通过 boost 调整。3.4 评估 Prompt 的设计与迭代评估 Prompt 我改过十几版最终稳定下来的结构是这样的你是一个检索质量评估助手。给定一个问题和一组检索结果请判断每条结果与问题的相关性0-1 分当前结果是否足以回答问题是/否如果不足还缺什么信息输出 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 适合处理模糊的、需要语义理解的决策不适合处理确定性的逻辑判断。这个经验让我在后来的项目中少走了很多弯路。