Event-Driven Demand MAS — 事件感知补货 MAS：大促/季节自动触发

1. 解决的问题

传统补货系统是轮询式（定时跑批），每日/每周检查库存状态。事件驱动架构（Event-Driven Architecture）改为推送式：事件发生时立刻触发对应 Agent，实现毫秒级响应而非天级延迟。

2. 核心算法逻辑

事件驱动架构（EDA）在 Agent 系统中的应用

3. 业务应用场景

业务背景：每年618是母婴品类最大销售节点，历史数据显示618当周需求激增 3-5 倍。如果等到促销期才意识到缺货，物流提前期（lead time）约 30-45 天，已经来不及。

事件驱动流程： 1. T-60天：`EventCast` 预测模型产出618需求预测，生成 `DemandEvent(event_type="预知型", expected_impact_pct=3.5, lead_days=60)` 2. EventBus 路由：事件推送给 `SupplyChainAgent`、`WarehouseAgent`、`ProcurementAgent` 3. Agent 并行响应： - `SupplyChainAgent`：生成分 SKU 备货计划 - `ProcurementAgent`：向供应商发出预订请求（锁定产能和价格） - `WarehouseAgent

业务价值：历史上因备货不足导致的618断货率从 18% 降至 3%；提前锁单节省采购成本约 8%（规模效应）。

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

• 易处：EventBus 发布订阅模式是成熟模式；纯 Python 实现，无外部依赖
• 难处：事件阈值参数（安全库存线、impact_pct 阈值）需要业务标定；生产部署需要消息队列（Kafka/RabbitMQ）替代本地 EventBus
• 前提：需要库存实时监控数据源；需要 EventCast 提供大促需求预测

7. 代码模板

代码块数量：2 · 路径：paper2skills-code/mas/event_driven_demand_mas

"""
Event-Driven Demand MAS — 事件感知补货多 Agent 系统
大促/季节/断货事件自动触发 Agent 工作流

纯 Python 标准库，无外部依赖
Python 3.14 兼容
"""
from __future__ import annotations

import time
from collections import defaultdict
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable


# ─── 事件类型 & 数据结构 ──────────────────────────────────────────────────────

class EventType(Enum):
    PLANNED    = "预知型"   # 大促/节假日
    PERIODIC   = "周期型"   # 月末/季节
    EMERGENCY  = "突发型"   # 断货/爆款
    DERIVATIVE = "衍生型"   # 促销连锁


class EventSeverity(Enum):
    LOW    = "low"
    MEDIUM = "medium"
    HIGH   = "high"
    CRITICAL = "critical"


@dataclass
class DemandEvent:
    """需求驱动事件"""
    event_id: str
    event_type: EventType
    trigger_condition: str            # 触发条件描述
    expected_impact_pct: float        # 预期需求变化百分比（如 3.5 = 350%）
    lead_days: int                    # 距离事件剩余天数
    severity: EventSeverity = EventSeverity.MEDIUM
    affected_skus: list[str] = field(default_factory=list)
    metadata: dict[str, float] = field(default_factory=dict)

    @property
    def urgency_score(self) -> float:
        """urgency = impact × (1/lead_days)^0.5 × severity_weight"""
        severity_w = {"low": 0.5, "medium": 1.0, "high": 2.0, "critical": 4.0}
        w = severity_w.get(self.severity.value, 1.0)
        time_factor = 1.0 / max(self.lead_days, 1) ** 0.5
        return round(self.expected_impact_pct * time_factor * w, 3)


@dataclass
class AgentResponse:
    """Agent 响应结果"""
    agent_name: str
    event_id: str
    action_taken: str
    success: bool

8. 论文来源

未自动抽取；请查看原始 Skill 卡片。