仓储机器人库位优化算法 —— 基于存取频次的动态热区调度“那年电商大促仓库里最忙的几台 AGV 每天要在货架间跑 80km结果发现爆款商品全被放在最角落。后来我们用频次-距离加权算法重构了库位分配策略把高频物料‘吸’到出入口附近单仓整体搬运距离直接砍掉了 42%。”—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸一、实际应用场景描述在电商分拣中心、汽车零配件仓、冷链物流库等场景AGV/AMR 需要在海量货架中完成高频拣选┌──────────────────────────────────────────────┐│ 仓储机器人智能库位优化系统 ││ ││ [WMS 上层调度系统] ││ │ 入库请求 / 出库订单 / 库存状态 ││ ▼ ││ ┌────────────────────────────┐ ││ │ 数据分析层 │ ││ │ ┌──────────────────────┐ │ ││ │ │ 1. 历史订单挖掘 │ │ ││ │ │ (SKU 频次统计) │ │ ││ │ └──────────────────────┘ │ ││ │ ┌──────────────────────┐ │ ││ │ │ 2. ABC 分类算法 │ │ ││ │ │ (帕累托分析) │ │ ││ │ └──────────────────────┘ │ ││ │ ┌──────────────────────┐ │ ││ │ │ 3. 关联规则挖掘 │ │ ││ │ │ (捆绑商品共置) │ │ ││ └────────────┬───────────────┘ ││ │ 物料热度画像 ││ ┌───────┴───────┐ ││ ▼ ▼ ││ ┌─────────┐ ┌─────────┐ ││ │ 库位建模 │ │ 距离评估器 │ ││ │ • 坐标体系 │ │ • 曼哈顿距 │ ││ │ • 热区划分 │ │ • 路径代价 │ ││ │ • 容量约束 │ │ • 拥堵权重 │ ││ └────┬────┘ └────┬────┘ ││ │ 库位拓扑热度 │ 代价矩阵C[i][j] ││ ▼ ▼ ││ ┌────────────────────────────┐ ││ │ 库位分配优化引擎 │ ││ │ • 目标: min Σ(f_i × d_i) │ ││ │ • 约束: 容量/互斥/稳定性 │ ││ │ • 算法: 贪心局部搜索 │ ││ └────────────┬───────────────┘ ││ │ 最优库位映射 ││ ▼ ││ ┌────────────────────────────┐ ││ │ 物理世界 (立体仓库) │ ││ │ 热区 (出入口5m内) │ ││ │ ⚡ 温区 (出入口5~15m) │ ││ │ ❄️ 冷区 (出入口15m外) │ ││ │ SKU-A(高频)→热区 │ ││ │ SKU-B(低频)→冷区 │ ││ └───────────────────────────┘ ││ ││ 核心: 频次-距离加权 ABC分类 动态重排 │└──────────────────────────────────────────────┘传统固定库位 vs 智能热区调度维度 传统固定库位 智能热区调度爆款位置 ❌ 随机分布常在角落 ✅ 自动吸附至出入口搬运效率 ❌ 平均距离长拥堵严重 ✅ 整体距离缩短 30~50%库位调整 ❌ 人工盘点后手动挪货 ✅ 基于数据自动优化冷热均衡 ❌ 热品挤占通道 ✅ 分层分区动态平衡突发应对 ❌ 大促期间瘫痪 ✅ 弹性扩容热区二、引入痛点2.1 现场的真实困境场景 现场发生了什么 根因“爆款绕远” “每天出库 2000 次的电池放在最远端” 无频次感知“AGV 空跑” “70% 的时间花在路上而非作业” 库位未优化“通道堵塞” “热销品集中AGV 排队取货” 无拥堵建模“人工搬库” “半年一次大盘点挪动上万箱” 无动态重排“大促崩盘” “双11期间拣选效率下降 40%” 静态库位无法应对峰值2.2 核心矛盾仓库的本质不是“存”而是“取”。 传统库位管理只关注“放得下”而忽略了“好取出”。解决方案是建立“频次-距离”联合优化模型将高频物料动态映射到出入口附近的“热区”同时兼顾品类关联性与库位稳定性。2.3 我们要解决什么用一段精简的 Python 程序构建一个 仓储机器人库位优先优化系统实现1. 频次统计 —— 基于历史订单计算 SKU 热度2. ABC 分类 —— 帕累托法则划分热/温/冷区3. 距离建模 —— 曼哈顿距离 拥堵权重4. 优化分配 —— 贪心 局部搜索最小化总搬运代价5. 可视化 —— 展示库位热力图与优化效果三、核心逻辑讲解3.1 理论基础频次-距离加权优化本工具基于哈工程《工业过程控制》第四章“线性规划”、第十三章“最优控制”和运筹学基础① 优化目标函数设仓库有 N 个库位 M 种物料。定义- f_i 物料 i 的历史存取频次权重- d_{ij} 物料 i 分配到库位 j 的搬运距离- x_{ij} \in \{0,1\} 分配决策变量目标最小化加权总搬运距离\min Z \sum_{i1}^{M} \sum_{j1}^{N} f_i \cdot d_{ij} \cdot x_{ij}② 约束条件\begin{aligned}\sum_{j1}^{N} x_{ij} 1, \quad \forall i \in M \quad \text{(每物料仅一个库位)}\\\sum_{i1}^{M} x_{ij} \le 1, \quad \forall j \in N \quad \text{(每库位至多存一种物料)}\\x_{ij} \in \{0,1\} \quad \text{(整数约束)}\end{aligned}③ 距离模型曼哈顿距离考虑 AGV 只能沿通道直行/直角转弯d_{ij} |x_i - x_j| |y_i - y_j|引入拥堵权重 \omega_j 通道交叉口权重更高d_{ij} d_{ij} \cdot \omega_j④ ABC 分类帕累托法则按累计频次占比划分- A 类热区累计占比 70%~80%应放置在出入口 5m 内- B 类温区累计占比 15%~25%放置在 5~15m- C 类冷区累计占比 5%~10%放置在 15m 外3.2 算法架构总览┌─────────────┐│ 历史订单数据 ││ (SKU, 时间戳) │└──────┬──────┘│┌─────────▼─────────┐│ 频次统计分析 ││ • 周期聚合 ││ • 滑动窗口 ││ • 趋势预测 │└─────────┬─────────┘│ 频次向量 f┌─────────▼─────────┐│ ABC分类器 ││ • 帕累托排序 ││ • 动态阈值 │└─────────┬─────────┘│ A/B/C标签┌─────────▼─────────┐│ 库位拓扑建模 ││ • 坐标网格 ││ • 热区划分 ││ • 拥堵权重 │└─────────┬─────────┘│ 距离矩阵 D┌─────────▼─────────┐│ 优化分配引擎 ││ • 贪心初始化 ││ • 局部搜索优化 ││ • 交换/插入算子 │└─────────┬─────────┘│ 最优分配方案▼┌─────────────┐│ WMS/AGV调度 │└─────────────┘四、代码讲解面向对象设计4.1 类结构总览类名 职责 设计模式SKU 物料单元dataclass 值对象StorageLocation 库位单元dataclass 值对象WarehouseLayout 仓库拓扑与距离计算 组合模式FrequencyAnalyzer 频次统计与 ABC 分类 策略模式AllocationOptimizer 库位优化分配引擎 模板方法WarehouseSimulator 仿真与效果评估 聚合根VisualizationEngine 可视化引擎 封装4.2 核心代码完整可运行完整源码约 550 行包含 7 个类、仿真引擎、可视化、优化算法。以下为精简核心版完整代码可直接复制运行。detailssummary 完整源码点击展开/折叠/summary仓储机器人库位优化算法 —— 基于频次-距离加权参考哈尔滨工程大学《工业过程控制》第四章线性规划from dataclasses import dataclass, fieldfrom typing import List, Dict, Tuple, Optionalimport numpy as npimport matplotlib.pyplot as pltfrom collections import defaultdict, Counterimport mathimport random# # 1. 基础数据结构# dataclassclass SKU:物料单元 —— 值对象id: strname: strfrequency: int 0 # 历史存取频次category: str C # ABC分类: A(热)/B(温)/C(冷)volume: float 1.0 # 占用体积weight: float 1.0 # 重量affinity_group: Optional[str] None # 关联组捆绑商品def __lt__(self, other):return self.frequency other.frequency # 按频次降序dataclassclass StorageLocation:库位单元 —— 值对象id: strx: int # 网格坐标Xy: int # 网格坐标Yzone: str cold # 区域: hot/warm/coldcongestion_weight: float 1.0 # 拥堵权重capacity: float 1.0 # 容量current_sku: Optional[SKU] Nonepropertydef distance_to_entrance(self) - int:到出入口(0,0)的曼哈顿距离return abs(self.x) abs(self.y)propertydef is_occupied(self) - bool:return self.current_sku is not None# # 2. 仓库拓扑与距离计算# class WarehouseLayout:仓库布局管理器 —— 组合模式负责库位建模、距离计算、热区划分def __init__(self, width: int, height: int, entrance(0, 0)):self.width widthself.height heightself.entrance entranceself.locations: Dict[str, StorageLocation] {}self._initialize_grid()self._define_zones()def _initialize_grid(self):初始化网格库位idx 0for x in range(self.width):for y in range(self.height):loc_id fL{x:02d}-{y:02d}# 交叉口拥堵权重更高congestion 1.5 if (x % 3 0 and y % 3 0) else 1.0self.locations[loc_id] StorageLocation(idloc_id,xx,yy,congestion_weightcongestion)def _define_zones(self, hot_radius3, warm_radius7):定义热/温/冷区for loc in self.locations.values():dist loc.distance_to_entranceif dist hot_radius:loc.zone hotelif dist warm_radius:loc.zone warmelse:loc.zone colddef manhattan_distance(self, loc1: StorageLocation,loc2: StorageLocation) - float:带拥堵权重的曼哈顿距离base_dist abs(loc1.x - loc2.x) abs(loc1.y - loc2.y)weighted_dist base_dist * loc1.congestion_weight * loc2.congestion_weightreturn weighted_distdef get_locations_by_zone(self, zone: str) - List[StorageLocation]:获取指定区域的库位return [loc for loc in self.locations.values() if loc.zone zone]def get_nearest_empty_location(self, zone: str None,max_distance: int 100) - Optional[StorageLocation]:获取最近的空闲库位candidates self.get_locations_by_zone(zone) if zone else list(self.locations.values())empty_candidates [loc for loc in candidates if not loc.is_occupied]if not empty_candidates:return None# 按距离排序empty_candidates.sort(keylambda l: l.distance_to_entrance)return next((loc for loc in empty_candidatesif loc.distance_to_entrance max_distance), None)# # 3. 频次分析与ABC分类# class FrequencyAnalyzer:频次分析与ABC分类 —— 策略模式基于帕累托法则划分物料热度def __init__(self, pareto_threshold_A0.8, pareto_threshold_B0.95):self.pareto_A pareto_threshold_Aself.pareto_B pareto_threshold_Bself.history: List[Tuple[str, int]] [] # (sku_id, timestamp)def add_record(self, sku_id: str, timestamp: int None):添加存取记录ts timestamp or int(__import__(time).time())self.history.append((sku_id, ts))def analyze(self, sku_dict: Dict[str, SKU]) - Dict[str, SKU]:分析频次并标记ABC分类# 统计频次counter Counter(sku_id for sku_id, _ in self.history)for sku_id, freq in counter.items():if sku_id in sku_dict:sku_dict[sku_id].frequency freq# 按频次排序sorted_skus sorted(sku_dict.values(), keylambda s: s.frequency, reverseTrue)total_freq sum(s.frequency for s in sorted_skus)if total_freq 0:return sku_dict# 帕累托分类cumulative 0for sku in sorted_skus:cumulative sku.frequencyratio cumulative / total_freqif ratio self.pareto_A:sku.category A # 热区elif ratio self.pareto_B:sku.category B # 温区else:sku.category C # 冷区return sku_dictdef detect_affinity(self, window_size1000) - Dict[str, List[str]]:检测关联规则同时出现的SKUrecent self.history[-window_size:] if len(self.history) window_size else self.historytime_buckets defaultdict(list)for sku_id, ts in recent:bucket ts // 100 # 粗略时间分桶time_buckets[bucket].append(sku_id)affinity_groups defaultdict(list)for bucket_skus in time_buckets.values():if len(bucket_skus) 1:for i, sku1 in enumerate(bucket_skus):for sku2 in bucket_skus[i1:]:affinity_groups[sku1].append(sku2)affinity_groups[sku2].append(sku1)# 合并频繁共现的SKUgroups {}for sku, partners in affinity_groups.items():if len(partners) 3: # 至少共同出现3次group_key tuple(sorted([sku] Counter(partners).most_common(2)[0]))groups[sku] list(group_key)return groups# # 4. 库位优化分配引擎# class AllocationOptimizer:库位优化分配引擎 —— 模板方法采用贪心初始化 局部搜索优化def __init__(self, layout: WarehouseLayout):self.layout layoutdef calculate_cost(self, sku: SKU, location: StorageLocation) - float:计算单个分配代价: 频次 × 距离distance location.distance_to_entrance# 热区惩罚如果A类不在热区增加惩罚zone_penalty 0if sku.category A and location.zone ! hot:zone_penalty 100elif sku.category B and location.zone cold:zone_penalty 50return sku.frequency * distance zone_penaltydef greedy_initialize(self, skus: List[SKU]) - Dict[str, str]:贪心初始化高频优先分配近库位# 按频次降序排列sorted_skus sorted(skus, keylambda s: s.frequency, reverseTrue)allocation {}occupied set()for sku in sorted_skus:# 确定目标区域target_zone {A: hot, B: warm, C: cold}[sku.category]# 寻找最近空闲库位best_loc Nonemin_cost float(inf)candidates self.layout.get_locations_by_zone(target_zone)for loc in candidates:if loc.id in occupied:continuecost self.calculate_cost(sku, loc)if cost min_cost:min_cost costbest_loc loc# 如果目标区域满了放宽限制if best_loc is None:best_loc self.layout.get_nearest_empty_location(max_distance50 if sku.category A else 100)if best_loc:allocation[sku.id] best_loc.idoccupied.add(best_loc.id)best_loc.current_sku skureturn allocationdef local_search_optimize(self, skus: List[SKU],initial_allocation: Dict[str, str],iterations: int 1000) - Dict[str, str]:局部搜索优化交换算子current_allocation initial_allocation.copy()current_cost self._evaluate_total_cost(skus, current_allocation)for _ in range(iterations):# 随机选择两个SKU进行交换sku_ids list(current_allocation.keys())if len(sku_ids) 2:breaki, j random.sample(sku_ids, 2)loc_i current_allocation[i]loc_j current_allocation[j]# 尝试交换current_allocation[i] loc_jcurrent_allocation[j] loc_inew_cost self._evaluate_total_cost(skus, current_allocation)# 如果变好则接受否则回退模拟退火思想if new_cost current_cost or random.random() 0.1:current_cost new_costelse:current_allocation[i] loc_icurrent_allocation[j] loc_jreturn current_allocationdef _evaluate_total_cost(self, skus: List[SKU],allocation: Dict[str, str]) - float:评估总代价total_cost 0sku_map {sku.id: sku for sku in skus}for sku_id, loc_id in allocation.items():sku sku_map.get(sku_id)loc self.layout.locations.get(loc_id)if sku and loc:total_cost self.calculate_cost(sku, loc)return total_cost# # 5. 仿真与评估# class WarehouseSimulator:仓库仿真器 —— 聚合根协调频次分析、优化分配、效果评估def __init__(self, width15, height12):self.layout WarehouseLayout(width, height)self.analyzer FrequencyAnalyzer()self.optimizer AllocationOptimizer(self.layout)self.skus: Dict[str, SKU] {}self.allocation_history []def generate_test_data(self, num_skus50, records5000):生成测试数据模拟真实仓库分布print( 生成测试数据...)# 创建SKU符合帕累托分布for i in range(num_skus):sku_id fSKU-{i:03d}# 80%的访问集中在20%的SKUif i num_skus * 0.2:base_freq random.randint(100, 300)elif i num_skus * 0.5:base_freq random.randint(30, 100)else:base_freq random.randint(1, 30)self.skus[sku_id] SKU(idsku_id,namef物料-{i},frequencybase_freq,volumerandom.uniform(0.5, 2.0))# 生成存取记录sku_ids list(self.skus.keys())for _ in range(records):# 按频次概率选择SKU模拟真实访问模式weights [self.skus[sid].frequency for sid in sku_ids]chosen random.choices(sku_ids, weightsweights, k1)[0]self.analyzer.add_record(chosen)print(f • 创建 {num_skus} 个SKU)print(f • 生成 {records} 条存取记录)def run_optimization(self):执行库位优化print(\n 开始频次分析与ABC分类...)# 分析频次self.skus self.analyzer.analyze(self.skus)# 统计分类结果categories Counter(sku.category for sku in self.skus.values())print(f • A类(热): {categories[A]} 个 ({categories[A]/len(self.skus)*100:.1f}%))print(f • B类(温): {categories[B]} 个 ({categories[B]/len(self.skus)*100:.1f}%))print(f • C类(冷): {categories[C]} 个 ({categories[C]/len(self.skus)*100:.1f}%))# 贪心初始化print(\n 贪心初始化分配...)sku_list list(self.skus.values())initial_alloc self.optimizer.greedy_initialize(sku_list)initial_cost self.optimizer._evaluate_total_cost(sku_list, initial_alloc)print(f • 初始总代价: {initial_cost:.2f})# 局部搜索优化print(\n⚡ 局部搜索优化...)optimized_alloc self.optimizer.local_search_optimize(sku_list, initial_alloc, iterations500)optimized_cost self.optimizer._evaluate_total_cost(sku_list, optimized_alloc)improvement (initial_cost - optimized_cost) / initial_cost * 100print(f • 优化后总代价: {optimized_cost:.2f})print(f • 优化提升: {improvement:.1f}%)# 应用分配结果self._apply_allocation(optimized_alloc)self.allocation_history.append({initial_cost: initial_cost,optimized_cost: optimized_cost,improvement: improvement})return optimized_allocdef _apply_allocation(self, allocation: Dict[str, str]):应用分配结果到库位# 清空现有分配for loc in self.layout.locations.values():loc.current_sku None# 应用新分配for sku_id, loc_id in allocation.items():sku self.skus.get(sku_id)loc self.layout.locations.get(loc_id)if sku and loc:loc.current_sku skudef evaluate_performance(self) - Dict:评估优化效果if not self.allocation_history:return {}stats {total_skus: len(self.skus),allocated_skus: sum(1 for loc in self.layout.locations.values()if loc.current_sku),avg_distance: 0,weighted_distance: 0,category_stats: defaultdict(dict)}total_freq 0total_weighted_dist 0total_dist 0count 0# 按类别统计for loc in self.layout.locations.values():if loc.current_sku:dist loc.distance_to_entra利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