AI供应链与物流优化完全教程
1. AI供应链优化概述与技术架构
传统供应链管理依赖人工经验与静态规则，面对需求波动、突发事件和全球化复杂网络时往往力不从心。AI技术的引入正在从根本上重塑供应链的运作方式——从被动响应转向主动预测，从局部优化转向全局协同。
技术架构全景
一套完整的AI供应链系统通常包含以下层次：
┌─────────────────────────────────────────────────┐
│ 业务应用层 │
│ 需求预测 │ 库存优化 │ 路径规划 │ 质量检测 │
├─────────────────────────────────────────────────┤
│ AI模型服务层 │
│ 时序预测 │ 优化求解 │ 计算机视觉 │ NLP │
├─────────────────────────────────────────────────┤
│ 数据处理层 │
│ 实时流处理 │ 批处理 │ 特征工程 │ 数据湖 │
├─────────────────────────────────────────────────┤
│ 数据采集层 │
│ IoT传感器 │ ERP系统 │ WMS系统 │ 外部数据源 │
└─────────────────────────────────────────────────┘
核心技术栈包括：
- 时序预测：Prophet、DeepAR、Temporal Fusion Transformer
- 运筹优化：Google OR-Tools、Gurobi、CPLEX
- 计算机视觉：YOLO、ResNet（用于质检与仓储）
- 图神经网络：用于供应链网络风险传导分析
- 数字孪生：AnyLogic、SimPy 离散事件仿真
数据基础建设
供应链AI的第一步是打通数据孤岛。典型的集成代码如下：
import pandas as pd
from datetime import datetime, timedelta
class SupplyChainDataHub:
"""供应链数据集成中心"""
def __init__(self):
self.connections = {}
def register_source(self, name, connector):
self.connections[name] = connector
print(f"数据源 [{name}] 已注册")
def fetch_inventory(self, warehouse_id: str) -> pd.DataFrame:
"""从WMS系统拉取库存数据"""
wms = self.connections.get('wms')
if not wms:
raise ValueError("WMS数据源未注册")
query = f"""
SELECT sku_id, quantity, location, last_updated,
reorder_point, lead_time_days
FROM inventory
WHERE warehouse_id = '{warehouse_id}'
AND quantity > 0
"""
return pd.DataFrame(wms.execute(query))
def fetch_orders(self, days: int = 90) -> pd.DataFrame:
"""从ERP系统拉取历史订单"""
erp = self.connections.get('erp')
since = datetime.now() - timedelta(days=days)
query = f"""
SELECT order_id, sku_id, quantity, order_date,
customer_id, delivery_date
FROM orders WHERE order_date >= '{since.date()}'
"""
return pd.DataFrame(erp.execute(query))
def fetch_iot_stream(self, sensor_type: str):
"""实时IoT传感器数据流（温度/湿度/GPS）"""
import json
stream = self.connections.get('iot_stream')
for message in stream.subscribe(sensor_type):
yield json.loads(message)
2. 需求预测与库存优化
需求预测是供应链AI的核心起点。预测精度直接影响库存成本、缺货率和客户满意度。
基于Prophet的基线预测
from prophet import Prophet
import pandas as pd
import numpy as np
def build_demand_forecast(sales_df: pd.DataFrame,
forecast_days: int = 30) -> pd.DataFrame:
"""
构建需求预测模型
sales_df: 包含 ds(日期) 和 y(销量) 列
"""
# 添加节假日效应
holidays = pd.DataFrame({
'holiday': 'promo_event',
'ds': pd.to_datetime(['2025-06-18', '2025-11-11', '2025-12-12']),
'lower_window': -3,
'upper_window': 3,
})
model = Prophet(
yearly_seasonality=True,
weekly_seasonality=True,
holidays=holidays,
changepoint_prior_scale=0.1,
seasonality_mode='multiplicative'
)
# 添加外部回归变量（如促销标记、价格变动）
if 'is_promotion' in sales_df.columns:
model.add_regressor('is_promotion')
if 'price_ratio' in sales_df.columns:
model.add_regressor('price_ratio')
model.fit(sales_df)
future = model.make_future_dataframe(periods=forecast_days)
forecast = model.predict(future)
return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']]
库存安全库存计算
预测给出需求分布后，需要结合服务水平目标计算安全库存：
import scipy.stats as stats
def calculate_safety_stock(forecast_std: float,
lead_time_days: float,
service_level: float = 0.95) -> float:
"""
计算安全库存
:param forecast_std: 预测误差标准差
:param lead_time_days: 补货提前期（天）
:param service_level: 服务水平目标（如0.95表示95%不缺货）
:return: 安全库存数量
"""
z_score = stats.norm.ppf(service_level)
safety_stock = z_score * forecast_std * np.sqrt(lead_time_days)
return np.ceil(safety_stock)
def calculate_reorder_point(avg_daily_demand: float,
lead_time_days: float,
safety_stock: float) -> float:
"""计算再订货点"""
return avg_daily_demand * lead_time_days + safety_stock
# 示例：SKU-001 的库存参数
avg_demand = 150 # 日均需求
forecast_error_std = 30 # 预测误差标准差
lead_time = 7 # 补货提前期7天
safety = calculate_safety_stock(forecast_error_std, lead_time, 0.97)
rop = calculate_reorder_point(avg_demand, lead_time, safety)
print(f"安全库存: {safety:.0f} 件")
print(f"再订货点: {rop:.0f} 件")
print(f"当库存降至 {rop:.0f} 件时触发补货")
多级库存联合优化
实际供应链往往包含多个层级（中央仓→区域仓→前置仓），需要联合优化：
from scipy.optimize import minimize
def multi_echelon_inventory(params):
"""
多级库存成本优化
params: [central_ss, regional_ss_1, regional_ss_2, ...]
"""
n_regional = 2
central_ss = params[0]
regional_ss = params[1:1+n_regional]
# 中央仓持有成本
central_holding = central_ss * 5.0 # 单位持有成本5元/天
# 区域仓持有成本（通常低于中央仓）
regional_holding = sum(ss * 3.0 for ss in regional_ss)
# 缺货惩罚成本（服务水平不达标时的罚款）
target_sl = 0.95
actual_sl_central = 1 - np.exp(-central_ss / 100)
actual_sl_regional = [1 - np.exp(-ss / 50) for ss in regional_ss]
shortage_penalty = 0
if actual_sl_central < target_sl:
shortage_penalty += (target_sl - actual_sl_central) * 10000
for sl in actual_sl_regional:
if sl < target_sl:
shortage_penalty += (target_sl - sl) * 5000
# 转运成本（区域间调拨）
transfer_cost = max(0, regional_ss[0] - regional_ss[1]) * 2.0
total = central_holding + regional_holding + shortage_penalty + transfer_cost
return total
# 优化求解
initial_guess = [200, 80, 80]
bounds = [(50, 500), (20, 200), (20, 200)]
result = minimize(multi_echelon_inventory, initial_guess, bounds=bounds, method='L-BFGS-B')
print(f"中央仓安全库存: {result.x[0]:.0f}")
print(f"区域仓1安全库存: {result.x[1]:.0f}")
print(f"区域仓2安全库存: {result.x[2]:.0f}")
print(f"最优总成本: {result.fun:.2f} 元/天")
3. 智能仓储与自动化分拣
智能仓储利用机器人、计算机视觉和路径规划算法，大幅提升仓库作业效率。
基于计算机视觉的货架检测
from ultralytics import YOLO
import cv2
class ShelfDetector:
"""货架空位检测系统"""
def __init__(self, model_path: str = 'yolov8n.pt'):
self.model = YOLO(model_path)
self.empty_threshold = 0.3 # 空置率报警阈值
def detect_shelf_status(self, image_path: str) -> dict:
"""分析货架图片，返回各层位状态"""
results = self.model(image_path, conf=0.5)
shelf_status = {
'total_slots': 0,
'occupied': 0,
'empty': 0,
'misplaced': [],
'alerts': []
}
for result in results:
boxes = result.boxes
for box in boxes:
cls = int(box.cls[0])
conf = float(box.conf[0])
label = self.model.names[cls]
shelf_status['total_slots'] += 1
if label == 'empty_slot':
shelf_status['empty'] += 1
elif label == 'product':
shelf_status['occupied'] += 1
elif label == 'misplaced':
shelf_status['misplaced'].append({
'confidence': conf,
'bbox': box.xyxy[0].tolist()
})
# 空置率报警
if shelf_status['total_slots'] > 0:
empty_rate = shelf_status['empty'] / shelf_status['total_slots']
if empty_rate > self.empty_threshold:
shelf_status['alerts'].append(
f"空置率 {empty_rate:.1%} 超过阈值，请补货"
)
return shelf_status
仓库拣货路径优化
拣货是仓库中最耗时的环节。使用遗传算法优化拣货路径：
import numpy as np
import random
class PickingRouteOptimizer:
"""拣货路径优化器 - 遗传算法求解TSP"""
def __init__(self, locations: list, depot: tuple = (0, 0)):
self.locations = np.array([depot] + locations)
self.n = len(locations)
self.dist_matrix = self._compute_distance_matrix()
def _compute_distance_matrix(self):
n = len(self.locations)
matrix = np.zeros((n, n))
for i in range(n):
for j in range(i + 1, n):
d = np.linalg.norm(self.locations[i] - self.locations[j])
matrix[i][j] = d
matrix[j][i] = d
return matrix
def _total_distance(self, route):
dist = 0
for i in range(len(route) - 1):
dist += self.dist_matrix[route[i]][route[i+1]]
return dist
def optimize(self, pop_size=100, generations=500, mutation_rate=0.02):
"""遗传算法优化"""
# 初始化种群（排除起点0，对1..n排列）
population = []
for _ in range(pop_size):
route = list(range(1, self.n))
random.shuffle(route)
population.append([0] + route + [0])
best_distance = float('inf')
best_route = None
for gen in range(generations):
# 计算适应度
distances = [(self._total_distance(r), r) for r in population]
distances.sort(key=lambda x: x[0])
if distances[0][0] < best_distance:
best_distance = distances[0][0]
best_route = distances[0][1][:]
# 选择前50%
survivors = [r for _, r in distances[:pop_size // 2]]
# 交叉与变异产生新种群
new_pop = survivors[:]
while len(new_pop) < pop_size:
p1, p2 = random.sample(survivors, 2)
child = self._order_crossover(p1, p2)
if random.random() < mutation_rate:
child = self._swap_mutation(child)
new_pop.append(child)
population = new_pop
return best_route, best_distance
def _order_crossover(self, p1, p2):
"""有序交叉算子"""
size = len(p1) - 2
start, end = sorted(random.sample(range(size), 2))
child_inner = p1[1:-1][:]
for gene in p2[1:-1]:
if gene not in child_inner[start:end+1]:
for i in range(size):
if child_inner[i] is None or (i < start or i > end):
if gene not in child_inner:
child_inner[i] = gene
break
return [0] + child_inner + [0]
def _swap_mutation(self, route):
"""交换变异"""
inner = route[1:-1]
i, j = random.sample(range(len(inner)), 2)
inner[i], inner[j] = inner[j], inner[i]
return [0] + inner + [0]
# 使用示例
pick_points = [(10, 20), (35, 45), (60, 10), (25, 70), (80, 55), (45, 85)]
optimizer = PickingRouteOptimizer(pick_points)
best_route, distance = optimizer.optimize()
print(f"最优拣货路线: {best_route}")
print(f"总距离: {distance:.1f} 米")
4. 路径优化与车辆调度（VRP）
车辆路径问题（Vehicle Routing Problem, VRP）是物流优化中最经典的问题之一。实际场景中还需考虑时间窗、载重限制、多车型等约束。
使用Google OR-Tools求解VRPTW
from ortools.constraint_solver import routing_enums_pb2, pywrapcp
def solve_vrptw(distance_matrix, time_windows, demands,
vehicle_capacity, num_vehicles):
"""
求解带时间窗的车辆路径问题
:param distance_matrix: 距离矩阵
:param time_windows: 每个客户的时间窗 [(最早, 最晚), ...]
:param demands: 每个客户需求量
:param vehicle_capacity: 车辆容量
:param num_vehicles: 车辆数
"""
manager = pywrapcp.RoutingIndexManager(
len(distance_matrix), num_vehicles, 0 # 0为配送中心
)
routing = pywrapcp.RoutingModel(manager)
# 距离回调
def distance_callback(from_idx, to_idx):
from_node = manager.IndexToNode(from_idx)
to_node = manager.IndexToNode(to_idx)
return distance_matrix[from_node][to_node]
transit_cb = routing.RegisterTransitCallback(distance_callback)
routing.SetArcCostEvaluatorOfAllVehicles(transit_cb)
# 容量约束
def demand_callback(idx):
return demands[manager.IndexToNode(idx)]
demand_cb = routing.RegisterUnaryTransitCallback(demand_callback)
routing.AddDimensionWithVehicleCapacity(
demand_cb, 0, [vehicle_capacity] * num_vehicles, True, 'Capacity'
)
# 时间窗约束
def time_callback(from_idx, to_idx):
from_node = manager.IndexToNode(from_idx)
to_node = manager.IndexToNode(to_idx)
travel_time = distance_matrix[from_node][to_node]
return travel_time + 10 # 10分钟服务时间
time_cb = routing.RegisterTransitCallback(time_callback)
routing.AddDimension(time_cb, 30, 480, False, 'Time')
time_dim = routing.GetDimensionOrDie('Time')
for location_idx, (earliest, latest) in enumerate(time_windows):
index = manager.NodeToIndex(location_idx)
time_dim.CumulVar(index).SetRange(earliest, latest)
# 求解参数
search_params = pywrapcp.DefaultRoutingSearchParameters()
search_params.first_solution_strategy = (
routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
)
search_params.local_search_metaheuristic = (
routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
)
search_params.time_limit.FromSeconds(30)
solution = routing.SolveWithParameters(search_params)
if solution:
routes = []
for v in range(num_vehicles):
route = []
index = routing.Start(v)
while not routing.IsEnd(index):
route.append(manager.IndexToNode(index))
index = solution.Value(routing.NextVar(index))
route.append(manager.IndexToNode(index))
if len(route) > 2: # 排除空路线
routes.append(route)
return routes
return None
# 示例数据
dist_matrix = [
[0, 10, 15, 20, 25, 30],
[10, 0, 35, 25, 20, 15],
[15, 35, 0, 30, 20, 25],
[20, 25, 30, 0, 15, 35],
[25, 20, 20, 15, 0, 10],
[30, 15, 25, 35, 10, 0],
]
time_windows = [(0, 480), (60, 180), (120, 300), (0, 240), (180, 360), (60, 420)]
demands = [0, 30, 20, 25, 35, 15]
routes = solve_vrptw(dist_matrix, time_windows, demands, vehicle_capacity=80, num_vehicles=2)
if routes:
for i, r in enumerate(routes):
print(f"车辆{i+1}路线: {' → '.join(map(str, r))}")
动态路径重规划
实际配送中常遇到交通拥堵、新增订单等突发情况，需要实时重规划：
import heapq
class DynamicRouter:
"""动态路径重规划引擎"""
def __init__(self, graph: dict):
self.graph = graph # 邻接表 {node: [(neighbor, weight), ...]}
def dijkstra(self, start, end):
"""最短路径算法"""
distances = {node: float('inf') for node in self.graph}
distances[start] = 0
prev = {node: None for node in self.graph}
pq = [(0, start)]
while pq:
d, u = heapq.heappop(pq)
if d > distances[u]:
continue
if u == end:
break
for v, w in self.graph[u]:
new_dist = d + w
if new_dist < distances[v]:
distances[v] = new_dist
prev[v] = u
heapq.heappush(pq, (new_dist, v))
path = []
node = end
while node:
path.append(node)
node = prev[node]
return list(reversed(path)), distances[end]
def reroute(self, current_pos, remaining_stops,
blocked_edges: list = None):
"""
当前车辆位置重新规划剩余配送点的最优路径
:param blocked_edges: 因拥堵/事故需要避开的边
"""
if blocked_edges:
for u, v in blocked_edges:
self.graph[u] = [(n, w) for n, w in self.graph[u] if n != v]
self.graph[v] = [(n, w) for n, w in self.graph[v] if n != u]
# 对剩余配送点做贪心最近邻排序
route = [current_pos]
unvisited = set(remaining_stops)
current = current_pos
while unvisited:
nearest = min(unvisited,
key=lambda s: self.dijkstra(current, s)[1])
path, dist = self.dijkstra(current, nearest)
route.extend(path[1:])
unvisited.remove(nearest)
current = nearest
return route
5. 供应商风险评估
供应链中的供应商风险可能来自自然灾害、地缘政治、财务状况恶化等多种因素。图神经网络可以建模供应商网络中的风险传导。
基于GNN的供应商风险传导模型
import torch
import torch.nn.functional as F
from torch_geometric.nn import GCNConv
class SupplierRiskGNN(torch.nn.Module):
"""基于图卷积网络的供应商风险评估模型"""
def __init__(self, num_features, hidden_dim=64, num_classes=3):
"""
num_classes: 3类风险等级(低/中/高)
"""
super().__init__()
self.conv1 = GCNConv(num_features, hidden_dim)
self.conv2 = GCNConv(hidden_dim, hidden_dim)
self.conv3 = GCNConv(hidden_dim, num_classes)
self.dropout = torch.nn.Dropout(0.3)
def forward(self, x, edge_index):
x = F.relu(self.conv1(x, edge_index))
x = self.dropout(x)
x = F.relu(self.conv2(x, edge_index))
x = self.dropout(x)
x = self.conv3(x, edge_index)
return F.log_softmax(x, dim=1)
# 供应商特征示例
# [财务评分, 交货准时率, 地理风险, 历史违约次数, 产能利用率, 合同金额占比]
def build_supplier_features(suppliers: list) -> torch.Tensor:
features = []
for s in suppliers:
features.append([
s['financial_score'],
s['delivery_rate'],
s['geo_risk'],
s['default_count'],
s['capacity_utilization'],
s['contract_ratio']
])
return torch.tensor(features, dtype=torch.float)
def assess_supplier_risk(model, features, edge_index):
"""评估供应商风险"""
model.eval()
with torch.no_grad():
output = model(features, edge_index)
predictions = torch.argmax(output, dim=1)
risk_labels = ['低风险', '中风险', '高风险']
results = []
for i, pred in enumerate(predictions):
results.append({
'supplier_id': i,
'risk_level': risk_labels[pred.item()],
'risk_scores': torch.exp(output[i]).tolist()
})
return results
供应商综合评分卡
def supplier_scorecard(supplier_data: dict) -> dict:
"""
供应商综合评分卡
满分100分，各维度加权
"""
weights = {
'quality': 0.25, # 质量
'delivery': 0.25, # 交付
'cost': 0.20, # 成本
'flexibility': 0.15, # 灵活性
'sustainability': 0.15 # 可持续性
}
scores = {}
# 质量评分（基于良品率和退货率）
quality_rate = supplier_data.get('quality_rate', 0.95)
return_rate = supplier_data.get('return_rate', 0.02)
scores['quality'] = min(100, quality_rate * 100 - return_rate * 200)
# 交付评分（基于准时率和提前期）
on_time_rate = supplier_data.get('on_time_rate', 0.90)
avg_lead_time = supplier_data.get('avg_lead_time', 7)
scores['delivery'] = on_time_rate * 70 + max(0, (14 - avg_lead_time)) * 2.5
# 成本评分（基于价格竞争力）
price_index = supplier_data.get('price_index', 1.0) # 相对市场均价
scores['cost'] = max(0, min(100, (2 - price_index) * 60))
# 灵活性评分
min_order = supplier_data.get('min_order_qty', 100)
response_time = supplier_data.get('response_hours', 24)
scores['flexibility'] = max(0, 100 - min_order * 0.2 - response_time * 1.5)
# 可持续性评分
esg_score = supplier_data.get('esg_score', 50)
carbon_footprint = supplier_data.get('carbon_tons', 100)
scores['sustainability'] = esg_score * 0.6 + max(0, (200 - carbon_footprint)) * 0.2
# 加权总分
total = sum(scores[k] * weights[k] for k in weights)
# 等级判定
if total >= 85:
grade = 'A (优秀供应商)'
elif total >= 70:
grade = 'B (合格供应商)'
elif total >= 55:
grade = 'C (需改进)'
else:
grade = 'D (淘汰风险)'
return {
'total_score': round(total, 1),
'grade': grade,
'dimension_scores': {k: round(v, 1) for k, v in scores.items()},
'improvement_areas': [k for k, v in scores.items() if v < 60]
}
6. 质量检测与缺陷识别
在制造业供应链中，AI视觉检测正在替代大量人工质检工作，检测精度和速度都有显著提升。
基于深度学习的缺陷检测
import torch
import torchvision.models as models
import torchvision.transforms as transforms
from PIL import Image
class QualityInspector:
"""产品质量AI检测器"""
# 常见缺陷类型
DEFECT_TYPES = [
'scratch', # 划痕
'dent', # 凹陷
'discoloration',# 色差
'crack', # 裂纹
'bubble', # 气泡
'foreign_body', # 异物
'normal' # 正常
]
def __init__(self, model_path: str = None):
self.model = models.resnet50(pretrained=True)
self.model.fc = torch.nn.Linear(2048, len(self.DEFECT_TYPES))
if model_path:
self.model.load_state_dict(torch.load(model_path))
self.model.eval()
self.transform = transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(
mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]
)
])
def inspect(self, image_path: str, threshold: float = 0.7) -> dict:
"""
检测单张产品图片
:return: 检测结果，包含缺陷类型和置信度
"""
image = Image.open(image_path).convert('RGB')
input_tensor = self.transform(image).unsqueeze(0)
with torch.no_grad():
output = self.model(input_tensor)
probabilities = torch.softmax(output, dim=1)[0]
max_prob, max_idx = torch.max(probabilities, 0)
defect_type = self.DEFECT_TYPES[max_idx.item()]
result = {
'defect_type': defect_type,
'confidence': max_prob.item(),
'is_defective': defect_type != 'normal' and max_prob.item() > threshold,
'all_scores': {
self.DEFECT_TYPES[i]: round(probabilities[i].item(), 4)
for i in range(len(self.DEFECT_TYPES))
}
}
# 多缺陷检测（多个类别超过阈值）
for i, prob in enumerate(probabilities):
if prob.item() > threshold and self.DEFECT_TYPES[i] != 'normal':
if 'secondary_defects' not in result:
result['secondary_defects'] = []
result['secondary_defects'].append({
'type': self.DEFECT_TYPES[i],
'confidence': prob.item()
})
return result
def batch_inspect(self, image_paths: list) -> list:
"""批量检测"""
results = []
for path in image_paths:
try:
result = self.inspect(path)
result['image_path'] = path
results.append(result)
except Exception as e:
results.append({
'image_path': path,
'error': str(e)
})
# 统计汇总
defective = sum(1 for r in results if r.get('is_defective'))
print(f"检测完成: 共{len(results)}件, 不良{defective}件, "
f"不良率 {defective/len(results):.2%}")
return results
7. 供应链可视化与数字孪生
数字孪生技术将物理供应链映射到虚拟空间，实现实时监控和仿真推演。
供应链网络仿真
import simpy
import random
import statistics
class SupplyChainSimulation:
"""供应链离散事件仿真 - 数字孪生"""
def __init__(self, env):
self.env = env
self.metrics = {
'orders_processed': 0,
'total_lead_time': [],
'stockouts': 0,
'backorders': []
}
def supplier_process(self, env, supplier_name, lead_time_mean, lead_time_std):
"""供应商供货流程"""
while True:
lead_time = max(1, random.gauss(lead_time_mean, lead_time_std))
yield env.timeout(lead_time)
# 供货到达
return lead_time
def warehouse_process(self, env, capacity, reorder_point, reorder_qty):
"""仓库库存管理流程"""
inventory = capacity
pending_orders = []
while True:
# 检查是否需要补货
if inventory <= reorder_point and not pending_orders:
# 触发补货
lead_time = yield env.process(
self.supplier_process(env, 'S1', 7, 2)
)
inventory += reorder_qty
yield env.timeout(1) # 每天检查一次
def customer_demand_process(self, env, demand_mean, demand_std):
"""客户需求生成"""
while True:
demand = max(0, int(random.gauss(demand_mean, demand_std)))
if demand > 0:
self.metrics['orders_processed'] += demand
yield env.timeout(1) # 每天产生需求
def run_simulation(self, days: int = 365):
"""运行仿真"""
env = simpy.Environment()
# 初始化各节点
env.process(self.customer_demand_process(env, 50, 15))
env.process(self.warehouse_process(env, 1000, 200, 500))
env.run(until=days)
return {
'total_orders': self.metrics['orders_processed'],
'avg_lead_time': statistics.mean(self.metrics['total_lead_time'])
if self.metrics['total_lead_time'] else 0,
'stockout_rate': self.metrics['stockouts'] / max(1, self.metrics['orders_processed'])
}
实时仪表盘数据服务
from dataclasses import dataclass, field
from typing import List, Dict
import json
@dataclass
class SupplyChainDashboard:
"""供应链实时仪表盘"""
warehouses: Dict[str, dict] = field(default_factory=dict)
shipments: List[dict] = field(default_factory=list)
alerts: List[dict] = field(default_factory=list)
def update_warehouse(self, warehouse_id: str, data: dict):
self.warehouses[warehouse_id] = {
**data,
'updated_at': 'now'
}
self._check_alerts(warehouse_id)
def _check_alerts(self, warehouse_id: str):
wh = self.warehouses[warehouse_id]
# 低库存报警
if wh.get('utilization', 0) > 90:
self.alerts.append({
'type': 'HIGH_UTILIZATION',
'warehouse': warehouse_id,
'message': f"仓库 {warehouse_id} 利用率超过90%",
'severity': 'warning'
})
# 库存不足报警
for sku, qty in wh.get('sku_quantities', {}).items():
if qty < wh.get('reorder_points', {}).get(sku, 0):
self.alerts.append({
'type': 'LOW_STOCK',
'warehouse': warehouse_id,
'sku': sku,
'message': f"SKU {sku} 库存低于再订货点",
'severity': 'critical'
})
def add_shipment(self, shipment: dict):
self.shipments.append(shipment)
def get_kpi_summary(self) -> dict:
total_inventory = sum(
sum(wh.get('sku_quantities', {}).values())
for wh in self.warehouses.values()
)
active_shipments = len([s for s in self.shipments if s.get('status') == 'in_transit'])
critical_alerts = len([a for a in self.alerts if a.get('severity') == 'critical'])
return {
'total_inventory_units': total_inventory,
'active_shipments': active_shipments,
'critical_alerts': critical_alerts,
'warehouse_count': len(self.warehouses),
'alert_summary': {
'critical': critical_alerts,
'warning': len([a for a in self.alerts if a.get('severity') == 'warning'])
}
}
def to_json(self) -> str:
return json.dumps({
'warehouses': self.warehouses,
'shipments_count': len(self.shipments),
'kpi': self.get_kpi_summary(),
'recent_alerts': self.alerts[-10:]
}, ensure_ascii=False, indent=2)
8. 智能客服与售后管理
供应链末端的客户服务同样受益于AI技术。智能客服可以处理订单查询、物流追踪、退换货等常见问题。
供应链智能客服系统
from dataclasses import dataclass
from typing import Optional
import re
@dataclass
class CustomerQuery:
query_text: str
customer_id: str
order_id: Optional[str] = None
intent: Optional[str] = None
sentiment: Optional[str] = None
class SupplyChainChatbot:
"""供应链智能客服"""
INTENT_PATTERNS = {
'track_order': [
r'物流|快递|发货|到哪了|运单|追踪',
r'order.*track|shipping.*status|delivery.*where'
],
'return_exchange': [
r'退|换货|退货|退款|质量问题|破损',
r'return|exchange|refund|defect'
],
'complaint': [
r'投诉|不满|差评|太慢|等很久|态度差',
r'complaint|unacceptable|too slow'
],
'order_modify': [
r'改地址|取消订单|修改|换地址',
r'change.*address|cancel.*order|modify'
],
'product_inquiry': [
r'有没有|库存|到货|补货|什么时候有',
r'in stock|availability|restock|when.*available'
]
}
def classify_intent(self, query: CustomerQuery) -> str:
"""意图识别"""
text = query.query_text.lower()
for intent, patterns in self.INTENT_PATTERNS.items():
for pattern in patterns:
if re.search(pattern, text):
return intent
return 'general_inquiry'
def analyze_sentiment(self, text: str) -> str:
"""简单情感分析"""
negative_words = ['差', '慢', '坏', '烂', '垃圾', '失望', '生气', '投诉']
positive_words = ['好', '快', '满意', '棒', '感谢', '不错', '优秀']
neg_count = sum(1 for w in negative_words if w in text)
pos_count = sum(1 for w in positive_words if w in text)
if neg_count > pos_count:
return 'negative'
elif pos_count > neg_count:
return 'positive'
return 'neutral'
def generate_response(self, query: CustomerQuery) -> str:
"""生成回复"""
query.intent = self.classify_intent(query)
query.sentiment = self.analyze_sentiment(query.query_text)
responses = {
'track_order': self._handle_tracking(query),
'return_exchange': self._handle_return(query),
'complaint': self._handle_complaint(query),
'order_modify': self._handle_modify(query),
'product_inquiry': self._handle_inquiry(query),
'general_inquiry': "您好，请问有什么可以帮您？您可以询问物流状态、退换货、订单修改等问题。"
}
return responses.get(query.intent, responses['general_inquiry'])
def _handle_tracking(self, query):
if query.order_id:
return (f"正在为您查询订单 {query.order_id} 的物流状态...\n"
f"📦 当前状态：运输中\n"
f"🚚 预计送达：明天下午\n"
f"📍 最新位置：XX转运中心")
return "请提供您的订单号，我来帮您查询物流状态。"
def _handle_return(self, query):
return ("退货流程如下：\n"
"1. 在订单详情页点击「申请退换货」\n"
"2. 选择退换货原因并上传照片\n"
"3. 审核通过后获取退货地址\n"
"4. 寄回商品，退款将在收到后3个工作日内处理")
def _handle_complaint(self, query):
if query.sentiment == 'negative':
return ("非常抱歉给您带来了不好的体验！🙏\n"
"我会立即为您升级处理，请您提供订单号和具体问题，"
"我们的高级客服将在30分钟内联系您。")
return "感谢您的反馈，我们会认真对待。请问具体是什么问题呢？"
def _handle_modify(self, query):
return ("订单修改说明：\n"
"- 未发货订单：可直接在订单页修改地址或取消\n"
"- 已发货订单：请联系客服拦截，可能产生额外费用\n"
"- 请提供您的订单号和需要修改的内容")
def _handle_inquiry(self, query):
return ("商品库存查询：请告诉我您想了解的商品名称或SKU编号，"
"我来帮您查询实时库存和预计到货时间。")
9. 可持续供应链优化
绿色供应链是企业ESG战略的重要组成部分。AI可以帮助优化碳排放、减少浪费、选择环保供应商。
碳排放追踪与优化
from dataclasses import dataclass
from typing import List
@dataclass
class CarbonFootprint:
"""碳足迹计算与优化"""
# 各运输方式碳排放因子 (kg CO2 / 吨公里)
EMISSION_FACTORS = {
'truck': 0.062,
'rail': 0.022,
'sea': 0.008,
'air': 0.602,
'electric_truck': 0.025
}
def calculate_transport_emission(self, weight_tons: float,
distance_km: float,
transport_mode: str) -> float:
"""计算单次运输碳排放（kg CO2）"""
factor = self.EMISSION_FACTORS.get(transport_mode, 0.062)
return weight_tons * distance_km * factor
def optimize_transport_mode(self, weight_tons: float,
distance_km: float,
max_hours: float,
carbon_budget_kg: float) -> dict:
"""
在时间和碳排放约束下选择最优运输方式
"""
MODE_SPEEDS = {
'truck': 60, 'rail': 80, 'sea': 30,
'air': 800, 'electric_truck': 50
}
feasible_modes = []
for mode in self.EMISSION_FACTORS:
hours = distance_km / MODE_SPEEDS[mode]
emission = self.calculate_transport_emission(
weight_tons, distance_km, mode
)
if hours <= max_hours and emission <= carbon_budget_kg:
feasible_modes.append({
'mode': mode,
'hours': round(hours, 1),
'emission_kg': round(emission, 1),
'cost_index': self._cost_index(mode, distance_km, weight_tons)
})
if not feasible_modes:
return {'error': '无可行运输方案，请放宽时间或碳排放约束'}
# 按碳排放排序，推荐最绿色方案
feasible_modes.sort(key=lambda x: x['emission_kg'])
return {
'recommended': feasible_modes[0],
'alternatives': feasible_modes[1:],
'emission_saved_vs_truck': round(
self.calculate_transport_emission(weight_tons, distance_km, 'truck')
- feasible_modes[0]['emission_kg'], 1
)
}
def _cost_index(self, mode, distance, weight):
costs = {'truck': 1.0, 'rail': 0.6, 'sea': 0.3, 'air': 5.0, 'electric_truck': 0.8}
return round(costs.get(mode, 1.0) * distance * weight / 1000, 2)
def generate_carbon_report(self, shipments: list) -> dict:
"""生成碳排放报告"""
total_emission = 0
by_mode = {}
for s in shipments:
emission = self.calculate_transport_emission(
s['weight'], s['distance'], s['mode']
)
total_emission += emission
by_mode[s['mode']] = by_mode.get(s['mode'], 0) + emission
return {
'total_emission_kg': round(total_emission, 1),
'total_emission_tons': round(total_emission / 1000, 2),
'by_mode': {k: round(v, 1) for k, v in by_mode.items()},
'trees_to_offset': int(total_emission / 21.77), # 每棵树年均吸收21.77kg CO2
'recommendation': self._green_recommendations(by_mode)
}
def _green_recommendations(self, by_mode):
recs = []
if by_mode.get('air', 0) > by_mode.get('sea', 0) * 0.5:
recs.append("空运占比过高，建议非紧急货物改用海运或铁路")
if by_mode.get('truck', 0) > sum(by_mode.values()) * 0.7:
recs.append("公路运输占比超70%，建议中长途改用铁路联运")
if not recs:
recs.append("当前运输结构较为合理，继续保持")
return recs
10. 实战案例：智能物流调度系统
将前面的技术整合为一个完整的物流调度系统：
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import random
@dataclass
class Vehicle:
id: str
capacity: float # 载重量(吨)
current_location: tuple # 当前位置
available_at: datetime # 可用时间
cost_per_km: float # 每公里成本
vehicle_type: str = 'truck'
@dataclass
class Order:
id: str
pickup_location: tuple
delivery_location: tuple
weight: float
deadline: datetime
priority: int = 1 # 1=普通, 2=加急, 3=特急
class SmartLogisticsDispatcher:
"""智能物流调度系统"""
def __init__(self):
self.vehicles: List[Vehicle] = []
self.pending_orders: List[Order] = []
self.dispatch_history: List[dict] = []
def add_vehicle(self, vehicle: Vehicle):
self.vehicles.append(vehicle)
def submit_order(self, order: Order):
self.pending_orders.append(order)
# 按优先级和截止时间排序
self.pending_orders.sort(
key=lambda o: (-o.priority, o.deadline)
)
def _distance(self, p1: tuple, p2: tuple) -> float:
return ((p1[0]-p2[0])**2 + (p1[1]-p2[1])**2) ** 0.5
def _estimate_time(self, distance: float, speed: float = 40) -> float:
"""估算行驶时间（小时）"""
return distance / speed
def dispatch(self) -> List[dict]:
"""执行智能调度"""
assignments = []
for order in self.pending_orders[:]:
best_vehicle = None
best_score = float('inf')
for vehicle in self.vehicles:
# 检查容量
if vehicle.capacity < order.weight:
continue
# 计算距离和成本
to_pickup = self._distance(vehicle.current_location,
order.pickup_location)
pickup_to_delivery = self._distance(order.pickup_location,
order.delivery_location)
total_km = to_pickup + pickup_to_delivery
# 估算到达时间
eta_hours = self._estimate_time(total_km)
eta_time = vehicle.available_at + timedelta(hours=eta_hours)
# 检查是否满足截止时间
if eta_time > order.deadline:
continue
# 综合评分（成本 + 时间紧迫度 + 优先级匹配）
time_buffer = (order.deadline - eta_time).total_seconds() / 3600
cost_score = total_km * vehicle.cost_per_km
urgency_score = max(0, 10 - time_buffer) * 10
priority_bonus = -order.priority * 20 if order.priority >= 2 else 0
total_score = cost_score + urgency_score + priority_bonus
if total_score < best_score:
best_score = total_score
best_vehicle = vehicle
if best_vehicle:
to_pickup = self._distance(best_vehicle.current_location,
order.pickup_location)
pickup_to_delivery = self._distance(order.pickup_location,
order.delivery_location)
total_km = to_pickup + pickup_to_delivery
assignment = {
'order_id': order.id,
'vehicle_id': best_vehicle.id,
'total_km': round(total_km, 1),
'estimated_cost': round(total_km * best_vehicle.cost_per_km, 2),
'eta': (best_vehicle.available_at +
timedelta(hours=self._estimate_time(total_km))).isoformat(),
'route': [
best_vehicle.current_location,
order.pickup_location,
order.delivery_location
]
}
assignments.append(assignment)
# 更新车辆状态
best_vehicle.current_location = order.delivery_location
best_vehicle.available_at += timedelta(
hours=self._estimate_time(total_km) + 0.5 # 加30分钟装卸时间
)
best_vehicle.capacity -= order.weight
self.pending_orders.remove(order)
self.dispatch_history.extend(assignments)
return assignments
def get_dashboard(self) -> dict:
"""调度仪表盘"""
return {
'available_vehicles': len([v for v in self.vehicles
if v.available_at <= datetime.now()]),
'pending_orders': len(self.pending_orders),
'total_dispatched': len(self.dispatch_history),
'total_distance_km': sum(a['total_km'] for a in self.dispatch_history),
'total_cost': sum(a['estimated_cost'] for a in self.dispatch_history),
'avg_cost_per_order': (
sum(a['estimated_cost'] for a in self.dispatch_history) /
max(1, len(self.dispatch_history))
)
}
# 使用示例
dispatcher = SmartLogisticsDispatcher()
# 添加车辆
for i in range(5):
dispatcher.add_vehicle(Vehicle(
id=f'TRUCK-{i+1:03d}',
capacity=random.uniform(5, 15),
current_location=(random.uniform(0, 100), random.uniform(0, 100)),
available_at=datetime.now(),
cost_per_km=random.uniform(2.0, 4.0)
))
# 提交订单
for i in range(8):
dispatcher.submit_order(Order(
id=f'ORD-{i+1:04d}',
pickup_location=(random.uniform(0, 100), random.uniform(0, 100)),
delivery_location=(random.uniform(0, 100), random.uniform(0, 100)),
weight=random.uniform(0.5, 5),
deadline=datetime.now() + timedelta(hours=random.randint(4, 24)),
priority=random.choice([1, 1, 1, 2, 3])
))
# 执行调度
assignments = dispatcher.dispatch()
dashboard = dispatcher.get_dashboard()
print("=== 调度结果 ===")
for a in assignments:
print(f" 订单 {a['order_id']} → 车辆 {a['vehicle_id']} | "
f"{a['total_km']}km | ¥{a['estimated_cost']}")
print(f"\n=== 仪表盘 ===")
for k, v in dashboard.items():
label = {
'available_vehicles': '可用车辆',
'pending_orders': '待处理订单',
'total_dispatched': '已调度数',
'total_distance_km': '总里程(km)',
'total_cost': '总成本(¥)',
'avg_cost_per_order': '平均成本/单(¥)'
}.get(k, k)
print(f" {label}: {v}")
11. 企业级落地经验
技术选型建议
| 场景 | 推荐技术 | 原因 |
|---|---|---|
| 需求预测 | Prophet / LightGBM | 速度快，可解释性好 |
| 路径优化 | Google OR-Tools | 开源，工业级求解器 |
| 质量检测 | YOLOv8 / ResNet | 实时性好，精度高 |
| 风险评估 | GNN / XGBoost | 图关系建模能力强 |
| 仿真推演 | SimPy / AnyLogic | 灵活度高，支持复杂场景 |
分阶段实施路线
第一阶段（0-3月）：数据基础建设
├── 打通ERP/WMS/TMS数据接口
├── 建立统一数据仓库
└── 基础数据清洗与标准化
第二阶段（3-6月）：核心场景落地
├── 需求预测模型上线
├── 库存优化策略实施
└── 基础路径优化
第三阶段（6-12月）：智能化升级
├── 动态调度系统
├── 智能质检系统
└── 供应商风险监控
第四阶段（12月+）：全链路智能化
├── 数字孪生平台
├── 自主决策系统
└── 持续优化闭环
常见踩坑与应对
数据质量问题：供应链数据往往存在缺失、不一致、延迟等问题。建议在AI模型之前建立严格的数据质量监控层，设置数据完整性、时效性和准确性指标。当数据质量低于阈值时自动回退到规则引擎。
模型漂移：需求模式会随季节、市场变化而改变。需要建立模型监控机制，跟踪预测精度变化，当MAPE超过预设阈值时自动触发模型重训练。
组织阻力：AI系统改变了许多传统工作流程，一线员工可能产生抵触。建议从小范围试点开始，用实际数据证明价值，逐步扩大应用范围。始终保留人工干预通道。
系统集成：与遗留系统的集成通常是最大的工程挑战。建议采用API Gateway模式，通过中间层适配新旧系统，避免大规模改造现有IT基础设施。
关键指标体系
class SupplyChainKPI:
"""供应链核心KPI指标"""
@staticmethod
def forecast_accuracy(actual: list, predicted: list) -> float:
"""预测准确率 (1 - MAPE)"""
errors = [abs(a - p) / max(a, 1) for a, p in zip(actual, predicted)]
return 1 - sum(errors) / len(errors)
@staticmethod
def fill_rate(orders_fulfilled: int, orders_total: int) -> float:
"""订单满足率"""
return orders_fulfilled / max(orders_total, 1)
@staticmethod
def inventory_turnover(cogs: float, avg_inventory_value: float) -> float:
"""库存周转率"""
return cogs / max(avg_inventory_value, 1)
@staticmethod
def on_time_delivery_rate(ontime: int, total: int) -> float:
"""准时交付率"""
return ontime / max(total, 1)
@staticmethod
def perfect_order_rate(total_orders: int,
ontime: int, complete: int,
damage_free: int, accurate_docs: int) -> float:
"""完美订单率"""
return (ontime * complete * damage_free * accurate_docs) / max(total_orders ** 4, 1)
供应链AI优化是一项系统工程，需要技术、业务和组织的深度协同。从数据基础做起，选择高价值场景快速验证，逐步构建完整的智能供应链体系，是最务实的落地路径。
