项目介绍
FastMCP供应链优化器
这是一个使用Gemini AI进行实时供应链优化的自定义实现。该项目展示了受Anthropic内部FastMCP系统启发的低延迟、多工具编排。
🎯 本项目演示了什么
- 自定义FastMCP实现：在每个LLM处理步骤中调用多个工具（非顺序）
- 实时事件处理：供应链事件流与实时AI响应
- 智能推荐：通过AI驱动的库存优化提供可操作见解
- 实时Web界面：实时监控和控制，具有美观的UI
- 模块化工具架构：易于扩展和修改以适应不同用例
🔧 关于FastMCP与MCP的区别
FastMCP不是开源的——它是Anthropic的内部实现。这个项目是对FastMCP关键创新的一个最小模拟：
核心差异：并行工具调用
- 标准MCP：在1个LLM调用→1个工具调用→1个LLM调用之间顺序交替
- FastMCP：在LLM处理的每一步调用多个工具
- 此实现：模拟FastMCP的方法，每个事件执行多个工具
FastMCP不是开源的，所以我构建了一个基于它的低延迟多工具编排堆栈的最小模拟——展示了一个LLM代理如何通过路由工具对实时供应链更新做出可操作建议的响应。
🚀 快速开始
1. 安装依赖项
pip install -r requirements.txt
2. 运行应用程序
python3 flask_app.py
3. 打开浏览器
导航到 http://localhost:5000
4. 可选：使用本地LLM
为了数据隐私和内部工具使用，你可以用local-llm-api替换Gemini API：
# 克隆并设置本地LLM 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服务器：点击“启动FastMCP服务器”初始化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"]
)
📊 发生了什么
处理的事件类型：
- DEMAND_SPIKE：产品需求突然增加
- DELAY：供应商交货延误
- COST_INCREASE：供应商价格变化
AI行动：
- 库存转移：在仓库之间移动库存
- 补货建议：建议新的订单数量
- 缺货预测：预测产品何时会缺货
- 成本优化：分析供应商替代方案
🏗️ 架构
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ Flask Web │ │ 自定义 │ │ Gemini AI │
│ 界面 │◄──►│ FastMCP │◄──►│ （或本地 │
│ │ │ 服务器 │ │ LLM API） │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│ │
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ 事件流 │ │ 供应链 │
│ （CSV 数据） │ │ 工具 │
└─────────────────┘ └─────────────────┘
📁 项目结构
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 # 本文件
🎯 示例工作流程
- 事件：
P001需求激增 - 40单位 - 分析：AI检查所有仓库中的当前库存
- 预测：识别潜在缺货风险
- 行动：建议从仓库B转移到仓库A
- 执行：更新库存并记录行动
示例对话流程：
事件流 → MCP客户端："P001需求激增，40单位"
MCP客户端 → get_inventory_status：{"product_id": "P001"}
MCP客户端 → predict_stockout：{"product_id": "P001", "warehouse": "warehouse_A"}
MCP客户端 → calculate_transfer：{"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}
MCP客户端 → recommend_reorder：{"product_id": "P001", "quantity": 50}
MCP客户端 → 用户："从B转移20单位到A，从供应商X补货50单位"
🔍 监控
- 终端输出：实时服务器日志和处理状态
- 行动日志：所有AI建议和执行的行动
- 状态指示器：服务器和事件流状态
- 事件进度：当前正在处理的事件
🚀 主要功能
- 实时处理：事件到达时即被处理
- 智能推荐：AI驱动的决策制定
- 实时更新：Web界面实时更新
- 简单设置：最少的依赖项和配置
- 可扩展性：轻松添加新工具和事件类型
- 隐私选项：可以使用本地LLM而不是云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]:
"""根据延误/成本推荐供应路线调整"""
# 使用成本分析实现
示例对话：
事件："供应商X延误，3天"
MCP客户端："分析P001、P002、P003的影响..."
调用工具：[get_inventory_status, calculate_correlation, forecast_demand, recommend_reroute]
响应："将P001从供应商X调整到供应商Y。P002和P003显示0.8的相关性 - 据此调整P002订单。"
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]:
"""生成面向客户的总结报告"""
示例对话：
用户："分析关于产品X的500份调查回复"
MCP客户端："使用多个代理处理回复..."
调用工具：[extract_themes, compute_frequencies, generate_summary_report]
响应："主要主题：UI/UX（45%），性能（32%），价格（23%）。满意度78%（±3% CI）。报告已生成。"
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]:
"""生成医生笔记"""
示例对话：
患者数据：{"症状"：["胸痛"，"呼吸困难"]，"生命体征"：{"血压"："140/90"}}
MCP客户端："分析患者数据..."
调用工具：[check_symptoms, classify_risk, score_triage_priority, generate_doctor_note]
响应："高风险 - 检测到心脏症状。需立即分诊。医生笔记：'患者出现胸痛和血压升高...' "
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解释"""
示例对话：
事件："竞争者价格变化：产品X，$25.99 → $22.99"
MCP客户端："分析竞争环境..."
调用工具：[find_competitor_match, calculate_optimal_price, generate_markdown_explanation]
响应："竞争者降价12%。建议行动：将价格降至$23.99。解释：'我们已调整价格以保持竞争力同时维持健康的利润率...' "
🔧 开发
添加新工具
- 在
supply_chain_tools.py中添加函数 - 在
fastmcp_server.py中注册工具 - 根据需要更新系统提示
添加新事件类型
- 在
data/events.csv中添加事件 - 更新
fastmcp_server.py中的事件处理逻辑 - 使用Web界面测试
切换到本地LLM
- 设置local-llm-api
- 修改
fastmcp_server.py以使用本地API端点 - 更新本地模型兼容性的提示
📝 注意事项
- 这是一个使用模拟数据的演示
- 库存更改会在服务器停止时保存回CSV
- 使用Gemini API免费层（有速率限制）
- 设计用于简单性和教育目的
- FastMCP不是开源的——这是自定义实现
- 可以通过本地LLM扩展以保护数据隐私
🤝 贡献
欢迎扩展此项目：
- 更复杂的AI模型
- 真实数据库集成
- 额外的供应链工具
- 增强的Web界面功能
- 并行工具执行优化
- 实时数据流
🔗 相关项目
- local-llm-api：用于数据隐私的本地LLM API
- MCP-RAG：参考MCP实现
准备好用AI优化您的供应链了吗？ 启动服务器，见证奇迹的发生！🚀
此项目展示了如何构建一个类似FastMCP的自定义系统，用于实时、多工具AI编排。