🚀 MCP驱动的Agentic RAG系统实战指南

🌟 概述

Agentic RAG（代理驱动的检索增强生成）系统通过引入智能代理（AI Agent）来动态协调检索与生成过程，突破了传统RAG系统在检索策略上的静态性和被动性。而模型上下文协议（Model Context Protocol, MCP） 由Anthropic提出，是一种开放标准，旨在为AI应用连接外部数据源和工具提供统一、可扩展的接口，如同AI世界的“USB-C端口”。将两者结合，可以构建出能够智能路由查询、灵活调用多种工具（如向量数据库搜索、网络搜索、计算器等）的强大系统。

本指南将带你一步步实现一个基于MCP的Agentic RAG系统。该系统能根据查询内容，智能选择是检索向量数据库还是发起网络搜索（使用Firecrawl），并以Qdrant作为向量数据库，Cursor作为MCP客户端。

🧠 核心概念与优势

在深入实现细节之前，理解以下几个核心概念和它们带来的优势至关重要：

  1. Agentic RAG: 其核心是智能代理（AI Agent）的引入。这个代理具备规划推理能力，可以分析用户查询意图，动态决定检索策略（例如，是否需要重构查询、选择哪个数据源、是否需要进行多步检索）；工具集成能力，能够调用各种外部工具（如不同的向量数据库、搜索引擎、计算器API等）；以及记忆管理能力，利用短期记忆维护对话上下文，借助长期记忆存储和回顾历史交互信息，从而显著提升复杂查询处理的准确性和灵活性。
  2. Model Context Protocol (MCP): MCP采用客户端-服务器（Client-Server）模式，通过标准化协议将外部数据源和工具封装成统一的接口。AI应用（客户端）只需与MCP服务器通信，而无需关心底层工具的具体实现细节。这种架构带来了巨大的模块化优势和技术灵活性：开发者可以独立开发、扩展或替换MCP服务器，客户端代码无需改动；同时，它也极大地提升了互操作性，任何符合MCP标准的工具都可以被轻松集成到AI系统中。
  3. 协同价值: MCP的标准化接口天然契合Agentic RAG中智能代理需要灵活调用多种工具的需求。代理不再需要为每个工具编写特定的适配代码，只需通过MCP客户端发送标准化请求即可。这使得系统构建更简洁，代理的工具调用能力更容易扩展，从而支撑更复杂的应用场景。

🏗️ 系统架构设计

一个典型的基于MCP的Agentic RAG系统总体架构如下，它清晰地展示了客户端、MCP服务器以及外部工具之间的协作关系：

在设计MCP服务器提供的工具时，需要仔细权衡功能粒度。工具既不能过于庞大而失去模块化优势，也不能过于琐碎而导致系统复杂化和交互开销增加。

⚙️ 环境准备与依赖安装

安装必要的Python包

确保你的Python版本在3.11及以上。我们将使用mcp、qdrant-client等核心库。

# 安装核心MCP和向量数据库客户端
pip install mcp qdrant-client

# 安装LlamaIndex用于RAG管道构建（如果采用类似方案）
pip install llama-index

# 安装网络请求和环境变量管理库
pip install requests python-dotenv

获取并配置API密钥

系统需要一些外部服务的API密钥来正常运行。

  1. Firecrawl API密钥: 访问Firecrawl官网注册并获取API密钥，用于网页抓取。
  2. 博查搜索/BrightData API密钥 (可选): 如果你计划使用博查搜索或BrightData作为网络搜索工具，也需要注册相应服务并获取密钥。

将获取到的密钥存储在项目根目录下的.env文件中，以便安全地读取：

# .env 文件示例
FIRECRAWL_API_KEY="your_firecrawl_api_key_here"
BOCHAAI_API_KEY="your_bochaai_api_key_here" # 或其他搜索引擎的密钥
# 其他配置...

启动Qdrant向量数据库

使用Docker快速运行一个Qdrant实例是最简单的方式。

# 使用Docker运行Qdrant，映射端口并持久化数据
docker run -p 6333:6333 -p 6334:6334 \
  -v $(pwd)/qdrant_storage:/qdrant/storage:z \
  qdrant/qdrant

这条命令会在本地启动Qdrant服务，客户端可以通过http://localhost:6333进行连接。

🖥️ 实现MCP服务器

MCP服务器是系统的核心，它封装了具体的工具逻辑，并向客户端提供标准化的接口。

初始化MCP服务器

我们使用FastMCP来快速创建服务器实例。

# server.py
from mcp.server.fastmcp import FastMCP
import os
from dotenv import load_dotenv
from your_rag_code import *  # 导入你自定义的RAG相关代码，例如检索器、嵌入模型等

