aigroup-rag-mcp

RAG MCP Server - 智能知识库检索系统

---

📋 目录

• 项目简介
• 功能特性
• 技术栈
• 快速开始
• 架构设计
• 使用文档
• 配置说明
• MCP服务器
• 检索模式
• 开发指南
• 常见问题
• 贡献指南

---

🎯 项目简介

RAG MCP Server 是一个基于 LlamaIndex 框架和 **MCP（Model Context Protocol）**协议的企业级检索增强生成（RAG）系统。该系统提供了强大的知识库管理和智能检索能力，支持多种检索模式，可通过 MCP 协议与 AI 助手（如 Claude Desktop）无缝集成。

核心优势

✅ 多模式检索：支持向量检索、BM25检索和混合检索
✅ 模块化设计：清晰的代码组织，易于维护和扩展
✅ 双传输支持：STDIO 和 HTTP/SSE 两种传输方式
✅ 中文优化：使用 jieba 分词，完美支持中文文档
✅ 生产就绪：支持 Qdrant 云存储，可横向扩展
✅ 开发友好：完整的日志系统和错误处理

---

✨ 功能特性

🔍 智能检索系统

• 向量检索（Vector Search）

  • 基于语义相似度的密集向量检索
  • 使用 SiliconFlow Embedding API
  • 支持 Qwen3-Embedding-8B 等先进模型

• BM25检索（Sparse Retrieval）

  • 基于关键词匹配的稀疏检索
  • 使用 jieba 进行中文分词
  • 支持自定义停用词过滤

• 混合检索（Hybrid Search） 🌟

  • 融合向量检索和BM25检索
  • 使用 QueryFusionRetriever 进行结果融合
  • 提供最佳的检索效果

📦 向量存储

• 本地存储（Simple Vector Store）

  • 轻量级本地文件存储
  • 适合开发和小规模部署

• Qdrant云存储

  • 支持 Qdrant 云向量数据库
  • 生产级性能和可靠性
  • 支持分布式部署

🔌 MCP协议支持

• STDIO传输

  • 标准输入/输出流通信
  • 适合本地集成（如 Claude Desktop）

• HTTP/SSE传输

  • 基于 FastAPI 的 HTTP 服务器
  • 支持 Server-Sent Events (SSE)
  • 适合远程访问和微服务架构

🛠️ MCP工具集

📦 MCP资源集

📄 文档支持

支持的文档格式：

• .txt - 纯文本文件
• .md - Markdown 文档
• .pdf - PDF 文档
• .docx - Word 文档

---

🔧 技术栈

核心框架

• LlamaIndex - RAG 框架
• FastAPI - Web 服务器框架
• MCP - Model Context Protocol
• Uvicorn - ASGI 服务器

AI 模型

• LLM: DeepSeek Chat (deepseek-chat)
• Embedding: Qwen3-Embedding-8B (via SiliconFlow)

检索技术

• llama-index-retrievers-bm25 - BM25 检索器
• jieba - 中文分词
• PyStemmer - 英文词干提取

向量数据库

• Qdrant - 云向量数据库
• SimpleVectorStore - 本地向量存储

工具库

• loguru - 日志系统
• python-dotenv - 环境变量管理
• httpx - HTTP 客户端
• watchdog - 文件监控

---

🚀 快速开始

环境要求

• Python 3.10 或更高版本
• pip 包管理器
• （可选）Qdrant 云服务账号

1. 安装依赖

# 克隆项目
git clone <repository-url>
cd aigroup-rag-mcp

# 安装依赖
pip install -r requirements.txt

2. 配置环境变量

创建 .env 文件并配置必要参数：

# ========== API 密钥 ==========
# DeepSeek API（必需）
DEEPSEEK_API_KEY=your_deepseek_api_key
DEEPSEEK_BASE_URL=https://api.deepseek.com/v1

# SiliconFlow API（必需，用于 Embedding）
SILICONFLOW_API_KEY=your_siliconflow_api_key
SILICONFLOW_BASE_URL=https://api.siliconflow.cn/v1

