Agentic RAG实战：如何让企业知识库真正"理解"复杂问题

前言

在语核科技的制造业知识库落地项目中，我们反复遇到同一类问题：用户提出的问题越复杂，传统RAG的答案质量越差。"找出去年因交期延误导致客户投诉的订单并分析原因"——这类问题涉及多步推理、跨文档聚合，单次向量检索根本无法覆盖。

这促使我们系统研究并工程化实现了Agentic RAG方案。本文总结这一过程中的关键架构决策、核心代码实现与实测数据，供有类似场景的技术团队参考。

---

一、传统RAG的失效边界

传统RAG（Retrieval-Augmented Generation）的主流实现路径极为简洁：将用户Query向量化，Top-K召回，拼接上下文，交给LLM生成答案。这条链路在简单问答场景下效果尚可，但在企业级知识库中，有四类问题会让它明显失效。

1.1 多跳问题

多跳问题要求系统先找到中间事实，再基于中间结果进行下一步检索。例如：

"找出去年所有因交期延误导致客户投诉的订单，并分析主要原因。"

这个问题隐含了三个子步骤：① 检索"客户投诉记录"→ ② 从投诉记录中筛选"交期延误"相关条目 → ③ 关联"订单表"找出具体订单号，再分析原因。

传统RAG把原始Query直接向量化，一步召回的文档块无法同时覆盖这三层信息，结果要么答非所问，要么漏掉关键上下文。

1.2 模糊查询

用户问"最近几个月有没有质量问题"，"最近几个月"是时间语义，"质量问题"边界模糊。向量检索依赖语义相似度，模糊表述的嵌入向量与精确文档块之间的余弦相似度偏低，召回率严重下降。

1.3 跨文档推理

答案散布在多个文档中，需要跨文档聚合。例如"对比A供应商和B供应商过去一年的交货准时率"，相关数据可能分布在几十份月报PDF中，单次Top-K召回天然无法处理聚合逻辑。

1.4 长文档理解

企业技术文档常见几十页的规格书或工艺手册。Embedding模型的上下文窗口有限（通常512~8192 token），超长文档切片后语义割裂，关键信息跨越切片边界时会被漏掉。

这四类失效场景的共同根因是：传统RAG是单步、无状态的检索，无法根据中间结果动态调整检索策略。

---

二、Agentic RAG的核心差异

2.1 架构对比

传统RAG的链路极短：

Query → 向量化 → Top-K检索 → 拼接上下文 → LLM生成

Agentic RAG引入了Agent作为规划与控制中枢：

Query → Agent规划 → 查询分解 → 多步检索（可迭代）→ 工具调用 → 自我反思 → 推理合成 → 生成

完整架构如下：

用户问题
    │
    ▼
┌─────────────────────┐
│    Agent Planner     │  ← LLM驱动的规划器，理解意图、制定策略
└──────────┬──────────┘
           │ 查询分解（Query Decomposition）
    ┌──────┴──────┐
    │             │
    ▼             ▼
┌────────┐   ┌──────────────┐
│向量检索 │   │ 结构化查询    │  ← SQL / API / 计算器等工具
│(语义)  │   │ (精确匹配)    │
└────┬───┘   └──────┬───────┘
     │               │
     └──────┬────────┘
            │ 中间检索结果
            ▼
┌─────────────────────┐
│   反思 & 判断模块    │  ← 当前信息是否足够回答问题？
│  (Self-Reflection)  │
└──────────┬──────────┘
           │
    ┌──────┴──────────────────┐
    │                         │
    ▼                         ▼
信息不足                   信息充足
（继续检索，调整策略）      （进入合成阶段）
    │                         │
    └──────────┬──────────────┘
               ▼
        最终回答生成

2.2 四项核心能力

① 查询分解（Query Decomposition）
Agent将复杂问题拆解为若干可独立检索的子问题。这一步本身由LLM完成，输出结构化的子问题列表，每个子问题对应一次检索动作。