# 加载环境变量
load_dotenv()

# 创建MCP服务器实例，指定名称、主机和端口
mcp = FastMCP("MCP-RAG-app", host="127.0.0.1", port=8080, timeout=30)

创建向量数据库查询工具

这个工具用于处理与机器学习FAQ相关的查询，从Qdrant向量数据库中检索最相关的文档。

@mcp.tool()
def machine_learning_faq_retrieval_tool(query: str) -> str:
    """
    从机器学习FAQ集合中检索最相关的文档。
    当用户询问关于机器学习（ML）的问题时，使用此工具。

    参数:
        query (str): 用户查询，用于检索最相关的文档

    返回:
        context (str): 从向量数据库检索到的最相关文档
    """
    # 输入验证：确保查询是字符串
    if not isinstance(query, str):
        raise ValueError("查询必须是字符串")
    
    # 初始化检索器：指定向量数据库和嵌入模型
    # 假设 Retriever, QdrantVDB, EmbedData 已在 your_rag_code 中定义
    retriever = Retriever(QdrantVDB("ml_faq_collection"), EmbedData())
    
    # 执行搜索
    response = retriever.search(query)
    
    return response

创建网络搜索工具

当查询不属于特定领域（如机器学习）或需要最新信息时，使用此工具进行网络搜索。

@mcp.tool()
def firecrawl_web_search_tool(query: str) -> list[str]:
    """
    使用Firecrawl在给定主题上搜索信息。
    当用户询问特定主题或问题，且该问题与通用机器学习不相关时，使用此工具。

    参数:
        query (str): 用户查询，用于搜索信息

    返回:
        context (list[str]): 最相关的网页搜索结果列表
    """
    # 输入验证：确保查询是字符串
    if not isinstance(query, str):
        raise ValueError("查询必须是字符串")
    
    import requests
    import json
    
    # 获取Firecrawl API密钥
    api_key = os.getenv("FIRECRAWL_API_KEY")
    if not api_key:
        raise ValueError("请在.env文件中设置FIRECRAWL_API_KEY")
    
    # 配置Firecrawl API请求
    url = "https://api.firecrawl.dev/v1/search" # 假设的Firecrawl搜索端点
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Content-Type': 'application/json'
    }
    payload = json.dumps({
        "q": query,
        "limit": 10  # 限制返回结果数量
    })
    
    # 发送请求
    response = requests.post(url, headers=headers, data=payload)
    response.raise_for_status()  # 如果请求失败则抛出异常
    
    # 解析返回的JSON数据并提取文本内容
    search_results = response.json()
    # 假设返回结构中含有 `results` 列表，每个结果有 `content` 字段
    organic_results = [result.get('content', '') for result in search_results.get('results', [])]
    
    return organic_results

运行MCP服务器

在脚本末尾添加代码来启动服务器。

# server.py (续)
if __name__ == "__main__":
    print("启动MCP服务器在 http://127.0.0.1:8080, 端口 8080")
    mcp.run()

使用命令python server.py来启动你的MCP服务器。

📱 配置MCP客户端（Cursor）

Cursor作为MCP客户端，需要配置以连接到我们刚创建的MCP服务器。

  1. 打开Cursor编辑器。
  2. 进入Settings（设置）菜单。
  3. 找到MCP（Model Context Protocol）设置部分。
  4. 点击Add New Global MCP Server（添加新的全局MCP服务器）。
  5. 在JSON配置中，填入服务器的详细信息：
{
  "mcpServers": {
    "mcp-rag-app": {
      "command": "python",
      "args": ["/absolute/path/to/your/server.py"], // 替换为你的server.py的绝对路径
      "host": "127.0.0.1",
      "port": 8080,
      "timeout": 30000
    }
  }
}

保存配置后，Cursor将会尝试连接到你的MCP服务器。如果连接成功，你就可以在Cursor中调用machine_learning_faq_retrieval_tool和firecrawl_web_search_tool等工具了。

🔧 实现智能代理与查询路由

为了让系统能够自动判断该使用向量搜索还是网络搜索，我们需要一个智能代理。虽然Cursor本身可以作为客户端调用工具，但更复杂的路由逻辑通常需要借助像LangGraph或LangChain这样的代理框架来构建一个独立的客户端应用。

定义代理工作流程

这个代理的工作流程基于ReAct（Reasoning + Acting）模式：

  1. 理解查询：代理接收用户查询。
  2. 决策（路由）：代理根据查询内容决定最合适的工具。
    • 如果查询是关于机器学习（ML）的特定知识，选择machine_learning_faq_retrieval_tool。
    • 如果查询是关于一般性知识、新闻或非常具体的最新信息，选择firecrawl_web_search_tool。
  3. 执行工具：代理调用选择的工具并获取结果。
  4. 合成答案：代理将工具返回的结果整合到提示中，并生成最终回答返回给用户。

