让RAG长出脑子，复合问题不再硬编答案

到第5篇，你的 RAG 已经能上线了——检索准、能评估、能更新、不泄密。但它还有个死穴：只会查一次。

用户问“我们公司 Q3 的毛利率是多少“，一次检索就够了。但如果用户问“对比一下我们 Q3 和 Q2 的毛利率，哪个变化更大，主要原因是什么“——一次检索就答不准了。这问题里藏着三个子问题：Q3 毛利率、Q2 毛利率、变化原因。你得查三次，还得算一次差值，最后再综合。

Ando@ando_w·Jul 9 Article私有知识库第5篇：企业RAG的权限隔离，从检索时就该管住我杭州有个朋友做企业知识库，上线第二周出了事。 老板把一份薪酬调整方案扔进了知识库，想着高管层方便查。结果一个普通员工问“今年调薪幅度多少“，RAG…24256.1K

单轮 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）。思路很朴素：

• 思考：LLM 分析当前手上有什么信息，还缺什么

• 行动：缺什么就去查什么，或者调个工具

• 观察：看行动返回了什么结果

• 再思考：信息够了就答，不够就回到第 1 步

循环直到 LLM 觉得“我能答了“，或者达到步数上限。

听起来玄乎，其实就是一个 while 循环。难点不在算法，在工程——怎么让 LLM 的每一步都可控、可观测、可中断。

三、第一步：把检索封装成工具

要让 LLM 调用检索，得先把检索变成一个“工具“。工具的本质是一份说明书：告诉 LLM 这个工具干嘛的、怎么传参、返回什么。

用 Qwen3.7-Max 的 function calling，定义两个工具：一个检索知识库，一个做计算（算差值、算百分比这类）。

pythonimport 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 融合 + 重排 + 权限过滤，一个不少。

pythondef 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 思考 → 调工具 → 看结果 → 再思考，直到它能答或者步数耗尽。

pythondef 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 自己怎么调度：

pythonanswer = 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 检索过就直接返回，别再打数据库。既省成本又防循环。

python_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，超过阈值就强制收尾。企业环境里不装这个，月底账单会让你怀疑人生。

pythontotal_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，也可以靠规则（问题长度、有没有“对比““分析”“原因“这类词）。

pythondef 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篇：权限模型，检索后过滤是错的