# ========== 模型配置 ==========
DEEPSEEK_MODEL=deepseek-chat
EMBEDDING_MODEL=Qwen/Qwen3-Embedding-8B

# ========== 检索配置 ==========
# 检索模式：vector, bm25, hybrid
RETRIEVAL_MODE=hybrid
DEFAULT_TOP_K=3

# BM25 配置
USE_BM25=True
BM25_TOP_K=5
BM25_LANGUAGE=chinese

# ========== 向量存储配置 ==========
USE_QDRANT=True
QDRANT_URL=your_qdrant_url
QDRANT_API_KEY=your_qdrant_api_key
QDRANT_COLLECTION_NAME=knowledge_base

# ========== 其他配置 ==========
CHUNK_SIZE=1024
CHUNK_OVERLAP=200
LLM_TEMPERATURE=0.1
LOG_LEVEL=INFO

3. 准备知识库文档

将你的文档放入 my_knowledge_base 目录：

my_knowledge_base/
├── document1.txt
├── document2.md
├── guide.pdf
└── manual.docx

4. 构建索引

# 使用 Python 脚本
from src.core import RAGSystem

rag = RAGSystem()
rag.build_index('./my_knowledge_base')

5. 启动 MCP 服务器

STDIO 模式（适合 Claude Desktop）

python mcp_server.py

在 Claude Desktop 配置文件中添加：

{
  "mcpServers": {
    "rag-knowledge-base": {
      "command": "python",
      "args": ["D:/aigroup-rag-mcp/mcp_server.py"],
      "env": {}
    }
  }
}

HTTP 模式（适合远程访问）

# 启动 HTTP 服务器
python mcp_http_server.py

# 或使用 Uvicorn
uvicorn mcp_http_server:app --host 0.0.0.0 --port 8000

访问 http://localhost:8000 查看服务状态。

6. 测试查询

from src.core import RAGSystem

rag = RAGSystem()
result = rag.query("什么是RAG技术？", mode="hybrid", top_k=3)
print(result['answer'])

---

🏗️ 架构设计

项目结构

aigroup-rag-mcp/
├── src/                          # 源代码目录
│   ├── config/                   # 配置模块
│   │   └── config.py            # 系统配置类
│   ├── core/                     # 核心业务逻辑
│   │   └── rag_system.py        # RAG系统核心实现
│   ├── embeddings/               # 嵌入模型
│   │   └── siliconflow_embedding.py
│   ├── tokenizers/               # 分词器
│   │   └── jieba_tokenizer.py   # 中文分词器
│   ├── storage/                  # 存储管理
│   │   ├── vector_store.py      # 向量存储管理
│   │   └── bm25_store.py        # BM25存储管理
│   ├── mcp/                      # MCP服务器
│   │   └── handlers/            # MCP处理器
│   │       ├── tools_handler.py     # 工具处理器
│   │       ├── resources_handler.py # 资源处理器
│   │       └── prompts_handler.py   # 提示词处理器
│   └── utils/                    # 工具函数
│       ├── logger.py            # 日志配置
│       └── patches.py           # 猴子补丁
├── storage/                      # 存储目录
│   ├── vector/                  # 向量索引存储
│   │   └── simple/              # 本地向量存储
│   └── bm25/                    # BM25索引存储
├── doc/                          # 文档目录
│   ├── BM25_FEATURE_README.md   # BM25功能文档
│   ├── MCP_HTTP_SERVER_README.md # HTTP服务器文档
│   └── REFACTORING_GUIDE.md     # 重构指南
├── logs/                         # 日志文件
├── my_knowledge_base/           # 知识库目录
├── mcp_server.py                # STDIO MCP服务器
├── mcp_http_server.py           # HTTP MCP服务器
├── requirements.txt             # 依赖清单
├── .env                         # 环境变量配置
├── .gitignore                   # Git忽略文件
└── README.md                    # 本文档

模块说明

