MCP Rag

🚀 MCP - RAG：用于商业分析的智能AI编排

本项目是一个轻量级演示，展示了如何将 模型上下文协议（Model Context Protocol，MCP） 与 检索增强生成（Retrieval - Augmented Generation，RAG） 相结合，为商业分析编排多智能体AI工作流。

🚀 快速开始

本项目展示了如何构建能够实现以下功能的 智能AI系统：

1. 编排多个智能体：MCP服务器协调不同的专业工具。
2. 检索商业知识：RAG提供上下文感知的信息检索。
3. 进行统计分析：通过自然语言查询实现自动化数据分析。
4. 保持模块化：易于更换大语言模型（LLM）后端并添加新功能。

✨ 主要特性

• 基于MCP的协调：多个专业服务器协同工作。
• 商业分析工具：支持均值、标准差、相关性、线性回归等分析。
• RAG知识库：包含商业术语、政策和分析指南。
• 模块化设计：便于使用新工具进行扩展或更换LLM后端。
• 自然语言接口：可以提出如“第一季度的平均收入是多少？”这样的问题。

📦 安装指南

📋 前提条件

• Python 3.8及以上版本
• Google Gemini API密钥（有免费套餐） - 用于未来的LLM集成
• 对MCP和RAG概念有基本了解

🛠️ 安装步骤

1. 克隆仓库：

   git clone https://github.com/ANSH-RIYAL/MCP-RAG.git
   cd MCP-RAG
   

2. 安装依赖项：

   pip install -r requirements.txt
   

3. 设置环境变量： 使用Gemini API（默认）：

   export LLM_MODE="gemini"
   export GEMINI_API_KEY="your-gemini-api-key"
   

   使用自定义本地主机API：

   export LLM_MODE="custom"
   export CUSTOM_API_URL="http://localhost:8000"
   export CUSTOM_API_KEY="your-api-key"  # 可选
   

💻 使用示例

快速演示

运行演示脚本以查看两个MCP服务器的实际运行情况：

python main.py

这将展示：

• 商业分析工具对示例数据的处理。
• RAG对商业术语的知识检索。
• 系统如何协同工作。
• LLM与所选后端的集成。

LLM后端选择

系统支持两种LLM后端：

选项1：Google Gemini API（默认）

export LLM_MODE="gemini"
export GEMINI_API_KEY="your-gemini-api-key"
python main.py

选项2：自定义本地主机API

export LLM_MODE="custom"
export CUSTOM_API_URL="http://localhost:8000"
export CUSTOM_API_KEY="your-api-key"  # 可选
python main.py

自定义API要求：

• 必须支持与OpenAI兼容的聊天完成端点（/v1/chat/completions）。
• 应接受工具/函数调用格式。
• 预计在本地主机8000端口运行（可配置）。

对话场景

运行对话场景脚本以查看实际使用示例：

python test_scenarios.py

这将展示LinkedIn帖子场景，说明非技术用户如何与系统交互。

商业分析工具

系统提供以下分析功能：

• 数据探索：获取数据集信息和示例数据。
• 统计分析：计算均值、标准差并支持过滤。
• 相关性分析：查找变量之间的关系。
• 预测建模：使用线性回归进行预测。

RAG知识检索

可通过以下方式访问商业知识：

• 术语定义：查找商业概念。
• 政策信息：检索公司程序。
• 分析指南：获取数据解释的上下文。

📚 详细文档

场景与用例

场景1：销售分析

经理：“第一季度的平均收入是多少？”
MCP - RAG系统：
1. 分析服务器：calculate_mean(column='earnings', filter_column='quarter', filter_value='Q1-2024')
   → 收入均值：101666.67
2. RAG服务器：get_business_terms(term='earnings')
   → 收入：部门或公司在给定时期内产生的总收入
3. 回复：“2024年第一季度的平均收入：101,667美元”

场景2：绩效相关性

经理：“销售额和费用之间的相关性是多少？”
MCP - RAG系统：
1. 分析服务器：calculate_correlation(column1='sales', column2='expenses')
   → 销售额和费用之间的相关性：0.923
2. 回复：“相关性：0.923（强正相关）”

场景3：预测建模

经理：“建立一个模型，根据销售额和员工数量预测收入”
MCP - RAG系统：
1. 分析服务器：linear_regression(target_column='earnings', feature_columns=['sales', 'employees'])
   → 线性回归结果：
      目标：收入
      特征：['sales', 'employees']
      截距：15000.00
      销售额系数：0.45
      员工数量系数：1250.00
      决定系数（R²）：0.987
2. 回复：“已创建模型，R² = 0.987”

场景4：商业知识

经理：“利润率是什么意思？”
MCP - RAG系统：
1. RAG服务器：get_business_terms(term='profit margin')
   → 利润率：扣除费用后作为利润保留的收入百分比，计算公式为 (收入 - 费用) / 收入
2. 回复：“利润率：扣除费用后作为利润保留的收入百分比”

场景5：政策信息

