基础模型已经让“给整个商品目录做需求预测”变得相对容易，但预测本身不会让货架恢复库存。真正棘手的是下一步：识别需求激增、核对供应商的实时可供货量、决定补多少，并在条件满足时自动下单。
这篇文章讨论一种部署在 Databricks 和 Amazon Quick 上的闭环：检测需求变化，做出补货决策，执行采购动作；只有没有任何供应商能够覆盖需求时，才升级给人工处理。
从预测结果走向可执行决策
可以把补货流程拆成三类事实：
- 需求事实：未来窗口内的预测需求、当前库存、已承诺订单和安全库存。
- 供应事实：供应商当前可供数量、交期、最低起订量和价格。
- 执行事实：是否已经创建采购订单、是否需要人工审批、异常是否已关闭。
预测模型只负责提供需求信号。后续系统还必须回答三个问题：
- 这次变化是不是足以构成需求激增？
- 哪个供应商能在目标交期内覆盖缺口？
- 这次动作可以无人值守执行，还是必须交给采购人员？
MMF 可以用于生成目录级预测，Databricks 负责组织数据、运行规则和保存决策结果，Genie 适合让业务人员用自然语言查询补货状态，Amazon Quick 则可以承载面向运营团队的分析和行动入口。关键不是把这些工具简单串联，而是让每个动作都能回溯到预测、库存和供应商数据。
一条可审计的 detect-decide-act 链路
一个实用的闭环可以按下面的顺序运行：
1. Detect：识别需求激增和库存缺口
将最新预测与基线预测或历史需求进行比较。例如，未来 14 天预测需求比基线高出 30%，并且可用库存低于需求加安全库存，就生成一个 replenishment event。
这里不要只看预测增幅。促销、季节性和数据延迟都可能造成假警报，因此事件中应保存预测版本、计算时间和输入数据时间窗。
2. Decide：把缺口与供应能力对齐
缺口可以用一个透明的公式计算：
replenishment_qty = max(
forecast_demand + safety_stock - on_hand - inbound_confirmed,
0
)
然后根据交期、可供数量、最低起订量和供应商优先级排序。若一个供应商无法覆盖全部缺口，可以按策略拆分订单；若所有合格供应商的总可供量仍然不足，就产生人工升级事件。
3. Act：创建订单或升级异常
自动动作必须具备幂等性。相同的事件重复执行时，不能创建两张采购订单。实践中可以使用 event_id 作为幂等键，并把 planned、ordered、escalated 等状态写回 Delta 表或其他审计存储。
一个可以改造的补货决策示例
下面的 Python 示例是一个最小的服务端决策骨架。它不绑定具体供应商 API：/availability 和 /purchase-orders 是示例接口，运行前需要替换成企业实际的供应商网关。预测和库存数据也可以改成从 Databricks SQL Warehouse 查询。
from dataclasses import dataclass
from typing import List, Optional
import os
import requests
SUPPLIER_API = os.environ.get("SUPPLIER_API", "https://supplier-gateway.example")
PURCHASE_API = os.environ.get("PURCHASE_API", "https://procurement.example")
@dataclass
class SupplierOffer:
supplier_id: str
available_qty: int
lead_time_days: int
unit_price: float
def required_qty(forecast_demand: int, safety_stock: int,
on_hand: int, inbound_confirmed: int) -> int:
return max(forecast_demand + safety_stock - on_hand - inbound_confirmed, 0)
def get_offers(sku: str, needed_qty: int) -> List[SupplierOffer]:
response = requests.get(
f"{SUPPLIER_API}/availability",
params={"sku": sku, "quantity": needed_qty, "max_lead_time_days": 14},
timeout=10,
)
response.raise_for_status()
return [SupplierOffer(**item) for item in response.json()["offers"]]
def create_order(event_id: str, sku: str, supplier_id: str, quantity: int) -> dict:
response = requests.post(
f"{PURCHASE_API}/purchase-orders",
headers={"Idempotency-Key": event_id},
json={
"event_id": event_id,
"sku": sku,
"supplier_id": supplier_id,
"quantity": quantity,
"source": "automated-replenishment",
},
timeout=10,
)
response.raise_for_status()
return response.json()
def decide_and_act(event: dict) -> dict:
needed = required_qty(
event["forecast_demand"],
event["safety_stock"],
event["on_hand"],
event["inbound_confirmed"],
)
if needed == 0:
return {"status": "no_action", "sku": event["sku"], "quantity": 0}
offers = sorted(
get_offers(event["sku"], needed),
key=lambda x: (x.lead_time_days, x.unit_price),
)
remaining = needed
orders = []
for offer in offers:
quantity = min(remaining, offer.available_qty)
if quantity <= 0:
continue
order = create_order(
event["event_id"], event["sku"], offer.supplier_id, quantity
)
orders.append(order)
remaining -= quantity
if remaining == 0:
break
if remaining > 0:
return {
"status": "escalated",
"sku": event["sku"],
"needed": needed,
"covered": needed - remaining,
"uncovered": remaining,
"orders": orders,
"reason": "no eligible supplier can cover the full surge",
}
return {
"status": "ordered",
"sku": event["sku"],
"quantity": needed,
"orders": orders,
}
if __name__ == "__main__":
sample_event = {
"event_id": "surge-2025-001-ABC123",
"sku": "ABC123",
"forecast_demand": 1200,
"safety_stock": 200,
"on_hand": 350,
"inbound_confirmed": 100,
}
print(decide_and_act(sample_event))
这个示例有几个值得保留的边界：
- 供应商可供量来自实时接口，而不是预测快照。
- 订单请求带有幂等键，避免重试造成重复采购。
- 供应量不足时不会“猜测”一个供应商，而是明确返回
escalated。 - 每个事件都带有
event_id，便于在 Databricks 中追踪输入、决策和结果。
如果业务允许拆单，还需要在代码中加入最低起订量、包装倍数、预算上限和供应商配额等约束。示例中的逐家分配逻辑只是起点，不能直接替代企业采购策略。
用 Genie 和 Amazon Quick 让人看懂自动化结果
自动化并不意味着采购团队失去可见性。相反，系统应该把决策过程做成可以查询和解释的记录。
可以在 Databricks 中保存一张类似下面的决策表：
CREATE TABLE IF NOT EXISTS replenishment_decisions (
event_id STRING,
sku STRING,
detected_at TIMESTAMP,
forecast_demand INT,
on_hand INT,
inbound_confirmed INT,
required_qty INT,
supplier_coverage INT,
status STRING,
reason STRING,
order_ids ARRAY<STRING>
) USING DELTA;
在这张表之上，可以让 Genie 回答类似问题：
过去 24 小时有哪些 SKU 因供应不足而升级？
按供应商汇总未覆盖数量，并列出预计缺货日期。
哪些自动采购订单的供应商交期超过了目标窗口？
Amazon Quick 的页面则可以围绕运营动作组织，而不是只展示一张预测折线图。例如展示：
- 当前待处理的需求激增事件；
- 已自动下单金额和订单数量；
- 因供应不足而升级的 SKU；
- 按供应商划分的覆盖率和交期；
- 自动化动作的失败、重试和人工接管记录。
每一张图都应能跳转到事件明细。采购人员需要知道“为什么下了这张单”，而不仅是“系统下了一张单”。
不要把预测置信度当成采购授权
闭环自动化最容易踩的坑，是把高置信度预测直接等同于可以下单。两者不是一回事。
建议至少设置以下护栏：
- 金额上限：超过预算的订单必须人工审批。
- 供应商白名单：只允许合同有效且通过合规检查的供应商自动接单。
- 交期约束：供应商能供货，但无法在需求窗口内交付时，不应算作有效覆盖。
- 库存保护：关键 SKU 不允许因为一次异常预测而消耗全部安全库存。
- 异常回滚：供应商接口超时、重复响应或返回不一致时，暂停自动下单并升级。
- 审计记录：保留预测版本、规则版本、供应商响应和最终订单内容。
还应定期评估三类指标：预测误差、供应覆盖率和自动动作成功率。只看缺货率，可能会让系统过度补货；只看自动化率，又可能掩盖错误订单。
落地时的一份小清单
可以按以下顺序推进，而不是一开始就让整个目录无人值守：
- 先选择少量 SKU 和一个可控的供应商群组。
- 把预测、库存、在途和供应商可供量统一到同一个事件模型。
- 先只生成建议订单，验证缺口计算和供应商排序。
- 为事件和订单加入幂等键、状态机和审计表。
- 让自动化只覆盖低金额、规则清晰的场景。
- 将“无人供应可覆盖”的事件明确路由给采购人员。
- 用 Genie 和 Amazon Quick 检查结果，再逐步扩大自动下单范围。
真正有价值的系统不是“预测得更复杂”，而是能把预测转成受约束、可解释、可回滚的行动。MMF 提供需求信号，Databricks Genie 和 Amazon Quick 提供查询与运营入口，供应商接口和采购系统则完成动作。只有这几部分共同形成闭环，需求激增才不会停留在报表里。