Flowr — 零售供应链多 Agent 端到端自动化

1. 解决的问题

母婴品牌在亚马逊/独立站同时运营，SKU 达 500+，跨境仓（海外仓 + 国内直发）补货涉及 DHL/UPS 运输周期（15-30 天）、海关清关（3-7 天）、Amazon FBA 入仓（1-5 天），任何一环延误都导致断货（Lost Buy Box，单 SKU 日损失 2,000-8,000 元）

2. 核心算法逻辑

Flowr 将人工密集型零售供应链操作系统性地分解为 6 个专业化 AI Agent 联盟，由中央 Reasoning LLM 协调编排，供应链经理通过 MCP（Model Context Protocol）接口实施人工监督（HITL）。核心洞察是：供应链决策具有天然的功能模块化特征，不同子任务所需的专业知识和数据源存在边界清晰的分工，因此适合由专业化 LLM（而非单一通用模型）各司其职，再由 Reasoning LLM 统一协调。

3. 业务应用场景

业务问题：母婴品牌在亚马逊/独立站同时运营，SKU 达 500+，跨境仓（海外仓 + 国内直发）补货涉及 DHL/UPS 运输周期（15-30 天）、海关清关（3-7 天）、Amazon FBA 入仓（1-5 天），任何一环延误都导致断货（Lost Buy Box，单 SKU 日损失 2,000-8,000 元）。

| Flowr Agent | 母婴对应角色 | 专属数据域 | |------------|------------|-----------| | Demand Forecasting | 销量预测 Agent（含节假日/季节/营销）| Amazon API 销售历史、促销日历 | | Inventory Monitoring | 库存状态 Agent（FBA + 海外仓 + 在途）| FBA Inventory Report、WMS 系统 | | Procurement & Ordering | 采购下单 Agent（MOQ/阶梯价格优化）| ERP 采购历史、供应商价格表 | | Sup

数据要求： - 销售历史：SKU × 日期 × 渠道，过去 12 个月，≥80% 数据完整率 - 库存快照：每日实时同步（FBA/海外仓/在途 三库区） - 运输时效：历史货代记录（含清关延误标记）

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

• 易：框架已有模板（本 Skill 提供完整代码骨架），6 个 Agent 边界清晰，可分阶段上线
• 中：需对接 Amazon API、WMS、ERP 等数据源（通常 2-4 周集成工期）
• 难：专业化 LLM 微调需要供应链领域语料（可用 RAG 替代降低门槛）

7. 代码模板

代码块数量：3 · 路径：未检测到

"""
Flowr Supply Chain MAS — 多 Agent 供应链协作框架（母婴出海版）
arXiv:2604.05987 | Python 3.14+ | 仅标准库，无需额外安装
"""
from __future__ import annotations
import json
import random
from dataclasses import dataclass, field
from datetime import date, timedelta
from typing import Any


# ── 数据结构 ──────────────────────────────────────────────────────────────────

@dataclass
class SKU:
    sku_id: str
    name: str
    current_stock: int        # 件
    daily_sales_avg: float    # 件/日
    lead_time_days: int       # 补货提前期（天）
    moq: int                  # 最小起订量
    unit_cost: float          # 元/件


@dataclass
class AgentResult:
    agent: str
    status: str               # "ok" | "warning" | "error"
    data: dict[str, Any]
    message: str = ""


@dataclass
class HITLGate:
    """人工监督门控记录"""
    gate_id: str
    agent: str
    decision_summary: str
    approved: bool = False
    approver: str = ""


# ── 6 个专业化 Agent ─────────────────────────────────────────────────────────

class DemandForecastingAgent:
    """需求预测 Agent — 基于移动平均 + 季节因子（生产环境可替换为 TFT/Prophet）"""

    def run(self, sku: SKU, horizon_days: int = 30, seasonality: float = 1.0) -> AgentResult:
        forecast_daily = sku.daily_sales_avg * seasonality
        forecast_total = round(forecast_daily * horizon_days)
        ci_lower = round(forecast_total * 0.8)
        ci_upper = round(forecast_total * 1.25)
        return AgentResult(
            agent="DemandForecasting",
            status="ok",
            data={
                "horizon_days": horizon_days,
                "forecast_total": forecast_total,
                "ci_lower": ci_lower,

8. 论文来源

• 2307.03595
• 2310.17168
• 2604.05987