经理：“预算分配政策是什么？”
MCP - RAG系统：
1. RAG服务器：get_company_policies(policy_type='budget')
   → 预算分配：营销部门占总预算的25%，工程部门占30%，销售部门占45%
2. 回复：“预算分配：营销部门25%，工程部门30%，销售部门45%”

定制指南

为您的组织定制

步骤1：替换示例数据

1. 更新商业数据：将 data/sample_business_data.csv 替换为您的实际数据。
   • 确保分析工具使用的列是数值类型。
   • 添加用于过滤的分类列。
   • 包含用于趋势分析的时间列。
2. 更新知识库：将 data/business_knowledge.txt 替换为您组织的：
   • 商业术语和定义。
   • 公司政策和程序。
   • 分析指南和最佳实践。

步骤2：添加自定义分析工具

需修改的文件：src/servers/business_analytics_server.py

1. 添加新工具：在 handle_list_tools() 函数（约第29行）中，将新工具添加到工具列表中：

   @server.list_tools()
   async def handle_list_tools() -> ListToolsResult:
       return ListToolsResult(
           tools=[
               # ... 现有工具 (calculate_mean, calculate_std, calculate_correlation, linear_regression) ...
               Tool(
                   name="your_custom_analysis",
                   description="Your custom analysis tool",
                   inputSchema={
                       "type": "object",
                       "properties": {
                           "parameter": {"type": "string"}
                       },
                       "required": ["parameter"]
                   }
               )
           ]
       )
   

2. 实现工具逻辑：在 handle_call_tool() 函数（约第140行）中，添加相应的处理程序：

   elif name == "your_custom_analysis":
       parameter = arguments["parameter"]
       # 此处添加您的自定义分析逻辑
       result = f"Custom analysis result for {parameter}"
       return CallToolResult(
           content=[TextContent(type="text", text=result)]
       )
   

步骤3：扩展RAG功能

需修改的文件：src/servers/rag_server.py

1. 添加新的知识源：修改 load_business_knowledge() 函数（约第25行）以包含：
   • 数据库连接。
   • 文档处理（PDF、Word文档等）。
   • API集成（Salesforce、HubSpot等）。
2. 添加新的RAG工具：在 handle_list_tools() 函数（约第50行）中，添加新工具：

   Tool(
       name="your_custom_rag_tool",
       description="Your custom knowledge retrieval tool",
       inputSchema={
           "type": "object",
           "properties": {
               "query": {"type": "string"}
           },
           "required": ["query"]
       }
   )
   

3. 实现RAG工具逻辑：在 handle_call_tool() 函数（约第90行）中，添加处理程序：

   elif name == "your_custom_rag_tool":
       query = arguments["query"]
       # 此处添加您的自定义RAG逻辑
       result = f"Custom RAG result for {query}"
       return CallToolResult(
           content=[TextContent(type="text", text=result)]
       )
   

步骤4：集成LLM后端

需创建的文件：src/servers/llm_server.py（新文件） 系统已经包含一个灵活的LLM客户端（src/core/llm_client.py），支持Gemini和自定义本地主机API。

1. 使用现有的LLM客户端：src/core/gemini_rag_agent.py 中的 FlexibleRAGAgent 已经支持：
   • Google Gemini API。
   • 自定义本地主机API（与OpenAI兼容的格式）。
2. 创建自定义LLM服务器（可选）：如果您需要一个专门的MCP服务器来进行LLM操作：

   import asyncio
   from mcp.server import Server
   from mcp.server.stdio import stdio_server
   from mcp.types import Tool, TextContent, CallToolResult
   
   server = Server("llm-server")
   
   @server.list_tools()
   async def handle_list_tools():
       return ListToolsResult(
           tools=[
               Tool(
                   name="process_natural_language",
                   description="Convert natural language to tool calls",
                   inputSchema={
                       "type": "object",
                       "properties": {
                           "query": {"type": "string"}
                       },
                       "required": ["query"]
                   }
               )
           ]
       )
   
   @server.call_tool()
   async def handle_call_tool(name: str, arguments: dict):
       if name == "process_natural_language":
           query = arguments["query"]
           # 与OpenAI、Gemini或本地模型集成
           # 将自然语言转换为适当的工具调用
           return CallToolResult(
               content=[TextContent(type="text", text=f"Processed: {query}")]
           )
   

3. 添加到requirements.txt：

   openai>=1.0.0
   google-genai>=0.3.0
   httpx>=0.24.0
   

步骤5：添加新的数据源

需修改的文件：src/servers/business_analytics_server.py 和 src/servers/rag_server.py

1. 数据库连接器：添加连接以下数据库的工具：
   • PostgreSQL、MySQL、SQLite。
   • MongoDB、Redis。
   • 数据仓库（Snowflake、BigQuery）。
2. API集成：连接到商业系统：
   • CRM系统（Salesforce、HubSpot）。
   • 营销平台（Google Analytics、Facebook Ads）。
   • 财务系统（QuickBooks、Xero）。