② 迭代检索（Iterative Retrieval）
每次检索的结果会作为上下文反馈给Agent，Agent基于当前已知信息决定下一步检索方向。这是与传统RAG最本质的区别——检索是有状态的。

③ 工具调用（Tool Use）
Agentic RAG不局限于向量检索，Agent可以调用任意工具：SQL查询、REST API、Python计算器、正则抽取器等。工具的扩展性决定了系统能处理的问题域宽度。

④ 自我反思（Self-Reflection）
每次检索后，Agent评估当前掌握的信息是否足够回答原始问题。若置信度低于阈值，则重新规划检索策略；若足够，则进入最终合成阶段。

---

三、工程化实现路径

下面给出基于LangChain的核心实现示例，覆盖查询分解、迭代检索循环与置信度判断三个关键模块。

from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from typing import List, Dict, Any
import json

# ── 1. 工具注册 ────────────────────────────────────────────────────────────
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="enterprise_kb",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

def vector_search(query: str, k: int = 5) -> List[Dict]:
    """语义向量检索，返回Top-K文档块及其元数据"""
    docs = vectorstore.similarity_search_with_score(query, k=k)
    return [
        {"content": doc.page_content, "source": doc.metadata.get("source", ""), "score": score}
        for doc, score in docs
    ]

def structured_query(sql: str) -> str:
    """结构化数据查询（示例：接入企业内部数据库）"""
    # 实际生产中替换为真实数据库连接
    # conn = get_db_connection()
    # result = conn.execute(sql).fetchall()
    # return json.dumps(result, ensure_ascii=False)
    raise NotImplementedError("请接入实际数据库")

# 注册为Agent可调用工具
tools = [
    Tool(
        name="vector_search",
        func=vector_search,
        description="语义检索企业知识库，适用于非结构化文档、技术手册、FAQ等内容"
    ),
    Tool(
        name="structured_query",
        func=structured_query,
        description="精确查询结构化数据库，适用于订单、库存、质检等结构化数据"
    ),
]


# ── 2. 查询分解器 ──────────────────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o", temperature=0)

DECOMPOSE_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个企业知识库问答系统的查询规划器。
将用户的复杂问题分解为若干个可以独立检索的子问题。
输出格式为JSON列表，每个子问题包含：
- sub_query: 子问题文本
- tool: 推荐使用的工具（vector_search 或 structured_query）
- depends_on: 依赖的前序子问题索引列表（空列表表示无依赖）
"""),
    ("human", "原始问题：{question}")
])

def decompose_query(question: str) -> List[Dict]:
    """将复杂问题拆解为子问题列表"""
    chain = DECOMPOSE_PROMPT | llm
    response = chain.invoke({"question": question})
    # 提取JSON内容
    content = response.content
    start = content.find("[")
    end = content.rfind("]") + 1
    if start == -1 or end == 0:
        # 无法分解则作为单一问题处理
        return [{"sub_query": question, "tool": "vector_search", "depends_on": []}]
    return json.loads(content[start:end])


# ── 3. 置信度判断（自我反思模块）────────────────────────────────────────────
REFLECTION_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个信息充分性评估器。
根据已收集到的上下文，判断是否足以回答原始问题。
输出JSON：{{"sufficient": true/false, "confidence": 0.0-1.0, "missing": "缺失信息描述"}}
如果sufficient为true，confidence应≥0.75。"""),
    ("human", """原始问题：{question}

已收集上下文：
{context}

请评估信息是否充分。""")
])

def assess_sufficiency(question: str, context: str) -> Dict:
    """评估当前上下文是否足以回答原始问题"""
    chain = REFLECTION_PROMPT | llm
    response = chain.invoke({"question": question, "context": context})
    content = response.content
    start = content.find("{")
    end = content.rfind("}") + 1
    if start == -1 or end == 0:
        return {"sufficient": False, "confidence": 0.0, "missing": "解析失败"}
    return json.loads(content[start:end])


