AI在智能仓储中的应用：从库存预测到拣货路径优化的技术复盘

AI在智能仓储中的应用：从库存预测到拣货路径优化的技术复盘

仓储智能化不是堆算法，而是把每个决策点的几秒钟优化加起来——拣货员每天走3万步降到1.8万步，这就是AI的价值。

一、仓储的四个AI决策点

2023年我们参与了一个日订单量5万单的电商仓储改造项目。仓库面积2万平米，SKU超过15万个，拣货员120人，日均出库8万件。核心痛点：库存周转慢、拣货路径长、爆仓与断货交替出现。

我们将仓储AI化拆解为四个独立又关联的决策问题：

二、库存需求的时序预测

我们选用了两种模型做对比：

2.1 Prophet vs DeepAR

import pandas as pd
from prophet import Prophet
from gluonts.model.deepar import DeepAREstimator
from gluonts.mx import Trainer

class InventoryForecastService:
"""
库存需求预测
Prophet: 适合有明显季节性和趋势的SKU
DeepAR: 适合SKU数量多且需要联合建模的场景
"""

def forecast_with_prophet(self, sku_id: str, history: pd.DataFrame,
periods: int = 30) -> pd.DataFrame:
"""Prophet单SKU预测，适合TOP 3000高频SKU"""
df = history[['ds', 'y']].copy()

model = Prophet(
growth='linear',
yearly_seasonality=True,
weekly_seasonality=True,
daily_seasonality=False,
changepoint_prior_scale=0.05,
seasonality_prior_scale=10.0,
holidays=self._build_holidays_df(), # 618/双11/年货节
interval_width=0.90 # 90%置信区间
)

# 添加促销事件作为额外回归器
df['is_promotion'] = history['is_promotion'].values
model.add_regressor('is_promotion', mode='additive')

model.fit(df)

future = model.make_future_dataframe(periods=periods, freq='D')
future['is_promotion'] = self._get_future_promotions(periods)

forecast = model.predict(future)

# 返回预测分位数（用于安全库存计算）
return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(periods)

def forecast_with_deepar(self, sku_list: List[str],
history_dict: Dict[str, pd.DataFrame],
periods: int = 30) -> Dict[str, pd.DataFrame]:
"""DeepAR多SKU联合预测，适合长尾SKU"""
# 构建GluonTS数据集
from gluonts.dataset.common import ListDataset

train_data = []
for sku_id in sku_list:
series = history_dict[sku_id]
train_data.append({
'start': series['ds'].min(),
'target': series['y'].values,
'feat_static_cat': [self._get_category_id(sku_id)],
'item_id': sku_id
})

estimator = DeepAREstimator(
freq='D',
prediction_length=periods,
context_length=min(90, len(series)),
num_layers=3,
num_cells=64,
cell_type='lstm',
dropout_rate=0.1,
trainer=Trainer(epochs=100, learning_rate=1e-3)
)

predictor = estimator.train(ListDataset(train_data, freq='D'))

results = {}
for forecast in predictor.predict(ListDataset(train_data, freq='D')):
sku_id = forecast.item_id
samples = forecast.samples # shape: (num_samples, prediction_length)
results[sku_id] = pd.DataFrame({
'ds': pd.date_range(start=forecast.start_date, periods=periods, freq='D'),
'yhat': samples.mean(axis=0),
'yhat_lower': np.percentile(samples, 5, axis=0),
'yhat_upper': np.percentile(samples, 95, axis=0),
})

return results

2.2 ABC分类与安全库存

def calculate_safety_stock(sku_id: str, forecast: pd.DataFrame,
service_level: float = 0.95) -> dict:
"""
安全库存 = Z_score × σ_demand × √(lead_time)

– A类（前20% SKU，占80%销售额）：service_level=0.99
– B类（中30% SKU，占15%销售额）：service_level=0.95
– C类（后50% SKU，占5%销售额）：service_level=0.85
"""
abc_class = get_abc_class(sku_id)

service_level_map = {'A': 0.99, 'B': 0.95, 'C': 0.85}
sl = service_level_map.get(abc_class, 0.90)

# Z-score from standard normal distribution
from scipy.stats import norm
z_score = norm.ppf(sl)

# 需求标准差（从90%预测区间反推）
sigma_demand = (forecast['yhat_upper'] – forecast['yhat_lower']).mean() / (2 * 1.645)

# 补货提前期(天)
lead_time = get_lead_time(sku_id)

safety_stock = z_score * sigma_demand * np.sqrt(lead_time)

return {
'sku_id': sku_id,
'abc_class': abc_class,
'service_level': sl,
'avg_daily_demand': forecast['yhat'].mean(),
'safety_stock': int(np.ceil(safety_stock)),
'reorder_point': int(np.ceil(forecast['yhat'].mean() * lead_time + safety_stock)),
'suggested_order_qty': int(np.ceil(max(
forecast['yhat'].sum() – get_current_stock(sku_id) + safety_stock, 0
)))
}

三、拣货路径的TSP优化

一张拣货单通常包含15-30个SKU，拣货员需要走过所有这些库位。这就是经典的TSP（旅行商问题）。

