仓储移动机器人货位调度与保质期优化系统 —— 基于多目标优先级队列的OOP实战

 

"在原料药仓库做技改时，库管员指着堆得密密麻麻的原料桶跟我说：'这批柠檬酸还有3天过期，那批还有两年，但ERP只按入库时间排序，机器人总是先拿最里面、最不好取的。结果就是——好取的没过期，难取的过期了，一个月报废十几万的原料。'后来我重构了调度算法，把保质期剩余天数、工艺配方优先级、货位取放难度三个维度加权，机器人现在专挑'临期+好取+急需'的原料先出库，报废率直接降到了零。"

—— 哈尔滨工程大学《工业过程控制》课程核心思想延伸

 

一、实际应用场景描述

 

在制药、食品、精细化工等行业，原料仓库存放着大量带有保质期限制的物料。移动机器人（AGV/RGV/堆垛机）负责原料的自动存取。传统的仓储管理系统（WMS）通常只按先进先出（FIFO）或入库时间排序，忽略了物料有效期和当前生产工艺的紧急程度，导致高价值原料过期浪费或生产线待料。

 

                    ┌──────────────────────────────────────────────┐

                    │ 智能仓储移动机器人调度系统 │

                    │ │

                    │ [MES/ERP 生产计划] │

                    │ │ 工艺配方、原料需求 │

                    │ ▼ │

                    │ ┌────────────────────────────┐ │

                    │ │ 调度优化引擎 │ │

                    │ │ ┌──────────────────────┐ │ │

                    │ │ │ 1. 保质期权重计算 │ │ │

                    │ │ │ 2. 工艺优先级映射 │ │ │

                    │ │ │ 3. 货位路径成本评估 │ │ │

                    │ │ │ 4. 多目标加权排序 │ │ │

                    │ │ └──────────────────────┘ │ │

                    │ └────────────┬───────────────┘ │

                    │ │ 最优任务序列 │

                    │ ┌───────▼────────┐ │

                    │ │ 移动机器人控制 │ │

                    │ │ ┌──────────────┐ │ │

                    │ │ │ 路径规划(A*) │ │ │

                    │ │ └──────┬───────┘ │ │

                    │ │ ▼ │ │

                    │ │ ┌──────────────┐ │ │

                    │ │ │ 运动执行 │ │ │

                    │ │ └──────┬───────┘ │ │

                    │ └────────┼────────┘ │

                    │ │ │

                    │ ┌────────▼────────┐ │

                    │ │ 立体仓库货架 │ │

                    │ │ ┌───┬───┬───┐ │ │

                    │ │ │A1 │A2 │A3 │ │ ← 高层难取 │

                    │ │ ├───┼───┼───┤ │ │

                    │ │ │B1 │B2 │B3 │ │ ← 中层一般 │

                    │ │ ├───┼───┼───┤ │ │

                    │ │ │C1 │C2 │C3 │ │ ← 底层好取 │

                    │ │ └───┴───┴───┘ │ │

                    │ └──────────────────┘ │

                    │ │

                    │ 核心: 保质期 + 工艺优先级 + 路径成本 = 最优调度 │

                    └──────────────────────────────────────────────┘

 

传统FIFO vs 智能优化调度

 

维度 传统 FIFO 智能优化调度

原料浪费 高（临期货积压） 极低（临期优先）

产线响应 慢（不一定匹配需求） 快（工艺驱动）

机器人效率 低（盲目寻址） 高（综合成本最低）

管理复杂度 简单 可配置、可量化

经济效益 隐性损失大 直接降本增效

 

二、引入痛点

 

2.1 现场的真实困境

 

场景 现场发生了什么 根因

"原料过期了" "才发现那批货过期半个月" FIFO未考虑保质期

"急料拿不到" "产线等着，机器人在拿不急用的料" 未关联工艺优先级

"好取的料不动" "机器人总往角落钻" 未考虑货位成本

"先进后出" "新料放外面，旧料压里面" 上架策略不合理

"账实不符" "系统说有，实际找不到" 调度与库存脱节

 

2.2 核心矛盾

 

仓储管理的目标是"既要保证先进先出，又要保证临期先出，还要保证急料先出"。 这三个目标往往是冲突的：最早入库的原料不一定最急用，也不一定最好取。单纯的FIFO或简单的权重算法无法解决这一多目标优化问题。我们需要的是一个可配置权重的综合评分模型，让管理者根据实际情况动态调整策略。

2.3 我们要解决什么

 

用一段精简的 Python 程序，构建一个仓储移动机器人智能调度系统，实现：

 

1. 原料模型 —— 包含SKU、保质期、入库时间、工艺优先级

