简介这是一份面向软件开发者与 AI 研究者的可运行源码包围绕代理式检索增强生成Agentic RAG展开解决传统 RAG 依赖单一知识源、一次性检索带来的信息覆盖不足与结果固化问题对构建智能检索系统具有直接参考价值。 压缩包共3个文件仅8KB含 html 说明页、inscode 运行配置和 gitignore 版本管理文件分别用于查看概念解析、准备运行环境与维护项目轻量易用。 目前已有147人学习适合中等水平开发者快速建立整体认识。 代码包结合说明文档系统梳理了 Agentic RAG 的工作原理、单/多代理架构差异及与传统 RAG 的区别并覆盖函数调用语言模型与 DSPy、LangChain 等框架落地路径读者可边运行边理解代理如何动态选择信息源、规划检索同时评估准确性与开发成本等局限为智能问答、知识管理等企业场景提供参考。1. 为什么说 Agentic RAG 是 RAG 从“一次问答”走向“多步推理”的分水岭场景用户问“Q2 的在线订单涨了 30%同时故障工单也涨了这俩是不是有关系”。传统 RAG 的做法是拿问题去做一次相似度检索把最像的两段文本拼给大模型生成。它能告诉你“订单涨了”“工单也涨了”但说不清两者之间是否有因果因为答案分散在多份数据源里一次检索取不齐。Agentic RAG 改变了这个流程它把“检索”本身交给大模型调度由模型决定先查订单数据、再查故障模块、必要时跑一次关联统计最后带着中间证据一起生成回答。检索不再是单点动作而是一个可观察、可干预、可追溯的推理循环。对这个方向感兴趣的通常是被复杂问题卡住的开发者和想把 RAG 从 Demo 推向生产的工程师。这套方案解决的核心问题很简单让回答不再依赖“第一次检索运气好不好”而是靠一个可控的决策闭环来逼近真实答案。2. 把 RAG 变成 Agentic核心循环、路由与反思机制2.1 核心循环观察、行动、反思三步缺一不可Agentic RAG 不是某个具体框架而是把 RAG 放进一个代理循环里执行。这个循环最常见的形态是 ReAct先“观察”当前问题与已有上下文再“行动”调用某一个工具拿到结果后“反思”是否足够不够就再行动够了就进入最终生成。工具可以是向量检索、数据库查询、代码搜索、计算器甚至外部 API。技术上的关键点在于每个动作都要把结果追加进上下文记忆下一次决策能看到之前所有尝试否则循环就退化成盲人摸象。我在实现最小版本时习惯把循环拆成三个函数一个负责生成“下一步做什么”的决策一个负责执行工具并返回结构化结果一个负责判断终止条件。这样三个函数互相独立调试某个环节时不会牵一发动全身。很多人在第一步就犯的错是让模型输出自然语言“下一步该查订单表了”而不是结构化的工具调用 JSON。自然语言不利于程序解析更容易在循环里累积歧义后面会让引用和日志都不好处理。再往下拆终止条件通常有三类模型自己认为证据已足够返回一个final动作循环达到预先设定的最大步数强制收尾反思模块判定当前已收集的上下文与问题无关需要重新路由。实际项目中前两类用得最多第三类适合在检索质量波动大的场景里做兜底。一个设计良好的循环会同时保留这三条退出路径避免代理在一个错误方向上无限空转。2.2 查询路由与查询重写在检索前把请求塑形路由解决的是“这个问题应该去问谁”。一条问题进入循环后模型先判断它是问文档、问代码、问数据还是问计算方法然后只会把请求发给对应工具。这个判断我建议做成结构化输出{tool: search_docs, query: ...}由代码执行而不是让模型直接调函数。把函数调用权留给框架层代理只发出意图能大幅降低模型输出格式不稳带来的偶发故障。路由粒度需要根据你的数据源来定。常见做法是给每个数据源配置一个独立工具文档库一个、代码库一个、业务数据库一个、计算器一个。工具之间不要做隐式转发例如“文档库内部再自动查数据库”这种设计会让排查链路变得非常绕。我见过一个翻车案例某开发者把数据库查询能力塞进了文档检索工具里结果代理以为每次查文档都能拿到实时数据生成的报表全是对不上号的这就是路由边界不清晰带来的黑匣子效应。重写解决的是“问题本身要不要拆”。复杂问题像“对比 Q2 与 Q3 的延迟并说明根因”模型就应该拆成“Q2 延迟数据”“Q3 延迟数据”“根因分析”三个子查询分别检索再合并。我看到不少人把重写这一步当作可选优化实际在 Agentic RAG 里它是质量的双保险不拆一次检索容易漏信息拆了生成的上下文更完整。但要注意重写不是改写同义词而是把问题拆成可检索的原子问题每个子问题都应能被某个工具单独回答。这步做对后面多步检索才不至于白跑。除了拆解重写还承担“补信息”的职责。当代理在反思阶段发现证据不足时可以把原问题改写得更具体比如加入时间范围、实体名称、指标口径。这个动作相当于给人一次重新提问的机会是循环里典型的后悔药。补信息补得对不对依赖模型对上下文的判断所以重写函数的输入不能只给原始问题还要给当前已经收集到的证据列表否则它不知道缺什么。2.3 混合检索与引用为忠实度设一道不撤的闸门代理循环鼓励模型多尝试但多尝试也意味着更可能引入无关信息。为了让最终回答能追溯到证据我在每个工具结果上都会保留来源标识并在生成时要求模型只引用收集到的 chunk编号形如[1]、[2]。引用不是事后格式化而是在生成提示词里就把上下文限制在“可见来源列表”并明确告诉模型未知信息不允许编造。需要留意的是代理循环里的检索结果不像传统 RAG 那样只来一批它可能分批到达。如果每批结果都直接拼进上下文模型生成时容易分不清哪条引用对应哪批来源。我的做法是给每个工具结果附带一个全局序号第一批检索到的 chunk 编为[1][2][3]第二批继续往后编[4][5]。这样最终引用天然有序也便于日志追踪。生成时只允许引用这些编号超出编号的引用要么是模型幻觉要么是提示词没约束住。混合检索在这里的意义是让召回更全。向量检索擅长语义相似关键词检索擅长精确命中。Agentic RAG 既可以把两者设计成不同工具让模型按需选择也可以让代理在一次行动里同时调用两种检索并合并去重。我的经验是先向量后关键词再按分数合并去重是稳定性和实现成本之间较好的折中。把两种召回直接拼进上下文会让 token 压力变大所以合并阶段最好做一次重排只保留分数最高的前若干个结果。3. 可运行源码一个最小 Agentic RAG 的完整实现3.1 依赖与取舍用什么和不用什么既然标题承诺可运行我就把“开箱即跑”作为第一优先级。完整实现分两层底层是模型接口上层是代理循环。为了让没有大模型 API Key 的读者也能立刻跑通我给模型接口加了一个回退实现检测到环境里有可用 Key 就调用真实模型没有的话就用规则逻辑模拟决策。向量检索也不引入重型框架用哈希词袋向量代替嵌入模型演示效果足够换真实嵌入时只替换一个函数。这种取舍的好处是零外部依赖坏处是模拟决策上限很低。它只能跑通流程、演示多步检索和引用格式不能像真实模型那样理解复杂语义。所以在实际项目中我不会把这里的模拟决策用到生产生产环境必须接真实模型。下面的代码我按“真实模型优先、模拟回退兜底”的结构给出你复制后不管有没有 Key 都能运行。依赖方面标准库加一个可选openai包即可。没有openai包时程序不会报错只是走到模拟分支。我一般会用pip install openai python-dotenv把依赖装齐然后在项目根目录放一个.env文件存模型密钥。.env里不要让密钥默认填进去留空也能跑。3.2 迷你向量存储与检索工具的实现先实现向量存储。代码里用的是词袋哈希嵌入每个词算一次 MD5取前 8 位十六进制映射到 512 维稀疏向量。这个设计纯粹为了可运行生产环境替换成sentence-transformers或它的在线接口即可search函数签名不用改。# agentic_rag_min.py 最小可运行 Agentic RAG标准库可跑有模型 Key 走真实模型没有走模拟。 import argparse import hashlib import json import math import re from typing import Dict, List, Tuple def _tokens(text: str) - List[str]: return re.findall(r[a-z0-9], text.lower()) def _embed(text: str) - Dict[int, int]: vec: Dict[int, int] {} for token in _tokens(text): h int(hashlib.md5(token.encode()).hexdigest()[:8], 16) % 512 vec[h] vec.get(h, 0) 1 return vec def _cosine(v1: Dict[int, int], v2: Dict[int, int]) - float: inter set(v1) set(v2) dot sum(v1[i] * v2.get(i, 0) for i in inter) n1 math.sqrt(sum(x * x for x in v1.values())) n2 math.sqrt(sum(x * x for x in v2.values())) return dot / (n1 * n2 1e-9) class MiniVectorStore: 极简向量库只保留 chunk 文本和对应的稀疏向量。 def __init__(self, chunks: List[str]): self.chunks chunks self.vectors [_embed(c) for c in chunks] def search(self, query: str, top_k: int 3) - List[Tuple[str, float]]: qv _embed(query) scored [(c, _cosine(qv, v)) for c, v in zip(self.chunks, self.vectors)] scored.sort(keylambda x: x[1], reverseTrue) return scored[:top_k]这个类有两个字段chunks保存原始文本vectors保存每个 chunk 对应的稀疏向量。search方法先计算查询向量再和每个 chunk 向量算余弦相似度最后按分数倒序取前top_k条。注意_cosine里分母加了1e-9避免空向量导致除零。这个细节在生产里同样重要因为真实嵌入模型对空文本可能输出全零向量。关键点在于search返回的不只是文本还带着分数。代理循环后续要用这个分数决定是否继续检索所以不要把分数丢掉。很多新手实现总喜欢[c for c, s in results]把分数剥掉结果反思阶段拿不到判断依据只能瞎猜证据够不够。3.3 代理决策循环与反思机制的实现决策循环是整个 Agentic RAG 的心脏。这里的_decide_next负责读入当前问题、已有历史、工具清单返回一个 JSON 字符串。真实模型模式下这个 JSON 由模型生成模拟模式下用规则判断。两条路径输出同样结构循环代码完全复用。def _call_llm(system_prompt: str, history: list) - str: 有 Key 时调用真实模型没 Key 时走规则模拟。 try: from openai import OpenAI client OpenAI() messages [{role: system, content: system_prompt}] messages.extend(history) resp client.chat.completions.create( modelgpt-4o-mini, messagesmessages, temperature0, ) return resp.choices[0].message.content or except Exception: return _mock_decide(history[-1][content]) def _mock_decide(user_text: str) - str: 模拟决策识别关键词制造一次多步检索效果。 low user_text.lower() if 检索 in low or 文档 in low or 原因 in low or 对比 in low: return json.dumps({thought: 需要先查文档, tool: search_docs, arg: user_text}) if 计算 in low or 统计 in low or 多少 in low: return json.dumps({thought: 需要跑计算, tool: run_calc, arg: user_text}) return json.dumps({thought: 证据足够, tool: final, arg: })_call_llm是适配层_mock_decide是回退层。模拟决策只看关键词所以它无法真正理解问题和证据是否充分。但它的价值在于你能在没有外部服务的情况下把循环、工具调用、引用编号这些框架逻辑完整跑通。生产环境接真实模型时except分支永远不会触发不用担心模拟逻辑污染结果。代理循环本体如下。每轮先取上一步的决策再执行对应工具结果追加进历史然后进入反思检查。反思检查在这里扮演终止条件如果tool是final或者步数超过max_steps就退出循环进入生成。AGENT_PROMPT 你是检索代理。根据问题与已有证据决定下一步动作。 工具列表 - search_docs: 检索文档 - run_calc: 做计算 - final: 结束检索准备生成 只输出 JSON格式: {thought: ..., tool: ..., arg: ...} def run_agent(query: str, store: MiniVectorStore, max_steps: int 6, top_k: int 3) - Dict: history [] evidence [] step 0 while step max_steps: if len(history) 0: decision _call_llm(AGENT_PROMPT, [{role: user, content: query}]) else: latest [{role: user, content: query}] for h in history[-4:]: latest.append({role: assistant, content: json.dumps(h, ensure_asciiFalse)}) latest.append({role: user, content: f执行结果: {json.dumps(h.get(result, ), ensure_asciiFalse)}}) decision _call_llm(AGENT_PROMPT, latest) try: parsed json.loads(decision) except Exception: parsed {thought: 解析失败直接终止, tool: final, arg: } if parsed.get(tool) final or step max_steps - 1: break if parsed.get(tool) search_docs: results store.search(parsed.get(arg, query), top_ktop_k) history.append({tool: search_docs, arg: parsed.get(arg), result: results}) evidence.extend(results) elif parsed.get(tool) run_calc: # 演示用只做文案统计 count len(_tokens(query)) history.append({tool: run_calc, arg: parsed.get(arg), result: {token_count: count}}) evidence.append((f查询文本共 {count} 个词, 1.0)) else: history.append({tool: search_docs, arg: query, result: []}) step 1 return {query: query, evidence: evidence, history: history}这里有几个值得说的参数。max_steps默认 6意味着代理最多尝试 6 次工具调用。history[-4:]只保留最近 4 条历史避免上下文无限膨胀代价是代理可能忘掉早期尝试实际操作里会配合摘要机制这里先不做。evidence是所有检索结果的统一池最终生成只从这个池子里取引用历史里的完整轨迹丢给日志。反思机制在这个最小实现里很朴素模型自己觉得够了就final或者循环走满主动退出。真实项目里我会再加一个显式检查比如最近两轮检索结果相似度持续低于 0.3就判定为“证据质量差”强制结束并让生成阶段给出不确定性声明。这个检查在模拟模式下看不出差别但接入真实模型后能挡住大量无意义的空转。3.4 带引用的最终生成与运行命令生成阶段把evidence里的每个 chunk 按顺序编成[1][2][3]只允许模型引用这些编号。没有真实模型时生成函数也走模拟直接拼接证据文本保留引用标号。这样读者一眼能看到“引用来自哪段文字”。def synthesize(query: str, evidence: List[Tuple[str, float]]) - str: if not evidence: return 未检索到可用的证据无法给出可靠回答。 parts [] for idx, (chunk, score) in enumerate(evidence): parts.append(f[{idx 1}] {chunk}) source_block \n.join(parts) try: from openai import OpenAI client OpenAI() resp client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 只依据引用回答禁止编造。引用用 [x] 标出。}, {role: user, content: f问题{query}\n引用资料\n{source_block}} ], temperature0.2, ) answer resp.choices[0].message.content or except Exception: best max(evidence, keylambda x: x[1]) answer f基于最相关的证据 {best[0][:80]}... 给出初步判断。 citation .join(f[{i 1}] for i in range(len(evidence))) return f{answer}\n\n参考来源{citation}\n\n{source_block}生成器的温度0.2是我习惯用的值。温度太高会让模型在引用编号上发挥创造力编出不存在的[5]温度太低又容易让回答机械复述证据。真实生成时还要在提示词里声明“引用只能来自上面列表不能伪造来源”这比事后做引用校验更省事。生成结束把source_block直接展示出来可以让用户人工核验引用是否真实。运行命令分为两种没有模型 Key 时用模拟跑通流程python agentic_rag_min.py 对比 Q2 与 Q3 延迟变化原因 --max-steps 6 --top-k 3有模型 Key 时程序自动走真实模型分支_call_llm和synthesize都会调用外部接口同一个命令不用改。.env里配置好密钥后整个脚本就是最小可运行的 Agentic RAG之后所有调优都是在这些参数上做文章。4. Agentic RAG 参数怎么调决定成败的关键数字4.1 max_steps 与反思阈值平衡迭代与退出max_steps决定了代理最多调多少次工具。设小了复杂问题还没收集够证据就被迫结束设大了代理容易陷入无意义的重复检索。我见过一个生产系统把max_steps设成 20结果有一条查询在文档库里反复检索了 9 次才放弃单次请求成本暴增。经验值是普通问答设 4 到 6附带代码检索或多数据源对比的场景设 8 到 10。反思阈值是终止条件的数值化。如果代理观察到最近一轮检索结果与之前结果重复度过高说明已经没有新信息可挖再检索也是浪费。这个阈值我用相似度去度量连续两轮命中同一批 chunk且文本重合度超过 0.7就强制进入生成。真实实现里还可以看检索分数的变化率但最简单可用的判断就是“有没有新 chunk 进来”。4.2 top_k 与相似度阈值决定召回质量的两个旋钮top_k控制每次向量检索返回多少条 chunk。传统 RAG 常用 3 到 5Agentic RAG 里我建议控制在 2 到 3。理由在于代理会多次检索每次带 3 条跑 4 轮就是 12 条证据生成阶段全塞进上下文token 压力很大。宁可每次少取、取准靠多轮弥补覆盖率也不能一次塞一堆无关文本污染生成。相似度阈值的作用是过滤低质量召回。向量检索在语义相似但主题脱靶的情况下会给一个不高不低的分数这类结果往往是幻觉的温床。我会在search返回后加一道过滤低于 0.3 的直接丢弃并在日志里标记为“低置信证据”。阈值定太高容易漏证据定太低则引入噪声0.25 到 0.35 是常见区间具体要看你的语料分布。参数默认值推荐区间调高时的副作用调低时的副作用max_steps64-10token 成本上升复杂问题回答不完整top_k32-5上下文膨胀召回覆盖不足相似度阈值0.30.25-0.35有效证据被过滤噪声引入幻觉温度0.20-0.3引用编造风险上升回答过于机械4.3 上下文预算与生成温度控制成本和风格上下文预算是最容易被忽略的一个参数。代理循环里历史轨迹和证据都会占用上下文窗口窗口不够时会触发截断或被迫压缩。常见做法是给证据块设一个最大字符数比如每块 800 字超过就按句截断。截断的位置要放在句子结束处而不是硬切字符否则下半截不完整会让模型理解出错。生成温度在 Agentic RAG 里比传统 RAG 更敏感因为任务已经不是单轮垫话术而是基于多来源证据做综合判断。温度过高模型可能为了流畅而忽略引用编号温度过低又可能复读证据而不是真正回答问题。另一个技巧是给生成阶段单独设一个比决策阶段更小的温度决策要探索所以要 0.3 左右生成要稳重所以压到 0.1 附近。不同阶段拆开配参是避免“决策灵活但生成飘”这组矛盾的有效手段。5. Agentic RAG 常见翻车现场现象、原因与排错路径5.1 代理无限循环反复检索同一批 chunk现象日志里看到代理对同一个查询连续调用search_docs四五次每次返回的 chunk 几乎一样然后才被迫超时退出。我把这个现象叫“原地打转”。原因反思机制没有判断“是否还有新信息”。模型觉得证据不够又没有能力意识到“再查也是同样的结果”于是反复尝试。另有一个常见原因是历史记录没传全代理忘了自己已经查过每次都当作第一次。解决在循环里加一条去重检查拿当前检索结果和已有证据对比文本重合度超过 0.7 就标记为“无新增量”直接把动作改为final。同时把最近几轮历史完整传给决策函数避免模型失忆。这个修复能挡住绝大多数原地打转。5.2 回答引用了与问题无关的检索结果现象问题问的是订单增长原因生成回答却引用了一段讲服务器延迟的 chunk而且引用编号还标得冠冕堂皇。原因检索阶段召回了语义相关但主题不符的文本生成阶段又被允许引用所有上下文模型没有能力逐条甄别每条 chunk 与问题的相关性。代理越活跃证据池越大这个问题就越明显。解决在检索返回后加一道相关性过滤不只看相似度分数还要看 chunk 与原始问题的关键词重叠度。重叠太低就让代理决定是否丢弃而不是直接进证据池。同时在生成提示词里收紧措辞只允许引用能直接支撑结论的证据。如果还不行就给证据池里的 chunk 加一个“信任分”生成时按分数加权提示引导模型优先参考高信任内容。5.3 一次请求延迟几十秒token 成本暴增现象一条简单问题代理来回调用工具六次加上反复把历史翻给模型看最终耗时突破三十秒账单数字也很吓人。原因max_steps设得过大同时历史记录没有压缩。每轮决策都要把前几轮的工具结果拼接给模型这些结果都是纯文本占用大量输入 token。工具调用本身如果走外部接口网络耗时还在成倍叠加。解决第一刀砍max_steps把上限压到实际场景所需的最小值。第二刀做历史摘要每轮只保留上一轮的动作、工具和核心结论而不是把原始检索结果原样重放。第三刀给工具调用加超时控制单个工具超过 5 秒就返回一个占位结果避免外部服务拖垮整个循环。5.4 多步检索把上下文窗口撑爆现象代理按计划拆解了子查询每个子查询返回 3 条 chunk跑完 4 轮之后上下文已经满了生成阶段只能收到被截断的残缺文本。原因证据池没有总量限制。每次检索的结果都无条件追加即使前一轮已经拿到足够信息代理也不会主动清理旧证据。解决给证据池设一个最大容量比如 5000 字或 10 条 chunk。新证据进来时如果总量超限就按相关度分数淘汰最弱的旧证据。这个淘汰动作比简单截断更智能能保住最关键的信息。生成阶段再对最终证据列表做一次排序相关性最高的排在前面防止模型只盯着前几条。5.5 测试用例太少上线后表现骤降现象开发阶段拿三五条精心设计的问题调得效果很好一上线面对真实用户问题回答质量立刻崩。我把这个叫“测试集自嗨”。原因Agentic RAG 的复杂度远高于传统 RAG行为空间大几条用例覆盖不了路由偏差、重写失效、反思误判这些失败模式。真实用户的问题语法不规范、意图不明确测试时根本没暴露。解决建一个至少二十条问题的回归集覆盖单轮、多轮、跨数据源、需要计算、问题含歧义五类场景。每次改动参数或者换模型都要跑一遍回归集记录“工具调用次数、引用命中率、人工评分”三个指标。这个回归集不用做得很重但它能拦住大部分“调好了这条、弄坏了那条”的翻车。6. 进阶用法把最小实现推向生产与可观测6.1 给代理循环加一条“轨迹日志”最小实现里的history只用来辅助决策生产环境应该把它改成结构化轨迹日志。每轮记录时间戳、工具名称、输入参数、返回结果摘要、相似度分数、耗时。排错时打开轨迹一眼就能看到代理在哪个环节走偏。这个日志就是黑匣子的钥匙没有它Agentic RAG 出了问题只能靠猜。6.2 用一组经典追问检验能力边界我会让测试覆盖这类问题“如果没有检索到任何证据模型会怎么回答”“检索到两条互相矛盾的数据模型怎么处理”“问题的答案需要三轮以上工具调用才能凑齐”。这三问能比较快暴露系统短板。试探性的做法是故意在语料里放一条和问题相关但在逻辑上错误的文本看模型是直接引用还是能识别矛盾。能识别并质疑证据说明循环够健康直接照抄错误文本就要强化反思环节。6.3 引入人工审核与工具权限控制Agentic RAG 生产化的最后一步是把“代理能做什么”和“代理不能做什么”用权限表约束住。比如run_calc只能对检索到的数据进行计算不能碰用户即兴给的数字search_code只允许查指定仓库不能跨目录检索。人工审核方面保留一份“高风险问题清单”命中时强制走人工确认再出最终答案。代理给出答案前还要附带完整的工具调用轨迹方便人工复核逻辑链。把轨迹日志接到可视化面板上之后我还养成了一个习惯每星期回看一次代理的失败轨迹。那些“检索了三轮还是答非所问”的记录比任何评测指标都更能说明问题出在路由还是反思。越是复杂系统越要保留它思考过程的每一帧画面。Agentic RAG 本质上是在用更多的推理换更好的回答代价是系统边界变宽、行为变复杂这时只有把每个动作都变成可见记录才有办法持续修正。这套可运行的最小实现就是那个起点沿着它加上日志、权限和回归集就是一个能真正用于生产的最小框架了。做 Agentic RAG 这一年多我最大的血泪经验是先让流程跑通再追求参数完美。很多同事一上来就调top_k和温度却连循环日志都没开等于在盲人摸象。你可以先把上面这段代码原样跑一遍然后换一个你自己业务里的复杂问题看看代理会走几步、引用了几条证据、错在哪里。把这套动作养成习惯希望你也能在自己的项目里少走弯路。希望帮到你。本文还有配套的精品资源点击获取
