FastMCP供应链优化器
A. 自定义实现 的 FastMCP 该项目展示了受Anthropic内部FastMCP系统启发的低延迟、多工具编排。
🎯 这表明了什么
- 自定义FastMCP实现：在每个LLM处理步骤进行多工具调用（非顺序）
- 实时事件处理：实时人工智能响应的供应链事件流
- 智能推荐：基于人工智能的库存优化，具有可操作的洞察力
- 实时Web界面：实时监控，界面美观
- 模块化工具架构：易于扩展和修改以适应不同的用例
🔧 关于FastMCP与MCP
FastMCP不是开源的 -这是Anthropic的内部实现。这个项目是 最小模拟 FastMCP的关键创新：
核心区别：并行工具调用
- 标准MCP：在1个LLM呼叫之间顺序交替→ 1 工具调用→ 1 LLM电话
- FastMCP：LLM处理的每个步骤都调用多个工具
- 这个实现：模拟FastMCP的方法，每个事件执行多个工具
FastMCP不是开源的，因此我受其启发构建了一个低延迟多工具编排堆栈的最小模拟，展示了LLM代理如何通过路由工具对实时供应链更新做出响应，并提供可操作的建议。
🚀 快速开始
1.安装依赖项
pip install -r requirements.txt2.运行应用程序
python3 flask_app.py3.打开浏览器
导航至 http://localhost:5000
4.替代方案：使用本地LLM
为了数据隐私和内部工具的使用，您可以使用将Gemini API替换为您自己的本地LLM 本地llm api:
# Clone and setup local LLM API
git clone https://github.com/ANSH-RIYAL/local-llm-api.git
cd local-llm-api
./run_server.sh
# Modify fastmcp_server.py to use local API instead of Gemini
# Replace GEMINI_API_KEY with CUSTOM_API_URL = "http://localhost:8050"🎮 如何使用
- 启动FastMCP服务器：点击“启动FastMCP服务器”以初始化AI代理
- 启动事件流：单击“启动事件流”开始处理供应链事件
- 监控结果：实时查看终端输出和行动建议
- 完成后停止：使用停止按钮优雅地关闭
🛠️ 已实施的工具
核心供应链工具
1. 获取库存状态
- 目的：检查所有仓库的当前库存水平
- 参数:
product_id（可选） - 退货：产品或所有产品的完整库存数据
- 示例:
{"product_id": "P001"}→ 返回仓库A/B/C库存水平
2. update_inventory
- 目的：修改仓库库存水平（增加/减少）
- 参数:
product_id,warehouse,quantity - 退货：成功状态和库存更改详细信息
- 示例:
{"product_id": "P001", "warehouse": "warehouse_A", "quantity": -10}
3. 计算传输
- 目的：在仓库之间移动库存
- 参数:
product_id,from_warehouse,to_warehouse,quantity - 退货：转账执行细节和新库存水平
- 示例:
{"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}
4. 预测库存
- 目的：预测产品何时缺货
- 参数:
product_id,warehouse - 退货：风险水平和预计缺货时间
- 示例:
{"product_id": "P001", "warehouse": "warehouse_A"}→ “高风险，1-2天”
5. 推荐订单
- 目的：建议重新订购数量和供应商
- 参数:
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]:
"""Description of what this tool does"""
# Implementation logic
return {"success": True, "result": "tool output"}- 在中注册工具
fastmcp_server.py:
Tool(
name="new_tool_function",
description="Description of what this tool does",
inputSchema={
"type": "object",
"properties": {
"param1": {"type": "string", "description": "Parameter 1"},
"param2": {"type": "integer", "description": "Parameter 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 # Product inventory data
│ └── events.csv # Supply chain events stream
├── templates/
│ └── index.html # Web interface
├── supply_chain_tools.py # Core business logic
├── fastmcp_server.py # Custom FastMCP implementation
├── flask_app.py # Web server and API
├── test_demo.py # Demo script
├── requirements.txt # Python dependencies
└── README.md # This file🎯 工作流示例
- 事件:
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建议和执行的行动
- 状态指示灯：服务器和事件流状态
- 活动进度：当前事件正在处理中
🚀 主要特点
- 实时处理：事件到达时已处理
- 智能推荐：人工智能驱动的决策
- 实时更新：Web界面实时更新
- 简单设置：最小的依赖关系和配置
- 可扩展：易于添加新工具和事件类型
- 隐私选项：可以使用本地LLM而不是云API
🎯 用例
- 供应链优化：实时库存管理
- 需求预测：人工智能驱动的股票预测
- 成本优化：供应商和定价分析
- 风险管理：预防和缓解缺货
🔄 场景修改
1.实时供应链优化器（流输入+实时代理校正）
当前实施情况: ✅ 部分执行
- ✅ 流式传输CSV事件
- ✅ 实时AI响应
- ✅ 基本库存工具
- ❌ 快速相关计算器
- ❌ 预测工具（ARIMA/指数平滑）
- ❌ 实时代理更正
很快可以添加什么:
# Add to supply_chain_tools.py
def calculate_correlation(self, product1: str, product2: str) -> Dict[str, Any]:
"""Calculate demand correlation between products"""
# Implementation using pandas correlation
def forecast_demand(self, product_id: str, periods: int) -> Dict[str, Any]:
"""Forecast demand using simple exponential smoothing"""
# Implementation using statsmodels
def recommend_reroute(self, from_supplier: str, to_supplier: str) -> Dict[str, Any]:
"""Recommend supply rerouting based on delays/costs"""
# Implementation with cost analysis对话示例:
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.交互式调查分析器（多代理和多工具）
需要修改:
# New tools in survey_tools.py
def extract_themes(self, responses: List[str]) -> Dict[str, Any]:
"""Extract common themes from survey responses"""
def compute_frequencies(self, data: pd.DataFrame) -> Dict[str, Any]:
"""Compute response frequencies and confidence intervals"""
def generate_summary_report(self, insights: Dict) -> Dict[str, Any]:
"""Generate client-facing summary reports"""对话示例:
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.临床分型助理（具有紧密延迟循环的工具选择）
需要修改:
# New tools in clinical_tools.py
def check_symptoms(self, symptoms: List[str]) -> Dict[str, Any]:
"""Check symptoms against medical database"""
def classify_risk(self, vitals: Dict) -> Dict[str, Any]:
"""Classify patient risk level"""
def score_triage_priority(self, risk: str, symptoms: List) -> Dict[str, Any]:
"""Score triage priority"""
def generate_doctor_note(self, patient_data: Dict) -> Dict[str, Any]:
"""Generate doctor notes"""对话示例:
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.电子商务定价代理（快速反馈循环）
需要修改:
# New tools in pricing_tools.py
def calculate_optimal_price(self, cost: float, margin: float, demand_factor: float) -> Dict[str, Any]:
"""Calculate optimal price using formula"""
def find_competitor_match(self, product_id: str) -> Dict[str, Any]:
"""Find nearest competitor product"""
def generate_markdown_explanation(self, price_change: Dict) -> Dict[str, Any]:
"""Generate markdown explanation for price changes"""对话示例:
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界面进行测试
切换到本地LLM
- 设置 本地llm api
- 修改
fastmcp_server.py使用本地API终结点 - 更新本地模型兼容性提示
📝 备注
- 这是一个 演示 使用模拟数据
- 服务器停止时，库存更改将保存回CSV
- 使用Gemini API免费等级（适用费率限制）
- 专为简单和教育目的而设计
- FastMCP不是开源的 -这是一个自定义实现
- 可以用本地LLM扩展数据隐私
🤝 贡献
请随时通过以下方式扩展此功能：
- 更复杂的人工智能模型
- 真正的数据库集成
- 其他供应链工具
- 增强的web界面功能
- 并行工具执行优化
- 实时数据流
🔗 相关项目
______________________________________________________________________
准备好用AI优化您的供应链了吗？ 启动服务器，观看魔术发生！ 🚀
*该项目演示了如何构建一个定制的类似FastMCP的系统，用于实时、多工具的人工智能编排。*