# ── 4. Agentic RAG 主循环 ───────────────────────────────────────────────────
CONFIDENCE_THRESHOLD = 0.75  # 置信度阈值，低于此值则继续检索
MAX_ITERATIONS = 4            # 最大迭代次数，防止无限循环

def agentic_rag(question: str) -> str:
    """
    Agentic RAG主流程：
    1. 查询分解
    2. 迭代检索（含自我反思）
    3. 最终答案合成
    """
    print(f"[AgenticRAG] 原始问题：{question}")

    # Step 1: 查询分解
    sub_queries = decompose_query(question)
    print(f"[AgenticRAG] 分解为 {len(sub_queries)} 个子问题")

    # Step 2: 迭代检索
    all_contexts: List[str] = []
    sub_results: Dict[int, str] = {}  # 存储每个子问题的检索结果，供后续依赖使用

    for iteration in range(MAX_ITERATIONS):
        iteration_contexts: List[str] = []

        for idx, sq in enumerate(sub_queries):
            # 处理依赖关系：将前序结果拼入当前子查询
            enriched_query = sq["sub_query"]
            for dep_idx in sq.get("depends_on", []):
                if dep_idx in sub_results:
                    enriched_query += f"\n（参考上下文：{sub_results[dep_idx]}）"

            # 调用对应工具
            if sq["tool"] == "vector_search":
                results = vector_search(enriched_query, k=4)
                chunk = "\n".join([r["content"] for r in results])
            else:
                # structured_query 场景需要Agent生成SQL，此处简化为向量搜索兜底
                results = vector_search(enriched_query, k=4)
                chunk = "\n".join([r["content"] for r in results])

            sub_results[idx] = chunk
            iteration_contexts.append(f"[子问题{idx+1}] {sq['sub_query']}\n{chunk}")

        all_contexts.extend(iteration_contexts)
        combined_context = "\n\n---\n\n".join(all_contexts)

        # Step 3: 自我反思——信息够了吗？
        assessment = assess_sufficiency(question, combined_context)
        print(f"[AgenticRAG] 第{iteration+1}轮检索，置信度：{assessment['confidence']:.2f}")

        if assessment["sufficient"] and assessment["confidence"] >= CONFIDENCE_THRESHOLD:
            print("[AgenticRAG] 信息充分，进入答案合成")
            break

        if iteration < MAX_ITERATIONS - 1:
            # 根据缺失信息调整子问题，补充检索
            missing = assessment.get("missing", "")
            if missing:
                sub_queries.append({
                    "sub_query": missing,
                    "tool": "vector_search",
                    "depends_on": list(sub_results.keys())
                })
                print(f"[AgenticRAG] 补充检索：{missing}")

    # Step 4: 最终答案合成
    synthesis_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是企业知识库问答助手。基于以下检索到的上下文，准确、完整地回答用户问题。如果信息不足，请如实说明。"),
        ("human", "问题：{question}\n\n上下文：\n{context}")
    ])
    chain = synthesis_prompt | llm
    answer = chain.invoke({"question": question, "context": combined_context})
    return answer.content


# ── 5. 使用示例 ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
    question = "找出去年所有因交期延误导致客户投诉的订单，并分析主要原因"
    answer = agentic_rag(question)
    print("\n最终答案：")
    print(answer)

上述代码有几个工程细节值得注意：

• depends_on 字段：子问题间的依赖关系让Agent在执行有序依赖的检索时，能将前序结果拼入当前查询，模拟人类"先找A，再用A去找B"的推理链。
• MAX_ITERATIONS 限制：防止Agent陷入无限检索循环，生产环境建议设为3~5。
• 置信度阈值 CONFIDENCE_THRESHOLD：0.75是我们在制造业场景下调试出的经验值，不同领域可能需要调整。
• 工具兜底逻辑：structured_query 场景在没有接入真实数据库时自动降级为向量检索，保证系统不崩溃。

---

四、实测数据对比

在语核科技的制造业技术文档知识库项目中，我们对传统RAG与Agentic RAG进行了系统对比测试，测试样本为生产环境中随机抽取的真实用户提问（非精选测试集）。

