AI导哈

检索增强生成系统

MCP RAG Agent是一个结合LangGraph代理和模型上下文协议（MCP）的生产级检索增强生成系统，支持混合搜索、上下文响应和自动化评估，适用于构建可靠的上下文感知AI代理。

  • by luisrodriguesphd
  • Python
  • 本地部署
  • 4 星标
  • 更新于
在 GitHub 查看 luisrodriguesphd/mcp-rag-agent

使用说明

MCP RAG代理

将LangGraph代理与模型上下文协议（MCP）集成的生产就绪RAG系统。通过MongoDB向量和全文搜索使用往复式排名融合（RRF）进行混合搜索，使用COSTAR提示进行固定响应，以及基于RAGAS的自动评估，以构建可靠、上下文感知的AI代理。

概述

MCP RAG Agent是一个复杂的问答系统，它：

  • 使用混合搜索从政策语料库中查找相关文档
  • 使用LangGraph代理来推理和检索信息
  • 通过模型上下文协议（MCP）集成模块化、可重用的组件
  • 使用COSTAR提示框架确保有根据的响应
  • 使用MongoDB Atlas矢量搜索存储和检索文档
  • 使用RAGAS指标提供全面的评估工具

主要特点

  • MCP集成：工具暴露和特工沟通的标准化协议
  • 混合搜索：使用往复式秩融合（RRF）将向量相似性和关键字搜索相结合
  • 语义搜索：使用OpenAI嵌入的基于矢量的文档检索
  • 文本搜索：带词干和相关性评分的全文关键字搜索
  • MongoDB 阿特拉斯：可扩展的矢量存储，具有高效的相似性搜索功能
  • 接地响应：严格的基于上下文的回答，没有幻觉
  • 成本激励：结构化的快速设计，实现一致的高质量输出
  • LangGraph代理：智能工具使用的推理和动作周期
  • 自动评估：基于RAGAS的答案质量评估指标

建筑

系统架构图说明了两个主要工作流程：

  1. 文档索引流程 （设置阶段）：使用OpenAI处理、嵌入文档，并将其存储在MongoDB Atlas Vector Search中，并使用适当的索引进行高效检索。

  2. 问答流程 （运行时）：用户查询触发LangGraph ReAct代理，该代理使用MCP工具通过语义搜索搜索相关文档，然后根据检索到的上下文制定有根据的响应。

此外，该系统还包括图中未显示的第三个工作流程：

  1. 评估流程 （质量保证）：系统为预定义的测试问题生成答案，并使用RAGAS指标（相关性、相似性、正确性）对其进行评估，以确保响应质量和准确性。

项目结构

mcp-rag-agent/
├── data/
│   ├── ingested_documents/         # Source documents (policies)
│   │   └── policies/
│   │       ├── 1 - Remote Working.txt
│   │       ├── 2 - Expenses.txt
│   │       ├── 3 - Annual Leave.txt
│   │       ├── 4 - IT Security.txt
│   │       └── 5 - Sustainability.txt
│   └── evaluation_documents/       # Test cases for evaluation
│       └── expected_behaviour.xlsx
├── evaluation/                     # Automated testing and metrics
│   ├── main.py                     # Main evaluation orchestration script
│   ├── answer_generator.py         # Generates answers using the agent
│   ├── metrics_evaluator.py        # Evaluates answers using RAGAS metrics
│   ├── metrics.py                  # RAGAS metrics wrapper and definitions
│   ├── results/                    # Evaluation output (CSV files)
│   └── README.md                   # Evaluation module documentation
├── src/mcp_rag_agent/
│   ├── agent/                      # LangChain agent implementation
│   │   ├── create_agent.py         # Agent creation and configuration
│   │   ├── prompts/                # COSTAR-based system prompts
│   │   │   ├── __init__.py         # Prompts module exports
│   │   │   └── system_prompt.py    # System prompt definitions
│   │   ├── utils/                  # Agent utility functions
│   │   │   ├── mcp_rag_agent_creator.py  # MCP-enabled agent factory
│   │   │   └── rag_agent_creator.py      # Base RAG agent factory
│   │   └── README.md               # Agent module documentation
│   ├── embeddings/                 # Document processing and indexing
│   │   ├── embedding_generator.py  # OpenAI embeddings generation
│   │   ├── index_documents.py      # Document indexing pipeline
│   │   ├── semantic_search.py      # Vector similarity search
│   │   ├── hybrid_search.py        # Hybrid search combining vector + text
│   │   └── README.md               # Embeddings module documentation
│   ├── mcp_server/                 # MCP server implementation
│   │   ├── server.py               # FastMCP server with tools
│   │   ├── tools.py                # MCP tool implementations
│   │   └── README.md               # MCP server documentation
│   ├── mongodb/                    # Database client
│   │   ├── client.py               # MongoDB wrapper with vector search
│   │   └── README.md               # MongoDB module documentation
│   └── core/                       # Configuration and utilities
│       ├── config.py               # Environment-based configuration
│       └── log_setup.py            # Logging configuration
├── tests/                          # Tests
│   └── unit_tests                  # Unit tests
├── .env.example                    # Example environment configuration
├── .gitignore                      # Git ignore patterns
├── requirements.txt                # Production dependencies
├── requirements_dev.txt            # Development dependencies
├── setup.py                        # Package installation configuration
├── start.cmd                       # Windows startup script
└── README.md                       # This file