示例代码结构（基于LangGraph）

以下是一个简化的示例，展示了如何使用LangGraph来创建这样一个代理：

# agent_client.py (示例概念)
from langgraph.graph import Graph
from langgraph.prebuilt import create_react_agent
from your_mcp_client_module import MultiServerMCPClient  # 假设有一个MCP客户端类
from langchain.tools import Tool
import asyncio

async def main():
    # 1. 初始化MCP客户端并连接到服务器
    client = MultiServerMCPClient.from_config('mcp_config.json')
    async with client as mcp_client:
        # 2. 获取MCP服务器提供的工具并将其转换为LangGraph可用的格式
        mcp_tools_list = await mcp_client.get_tools_for_langgraph()
        # 假设 get_tools_for_langgraph 返回一个Tool对象的列表
        
        # 3. 创建ReAct智能体，指定模型和工具
        agent = create_react_agent(
            model=your_llm_model, # 替换为你选择的LLM模型
            tools=mcp_tools_list,
            prompt=your_system_prompt # 包含路由决策指示的系统提示词
        )
        
        # 4. 运行代理
        user_query = "你的用户查询问题"
        final_state = await agent.ainvoke({"user_query": user_query})
        print(final_state['final_output'])

if __name__ == "__main__":
    asyncio.run(main())

系统提示词（System Prompt） 需要精心设计，以指导代理进行路由决策。例如：

“你是一个有帮助的AI助手。你可以使用工具。根据用户的问题决定是搜索机器学习知识库还是进行网络搜索。如果问题涉及机器学习概念、模型、算法或历史数据，使用机器学习FAQ工具。如果问题涉及新闻、实时事件或非常广泛的主题，使用网络搜索工具。”

🧪 端到端测试与演示

完成所有组件开发后，进行端到端测试以确保系统协同工作。

  1. 启动服务：确保Qdrant向量数据库和MCP服务器都在运行。

    # 终端1: 运行Qdrant (如果尚未运行)
docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant

# 终端2: 运行MCP服务器
python /absolute/path/to/your/server.py

  2. 配置客户端：确保Cursor已正确配置并连接到MCP服务器。

  3. 进行查询测试：

    • 测试向量搜索：在Cursor中输入一个机器学习相关的问题，例如“What is overfitting in machine learning?”。观察代理是否正确地调用了machine_learning_faq_retrieval_tool并返回了相关的FAQ答案。
    • 测试网络搜索：在Cursor中输入一个寻求最新信息的问题，例如“What were the latest results of the Paris Olympics 2024?”。观察代理是否正确地调用了firecrawl_web_search_tool并返回了网络搜索结果。

  4. 观察与调试：密切关注MCP服务器和客户端的日志输出，这有助于理解工具调用的流程和识别任何潜在问题。

🚀 性能优化与扩展建议

构建基本系统后，可以考虑以下方面来优化性能和扩展功能：

  • 缓存机制：为向量索引和文档解析结果实现缓存层。可以基于文档内容哈希和解析参数（如chunk_size）创建唯一键，避免重复解析和索引创建，显著提升性能。
  • 多代理协作：对于极其复杂的任务，可以考虑引入多种类型的代理（如路由代理、查询计划代理）协同工作，由路由代理初步分配任务，查询计划代理分解复杂查询并协调执行。
  • 扩展工具集：除了搜索，MCP服务器可以集成任何其他工具，例如计算器、代码执行器、电子邮件发送接口或企业内部系统API，使代理的能力无限延伸。
  • 高级检索技术：可以考虑实现RAG Fusion或GraphRAG等高级检索技术来进一步提升检索质量和答案准确性。

🎯 总结

通过本指南，功构建了一个基于Model Context Protocol (MCP) 的Agentic RAG系统。这个系统不仅能够通过向量数据库检索特定领域知识，还能在需要时智能地切换到网络搜索以获取最新信息，所有这些都是通过智能代理动态路由实现的。

MCP作为AI应用和外部工具之间的“通用接口”，极大地简化了集成复杂度，提升了系统的模块化和可扩展性。而Agentic RAG的引入，则赋予了系统推理和决策能力，使其从被动的检索工具转变为主动的问题解决者。

未来，可以探索将更多样化的工具集成到MCP服务器中，或者尝试更复杂的多代理协作架构，以构建更强大、更智能的AI应用。

最后更新: