Agentic RAG：基于多文档的 AI Agent 智能体构建指南

什么是 Agentic RAG

经典 RAG（Retrieval-Augmented Generation）应用的范式与架构已经非常流行，开发者甚至可以在很短的时间内借助成熟框架开发一个简单能用的 RAG 应用。其基本流程是：用户问题被输入 RAG 系统，应用执行检索操作，从被向量化的文档中检索相关知识块，最后送入大语言模型（LLM）进行合成响应。

然而，让我们考虑这样一个应用场景：企业中存在大量不同来源与类型的文档（在实际业务中并不一定代表'文件'，也可以是某种非文件形态的信息，比如存放在关系型数据库 RDBMS 中的数据），现在需要在这些文档之上构建一个依赖于它们的、知识密集型的应用或工具。这些需求通常包括：

基于全局理解的文档后回答问题：例如，对某知识内容进行总结摘要？
跨文档与知识库的回答问题：例如，比较不同文档内容的区别？
结合非知识工具的复合场景：例如，从文档提取产品介绍并发送给指定客户？

这种复杂需求的场景如果使用经典 RAG 架构，通过 chunks + 向量 + top_K 检索来获得并插入上下文，直接让 LLM 来给出答案，显然是不现实的。经典 RAG 在回答文档相关的事实性问题时可以工作得不错，但是实际的知识应用并不总是这种类型！当然你也可以借助一些改进的 RAG 范式来提高应用场景的适应性，比如 RAPTOR（基于文档树的多级检索机制，有利于回答从细节到高层理解的多级问题），但在一些跨文档或者需要结合工具的场景仍然无法胜任。

这里介绍 Agentic RAG 方案：一种基于 AI Agent 的方法，借助 Agent 的任务规划与工具能力，来协调完成对多文档的、多类型的问答需求。它既能提供 RAG 的基础查询能力，也能提供基于 RAG 之上更多样与复杂任务能力。概念架构如下：

Agentic RAG 架构图

在这里的 Agentic RAG 架构中：

RAG 应用退化成一个 Agent 使用的知识工具：你可以针对一个文档/知识库构建多种不同的 RAG 引擎，比如使用向量索引来回答事实性问题；使用摘要索引来回答总结性问题；使用知识图谱索引来回答需要更多关联性的问题等。
ToolAgent 设置：在单个文档/知识库的多个 RAG 引擎之上设置一个 ToolAgent，把 RAG 引擎作为该 Agent 的工具（tools），并利用 LLM 的能力由 ToolAgent 在自己'负责'的文档内使用这些 tools 来回答问题。
TopAgent 管理：设置一个总的顶级代理 TopAgent 来管理所有的低阶 ToolAgent，将 ToolAgent 看作自己的工具，仍然利用 LLM 来规划、协调、执行用户问题的回答方案。

以下使用 LlamaIndex 来实现这个架构（如果你是 LangChain 用户，也完全可以读懂并参考实现）。

实现 Agentic RAG

让我们来一步步实现简单的 Agentic RAG。

准备测试文档

首先这里准备三个 RAG 相关的测试 PDF 文档，其名称与路径分别保存。当然，在实际应用中，这里文档数量可以扩展到非常大（后面会看到针对大量文档的一个优化方法）：

names = ['c-rag', 'self-rag', 'kg-rag']  
files = ['../../data/c-rag.pdf', '../../data/self-rag.pdf', '../../data/kg-rag.pdf']

准备创建 Tool Agent 的函数

创建一个针对单个文档生成 Tool Agent 的函数，在这个函数中，将对一个文档创建两个索引与对应的 RAG 引擎：

针对普通事实性问题的向量索引与 RAG 引擎。
针对更高层语义理解的总结类问题的摘要索引与 RAG 引擎。

最后，我们把这两个引擎作为一个 Agent 可使用的两个 tool，构建一个 Tool Agent 返回。

# ...省略 import 部分与准备 llm 部分...

# 采用 chroma 向量数据库
chroma = chromadb.HttpClient(host="localhost", port=8000)
collection = chroma.get_or_create_collection(name="agentic_rag")   
vector_store = ChromaVectorStore(chroma_collection=collection)  

# 创建针对某个文档的 tool_agent
def create_tool_agent(file, name):
    
    # 文档拆分
    print(f'Starting to create tool agent for [{name}]...\n')
    docs = SimpleDirectoryReader(input_files=[file]).load_data()
    splitter = SentenceSplitter(chunk_size=500, chunk_overlap=50)
    nodes = splitter.get_nodes_from_documents(docs)
      
    # 创建向量索引，并做持久保存
    if not os.path.exists(f"./storage/{name}"):
        print('Creating vector index...\n')
        storage_context = StorageContext.from_defaults(vector_store=vector_store)
        vector_index = VectorStoreIndex(nodes, storage_context=storage_context)
        vector_index.storage_context.persist(persist_dir=f"./storage/{name}")
    else:
        print('Loading vector index...\n')
        storage_context = StorageContext.from_defaults(persist_dir=f"./storage/{name}", vector_store=vector_store)
        vector_index = load_index_from_storage(storage_context=storage_context)
  
    # 创建基于向量的查询引擎
    query_engine = vector_index.as_query_engine(similarity_top_k=5)
  
    # 创建摘要索引与对应的查询引擎
    summary_index = SummaryIndex(nodes)
    summary_engine = summary_index.as_query_engine(response_mode="tree_summarize")  
  
    # 将 RAG 引擎转化为两个 tool
    query_tool = QueryEngineTool.from_defaults(
        query_engine=query_engine,
        name=f'query_tool',
        description=
    )
    summary_tool = QueryEngineTool.from_defaults(
        query_engine=summary_engine,
        name=,
        description=
    )  
  
    
    tool_agent = ReActAgent.from_tools([query_tool, summary_tool], verbose=,  
                                  system_prompt=)
     tool_agent

这部分代码主要目的就是把两个查询的 RAG 引擎包装成工具（一个是 query_tool，用于回答事实性问题；一个是 summary_tool 用于回答总结性问题，当然你还可以构建更多类型的引擎），最后构建一个 ReAct 思考范式的 AI Agent，并把构建的 RAG tools 插入。

如果你了解 LlamaIndex，可能会使用路由 RouteQueryEngine 来代替这里的 Agent，实现接近的功能。但是要注意，Router 与 Agent 是有区别的：路由仅仅是起到一个'选择'工具与'转发'的作用，并不会做多次迭代；而 Agent 则会观察工具返回的结果，且有可能会使用多个工具通过迭代来完成任务。

批量创建 Tool Agent

有了上面的函数后，就可以批量创建好这些文档的 Tool Agent。这里把每一个文档名字和对应的 Agent 保存在一个 dict 中：

# 创建不同文档的 agent
print('===============================================\n')
print('Creating tool agents for different documents...\n')
tool_agents_dict = {}
for name, file in zip(names, files):  
    tool_agent = create_tool_agent(file, name)  
    tool_agents_dict[name] = tool_agent

创建 Top Agent

最后，我们需要创建一个顶层的 Top Agent，这个 Agent 的作用是接收客户的请求问题，然后规划这个问题的查询计划，并使用工具来完成，而这里的工具就是上面创建好的多个 Tool Agent：

# 首先将 Tool Agent 进行'工具化'
print('===============================================\n')
print('Creating tools from tool agents...\n')
all_tools = []  

for name in names:  
    agent_tool = QueryEngineTool.from_defaults(  
            # 注意，Agent 本身也是一种 Query Engine，所以直接转为 tool
           query_engine=tool_agents_dict[name],  

            # 这个工具的名字
            name=f"tool_{name.replace("-", "")}",  

            # 描述这个工具的作用和使用方法
            description=f"Use this tool if you want to answer any questions about {name}."
    )  

    all_tools.append(agent_tool)  

# 创建 Top Agent
print('Creating top agent...\n')
top_agent = OpenAIAgent.from_tools(tools=all_tools, verbose=True, system_prompt="""You are an agent designed to answer queries over a set of given papers.Please always use the tools provided to answer a question.Do not rely on prior knowledge.DO NOT fabricate answer""" )

注意这里我们创建的 Top Agent 使用了 OpenAIAgent，而不是 ReActAgent，这也展示了这种架构的灵活性：不同 Agent 可以按需使用不同的推理范式。

测试

现在来简单测试这个 Top Agent，并观察其执行的过程：

top_agent.chat_repl()

输入一个问题：Please introduce Retrieval Evaluator in C-RAG pattern?

注意观察这里红线与绿色部分内容，可以看出 Agent 的'思考'过程：