快速开始

先决条件

  • Python 3.8+
  • MongoDB Atlas帐户（用于矢量搜索）
  • OpenAI API密钥

安装

  1. 克隆仓库:
git clone <repository-url>
cd mcp-rag-agent
  1. 跑吧 start 文件:
# Windows:
start.cmd

# Linux/macOS:
chmod +x start.sh
./start.sh

此脚本将自动执行以下操作：

  • 安装并升级pip
  • 创建并激活虚拟环境
  • 安装所有开发依赖项
  • 以可编辑模式安装软件包
  1. 配置环境变量:
cp .env.example .env
# Edit .env with your settings

设置工作流

  1. 索引文档:
python -m mcp_rag_agent.embeddings.index_documents

这将：

  • 从以下位置读取文档 data/ingested_documents/
  • 使用OpenAI生成嵌入
  • 在MongoDB Atlas中存储向量
  • 创建矢量搜索索引
  1. 测试MCP服务器 （可选-需要Node.js）：
mcp dev src/mcp_rag_agent/mcp_server/server.py

这将打开一个UI来测试 search_documents 工具和其他资源。

  1. 运行代理:
python -m mcp_rag_agent.agent.create_agent

这将运行一个演示查询，显示正在运行的代理。

  1. 评估性能 （可选）：
python evaluation/main.py

使用RAGAS指标运行自动评估。

使用示例

基本代理查询

import asyncio
from mcp_rag_agent.agent.create_agent import create_mcp_rag_agent
from mcp_rag_agent.agent.prompts import system_prompt
from mcp_rag_agent.core.config import config

async def main():
    # Create agent
    agent = await create_mcp_rag_agent(
        system_prompt=system_prompt,
        config=config
    )
    
    # Query the agent
    result = await agent.ainvoke({
        "messages": [{
            "role": "user",
            "content": "What is the remote working policy?"
        }]
    })
    
    # Get the answer
    answer = result["messages"][-1].content
    print(answer)

asyncio.run(main())

直接语义搜索

import asyncio
from mcp_rag_agent.mongodb.client import MongoDBClient
from mcp_rag_agent.embeddings.embedding_generator import EmbeddingGenerator
from mcp_rag_agent.embeddings.semantic_search import SemanticSearch
from mcp_rag_agent.core.config import config

async def main():
    # Setup
    mongo_client = MongoDBClient(config.db_url, config.db_name)
    mongo_client.connect()
    
    embedder = EmbeddingGenerator(
        api_key=config.model_api_key,
        model=config.embedding_model
    )
    
    search = SemanticSearch(mongo_client, embedder)
    
    # Search
    results = await search.search(
        query="annual leave entitlement",
        limit=3
    )
    
    for doc in results:
        print(f"File: {doc['file_name']}")
        print(f"Score: {doc['score']:.3f}")
        print(f"Content: {doc['content'][:200]}...\n")
    
    mongo_client.disconnect()

asyncio.run(main())

混合搜索（推荐）

import asyncio
from mcp_rag_agent.mongodb.client import MongoDBClient
from mcp_rag_agent.embeddings.embedding_generator import EmbeddingGenerator
from mcp_rag_agent.embeddings.hybrid_search import HybridSearch
from mcp_rag_agent.core.config import config