2. 货位模型 —— 三维坐标、取放难度系数、占用状态

3. 多目标优化算法 —— 保质期权重 + 工艺优先级 + 路径成本

4. 调度引擎 —— 根据出库请求生成最优任务序列

5. 机器人模拟 —— 执行任务并记录路径

6. 可视化 —— 货架状态、任务序列、优化评分

 

三、核心逻辑讲解

 

3.1 理论基础：多目标加权评分模型

 

本工具基于哈工程《工业过程控制》第十三章"过程优化与控制"和运筹学基础：

 

① 综合评分函数

 

Score = w_1 \cdot S_{shelf\_life} + w_2 \cdot S_{priority} + w_3 \cdot S_{accessibility}

 

其中：

 

- S_{shelf\_life} ：保质期紧迫度评分（越高越紧急）

- S_{priority} ：工艺需求优先级评分

- S_{accessibility} ：货位可达性评分

- w_1, w_2, w_3 ：权重系数（ \sum w_i = 1 ）

 

② 保质期紧迫度计算

 

S_{shelf\_life} = \frac{1}{1 + e^{-k(t_{remain} - t_{mid})}}

 

③ 路径成本估算（曼哈顿距离）

 

Cost = |x_{robot} - x_{bin}| + |y_{robot} - y_{bin}| + |z_{robot} - z_{bin}|

 

四、代码讲解（面向对象设计）

 

4.1 类结构总览

 

类名 职责 设计模式

 

"MaterialSpec" 原料规格（dataclass） 值对象

 

"StorageBin" 货位模型（dataclass） 值对象

 

"TaskRequest" 出库任务请求（dataclass） 命令模式

 

"ScoringWeights" 评分权重配置（dataclass） 值对象

 

"ShelfLifeCalculator" 保质期评分计算器 策略模式

 

"PriorityMapper" 工艺优先级映射器 策略模式

 

"PathCostEstimator" 路径成本估算器 策略模式

 

"DispatchOptimizer" 调度优化引擎 模板方法

 

"RobotSimulator" 机器人运动模拟器 状态模式

 

"WarehouseVisualizer" 仓库可视化器 封装

 

"WarehouseSystem" 系统编排器（聚合根） 聚合根

 

4.2 数据模型层

 

from dataclasses import dataclass, field

from datetime import datetime, timedelta

from typing import List, Dict, Optional, Tuple

from enum import Enum, auto

import math

import random

 

class MaterialCategory(Enum):

    """原料类别"""

    API = auto() # 原料药

    EXCIPIENT = auto() # 辅料

    SOLVENT = auto() # 溶剂

    CATALYST = auto() # 催化剂

 

class PriorityLevel(Enum):

    """工艺优先级"""

    CRITICAL = 5 # 紧急生产

    HIGH = 4 # 重要订单

    MEDIUM = 3 # 常规生产

    LOW = 2 # 备货

    STOCK = 1 # 仅入库

 

@dataclass(frozen=True)

class MaterialSpec:

    """原料规格 —— 值对象"""

    sku: str

    name: str

    category: MaterialCategory

    base_unit: str = "kg"

    default_priority: PriorityLevel = PriorityLevel.MEDIUM

 

@dataclass(frozen=True)

class StorageBin:

    """货位模型 —— 值对象"""

    bin_id: str

    x: int # 列

    y: int # 排

    z: int # 层 (0=地面, 1=一层...)

    access_cost: float = field(init=False) # 取放难度系数

 

    def __post_init__(self):

        # 层数越高，成本越高（需要升降）

        object.__setattr__(self, 'access_cost', 1.0 + self.z * 0.5)

 

@dataclass

class InventoryItem:

    """库存物品（可变）"""

    material: MaterialSpec

    bin_location: StorageBin

    batch_number: str

    quantity: float

    manufacture_date: datetime

    expiry_date: datetime

    inbound_date: datetime

    reserved: bool = False

 

    @property

    def remaining_days(self) -> int:

        return (self.expiry_date - datetime.now()).days

 

    @property

    def is_expired(self) -> bool:

        return self.remaining_days <= 0

 

    @property

    def is_near_expiry(self, threshold_days: int = 30) -> bool:

        return self.remaining_days <= threshold_days

 

@dataclass(frozen=True)

class TaskRequest:

    """出库任务请求 —— 命令模式"""

    request_id: str

    material_sku: str

    required_quantity: float

    requested_priority: PriorityLevel

    requester: str # 工单号/产线名

    deadline: Optional[datetime] = None

 

@dataclass(frozen=True)

