供应链异常检测——AI 在运营场景的一个真实应用
供应链异常检测——AI 在运营场景的一个真实应用
适读人群：有运营数据分析需求的工程师/数据工程师 | 阅读时长：约17分钟 | 核心价值：供应链异常检测的完整技术方案，讲清楚AI在运营场景落地的真实挑战
这个项目我从头到尾参与了八个月，是我做过的最有挑战的AI项目之一。
不是因为模型复杂。是因为在运营场景里，"异常"这个词的定义，比我最初想象的难定义得多。
客户是一家做零售的中型公司，有200多个SKU，10多个供应商，5个仓库。他们遇到的问题是：供应链异常（缺货、滞销、供应商交货延迟、库存积压）经常被发现得太晚，等采购经理意识到不对劲，损失已经产生了。
他们想要的是：提前发现异常苗头，不是等异常已经发生了才报警。
第一步：定义"异常"
这是整个项目里花时间最多的部分，没有之一。
我和采购团队开了四次工作坊，每次2-3小时，专门讨论"什么叫异常"。他们给我讲了很多真实案例，我逐渐理解到：供应链异常不是一个单一的东西，而是一堆不同类型的问题。
最终我把异常分成五类：
1. 库存水位异常
- 安全库存告警（库存量低于安全库存线）
- 库存积压告警（超过正常周转天数的库存大量堆积）
- 库存突变（24小时内库存量变化超过正常波动的3倍，可能是系统错误或漏损）
2. 销售节奏异常
- 滞销预警（某SKU连续N天销量低于历史均值的30%）
- 销量暴涨（可能是促销效果好，也可能是数据错误）
- 区域异常（某仓库/渠道销售偏离同类仓库/渠道）
3. 供应商交货异常
- 交货延迟（实际到货时间晚于承诺时间超过X天）
- 交货数量异常（实际到货量与订单量差异超过阈值）
- 供应商发货节奏异常（突然停止发货或频率异常）
4. 需求预测偏差
- 实际销量持续高于或低于预测值（预测失效信号）
- 季节性规律异常（某季节销量不符合历史季节性规律）
5. 价格成本异常
- 采购成本异常波动（某SKU采购价格突然涨价或降价幅度超过阈值）
- 毛利率异常（某SKU毛利率偏离正常范围）
有了这个分类，才能针对每类异常设计具体的检测逻辑。
数据层：先搞清楚你有什么数据
在做任何模型之前，我做了完整的数据盘点：
# 数据来源清单和字段说明
DATA_SOURCES = {
"inventory": {
"source": "ERP系统",
"update_frequency": "每日",
"key_fields": [
"sku_id", "warehouse_id", "date",
"stock_qty", # 库存量
"safety_stock", # 安全库存线
"max_stock", # 最大库存
"unit_cost" # 单位成本
],
"history_depth": "2年",
"data_quality_issues": ["部分仓库有漏录问题（约3%的记录缺失）"]
},
"sales": {
"source": "销售系统",
"update_frequency": "实时（每小时同步）",
"key_fields": [
"sku_id", "warehouse_id", "date", "hour",
"sales_qty", "sales_amount", "return_qty"
],
"history_depth": "3年",
"data_quality_issues": ["退货数据延迟2-3天才能同步"]
},
"purchase_orders": {
"source": "采购系统",
"update_frequency": "每日",
"key_fields": [
"po_id", "supplier_id", "sku_id",
"order_date", "promised_delivery_date",
"actual_delivery_date", # 实际到货日期，可能为空（未到货）
"ordered_qty", "received_qty",
"unit_price"
]
}
}数据盘点做完，发现了几个问题：
- 退货数据有2-3天延迟，这意味着我不能基于当日销量做实时判断
- 部分仓库历史数据有缺口，需要用插值处理
- ERP和销售系统的sku_id编码不完全一致（有15个SKU在两个系统里的ID格式不同）
这些数据质量问题，在我接触的大多数企业里都存在。在做AI模型之前，花时间处理数据质量，比调参更重要。
核心检测逻辑：规则 + 统计 + LLM三层架构
我没有直接用LLM做异常检测，而是设计了三层架构：
第一层：规则检测（硬规则，高精度）
处理有明确业务定义的异常，不需要AI：
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict
class RuleBasedDetector:
"""
基于业务规则的异常检测
高精度，不需要机器学习
"""
def check_safety_stock(
self,
inventory_df: pd.DataFrame,
threshold_days: int = 3 # 如果库存不足以支撑N天销售，告警
) -> List[Dict]:
"""检查安全库存告警"""
alerts = []
# 计算最近7天平均日销量
sales_7d = self._get_avg_daily_sales(days=7)
for _, row in inventory_df.iterrows():
sku_id = row['sku_id']
warehouse_id = row['warehouse_id']
current_stock = row['stock_qty']
avg_daily_sales = sales_7d.get((sku_id, warehouse_id), 0)
if avg_daily_sales > 0:
days_of_stock = current_stock / avg_daily_sales
if days_of_stock < threshold_days:
alerts.append({
"type": "safety_stock_warning",
"severity": "high" if days_of_stock < 1 else "medium",
"sku_id": sku_id,
"warehouse_id": warehouse_id,
"current_stock": current_stock,
"days_of_stock": round(days_of_stock, 1),
"message": f"库存仅够{days_of_stock:.1f}天销售"
})
return alerts
def check_delivery_delay(
self,
po_df: pd.DataFrame,
delay_days_threshold: int = 3
) -> List[Dict]:
"""检查采购订单交货延迟"""
alerts = []
today = datetime.now().date()
# 未到货的订单
pending_orders = po_df[po_df['actual_delivery_date'].isna()].copy()
for _, order in pending_orders.iterrows():
promised_date = pd.to_datetime(order['promised_delivery_date']).date()
delay_days = (today - promised_date).days
if delay_days >= delay_days_threshold:
alerts.append({
"type": "delivery_delay",
"severity": "high" if delay_days >= 7 else "medium",
"po_id": order['po_id'],
"supplier_id": order['supplier_id'],
"sku_id": order['sku_id'],
"delay_days": delay_days,
"message": f"采购单{order['po_id']}已延迟{delay_days}天到货"
})
return alerts
def _get_avg_daily_sales(self, days: int) -> Dict:
# 实际实现从数据库查询，这里略
pass第二层：统计异常检测（处理"不寻常但没有硬规则"的情况）
from scipy import stats
class StatisticalDetector:
"""
基于统计方法的异常检测
处理时间序列中的异常波动
"""
def detect_sales_anomaly(
self,
sales_history: pd.Series, # 历史日销量时间序列
current_value: float,
sensitivity: float = 3.0 # Z-score阈值，越大越不敏感
) -> Dict:
"""
基于Z-score检测销量是否异常
使用滚动均值和标准差，对季节性有一定鲁棒性
"""
# 使用最近90天数据，计算均值和标准差
recent_history = sales_history.tail(90)
if len(recent_history) < 14: # 数据不足，无法判断
return {"is_anomaly": False, "reason": "历史数据不足"}
mean = recent_history.mean()
std = recent_history.std()
if std == 0:
return {"is_anomaly": False, "reason": "销量无波动"}
z_score = (current_value - mean) / std
return {
"is_anomaly": abs(z_score) > sensitivity,
"z_score": round(z_score, 2),
"direction": "spike" if z_score > 0 else "drop",
"deviation_pct": round((current_value - mean) / mean * 100, 1),
"historical_mean": round(mean, 1),
"current_value": current_value
}
def detect_trend_change(
self,
time_series: pd.Series,
window: int = 7
) -> Dict:
"""
检测时间序列的趋势变化
比较近期趋势和历史趋势是否有显著差异
"""
if len(time_series) < window * 3:
return {"trend_change": False, "reason": "数据不足"}
# 近期趋势（最近window天）
recent = time_series.tail(window)
recent_slope = np.polyfit(range(len(recent)), recent, 1)[0]
# 历史趋势（更早的window天）
historical = time_series.iloc[-(window*2):-window]
historical_slope = np.polyfit(range(len(historical)), historical, 1)[0]
slope_change = recent_slope - historical_slope
# 判断趋势是否发生显著变化
historical_std = time_series.std()
normalized_change = abs(slope_change) / (historical_std + 1e-10)
return {
"trend_change": normalized_change > 0.5,
"direction": "improving" if slope_change > 0 else "deteriorating",
"recent_daily_change": round(recent_slope, 2),
"historical_daily_change": round(historical_slope, 2)
}第三层：LLM综合分析（把多个信号整合成可读的诊断）
单个异常信号往往没有太大意义，真正有价值的是把多个信号放在一起看。这是LLM最擅长的地方：
from openai import OpenAI
client = OpenAI(api_key="your_key", base_url="your_base")
ANALYSIS_PROMPT = """你是一名有丰富经验的供应链分析师。
请分析以下供应链异常信号，给出综合诊断和建议。
SKU信息：
- SKU ID: {sku_id}
- SKU名称: {sku_name}
- 品类: {category}
- 主要仓库: {warehouse}
异常信号（过去24小时内检测到）：
{anomaly_signals}
历史背景：
- 过去30天平均日销量: {avg_daily_sales} 件
- 当前库存: {current_stock} 件（约可售{days_of_stock}天）
- 在途采购量: {incoming_qty} 件（预计{incoming_date}到货）
- 本月是否有促销活动: {has_promotion}
请给出：
1. **综合判断**：这些信号组合在一起意味着什么？是需要立即处理的危机，还是需要关注的苗头？
2. **根本原因推断**：最可能的原因是什么？（1-3个候选原因，按可能性排序）
3. **建议行动**：具体的、可执行的建议（区分"立即做"和"本周内做"）
4. **风险评估**：如果不处理，可能的业务影响是什么？
注意：只根据提供的数据做判断，不要做没有依据的猜测。如果信息不足以判断，明确说明需要哪些额外信息。
输出格式要简洁，方便采购经理在5分钟内读完并做决策。"""
def generate_supply_chain_insight(
sku_info: dict,
anomaly_signals: list,
context: dict
) -> str:
"""
使用LLM对多个异常信号进行综合分析
"""
if not anomaly_signals:
return "当前无异常信号"
# 格式化异常信号
signals_text = "\n".join([
f"- [{s['type']}] {s['message']} (严重程度: {s['severity']})"
for s in anomaly_signals
])
prompt = ANALYSIS_PROMPT.format(
sku_id=sku_info["sku_id"],
sku_name=sku_info["sku_name"],
category=sku_info["category"],
warehouse=sku_info.get("warehouse", "全部"),
anomaly_signals=signals_text,
avg_daily_sales=context.get("avg_daily_sales", "未知"),
current_stock=context.get("current_stock", "未知"),
days_of_stock=context.get("days_of_stock", "未知"),
incoming_qty=context.get("incoming_qty", 0),
incoming_date=context.get("incoming_date", "无在途采购"),
has_promotion=context.get("has_promotion", "否")
)
response = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": prompt}],
max_tokens=800,
temperature=0.3
)
return response.choices[0].message.content上线后遇到的特殊挑战
挑战1：误报太多导致告警疲劳
系统上线第一周，每天产生了80多条告警。采购团队完全无法处理，一个人一天的工作量就是看告警，全是误报。第三天他们就停用了系统，说"这东西给我添麻烦"。
解决方案：引入告警优先级过滤。只有以下情况的告警才发送通知：
- 单个SKU同时出现2个及以上异常信号
- 库存低于1天销售量（极度紧急）
- 供应商连续3天无发货记录
其余信号进入"关注"列表，每日汇总发送一次，不推送实时通知。
告警量从每天80+条降到每天8-12条，误报率降低，采购团队开始接受系统。
挑战2：季节性和促销活动干扰统计基线
春节、618、双11期间，销量基线完全不同。如果用全年平均做基线，这些节点全是误报（销量暴涨被识别为"异常"）。
解决方案：在系统里加入业务日历，让运营团队提前标注促销和节假日，在这些日期期间切换到同比基线（对比去年同期），而不是近期均值。
挑战3：AI分析的"瞎话"问题
LLM有时候会在信息不足的情况下"强行推理"，给出听起来合理但实际上没有依据的分析。
比如某SKU库存下降+销量下降，LLM推断"可能是产品出现质量问题导致退货增加"——但实际上只是该产品下架了，和质量无关。
解决方案：在Prompt里强调"如果信息不足以判断，明确说明需要哪些额外信息"，同时在UI上标注"以下分析基于有限数据，需结合实际情况判断"。把AI定位为"辅助分析"而不是"权威诊断"。
八个月的真实数据
系统全面上线后运行了三个月的数据：
| 指标 | 数据 |
|---|---|
| 检测到的异常信号总数 | 2,847个 |
| 其中有效异常（被采购确认） | 1,631个（57.3%） |
| 平均发现时间（vs 人工发现） | 提前2.1天 |
| 由此避免的缺货事件 | 34起 |
| 库存周转率改善 | +8% |
| 采购团队满意度（5分制） | 3.8分（上线初期2.1分） |
57%的有效率意味着43%是误报，这个比例不算低。但和上线前的状态（完全依赖人工，很多问题等到已经影响业务才发现）相比，采购团队认为这是可以接受的。
提前2.1天发现问题，在供应链场景里是很大的价值——很多补货动作需要3-5天的提前量，提前2天发现往往是来得及处理和来不及处理的分界线。
AI在运营场景落地的几个特殊挑战
这个项目让我对AI在运营场景的落地有了几个新认识：
"异常"的定义是业务知识，不是技术问题。 花了我最多时间的不是写代码，是和采购团队反复讨论"什么叫异常"。技术能力是基础，但不懂业务就没有好的AI应用。
运营人员的信任需要用准确率来建立，不能用演示来代替。 我们展示了很好的demo，但用户真正信任系统，是在他们发现"系统提前告警，我们行动了，避免了一次缺货"这样的真实案例之后。
告警疲劳是运营类AI工具最大的死因之一。 宁可少报，不可多报——这在医疗诊断场景可能相反，但在运营场景，误报太多会让用户彻底放弃工具。
AI只是工具，最终决策还是人。 这套系统从来没有设计成"AI自动触发补货"，而是"AI发现信号，采购经理决定怎么做"。运营场景里的业务规则和例外情况太多，完全自动化反而风险更大。