🚀 FastMCP供应链优化器
本项目基于Gemini AI，对FastMCP（模型上下文协议）进行了自定义实现，可用于实时供应链优化。该项目受Anthropic内部FastMCP系统的启发，展示了低延迟的多工具编排能力。
🚀 快速开始
1. 安装依赖
pip install -r requirements.txt
2. 运行应用程序
python3 flask_app.py
3. 打开浏览器
访问 http://localhost:5000
4. 替代方案：使用本地大语言模型
若出于数据隐私和内部工具使用的考虑，你可以使用 local-llm-api 将Gemini API替换为你自己的本地大语言模型：
# 克隆并设置本地大语言模型API
git clone https://github.com/ANSH-RIYAL/local-llm-api.git
cd local-llm-api
./run_server.sh
# 修改fastmcp_server.py以使用本地API而非Gemini
# 将GEMINI_API_KEY替换为CUSTOM_API_URL = "http://localhost:8050"
✨ 主要特性
- 自定义FastMCP实现：在每个大语言模型处理步骤中进行多工具调用（非顺序执行）
- 实时事件处理：处理供应链事件流，并通过实时AI响应
- 智能推荐：由AI驱动的库存优化，提供可操作的见解
- 实时Web界面：通过美观的用户界面进行实时监控和控制
- 模块化工具架构：易于扩展和修改，以适应不同的用例
🔧 关于FastMCP与MCP
FastMCP并非开源项目，它是Anthropic的内部实现。本项目是对FastMCP关键创新点的最小模拟：
核心区别：并行工具调用
- 标准MCP：按顺序交替进行1次大语言模型调用 → 1次工具调用 → 1次大语言模型调用
- FastMCP：在大语言模型处理的每个步骤中调用多个工具
- 本实现：针对每个事件模拟FastMCP的方法，执行多个工具
由于FastMCP未开源，我构建了一个低延迟多工具编排栈的最小模拟，展示了大语言模型代理如何通过路由工具对实时供应链更新做出可操作的建议。
💻 使用示例
基础用法
本项目的使用流程如下：
- 启动FastMCP服务器：点击“Start FastMCP Server”初始化AI代理
- 启动事件流：点击“Start Event Stream”开始处理供应链事件
- 监控结果：实时查看终端输出和行动建议
- 完成后停止：使用停止按钮正常关闭程序
高级用法
处理事件类型
- DEMAND_SPIKE：产品需求突然增加
- DELAY：供应商交货延迟
- COST_INCREASE：供应商价格变动
AI采取的行动
- 库存转移：在仓库之间转移库存
- 重新订购建议：建议新的订单和数量
- 缺货预测：预测产品何时会缺货
- 成本优化：分析供应商替代方案
📚 详细文档
🛠️ 已实现的工具
核心供应链工具
1. get_inventory_status
- 用途：检查所有仓库的当前库存水平
- 参数：
product_id（可选） - 返回值：产品或所有产品的完整库存数据
- 示例：
{"product_id": "P001"}→ 返回仓库A、B、C的库存水平
2. update_inventory
- 用途：修改仓库库存水平（增加/减少）
- 参数：
product_id、warehouse、quantity - 返回值：成功状态和库存变更详情
- 示例：
{"product_id": "P001", "warehouse": "warehouse_A", "quantity": -10}
3. calculate_transfer
- 用途：在仓库之间转移库存
- 参数：
product_id、from_warehouse、to_warehouse、quantity - 返回值：转移执行详情和新的库存水平
- 示例：
{"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}
4. predict_stockout
- 用途：预测产品何时会缺货
- 参数：
product_id、warehouse - 返回值：风险级别和预测的缺货时间线
- 示例：
{"product_id": "P001", "warehouse": "warehouse_A"}→ "高风险，1 - 2天"
5. recommend_reorder
- 用途：建议重新订购的数量和供应商
- 参数：
product_id、quantity - 返回值：包含成本计算的订单详情
- 示例：
{"product_id": "P001", "quantity": 50}→ "订单：从供应商X订购50件，单价5.50美元"
如何修改工具
添加新工具
- 在
supply_chain_tools.py中添加函数：
def new_tool_function(self, param1: str, param2: int) -> Dict[str, Any]:
"""描述此工具的作用"""
# 实现逻辑
return {"success": True, "result": "工具输出"}
- 在
fastmcp_server.py中注册工具：
Tool(
name="new_tool_function",
description="描述此工具的作用",
inputSchema={
"type": "object",
"properties": {
"param1": {"type": "string", "description": "参数1"},
"param2": {"type": "integer", "description": "参数2"}
},
"required": ["param1", "param2"]
}
)
- 在
handle_call_tool中添加处理程序：
elif name == "new_tool_function":
result = self.tools.new_tool_function(
arguments["param1"],
arguments["param2"]
)
🏗️ 架构
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Flask Web │ │ Custom │ │ Gemini AI │
│ Interface │◄──►│ FastMCP │◄──►│ (or Local │
│ │ │ Server │ │ LLM API) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ Event Stream │ │ Supply Chain │
│ (CSV Data) │ │ Tools │
└─────────────────┘ └─────────────────┘
📁 项目结构
FastMCP/
├── data/
│ ├── inventory.csv # 产品库存数据
│ └── events.csv # 供应链事件流
├── templates/
│ └── index.html # Web界面
├── supply_chain_tools.py # 核心业务逻辑
├── fastmcp_server.py # 自定义FastMCP实现
├── flask_app.py # Web服务器和API
├── test_demo.py # 演示脚本
├── requirements.txt # Python依赖项
└── README.md # 本文件
🎯 示例工作流程
- 事件：
DEMAND_SPIKE for P001 - 40 units - 分析：AI检查所有仓库的当前库存
- 预测：识别潜在的缺货风险
- 行动：建议从仓库B向仓库A转移库存
- 执行：更新库存并记录操作
示例对话流程
Event Stream → MCP Client: "DEMAND_SPIKE: P001, 40 units"
MCP Client → get_inventory_status: {"product_id": "P001"}
MCP Client → predict_stockout: {"product_id": "P001", "warehouse": "warehouse_A"}
MCP Client → calculate_transfer: {"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}
MCP Client → recommend_reorder: {"product_id": "P001", "quantity": 50}
MCP Client → User: "Transfer 20 units from B to A, reorder 50 units from Supplier X"
🔍 监控
- 终端输出：实时服务器日志和处理状态
- 操作日志：所有AI建议和执行的操作
- 状态指示器：服务器和事件流状态
- 事件进度：当前正在处理的事件
🚀 关键特性
- 实时处理：事件到达时立即处理
- 智能推荐：由AI驱动的决策
- 实时更新：Web界面实时更新
- 简单设置：依赖项和配置最少
- 可扩展：易于添加新工具和事件类型
- 隐私选项：可以使用本地大语言模型替代云API
🎯 用例
- 供应链优化：实时库存管理
- 需求预测：由AI驱动的库存预测
- 成本优化：供应商和定价分析
- 风险管理：预防和缓解缺货情况
🔄 场景修改
1. 实时供应链优化器（流式输入 + 实时代理校正）
当前实现情况：✅ 部分实现
- ✅ 流式CSV事件
- ✅ 实时AI响应
- ✅ 基本库存工具
- ❌ 快速相关性计算器
- ❌ 预测工具（ARIMA/指数平滑）
- ❌ 实时代理校正
近期可添加的功能：
# 添加到supply_chain_tools.py
def calculate_correlation(self, product1: str, product2: str) -> Dict[str, Any]:
"""计算产品之间的需求相关性"""
# 使用pandas相关性实现
def forecast_demand(self, product_id: str, periods: int) -> Dict[str, Any]:
"""使用简单指数平滑法预测需求"""
# 使用statsmodels实现
def recommend_reroute(self, from_supplier: str, to_supplier: str) -> Dict[str, Any]:
"""根据延迟/成本建议供应重新路由"""
# 实现成本分析
示例对话：
Event: "SUPPLIER_DELAY: Supplier X, 3 days"
MCP Client: "Analyzing impact on P001, P002, P003..."
Tools Called: [get_inventory_status, calculate_correlation, forecast_demand, recommend_reroute]
Response: "Reroute P001 from Supplier X to Supplier Y. P002 and P003 show 0.8 correlation - adjust P002 orders accordingly."
2. 交互式调查分析器（多代理 & 多工具）
需要修改的内容：
# survey_tools.py中的新工具
def extract_themes(self, responses: List[str]) -> Dict[str, Any]:
"""从调查响应中提取常见主题"""
def compute_frequencies(self, data: pd.DataFrame) -> Dict[str, Any]:
"""计算响应频率和置信区间"""
def generate_summary_report(self, insights: Dict) -> Dict[str, Any]:
"""生成面向客户的摘要报告"""
示例对话：
User: "Analyze 500 survey responses about Product X"
MCP Client: "Processing responses with multiple agents..."
Tools Called: [extract_themes, compute_frequencies, generate_summary_report]
Response: "Top themes: UI/UX (45%), Performance (32%), Price (23%). 78% satisfaction rate (±3% CI). Report generated."
3. 临床分诊助手（低延迟循环的工具选择）
需要修改的内容：
# clinical_tools.py中的新工具
def check_symptoms(self, symptoms: List[str]) -> Dict[str, Any]:
"""根据医学数据库检查症状"""
def classify_risk(self, vitals: Dict) -> Dict[str, Any]:
"""对患者风险级别进行分类"""
def score_triage_priority(self, risk: str, symptoms: List) -> Dict[str, Any]:
"""对分诊优先级进行评分"""
def generate_doctor_note(self, patient_data: Dict) -> Dict[str, Any]:
"""生成医生笔记"""
示例对话：
Patient Data: {"symptoms": ["chest pain", "shortness of breath"], "vitals": {"bp": "140/90"}}
MCP Client: "Analyzing patient data..."
Tools Called: [check_symptoms, classify_risk, score_triage_priority, generate_doctor_note]
Response: "HIGH RISK - Cardiac symptoms detected. Immediate triage required. Doctor note: 'Patient presents with chest pain and elevated BP...'"
4. 电子商务定价代理（快速反馈循环）
需要修改的内容：
# pricing_tools.py中的新工具
def calculate_optimal_price(self, cost: float, margin: float, demand_factor: float) -> Dict[str, Any]:
"""使用公式计算最优价格"""
def find_competitor_match(self, product_id: str) -> Dict[str, Any]:
"""查找最近的竞争对手产品"""
def generate_markdown_explanation(self, price_change: Dict) -> Dict[str, Any]:
"""为价格变化生成Markdown解释"""
示例对话：
Event: "COMPETITOR_PRICE_CHANGE: Product X, $25.99 → $22.99"
MCP Client: "Analyzing competitive landscape..."
Tools Called: [find_competitor_match, calculate_optimal_price, generate_markdown_explanation]
Response: "Competitor reduced price by 12%. Recommended action: Reduce price to $23.99. Explanation: 'We've adjusted our pricing to remain competitive while maintaining healthy margins...'"
🔧 开发
添加新工具
- 在
supply_chain_tools.py中添加函数 - 在
fastmcp_server.py中注册工具 - 根据需要更新系统提示
添加新事件类型
- 在
data/events.csv中添加事件 - 在
fastmcp_server.py中更新事件处理逻辑 - 使用Web界面进行测试
切换到本地大语言模型
- 设置 local-llm-api
- 修改
fastmcp_server.py以使用本地API端点 - 更新提示以适应本地模型
📄 许可证
本项目文档未提及许可证相关信息。
📝 注意事项
- 本项目是使用模拟数据的演示项目
- 服务器停止时，库存更改会保存回CSV文件
- 使用Gemini API免费层（适用速率限制）
- 设计目的是为了简单和教育
- FastMCP并非开源项目，本项目是自定义实现
- 可以使用本地大语言模型扩展以保护数据隐私
🤝 贡献
欢迎通过以下方式扩展本项目：
- 更复杂的AI模型
- 与真实数据库集成
- 更多供应链工具
- 增强Web界面功能
- 并行工具执行优化
- 实时数据流
🔗 相关项目
- local-llm-api：用于数据隐私的本地大语言模型API
- MCP-RAG：参考MCP实现
准备好使用AI优化你的供应链了吗？ 启动服务器，见证奇迹发生！🚀
本项目展示了如何构建一个类似FastMCP的系统，用于实时多工具AI编排。