class ScoringWeights:

    """评分权重配置 —— 值对象"""

    shelf_life_weight: float = 0.5 # 保质期权重

    priority_weight: float = 0.3 # 工艺优先级权重

    access_weight: float = 0.2 # 货位可达性权重

 

    def __post_init__(self):

        total = self.shelf_life_weight + self.priority_weight + self.access_weight

        if abs(total - 1.0) > 0.001:

            raise ValueError("Weights must sum to 1.0")

 

4.3 评分计算器（策略模式）

 

class ShelfLifeCalculator:

    """保质期评分计算器 —— 策略模式"""

 

    @staticmethod

    def calculate_score(item: InventoryItem,

                        mid_point_days: int = 90,

                        steepness: float = 0.05) -> float:

        """

        使用Sigmoid函数计算保质期紧迫度

        越接近过期，分数越高（越紧急）

 

        Args:

            item: 库存物品

            mid_point_days: Sigmoid中点（剩余天数）

            steepness: 曲线陡峭度

        """

        if item.is_expired:

            return 1.0 # 已过期，最高优先级

 

        remaining = item.remaining_days

        # Sigmoid: 1 / (1 + e^(-k*(x-x0)))

        # x0是中点，k控制陡度

        score = 1.0 / (1.0 + math.exp(-steepness * (remaining - mid_point_days)))

 

        # 反转：剩余天数越少，分数越高

        return 1.0 - score

 

class PriorityMapper:

    """工艺优先级映射器 —— 策略模式"""

 

    @staticmethod

    def map_priority_to_score(priority: PriorityLevel) -> float:

        """将离散优先级映射为连续分数"""

        mapping = {

            PriorityLevel.CRITICAL: 1.0,

            PriorityLevel.HIGH: 0.8,

            PriorityLevel.MEDIUM: 0.6,

            PriorityLevel.LOW: 0.3,

            PriorityLevel.STOCK: 0.1

        }

        return mapping.get(priority, 0.5)

 

    @staticmethod

    def adjust_by_request(item: InventoryItem,

                          request_priority: PriorityLevel) -> float:

        """根据请求动态调整"""

        base_score = PriorityMapper.map_priority_to_score(

            item.material.default_priority

        )

        request_score = PriorityMapper.map_priority_to_score(request_priority)

 

        # 请求优先级权重更高

        return base_score * 0.3 + request_score * 0.7

 

class PathCostEstimator:

    """路径成本估算器 —— 策略模式"""

 

    @staticmethod

    def estimate_manhattan_distance(robot_pos: Tuple[int, int, int],

                                    target_bin: StorageBin) -> float:

        """曼哈顿距离（AGV常用）"""

        dx = abs(robot_pos[0] - target_bin.x)

        dy = abs(robot_pos[1] - target_bin.y)

        dz = abs(robot_pos[2] - target_bin.z)

        return dx + dy + dz

 

    @staticmethod

    def estimate_accessibility_score(bin_location: StorageBin,

                                     robot_pos: Tuple[int, int, int]) -> float:

        """

        可达性评分：距离越近、层数越低，分数越高

        注意：这里是"易达性"，所以分数越高越好

        """

        distance = PathCostEstimator.estimate_manhattan_distance(

            robot_pos, bin_location

        )

        # 归一化（假设最大距离为20）

        normalized_dist = min(distance / 20.0, 1.0)

 

        # 层数惩罚

        level_penalty = bin_location.z * 0.1

 

        # 综合评分（越近越高，层数越低越高）

        score = 1.0 - normalized_dist - level_penalty

        return max(0.0, min(1.0, score))

 

4.4 调度优化引擎（模板方法）

 

