python的工业过程控制场景模拟第一百零三篇:仓储机器人库位优先算法，高频取用物料放置靠近出入口，缩短搬运距离。
仓储机器人库位优化算法 —— 基于存取频次的动态热区调度
“那年电商大促，仓库里最忙的几台 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_{i=1}^{M} \sum_{j=1}^{N} f_i \cdot d_{ij} \cdot x_{ij}
② 约束条件
\begin{aligned}
&\sum_{j=1}^{N} x_{ij} = 1, \quad \forall i \in M \quad &\text{(每物料仅一个库位)}\\
&\sum_{i=1}^{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 个类、仿真引擎、可视化、优化算法。
以下为精简核心版，完整代码可直接复制运行。
<details><summary>🔧 完整源码（点击展开/折叠）</summary>
"""
仓储机器人库位优化算法 —— 基于频次-距离加权
参考哈尔滨工程大学《工业过程控制》第四章线性规划
"""
from dataclasses import dataclass, field
from typing import List, Dict, Tuple, Optional
import numpy as np
import matplotlib.pyplot as plt
from collections import defaultdict, Counter
import math
import random
# ============================================================
# 1. 基础数据结构
# ============================================================
@dataclass
class SKU:
"""物料单元 —— 值对象"""
id: str
name: str
frequency: 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 # 按频次降序
@dataclass
class StorageLocation:
"""库位单元 —— 值对象"""
id: str
x: int # 网格坐标X
y: int # 网格坐标Y
zone: str = 'cold' # 区域: hot/warm/cold
congestion_weight: float = 1.0 # 拥堵权重
capacity: float = 1.0 # 容量
current_sku: Optional[SKU] = None
@property
def distance_to_entrance(self) -> int:
"""到出入口(0,0)的曼哈顿距离"""
return abs(self.x) + abs(self.y)
@property
def 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 = width
self.height = height
self.entrance = entrance
self.locations: Dict[str, StorageLocation] = {}
self._initialize_grid()
self._define_zones()
def _initialize_grid(self):
"""初始化网格库位"""
idx = 0
for x in range(self.width):
for y in range(self.height):
loc_id = f"L{x:02d}-{y:02d}"
# 交叉口拥堵权重更高
congestion = 1.5 if (x % 3 == 0 and y % 3 == 0) else 1.0
self.locations[loc_id] = StorageLocation(
id=loc_id,
x=x,
y=y,
congestion_weight=congestion
)
def _define_zones(self, hot_radius=3, warm_radius=7):
"""定义热/温/冷区"""
for loc in self.locations.values():
dist = loc.distance_to_entrance
if dist <= hot_radius:
loc.zone = 'hot'
elif dist <= warm_radius:
loc.zone = 'warm'
else:
loc.zone = 'cold'
def 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_weight
return weighted_dist
def 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(key=lambda l: l.distance_to_entrance)
return next((loc for loc in empty_candidates
if loc.distance_to_entrance <= max_distance), None)
# ============================================================
# 3. 频次分析与ABC分类
# ============================================================
class FrequencyAnalyzer:
"""
频次分析与ABC分类 —— 策略模式
基于帕累托法则划分物料热度
"""
def __init__(self, pareto_threshold_A=0.8, pareto_threshold_B=0.95):
self.pareto_A = pareto_threshold_A
self.pareto_B = pareto_threshold_B
self.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(), key=lambda s: s.frequency, reverse=True)
total_freq = sum(s.frequency for s in sorted_skus)
if total_freq == 0:
return sku_dict
# 帕累托分类
cumulative = 0
for sku in sorted_skus:
cumulative += sku.frequency
ratio = cumulative / total_freq
if ratio <= self.pareto_A:
sku.category = 'A' # 热区
elif ratio <= self.pareto_B:
sku.category = 'B' # 温区
else:
sku.category = 'C' # 冷区
return sku_dict
def detect_affinity(self, window_size=1000) -> Dict[str, List[str]]:
"""检测关联规则（同时出现的SKU）"""
recent = self.history[-window_size:] if len(self.history) > window_size else self.history
time_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[i+1:]:
affinity_groups[sku1].append(sku2)
affinity_groups[sku2].append(sku1)
# 合并频繁共现的SKU
groups = {}
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 = layout
def calculate_cost(self, sku: SKU, location: StorageLocation) -> float:
"""计算单个分配代价: 频次 × 距离"""
distance = location.distance_to_entrance
# 热区惩罚：如果A类不在热区，增加惩罚
zone_penalty = 0
if sku.category == 'A' and location.zone != 'hot':
zone_penalty = 100
elif sku.category == 'B' and location.zone == 'cold':
zone_penalty = 50
return sku.frequency * distance + zone_penalty
def greedy_initialize(self, skus: List[SKU]) -> Dict[str, str]:
"""贪心初始化：高频优先分配近库位"""
# 按频次降序排列
sorted_skus = sorted(skus, key=lambda s: s.frequency, reverse=True)
allocation = {}
occupied = set()
for sku in sorted_skus:
# 确定目标区域
target_zone = {'A': 'hot', 'B': 'warm', 'C': 'cold'}[sku.category]
# 寻找最近空闲库位
best_loc = None
min_cost = float('inf')
candidates = self.layout.get_locations_by_zone(target_zone)
for loc in candidates:
if loc.id in occupied:
continue
cost = self.calculate_cost(sku, loc)
if cost < min_cost:
min_cost = cost
best_loc = loc
# 如果目标区域满了，放宽限制
if best_loc is None:
best_loc = self.layout.get_nearest_empty_location(
max_distance=50 if sku.category == 'A' else 100
)
if best_loc:
allocation[sku.id] = best_loc.id
occupied.add(best_loc.id)
best_loc.current_sku = sku
return allocation
def 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:
break
i, j = random.sample(sku_ids, 2)
loc_i = current_allocation[i]
loc_j = current_allocation[j]
# 尝试交换
current_allocation[i] = loc_j
current_allocation[j] = loc_i
new_cost = self._evaluate_total_cost(skus, current_allocation)
# 如果变好则接受，否则回退（模拟退火思想）
if new_cost < current_cost or random.random() < 0.1:
current_cost = new_cost
else:
current_allocation[i] = loc_i
current_allocation[j] = loc_j
return current_allocation
def _evaluate_total_cost(self, skus: List[SKU],
allocation: Dict[str, str]) -> float:
"""评估总代价"""
total_cost = 0
sku_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, width=15, height=12):
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_skus=50, records=5000):
"""生成测试数据（模拟真实仓库分布）"""
print("📦 生成测试数据...")
# 创建SKU（符合帕累托分布）
for i in range(num_skus):
sku_id = f"SKU-{i:03d}"
# 80%的访问集中在20%的SKU
if 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(
id=sku_id,
name=f"物料-{i}",
frequency=base_freq,
volume=random.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, weights=weights, k=1)[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, iterations=500
)
optimized_cost = self.optimizer._evaluate_total_cost(sku_list, optimized_alloc)
improvement = (initial_cost - optimized_cost) / initial_cost * 100
print(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_alloc
def _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 = sku
def 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 = 0
total_weighted_dist = 0
total_dist = 0
count = 0
# 按类别统计
for loc in self.layout.locations.values():
if loc.current_sku:
dist = loc.distance_to_entra
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！