1. 配置模块 (src/config/config.py)

• 集中管理所有系统配置参数
• 支持环境变量和默认值
• 提供配置验证和显示功能

2. 核心模块 (src/core/rag_system.py)

• RAG系统的核心实现
• 索引构建、加载、查询和删除
• 集成向量存储和BM25存储管理器
• 支持多种检索模式

3. 嵌入模块 (src/embeddings/siliconflow_embedding.py)

• SiliconFlow API的嵌入实现
• 支持批量和单个文本的向量化
• 自动批处理和错误恢复

4. 分词模块 (src/tokenizers/jieba_tokenizer.py)

• 中文分词器实现
• 支持中英文混合文本分词
• 停用词过滤和词干提取

5. 存储模块

• 向量存储 (src/storage/vector_store.py)
  • 支持本地存储和Qdrant云存储
  • 提供统一的向量存储接口
• BM25存储 (src/storage/bm25_store.py)
  • 管理BM25检索器的创建、加载和持久化

6. MCP模块 (src/mcp/handlers/)

• tools_handler.py: 工具调用处理
• resources_handler.py: 资源读取处理
• prompts_handler.py: 提示词处理

---

📚 使用文档

Python API 使用

基础使用

from src.core import RAGSystem

# 创建RAG系统实例
rag = RAGSystem()

# 构建索引
success = rag.build_index("./my_knowledge_base")

# 查询（使用默认混合模式）
result = rag.query("什么是RAG技术？")
print(result['answer'])

# 查看源文档
for source in result['sources']:
    print(f"评分: {source['score']}")
    print(f"内容: {source['text']}")
    print(f"元数据: {source['metadata']}")

高级使用

from src.core import RAGSystem
from src.config import Config

# 创建RAG系统
rag = RAGSystem(config=Config)

# 使用不同检索模式查询
result_vector = rag.query("问题", mode="vector", top_k=5)
result_bm25 = rag.query("问题", mode="bm25", top_k=5)
result_hybrid = rag.query("问题", mode="hybrid", top_k=5)

# 获取索引统计信息
stats = rag.get_index_stats()
print(f"向量索引: {stats['vector_index_exists']}")
print(f"BM25索引: {stats['bm25_index_exists']}")
print(f"检索模式: {stats['retrieval_mode']}")

# 删除索引
rag.delete_index()

RAGSystem API 参考

build_index(data_dir, force_rebuild=False)

构建知识库索引。

参数：

• data_dir (str): 文档目录路径
• force_rebuild (bool): 是否强制重建索引

返回：

• bool: 构建是否成功

load_index()

从存储加载索引。

返回：

• bool: 加载是否成功

query(question, top_k=None, mode=None)

查询知识库。

参数：

• question (str): 用户问题
• top_k (int, optional): 检索的文档数量
• mode (str, optional): 检索模式 (vector/bm25/hybrid)

返回：

• dict: 包含答案和元数据的字典

{
    "success": bool,
    "answer": str,
    "sources": List[dict],
    "mode": str,
    "error": Optional[str]
}

get_index_stats()

获取索引统计信息。

返回：

• dict: 索引统计信息

delete_index()

删除所有索引。

返回：

• bool: 删除是否成功

---

⚙️ 配置说明

核心配置参数

API配置

模型配置

检索配置

向量存储配置

---

🌐 MCP服务器

STDIO 传输模式

适用场景：

• 本地开发和测试
• Claude Desktop等本地客户端
• 命令行工具集成

启动方式：

python mcp_server.py

HTTP/SSE 传输模式

适用场景：

• 远程访问MCP服务
• Web应用集成
• 微服务架构
• 多客户端同时访问

启动方式：

# 开发模式
python mcp_http_server.py

# 生产模式（多worker）
uvicorn mcp_http_server:app --workers 4 --host 0.0.0.0 --port 8000

API端点：

• GET / - 服务器信息
• GET /health - 健康检查
• GET /sse - SSE通信端点
• POST /messages - 消息发送端点
• GET /docs - API文档