@Service
public class PickingPathOptimizer {

/**
* 使用Lin-Kernighan启发式算法（LKH）求解TSP
* 对于20个拣货点，LKH能在100ms内找到最优解
*/
public PickingRoute optimize(List<PickLocation> locations, Location startPoint) {
int n = locations.size();

// 构建距离矩阵（实际行走距离，非欧氏距离）
double[][] distanceMatrix = buildDistanceMatrix(locations);

// LKH求解
LKHSolver solver = new LKHSolver(distanceMatrix);
int[] tour = solver.solve();

// 转换为实际路径
List<PickLocation> optimalPath = new ArrayList<>();
optimalPath.add(startPoint); // 起点：拣货台
for (int idx : tour) {
optimalPath.add(locations.get(idx));
}
optimalPath.add(startPoint); // 回到拣货台

double totalDistance = calculateTotalDistance(optimalPath, distanceMatrix);

return new PickingRoute(optimalPath, totalDistance,
estimatePickingTime(totalDistance, locations.size()));
}

/**
* 构建仓库实际行走距离矩阵
* 关键：库位之间的实际距离 ≠ 欧氏距离
* 需要考虑通道走向、单向通道、楼梯/货梯
*/
private double[][] buildDistanceMatrix(List<PickLocation> locations) {
int n = locations.size();
double[][] matrix = new double[n][n];

for (int i = 0; i < n; i++) {
for (int j = i + 1; j < n; j++) {
PickLocation a = locations.get(i);
PickLocation b = locations.get(j);

// 同通道内：直接距离
if (a.getAisleId().equals(b.getAisleId())) {
matrix[i][j] = Math.abs(a.getShelfPosition() – b.getShelfPosition());
} else {
// 跨通道：走到通道口 + 横向移动 + 走到目标位置
matrix[i][j] = crossAisleDistance(a, b);
}
matrix[j][i] = matrix[i][j];
}
}
return matrix;
}
}

3.1 优化效果

在真实仓库数据上的测试：

LKH优化后每人日均拣货效率提升81%，人力成本节省约40%。

四、AGV调度的多智能体协调

50台AGV在仓库中运行，核心挑战是死锁避免和路径冲突解决。

@Service
public class AGVScheduler {

// 仓库地图：将仓库建模为网格图
private final GridMap warehouseMap;
// 每台AGV的预留路径（时间维度的路径占用）
private final Map<String, ReservedPath> reservedPaths = new ConcurrentHashMap<>();

/**
* 基于时空A*的AGV路径规划
* 关键：路径不仅包含空间坐标，还包含时间维度
* 这样不同AGV可以在不同时间使用同一空间位置
*/
public AGVPath planPath(String agvId, Location from, Location to) {
// 时空A*：状态 = (x, y, t)
PriorityQueue<STNode> openSet = new PriorityQueue<>(
Comparator.comparingDouble(n -> n.gCost + n.hCost)
);

Map<String, STNode> closedSet = new HashMap<>();

STNode start = new STNode(from.x, from.y, currentTime(), 0, heuristic(from, to));
openSet.add(start);

while (!openSet.isEmpty()) {
STNode current = openSet.poll();
String key = current.key();

if (closedSet.containsKey(key)) continue;
closedSet.put(key, current);

// 到达目标位置
if (current.x == to.x && current.y == to.y) {
return reconstructPath(current);
}

// 扩展邻居（5个动作：上下左右 + 等待）
for (Action action : ACTIONS) {
int nx = current.x + action.dx;
int ny = current.y + action.dy;
long nt = current.time + action.duration;

// 边界检查
if (!warehouseMap.isValid(nx, ny)) continue;

// 冲突检查：该位置在该时间是否已被其他AGV预留
if (isReserved(nx, ny, nt, agvId)) continue;

double ng = current.gCost + action.cost;
STNode neighbor = new STNode(nx, ny, nt, ng, heuristic(nx, ny, to));
openSet.add(neighbor);
}
}

throw new NoPathFoundException(from, to);
}

/**
* 死锁检测与恢复
*/
@Scheduled(fixedDelay = 1000)
public void detectDeadlock() {
// 构建等待图：AGV A等待AGV B释放资源
WaitGraph graph = new WaitGraph();
for (AGV agv : agvRegistry.getAllActive()) {
if (agv.isWaiting()) {
List<String> blockingAgvs = findBlockingAgvs(agv);
for (String blocker : blockingAgvs) {
graph.addEdge(agv.getId(), blocker);
}
}
}

// 检测环 → 死锁
List<List<String>> cycles = graph.findCycles();
for (List<String> cycle : cycles) {
// 死锁恢复：让优先级最低的AGV重新规划路径
String victim = selectVictim(cycle);
agvRegistry.get(victim).replanPath();
log.warn("死锁检测到，牺牲AGV: {}, 死锁环: {}", victim, cycle);
}
}
}

五、总结

智能仓储的AI落地有三个核心认知：

最终效果：库存周转天数从45天降到28天，拣货效率提升81%，AGV利用率从62%提升到88%。仓储AI不是炫技，是每平方米、每分钟、每一步的优化累积。