当前工具实现

• 商业分析工具（src/servers/business_analytics_server.py）：
  • calculate_mean - 计算数值列的平均值。
  • calculate_std - 计算标准差。
  • calculate_correlation - 查找变量之间的关系。
  • linear_regression - 构建预测模型。
  • get_data_info - 获取数据集信息。
• RAG工具（src/servers/rag_server.py）：
  • get_business_terms - 查找商业定义。
  • get_company_policies - 检索政策信息。
  • search_business_knowledge - 进行通用知识搜索。
• LLM集成（src/core/llm_client.py）：
  • FlexibleRAGAgent - 支持Gemini和自定义本地主机API。
  • LLMClient - 处理两个后端的API通信。
  • 工具调用和对话管理。

模块化架构的好处

模块化设计允许您：

• 交换组件：更换任何服务器而不影响其他服务器。
• 添加功能：插入新工具而无需重写现有代码。
• 独立扩展：在不同的机器上运行不同的服务器。
• 根据用例定制：仅使用您需要的工具。

示例扩展

添加情感分析

需创建的文件：src/servers/sentiment_analysis_server.py

# 创建sentiment_analysis_server.py
@server.list_tool()
async def analyze_sentiment(text: str) -> CallToolResult:
    # 与情感分析API集成
    # 返回情感分数和见解

添加预测功能

需修改的文件：src/servers/business_analytics_server.py

# 添加到handle_list_tools()函数中
Tool(
    name="time_series_forecast",
    description="Forecast future values using time series analysis",
    inputSchema={
        "type": "object",
        "properties": {
            "column": {"type": "string"},
            "periods": {"type": "integer"}
        }
    }
)

添加文档处理功能

需创建的文件：src/servers/document_processor_server.py

# 创建document_processor_server.py
@server.list_tool()
async def process_document(file_path: str) -> CallToolResult:
    # 从PDF、Word文档等中提取文本
    # 添加到知识库中

架构

项目结构

MCP - RAG/
├── data/
│   ├── sample_business_data.csv    # 用于分析的商业数据集
│   └── business_knowledge.txt      # RAG知识库
├── src/
│   └── servers/
│       ├── business_analytics_server.py  # 统计分析工具
│       └── rag_server.py                 # 知识检索工具
├── main.py                         # 演示和编排脚本
├── test_scenarios.py               # 对话场景
├── requirements.txt                # 依赖项
└── README.md                       # 本文件

关键组件

1. 商业分析服务器：提供统计分析工具的MCP服务器。
2. RAG服务器：用于商业知识检索的MCP服务器。
3. 编排层：协调服务器和LLM（未来）。
4. 数据层：示例商业数据和知识库。

配置

环境变量

示例数据

系统包含：

• 季度商业数据：四个季度的销售、营销、工程指标。
• 商业知识库：术语、政策和分析指南。

用例

商业领导者

• 无代码分析：用自然语言询问商业数据相关问题。
• 快速洞察：无需技术专业知识即可获得统计分析结果。
• 上下文感知报告：将数据分析与商业知识相结合。

数据团队

• 模块化架构：易于添加新的分析工具。
• LLM集成：可进行自然语言查询处理。
• 可扩展框架：为特定需求构建自定义智能体。

AI工程师

• MCP协议：学习现代AI编排模式。
• RAG实现：了解知识检索系统。
• 智能体设计：构建多智能体AI工作流。

未来增强功能

计划功能

• [ ] LLM集成：连接Gemini、OpenAI或本地模型。
• [ ] 自然语言查询：处理复杂的商业问题。
• [ ] 高级分析：时间序列分析、聚类、预测。
• [ ] Web界面：为非技术用户提供用户友好的仪表板。
• [ ] 实时数据：连接实时数据源。
• [ ] 自定义知识库：上传公司特定的文档。

集成可能性

• 本地LLM API：使用 [Local LLM API](https://github.com/ANSH - RIYAL/local - llm - api) 运行开源模型。
• 数据库连接器：连接SQL数据库、数据仓库。
• API集成：Salesforce、HubSpot、Google Analytics。
• 文档处理：PDF、DOCX、电子邮件分析。

🤝 贡献

这是构建智能AI系统的基础，欢迎贡献：

• 新的分析工具：添加统计方法、机器学习模型。
• 知识库扩展：商业领域、特定行业的内容。
• LLM集成：支持不同的AI模型。
• 文档：教程、用例、最佳实践。

📄 许可证

本项目采用MIT许可证，您可以自由使用和修改以用于自己的项目！

🔗 相关项目

• [Local LLM API](https://github.com/ANSH - RIYAL/local - llm - api)：在本地运行开源大语言模型。
• MCP协议：官方文档

准备好构建自己的智能AI系统了吗？ 从这个基础开始，根据您的特定需求进行扩展。模块化设计使其易于添加新功能，同时保持清晰的架构。

#智能AI #MCP #RAG #商业分析 #开源AI