class DispatchOptimizer:

    """

    调度优化引擎 —— 模板方法模式

 

    固定流程: 筛选候选 → 计算评分 → 排序 → 分配任务

    """

 

    def __init__(self, weights: ScoringWeights):

        self.weights = weights

        self.shelf_calc = ShelfLifeCalculator()

        self.priority_mapper = PriorityMapper()

        self.path_estimator = PathCostEstimator()

 

    def optimize_dispatch(self, request: TaskRequest,

                          inventory: List[InventoryItem],

                          robot_position: Tuple[int, int, int]) -> List[InventoryItem]:

        """

        优化调度：找出最适合出库的物料序列

 

        Args:

            request: 出库请求

            inventory: 当前库存

            robot_position: 机器人当前位置

 

        Returns:

            按优先级排序的出库物料列表

        """

        # 1. 筛选候选物料（模板方法的第一步）

        candidates = self._filter_candidates(request, inventory)

 

        if not candidates:

            return []

 

        # 2. 计算每个候选的综合评分（模板方法的第二步）

        scored_candidates = []

        for item in candidates:

            score = self._calculate_composite_score(

                item, request.requested_priority, robot_position

            )

            scored_candidates.append((item, score))

 

        # 3. 按综合评分排序（模板方法的第三步）

        scored_candidates.sort(key=lambda x: x[1], reverse=True)

 

        # 4. 分配任务（模板方法的第四步）

        allocated = self._allocate_quantity(scored_candidates, request.required_quantity)

 

        return allocated

 

    def _filter_candidates(self, request: TaskRequest,

                           inventory: List[InventoryItem]) -> List[InventoryItem]:

        """筛选符合条件的候选物料"""

        candidates = []

        for item in inventory:

            # 条件1: SKU匹配

            if item.material.sku != request.material_sku:

                continue

            # 条件2: 未过期

            if item.is_expired:

                continue

            # 条件3: 有库存

            if item.quantity <= 0:

                continue

            # 条件4: 未被预留

            if item.reserved:

                continue

            candidates.append(item)

        return candidates

 

    def _calculate_composite_score(self, item: InventoryItem,

                                  request_priority: PriorityLevel,

                                  robot_pos: Tuple[int, int, int]) -> float:

        """计算综合评分"""

        # 1. 保质期评分

        shelf_score = self.shelf_calc.calculate_score(item)

 

        # 2. 优先级评分

        priority_score = self.priority_mapper.adjust_by_request(

            item, request_priority

        )

 

        # 3. 可达性评分

        access_score = self.path_estimator.estimate_accessibility_score(

            item.bin_location, robot_pos

        )

 

        # 4. 加权综合

        composite = (

            self.weights.shelf_life_weight * shelf_score +

            self.weights.priority_weight * priority_score +

            self.weights.access_weight * access_score

        )

 

        return composite

 

    def _allocate_quantity(self, scored_items: List[Tuple[InventoryItem, float]],

                           required_qty: float) -> List[InventoryItem]:

        """根据需求量分配出库物料"""

        allocated = []

        remaining_qty = required_qty

 

        for item, score in scored_items:

            if remaining_qty <= 0:

                break

 

            take_qty = min(item.quantity, remaining_qty)

            if take_qty > 0:

                # 创建副本（因为出库会减少库存）

                allocated_item = InventoryItem(

                    material=item.material,

                    bin_location=item.bin_location,

                    batch_number=item.batch_number,

                    quantity=take_qty,

                    manufacture_date=item.manufacture_date,

                    expiry_date=item.expiry_date,

                    inbound_date=item.inbound_date,

                    reserved=True

                )

                allocated.append(allocated_item)

                remaining_qty -= take_qty

 

        return allocated

 

4.5 机器人模拟器（状态模式）

 

class RobotSimulator:

    """

    移动机器人模拟器 —— 状态模式

 

    状态: IDLE → MOVING → PICKING → RETURNING → IDLE

    """

 

    class State(Enum):

        IDLE = auto()

        MOVING = auto()

        PICKING = auto()

        RETURNING = auto()

 

    def __init__(self, start_position: Tuple[int, int, int] = (0, 0, 0)):

        self.position = start_position

        self.state = self.State.IDLE

        self.current_task: Optional[InventoryItem] = None

        self.path_history: List[Tuple[int, int, int]] = []

        self.task_history: List[Dict] = []

        self.total_distance = 0.0

 

    def assign_task(self, item: InventoryItem):

        """分配任务"""

        if self.state == self.State.IDLE:

            self.current_task = item

            self.state = self.State.MOVING

            return True

        return False

 

    def step(self, dt: float) -> bool:

        """

        执行一步仿真

 

        Returns:

            True if task completed

        """

        if self.state == self.State.IDLE:

            return False

 

        self.path_history.append(self.position)

 

        if self.state == self.State.MOVING:

            # 移动到目标货位

            target = self.current_task.bin_location

            if self._move_towards(target):

                self.state = self.State.PICKING

                print(f" 🤖 机器人到达货位 {target.bin_id}")

 

        elif self.state == self.State.PICKING:

            # 模拟抓取时间

            time.sleep(0.5) # 仿真中的延时

            self.state = self.State.RETURNING

            print(f" 📦 取出 {self.current_task.material.name} "

                  f"({self.current_task.quantity}{self.current_task.material.base_unit})")

 

        elif self.state == self.State.RETURNING:

            # 返回起点

            if self._move_towards((0, 0, 0)):

                # 完成任务

                self.task_history.append({

                    'material': self.current_task.material.sku,

                    'batch': self.current_task.batch_number,

                    'quantity': self.current_task.quantity,

                    'bin': self.current_task.bin_location.bin_id,

                    'distance': self.total_distance

                })

                self.current_task = None

                self.state = self.State.IDLE

                print(f" ✅ 任务完成，返回起点")

                return True

 

        return False

 

    def _move_towards(self, target: Tuple[int, int, int]) -> bool:

        """向目标移动一步（曼哈顿路径）"""

        dx = target[0] - self.position[0]

        dy = target[1] - self.position[1]

        dz = target[2] - self.position[2]

 

        step_x = 1 if dx > 0 else (-1 if dx < 0 else 0)

        step_y = 1 if dy > 0 else (-1 if dy < 0 else 0)

        step_z = 1 if dz > 0 else (-1 if dz < 0 else 0)

 

        new_pos = (self.position[0] + step_x,

                   self.position[1] + step_y,

                   self.position[2] + step_z)

 

        self.total_distance += 1 # 曼哈顿距离每步为1

        self.position = new_pos

 

        # 检查是否到达

        return new_pos == target

 

    def get_status(self) -> Dict:

        return {

            'position': self.position,

            'state': self.state.name,

            'task': self.current_task.material.sku if self.current_task else None,

            'distance_traveled': self.total_distance

        }

 