详细文档：MCP HTTP服务器使用指南

---

🔍 检索模式

1. 向量检索 (vector)

特点：

• ✅ 语义理解能力强
• ✅ 能匹配相似概念
• ✅ 适合模糊查询
• ❌ 对精确关键词匹配不够敏感

适用场景：概念性问题、语义相似查询

result = rag.query("如何提升模型性能", mode="vector")

2. BM25检索 (bm25)

特点：

• ✅ 精确关键词匹配
• ✅ 速度快，无需embedding
• ✅ 适合专业术语查找
• ❌ 无法理解语义

适用场景：精确术语查找、关键词搜索

result = rag.query("BM25算法", mode="bm25")

3. 混合检索 (hybrid) 🌟

特点：

• ✅ 结合语义理解和关键词匹配
• ✅ 更全面的检索覆盖
• ✅ 提供最佳检索效果

适用场景：默认推荐、通用查询场景

result = rag.query("RAG技术的核心思想", mode="hybrid")

检索模式对比

详细文档：BM25混合检索功能使用指南

---

👨‍💻 开发指南

模块化架构

本项目采用模块化设计，各模块职责清晰：

配置层 (config) → 业务层 (core) → 存储层 (storage)
                        ↓
                   接口层 (mcp)

详细文档：代码重构指南

扩展开发

添加新的嵌入模型

from llama_index.core.embeddings import BaseEmbedding

class CustomEmbedding(BaseEmbedding):
    def _get_query_embedding(self, query: str) -> List[float]:
        # 实现查询嵌入
        pass
    
    def _get_text_embedding(self, text: str) -> List[float]:
        # 实现文本嵌入
        pass

添加新的MCP工具

在 src/mcp/handlers/tools_handler.py 中：

def list_tools(self) -> list[Tool]:
    return [
        Tool(
            name="custom_tool",
            description="自定义工具描述",
            inputSchema={...}
        )
    ]

---

❓ 常见问题

Q1: 如何选择检索模式？

A: 建议默认使用 hybrid 混合检索模式。对于特定场景：

• 语义理解 → vector
• 精确术语 → bm25
• 通用查询 → hybrid

Q2: BM25支持中文吗？

A: 完全支持！系统使用 jieba 进行中文分词，在配置中设置 BM25_LANGUAGE=chinese 即可。

Q3: 如何使用本地向量存储？

A: 在 .env 文件中设置 USE_QDRANT=False。

Q4: 索引构建失败怎么办？

A: 检查：

1. 文档目录是否存在且不为空
2. API密钥是否正确配置
3. 查看日志文件 logs/rag_system.log
4. 确认网络连接正常

Q5: 如何更新知识库？

A: 两种方式：

# 方式1：强制重建
rag.build_index("./my_knowledge_base", force_rebuild=True)

# 方式2：删除后重建
rag.delete_index()
rag.build_index("./my_knowledge_base")

Q6: HTTP服务器如何部署？

A: 参考 MCP HTTP服务器使用指南 中的生产部署章节。

---

🤝 贡献指南

欢迎贡献代码、报告问题或提出建议！

贡献流程

1. Fork 本仓库
2. 创建特性分支 (git checkout -b feature/AmazingFeature)
3. 提交更改 (git commit -m 'Add some AmazingFeature')
4. 推送到分支 (git push origin feature/AmazingFeature)
5. 打开 Pull Request

开发规范

• 遵循 PEP 8 代码风格
• 添加必要的注释和文档字符串
• 编写单元测试
• 更新相关文档

---

📄 许可证

本项目采用 MIT 许可证。详见 LICENSE 文件。

---

📞 联系方式

如有问题或建议，请通过以下方式联系：

• 提交 Issue
• 发送 Pull Request
• 邮件咨询（待补充）

---

🙏 致谢

感谢以下开源项目：

• LlamaIndex - RAG框架
• FastAPI - Web框架
• Qdrant - 向量数据库
• jieba - 中文分词

---