评估维度	传统RAG	Agentic RAG	备注
多跳问题准确率	43%	89%	提升 +46pct，样本量 n=120
简单单跳问答准确率	87%	88%	差异不显著
跨文档聚合准确率	31%	82%	提升 +51pct
模糊查询召回率	58%	79%	提升 +21pct
平均响应时间（简单问答）	1.2s	3.8s	Agentic模式多次LLM调用
平均响应时间（复杂问题）	1.5s	8.4s	含2~3轮迭代检索
平均Token消耗（每次查询）	~800	~3200	约4倍消耗

数据来源：语核科技内部测试，场景为制造业技术文档知识库，测试时间2026年Q1，样本为生产环境随机抽取的真实用户提问。

核心结论：

1. 复杂问题提升显著：多跳和跨文档场景准确率提升40~50个百分点，是Agentic RAG最核心的价值区间。
2. 简单问题收益有限：单跳问答两者准确率相当，但Agentic RAG的响应时间是传统RAG的3倍以上。
3. 成本是真实代价：Token消耗约为传统RAG的4倍，在高并发场景下成本压力不可忽视。

---

五、落地建议

基于我们的工程实践，给有意落地Agentic RAG的技术团队三条实用建议。

5.1 先判断问题复杂度，再选架构

不是所有场景都需要Agentic RAG。一个简单判断标准：

• 用一句话能直接检索到答案 → 传统RAG，够用且快
• 需要"先找A，再基于A找B" → Agentic RAG，必要
• 需要跨多个文档聚合统计 → Agentic RAG，必要
• 问题涉及时间窗口、条件筛选 → 优先考虑接入结构化查询工具

在语核科技的制造业项目中，我们的做法是先对历史Query做聚类，识别出多跳/聚合类问题的占比（约占总量的30%~40%），再决定是否引入Agentic RAG——这一部分问题对业务价值影响最大，值得投入工程成本。

5.2 做好成本控制

Agentic RAG的Token消耗是传统RAG的3~5倍，有几个控制手段：

• 分层路由：先用轻量分类器判断问题复杂度，简单问题走传统RAG链路，复杂问题才进入Agent规划。
• 限制迭代次数：MAX_ITERATIONS 建议不超过5，配合置信度阈值提前退出。
• 子模型降本：查询分解、置信度判断等辅助任务可以用GPT-4o-mini等小模型，仅最终合成使用强模型。

5.3 用RAGAS框架量化评估效果

主观感受不足以驱动迭代，推荐使用 RAGAS 框架进行量化评估，核心指标：

• Faithfulness：答案是否忠实于检索到的上下文，避免幻觉
• Answer Relevancy：答案与问题的相关性
• Context Recall：检索结果是否覆盖了回答所需的关键信息
• Context Precision：召回的文档块是否精准，噪声比例

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall, context_precision
from datasets import Dataset

# 构造评估数据集
eval_data = {
    "question": ["问题1", "问题2"],
    "answer": ["Agentic RAG生成的答案1", "答案2"],
    "contexts": [["检索到的上下文块1a", "1b"], ["上下文块2a"]],
    "ground_truth": ["标准答案1", "标准答案2"]
}

dataset = Dataset.from_dict(eval_data)
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_recall, context_precision]
)
print(results)

建议在上线前和每次迭代后各跑一次RAGAS评估，将核心指标纳入CI流程，防止回归。

---

总结

Agentic RAG不是传统RAG的小改良，而是从"单步检索"到"有状态推理"的范式转变。它真正解决了企业知识库中最有价值也最难处理的那类问题——需要多步推理、跨文档聚合的复杂查询。代价是更高的延迟与Token成本，因此合理的分层路由策略至关重要。

工程化落地的关键不在于框架选型，而在于三点：查询分解的粒度设计、工具集的覆盖范围，以及置信度阈值的调参。这三点都需要结合具体业务场景反复迭代。

欢迎有类似工程实践的同学在评论区交流。