4.6 仓库可视化器

 

class WarehouseVisualizer:

    """仓库可视化器"""

 

    @staticmethod

    def visualize_warehouse(inventory: List[InventoryItem],

                           robot_pos: Tuple[int, int, int],

                           tasks: List[InventoryItem],

                           output: str = "warehouse.png"):

        import matplotlib.pyplot as plt

        from mpl_toolkits.mplot3d import Axes3D

 

        fig = plt.figure(figsize=(14, 10))

        ax = fig.add_subplot(111, projection='3d')

 

        # 绘制货架

        bins = list(set(item.bin_location for item in inventory))

        for bin_loc in bins:

            # 货位框架

            ax.scatter(bin_loc.x, bin_loc.y, bin_loc.z,

                      s=200, alpha=0.2, color='gray', edgecolors='black')

 

        # 绘制库存物品

        colors = {'API': 'red', 'EXCIPIENT': 'blue',

                  'SOLVENT': 'green', 'CATALYST': 'purple'}

        for item in inventory:

            cat_name = item.material.category.name

            color = colors.get(cat_name, 'gray')

 

            # 大小表示剩余保质期（越小越红/紧急）

            size = 100 + max(0, item.remaining_days) * 2

 

            ax.scatter(item.bin_location.x,

                      item.bin_location.y,

                      item.bin_location.z,

                      s=size, alpha=0.7, color=color,

                      label=item.material.name if item.material.name not in [t.get_label() for t in ax.get_legend_handles_labels()[0]] else "")

 

            # 标注SKU

            ax.text(item.bin_location.x, item.bin_location.y, item.bin_location.z,

                   item.material.sku, fontsize=8, ha='center')

 

        # 绘制机器人

        ax.scatter(robot_pos[0], robot_pos[1], robot_pos[2],

                  s=300, color='yellow', marker='o', edgecolors='black',

                  label='Robot')

 

        # 绘制任务路径

        if tasks:

            path_x = [robot_pos[0]]

            path_y = [robot_pos[1]]

            path_z = [robot_pos[2]]

 

            for task in tasks:

                path_x.append(task.bin_location.x)

                path_y.append(task.bin_location.y)

                path_z.append(task.bin_location.z)

 

            ax.plot(path_x, path_y, path_z, 'r--', linewidth=2, label='Task Path')

 

        # 设置坐标轴

        ax.set_xlabel('X (Column)')

        ax.set_ylabel('Y (Row)')

        ax.set_zlabel('Z (Level)')

        ax.set_title('Smart Warehouse Dispatch Simulation')

 

        # 图例

        handles, labels = ax.get_legend_handles_labels()

        by_label = dict(zip(labels, handles))

        ax.legend(by_label.values(), by_label.keys(), loc='upper left', bbox_to_anchor=(1.05, 1))

 

        plt.tight_layout()

        plt.savefig(output, dpi=150, bbox_inches='tight')

        plt.close()

 

    @staticmethod

    def print_dispatch_report(tasks: List[InventoryItem],

                             optimizer: DispatchOptimizer,

                             robot: RobotSimulator):

        print("\n" + "=" * 60)

        print(" 智能调度执行报告")

        print("=" * 60)

 

        total_qty = sum(t.quantity for t in tasks)

        print(f"\n📋 本次出库任务:")

        print(f" 共 {len(tasks)} 个货位")

        print(f" 总出库量: {total_qty:.1f} kg")

 

 利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！