async def main():
    # Setup
    mongo_client = MongoDBClient(config.db_url, config.db_name)
    mongo_client.connect()
    
    embedder = EmbeddingGenerator(
        api_key=config.model_api_key,
        model=config.embedding_model
    )
    
    hybrid = HybridSearch(
        mongo_client=mongo_client,
        embedding_generator=embedder,
        default_collection=config.db_vector_collection
    )
    
    # Perform hybrid search (combines semantic + keyword matching)
    results = await hybrid.search(
        query="What are the sustainability initiatives?",
        limit=5,
        semantic_weight=0.7  # 70% semantic, 30% keyword (default)
    )
    
    for doc in results:
        print(f"RRF Score: {doc['rrf_score']:.4f}")
        print(f"Vector Rank: {doc['vector_rank']}, Text Rank: {doc['text_rank']}")
        print(f"Content: {doc['content'][:200]}...\n")
    
    mongo_client.disconnect()

asyncio.run(main())

为新文档编制索引

import asyncio
from mcp_rag_agent.embeddings.index_documents import index_documents
from mcp_rag_agent.core.config import config

async def main():
    await index_documents(
        directory_path="data/ingested_documents",
        config=config
    )

asyncio.run(main())

模块文档

每个模块都有详细的文档：

  • 代理：具有MCP集成的LangGraph ReAct代理
  • MCP服务器：提供RAG工具的FastMCP服务器
  • MongoDB：具有矢量、文本和混合搜索功能的数据库客户端
  • 嵌入：文档索引、语义搜索和混合搜索
  • 评估：使用RAGAS指标进行自动化测试

配置

配置通过两层进行管理：

  1. 环境变量(.env)：大多数设置都是通过环境变量配置的，尽管只有外部依赖关系包含在 .env.sample 文件。
  2. 代码配置(src/mcp_rag_agent/core/config.py)：一些高级设置直接在 Config 类，如文本生成参数（温度等）

注： 要修改这些设置，请编辑 src/mcp_rag_agent/core/config.py 直接。这 Config 类加载环境变量，并为所有配置参数提供默认值。

关键技术

发展

运行测试

pytest tests/

编码结构

  • 遵循Python最佳实践和PEP 8
  • 对所有函数使用类型提示
  • 将文档字符串添加到公共API
  • 保持模块的重点和连贯性

添加新功能

  1. 新MCP工具:

    • 添加 @mcp.tool() 装饰功能 server.py
    • MCP服务器README中的文档
    • 测试用 mcp dev

  2. 新建文档类型:

    • 更新 index_documents.py 处理新格式
    • 确保元数据得到保留
    • 重新索引文档

  3. 新指标:

    • 添加 evaluation/metrics.py
    • 更新评估器以计算和保存指标
    • 评估文档README

评估

该项目包括使用RAGAS的综合评估工具：

python evaluation/evaluator.py

计算指标:

结果保存到 evaluation/results/ 带有时间戳。

故障排除

常见问题

MongoDB连接失败:

  • 验证MongoDB Atlas集群是否正在运行
  • 查看Atlas中的IP白名单
  • 验证中的连接URI .env

MCP服务器无法启动:

  • 确保MongoDB已连接
  • 检查OpenAI API密钥是否有效
  • 验证是否已安装所有依赖项

没有搜索结果:

  • 跑 index_documents.py 填充数据库
  • 检查MongoDB Atlas中是否存在向量索引
  • 验证嵌入尺寸是否匹配

代理不调用工具:

  • 检查MCP服务器是否可访问
  • 审查系统提示鼓励使用工具
  • 如有需要，提高模型温度

评估错误:

  • 确保 expected_behaviour.xlsx 存在
  • 检查OpenAI API配额
  • 验证评估模型是否可访问

性能考量

  • 索引：每份文档约1-2秒（取决于文档大小）
  • 查询：每个查询约2-5秒（嵌入+搜索+生成）
  • 矢量搜索：对于高达10万份文档的收藏，次秒级
  • 批量操作：使用 insert_documents() 用于批量索引

最佳实践

  1. 提示工程：对所有提示使用COSTAR框架
  2. 错误处理：始终优雅地处理连接失败
  3. 日志记录：使用结构化日志进行调试
  4. 测试：在发生重大变化后运行评估
  5. 向量索引：在安装过程中创建，而不是在运行时创建
  6. 连接池：重用MongoDB客户端实例
  7. API费率限制：为OpenAI调用实现指数回退

安全

  • 永不承诺 .env 文件到版本控制
  • 定期旋转API键
  • 使用MongoDB Atlas IP白名单
  • 对生产部署实施速率限制
  • 在处理之前对用户输入进行消毒

贡献

  1. 分叉存储库
  2. 创建要素分支
  3. 进行更改
  4. 添加新功能的测试
  5. 更新文档
  6. 提交拉取请求

许可证

麻省理工学院

相关 MCP