在 TopAgent 这一层，由于我们使用了 OpenAIAgent，其是通过 OpenAI 的 function calling 来实现，因此这里显示 LLM 要求进行函数调用，需要调用 tool_crag，输入参数为"Retrieval Evaluator in C-RAG pattern"。而这里的函数名 tool_crag，也就是后端 Tool Agent 的名称。
然后来到 Tool Agent 层，Tool Agent 收到请求后，通过 ReAct 范式的思考过程，决定需要调用 query_tool 工具，也就是通过向量索引进行响应的 RAG 引擎。在调用这个引擎后，获得了返回内容（observation 的内容）。收到返回后 Tool Agent 通过观察与推理，认为可以回答这个问题，因此 Tool Agent 运行结束，并返回结果给 Top Agent。
Top Agent 收到函数调用的结果后，认为无需再次进行其他函数调用，因此直接输出了结果，整个迭代过程结束。

当然，你也可以自行测试更复杂的文档任务，比如：要求对比两个文档中某个知识点的区别等。

进一步优化 Agentic RAG

上面我们只用了三个文档，构建了针对他们的 Tool Agent。那么如果这里的文档数量是几十或者几百，过多的 Tool Agent 作为 Tools 塞给 Top Agent 进行推理选择时会带来一些问题：

LLM 产生困惑并推理错误的概率会提高。
过多的 Tools 信息导致上下文过大，成本与延迟增加。

一种可行的方法是：**利用 RAG 的思想对 Tools 进行检索，即只把本次输入问题语义相关的 Tools（即这里的多个 ToolAgent）交给 Top Agent 使用。**这里借助 LlamaIndex 中的 Object Index 来实现：Object Index 可以对任意 Python 对象构建向量化的索引，并通过输入问题来检索出相关的 Objects。

现在可以对上面的代码做简单的改造，给 Top Agent 在推理时增加 tools 检索功能，从而能够缩小 tools 选择的范围。只需要在创建 Top Agent 之前针对 tools 创建一个 Object Index 的检索器用来根据输入问题检索相关的 tools：

# 创建工具检索器
print('===============================================\n')
print('Creating tool retrieve index...\n')
obj_index = ObjectIndex.from_objects(all_tools, index_cls=VectorStoreIndex,)
tool_retriever = obj_index.as_retriever(similarity_top_k=5, verbose=True)

然后将创建 Top Agent 的代码做简单的修改，不再传入 all_tools，而是传入 tools 检索器：

......  
top_agent = OpenAIAgent.from_tools(tool_retriever=tool_retriever,                                   verbose=True,  
                                   system_prompt="""You are an agent designed to answer queries over a set of given papers.Please always use the tools provided to answer a question.Do not rely on prior knowledge.""")  
.......

现在如果你继续测试这个 Agent，会发现仍然可以达到相同的效果。当然，如果你需要验证这里检索出来的 tools 正确性，可以直接对 tool_retriever 调用检索方法来观察（输入相同的自然语言问题）输出的 tools 信息：

tools_needed = tool_retriever.retrieve("What is the Adaptive retrieval in the c-RAG?")  
print('Tools needed to answer the question:')
for tool in tools_needed:  
    print(tool.metadata.name)

Agentic RAG 总结与最佳实践

相对于更适用于对几个文档进行简单查询的经典 RAG 应用，Agentic RAG 的方法通过更具有自主能力的 AI Agent 来对其进行增强，具备了极大的灵活性与扩展性，几乎可以完成任意基于知识的复杂任务：

基于 RAG 之上的 Tool Agent 将不再局限于简单的回答事实性的问题，通过扩展更多的后端 RAG 引擎，可以完成更多的知识型任务。比如：整理、摘要生成、数据分析、甚至借助 API 访问外部系统等。
Top Agent 管理与协调下的多个 Tool Agent 可以通过协作完成联合型的任务。比如对两个不同文档中的知识做对比与汇总，这也是经典问答型的 RAG 无法完成的任务类型。

实施建议

成本控制：在 Top Agent 阶段引入工具检索（Object Index）是控制 Token 消耗的关键。随着文档库规模扩大，务必避免将所有工具暴露给主 Agent。
错误处理：ReActAgent 和 OpenAIAgent 在处理失败时行为不同，建议在系统提示词中明确定义失败时的回退策略，例如当所有工具都未找到相关信息时，应如何回复用户。
性能监控：记录每次 Agent 调用的工具链路径，分析哪些工具组合最常被使用，以便优化索引结构和工具描述。

Agentic RAG 协作示意图

通过上述架构与实践，开发者可以构建出适应企业级复杂知识场景的智能体系统，真正实现从'检索 - 生成'到'规划 - 执行'的跨越。

Agentic RAG：基于多文档的 AI Agent 智能体构建指南