让RAG长出脑子：Agentic RAG的多步推理实践

✍ Ando🕐 2026-07-10📦 17.6 KB 🟢 已读 𝕏 文章列表

文章指出传统单轮 RAG 在处理复合问题时存在检索不足或幻觉的缺陷，提出通过 Agentic RAG（多步推理）来解决。核心思路是将 LLM 从被动应答升级为主动调度的指挥官，利用 ReAct（推理-行动循环）机制，自主决定检索次数和工具调用。文中详细展示了如何封装检索工具、执行函数（含权限过滤、混合检索）以及编写多步推理循环代码，实现复杂问题的精准回答。

# 让RAG长出脑子，复合问题不再硬编答案 **作者**: Ando **日期**: 2026-07-10T06:35:35.000Z **来源**: [https://x.com/ando_w/status/2075468963098546520](https://x.com/ando_w/status/2075468963098546520) --- ![Image](/doc/2886/image/img_79e63efccd78.jpg) 到第5篇，你的 RAG 已经能上线了——检索准、能评估、能更新、不泄密。但它还有个死穴：只会查一次。用户问"我们公司 Q3 的毛利率是多少"，一次检索就够了。但如果用户问"对比一下我们 Q3 和 Q2 的毛利率，哪个变化更大，主要原因是什么"——一次检索就答不准了。这问题里藏着三个子问题：Q3 毛利率、Q2 毛利率、变化原因。你得查三次，还得算一次差值，最后再综合。单轮 RAG 拿到这种问题，要么硬编一个答案出来（幻觉），要么把三个子问题全塞进一个 query 去检索（啥也召回不来）。这篇讲怎么让 RAG 长出脑子——从"单轮问答"升级到"多步推理"，也就是 Agentic RAG。 # 一、单轮检索到底卡在哪先把问题说透。单轮 RAG 的流程是：用户提问 → 检索一次 → 把检索结果喂给 LLM → LLM 生成答案。一锤子买卖。这个流程在"事实查询"场景下没问题，因为事实查询的子问题只有一个。但真实业务里，用户的问题经常是复合的： - "对比 A 和 B 的优劣"——要查 A，要查 B，还要对比 - "最近三个月销售额下降的原因"——要查三个月数据，要查同期市场情况，要查内部动作 - "这个合同里有没有对我们不利的条款"——要分段检索整份合同，每段都要判断这类问题有个共同特征：一次检索拿不全，拿全了 LLM 也消化不了。你把 50 个切片全塞给 LLM，它会在噪音里迷失，最后给你一个四平八稳的废话答案。还有一个场景单轮 RAG 彻底搞不定：需要外部工具。用户问"这个客户上周下的单到了没"，检索只能查到合同信息，查不到物流状态——物流在另一个系统里，你得调接口。所以 Agentic RAG 要解决两件事：一是让 LLM 自己决定查几次、查什么；二是让 LLM 能调用工具，不只是检索。 # 二、Agentic RAG 的核心思路一句话：把 LLM 从"答题机器"升级成"指挥官"。单轮 RAG 里，LLM 是被动接球的——你喂什么它答什么。Agentic RAG 里，LLM 是主动调度的一方：它拿到用户问题后，自己思考"这个问题我需要查几次、先查什么、要不要调别的工具、查够了没"，然后一步步执行，直到它能答出来为止。这个过程叫"推理-行动循环"，学术上叫 ReAct（Reasoning + Acting）。思路很朴素： 1. 思考：LLM 分析当前手上有什么信息，还缺什么 2. 行动：缺什么就去查什么，或者调个工具 3. 观察：看行动返回了什么结果 4. 再思考：信息够了就答，不够就回到第 1 步循环直到 LLM 觉得"我能答了"，或者达到步数上限。听起来玄乎，其实就是一个 while 循环。难点不在算法，在工程——怎么让 LLM 的每一步都可控、可观测、可中断。 # 三、第一步：把检索封装成工具要让 LLM 调用检索，得先把检索变成一个"工具"。工具的本质是一份说明书：告诉 LLM 这个工具干嘛的、怎么传参、返回什么。用 Qwen3.7-Max 的 function calling，定义两个工具：一个检索知识库，一个做计算（算差值、算百分比这类）。 ``` import json from openai import OpenAI client = OpenAI( api_key="sk-你的key", base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) MODEL = "qwen-max-latest" # 工具定义：告诉 LLM 有哪些工具可用 tools = [ { "type": "function", "function": { "name": "search_knowledge_base", "description": "在公司私有知识库里检索信息。当需要查询公司内部文档、合同、报告、手册等内容时调用。传入一个具体的查询问题，返回相关文档片段。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "要检索的具体问题，尽量具体。例如'Q3毛利率'而不是'财务数据'" } }, "required": ["query"] } } }, { "type": "function", "function": { "name": "calculate", "description": "做数学计算。当需要算差值、百分比、求和等数值运算时调用。传入一个数学表达式。", "parameters": { "type": "object", "properties": { "expression": { "type": "string", "description": "数学表达式，例如'45.2 - 42.1'或'(45.2-42.1)/42.1*100'" } }, "required": ["expression"] } } } ] ``` 这里有个细节很多人会忽略：工具的 description 要写得 LLM 看得懂、知道什么时候用。别写"检索数据库"这种废话，要写清楚"查公司内部文档时用""传具体问题别传宽泛词"。LLM 是照着 description 决定调不调工具的，描述糊弄，调用就乱来。 # 四、第二步：实现工具的执行函数工具定义只是说明书，真正干活的还得你自己写。把前面 5 篇的检索管线接进来——向量检索 + BM25 + RRF 融合 + 重排 + 权限过滤，一个不少。 ``` def search_knowledge_base(query: str, user_id: str = "u1") -> str: """检索知识库，复用前面5篇的完整管线""" # 1. 向量检索（带权限JOIN） vec_sql = """ SELECT c.id, c.content, c.doc_id, 1 - (c.embedding <=> %s::vector) AS vec_score FROM chunks c JOIN user_doc_acl a ON a.doc_id = c.doc_id AND a.user_id = %s WHERE 1 - (c.embedding <=> %s::vector) > 0.3 ORDER BY c.embedding <=> %s::vector LIMIT 20 """ emb = get_embedding(query) # text-embedding-v3, 1024维 cur.execute(vec_sql, (emb, user_id, emb, emb)) vec_results = {r[0]: {"content": r[1], "vec_score": r[3]} for r in cur.fetchall()} # 2. BM25检索（带权限JOIN） bm25_sql = """ SELECT c.id, c.content, ts_rank_cd(c.tsv, plainto_tsquery('chinese', %s)) AS bm25_score FROM chunks c JOIN user_doc_acl a ON a.doc_id = c.doc_id AND a.user_id = %s WHERE c.tsv @@ plainto_tsquery('chinese', %s) ORDER BY bm25_score DESC LIMIT 20 """ cur.execute(bm25_sql, (query, user_id, query)) bm25_results = {r[0]: {"content": r[1], "bm25_score": r[2]} for r in cur.fetchall()} # 3. RRF融合 all_ids = set(vec_results.keys()) | set(bm25_results.keys()) rrf = {} for rank, cid in enumerate(sorted(vec_results.keys(), key=lambda x: -vec_results[x]["vec_score"])): rrf[cid] = rrf.get(cid, 0) + 1 / (60 + rank) for rank, cid in enumerate(sorted(bm25_results.keys(), key=lambda x: -bm25_results[x]["bm25_score"])): rrf[cid] = rrf.get(cid, 0) + 1 / (60 + rank) # 4. 取Top5做重排 top_ids = sorted(rrf.keys(), key=lambda x: -rrf[x])[:5] contents = [vec_results.get(cid, bm25_results.get(cid))["content"] for cid in top_ids] reranked = rerank(query, contents) # 第3篇的重排模型 # 5. 返回Top3 result = [] for idx in sorted(range(len(reranked)), key=lambda i: -reranked[i])[:3]: result.append(contents[idx]) return "\n---\n".join(result) if result else "未检索到相关内容" def calculate(expression: str) -> str: """数学计算""" try: # 只允许数字和运算符，防注入 safe = "".join(c for c in expression if c in "0123456789.+-*/() ") result = eval(safe, {"__builtins__": {}}, {}) return f"{expression} = {result}" except Exception as e: return f"计算失败: {e}" ``` 注意 search_knowledge_base 里权限 JOIN 还在——第5篇的权限模型不能丢。Agentic RAG 只是把单轮检索包了一层，底层的权限隔离、混合检索、重排一个都没少。很多人上 Agentic RAG 时把权限忘了，LLM 调用工具时带了用户身份，但工具内部没过滤，结果权限形同虚设。 # 五、第三步：多步推理循环这是 Agentic RAG 的心脏。一个 while 循环，LLM 思考 → 调工具 → 看结果 → 再思考，直到它能答或者步数耗尽。 ``` def agentic_rag(question: str, user_id: str = "u1", max_steps: int = 5) -> str: """Agentic RAG：多步推理循环""" messages = [ { "role": "system", "content": ( "你是一个企业知识库助手。回答用户问题时，如果信息不足，" "可以调用 search_knowledge_base 检索知识库，或调用 calculate 做计算。" "每次检索后先判断信息够不够，够了再回答，不够继续检索。" "最多检索5次，避免无限循环。" ) }, {"role": "user", "content": question} ] for step in range(max_steps): # LLM思考：决定是调工具还是直接答 resp = client.chat.completions.create( model=MODEL, messages=messages, tools=tools, tool_choice="auto" # auto让LLM自己决定 ) msg = resp.choices[0].message messages.append(msg) # 如果LLM决定调工具 if msg.tool_calls: for call in msg.tool_calls: name = call.function.name args = json.loads(call.function.arguments) # 执行工具 if name == "search_knowledge_base": result = search_knowledge_base(args["query"], user_id) elif name == "calculate": result = calculate(args["expression"]) else: result = f"未知工具: {name}" print(f" [步骤{step+1}] 调用 {name}({args}) → {result[:80]}...") # 工具结果塞回对话 messages.append({ "role": "tool", "tool_call_id": call.id, "content": result }) else: # LLM没调工具，说明它觉得信息够了，直接给答案 print(f" [步骤{step+1}] 直接回答") return msg.content # 步数耗尽，强制让LLM基于已有信息回答 messages.append({ "role": "user", "content": "已达到最大检索次数，请基于已有信息给出最佳回答。" }) resp = client.chat.completions.create(model=MODEL, messages=messages) return resp.choices[0].message.content ``` 跑一下那个毛利率对比的问题，看看 LLM 自己怎么调度： ``` answer = agentic_rag("对比我们公司Q3和Q2的毛利率，哪个变化更大，主要原因是什么") ``` 控制台会打印出类似这样的执行轨迹： ``` [步骤1] 调用 search_knowledge_base({'query': 'Q3毛利率'}) → 2024年Q3毛利率45.2%，主营... [步骤2] 调用 search_knowledge_base({'query': 'Q2毛利率'}) → 2024年Q2毛利率42.1%，主营... [步骤3] 调用 calculate({'expression': '45.2 - 42.1'}) → 45.2 - 42.1 = 3.1 [步骤4] 调用 search_knowledge_base({'query': 'Q3毛利率上升原因'}) → Q3毛利率上升主要受原材料... [步骤5] 直接回答 ``` LLM 自己拆了四个子问题，查了三次、算了一次，最后综合成答案。你不用提前告诉它"这是个对比题"，它自己看出来了。这就是 Agentic RAG 和单轮 RAG 的本质区别——调度权交给了 LLM。 # 六、防失控：别让 Agent 跑野了把调度权交给 LLM，风险也来了：它可能陷入死循环，查了又查、查了又查，永远不满意自己的答案。或者更糟，它每次检索都召回不到东西，还是坚持不懈地查十几次，token 烧了一大把，最后给你一句"查不到"。三个阀门，必须装上：第一个：最大步数限制。就是上面代码里的 max_steps=5。5 步对绝大多数问题够用了，再复杂的也该拆成多次对话。这个数字别开太大，开到 20 步，一个查询能烧掉你半个月的 token 预算。第二个：去重。 LLM 有个毛病，会重复检索同一个 query。你可以在工具里加个缓存，同一个 query 检索过就直接返回，别再打数据库。既省成本又防循环。 ``` _search_cache = {} def search_knowledge_base(query: str, user_id: str = "u1") -> str: cache_key = (query, user_id) if cache_key in _search_cache: return _search_cache[cache_key] # 命中缓存，直接返回 # ... 正常检索 ... _search_cache[cache_key] = result return result ``` 记得缓存 key 要带 user_id（第5篇踩过的坑，权限不能靠缓存泄露）。第三个：成本监控。每次循环记一下消耗的 token，超过阈值就强制收尾。企业环境里不装这个，月底账单会让你怀疑人生。 ``` total_tokens = 0 COST_LIMIT = 8000 # 单次问答最多烧8000 token for step in range(max_steps): resp = client.chat.completions.create(...) total_tokens += resp.usage.total_tokens if total_tokens > COST_LIMIT: break # 超预算，强制收尾 ``` # 七、什么时候该上 Agentic RAG，什么时候别上这是最容易被忽略的问题。Agentic RAG 不是万能药，它有代价：慢（一个答案要等十几秒）、贵（多次 LLM 调用）、不可控（LLM 可能瞎调度）。该上的场景：问题复合度高、需要多跳推理、需要调外部工具。比如"对比类""归因类""核查类"问题，单轮检索搞不定的。别上的场景：简单事实查询。用户问"公司地址是什么"，一次检索就答了，你非得让 LLM 思考三轮、调两次工具，响应时间从 2 秒变 15 秒，用户早跑了。工程上的做法是路由：先用一个轻量判断，把问题分成"简单"和"复杂"两类，简单的走单轮 RAG，复杂的走 Agentic RAG。这个判断可以靠 LLM，也可以靠规则（问题长度、有没有"对比""分析""原因"这类词）。 ``` def route_question(question: str) -> str: """路由：判断走单轮还是Agentic""" complexity_keywords = ["对比", "比较", "分析", "原因", "为什么", "差异", "变化"] if any(kw in question for kw in complexity_keywords) or len(question) > 30: return "agentic" return "simple" def rag(question: str, user_id: str = "u1") -> str: route = route_question(question) if route == "simple": return simple_rag(question, user_id) # 单轮，快 return agentic_rag(question, user_id) # 多步，准 ``` 别为了显技术栈高级就把所有问题都丢给 Agentic RAG，那是对用户耐心和你钱包的双重不负责。 # 八、这一篇的收获到这，RAG 系列走完了。回顾一下六篇串起来的完整链路： - 第1篇：最小可运行工程，把 RAG 跑起来 - 第2篇：千问模型链路，embedding + LLM 全国产 - 第3篇：查询改写 + 混合检索 + 重排，召回率拉满 - 第4篇：评估集 + 增量更新，能衡量能迭代 - 第5篇：权限模型，企业落地不泄密 - 第6篇：Agentic RAG，长出脑子，能多步推理一套下来，你的 RAG 从"能跑"到"能打"。每一篇都是上一篇的自然延伸，没有跳跃，代码都能直接拼起来跑。 Agentic RAG 是这个系列的终点，但不是 RAG 的终点。再往后还有 GraphRAG（知识图谱增强）、Self-RAG（模型自己判断要不要检索）、多模态 RAG（图文档表格一起查）。那些是另一个故事了，先把这套基础打扎实。真要做企业 RAG，这六篇够你上线了。剩下的，是真实用户和真实数据教会你的事。参考资料： - ReAct 原始论文（Reasoning + Acting）：https://arxiv.org/abs/2210.03629 - Qwen function calling 文档：https://help.aliyun.com/zh/model-studio/function-calling - OpenAI tools 接口规范（Qwen兼容）：https://platform.openai.com/docs/guides/function-calling - Agentic RAG 综述：https://arxiv.org/abs/2501.09136 上一篇：私有知识库第5篇：权限模型，检索后过滤是错的 ## 相关链接 - [Ando](https://x.com/ando_w) - [@ando_w](https://x.com/ando_w) - [177](https://x.com/ando_w/status/2075468963098546520/analytics) - [最小可运行工程，把 RAG 跑起来](https://x.com/ando_w/status/2073070705789329561) - [千问模型链路，embedding + LLM 全国产](https://x.com/ando_w/status/2073979240962228582) - [查询改写 + 混合检索 + 重排，召回率拉满](https://x.com/ando_w/status/2074321836117037349) - [评估集 + 增量更新，能衡量能迭代](https://x.com/ando_w/status/2074699344129789985) - [权限模型，企业落地不泄密](https://x.com/ando_w/status/2075048886972907736) - [https://arxiv.org/abs/2210.03629](https://arxiv.org/abs/2210.03629) - [https://help.aliyun.com/zh/model-studio/function-calling](https://help.aliyun.com/zh/model-studio/function-calling) - [https://platform.openai.com/docs/guides/function-calling](https://platform.openai.com/docs/guides/function-calling) - [https://arxiv.org/abs/2501.09136](https://arxiv.org/abs/2501.09136) - [私有知识库第5篇：权限模型，检索后过滤是错的](https://x.com/ando_w/status/2073070705789329561) - [Upgrade to Premium](https://x.com/i/premium_sign_up) - [2:35 PM · Jul 10, 2026](https://x.com/ando_w/status/2075468963098546520) - [177 Views](https://x.com/ando_w/status/2075468963098546520/analytics) --- *导出时间: 2026/7/10 15:51:58*

🔗 相关文档