1.2.2

下载 Notebook 下载完整案例

订单分批问题¶

类别： 路径优化

使用 OptAgent 的 Python 接口描述变量、约束与目标。

问题与原始示例来源：Hexaly Code Templates。

问题描述¶

订单分批问题出现在仓库物流场景中。在该问题中，一组订单（每个订单由位于具有平行通道的矩形仓库中特定位置的商品组成）需要被分组成批次。对于每个批次，一个拣货员在一次仓库巡回中收集所有商品，从仓库起点出发并返回。

一个关键约束是拣货员有限的承载能力：每个批次只能包含一定数量的商品。完成一个批次所行驶的距离由 S 形路径策略确定，即拣货员沿每条通道全长移动，每次往返改变方向（从前到后，然后从后到前）。

总体目标是以最小化拣货员总行驶距离的方式将订单组织成批次。

建模要点¶

• 使用 json 标准库 读取输入文件
• 使用 set 决策变量 建模批次
• 使用 lambda 函数 计算每个批次中的商品数量以及拣货员访问的通道

数据¶

订单分批问题的实例采用 JSON 格式。

每个文件包含以下字段：

• “instanceName”：实例的名称

• “nbOrders”：客户订单的总数

• “capacity”：每个批次允许的最大商品数量

• “nbAisles”：仓库中物理通道的数量

• “maxNbBatches”：批次数量的上界

• “aisleTraversal”：每条通道的长度（距离单位）

• “gap”：相邻两条通道中心线之间的距离

• “depotOffset”：从仓库起点到第一条通道的距离

• “orders”：该字段包含一个客户订单列表。每个订单的定义如下：

• “nbItems”：订单中的商品数量

• “aislesToVisit”：为收集该订单需要访问的通道索引

Python 实现使用标准库 json 读取实例。

建模思路¶

订单分批问题的 OptAgent 模型使用 set 决策变量，每个 set 表示分配给一个批次的订单集合。

首先，这些 set 被约束为构成一个 partition，即每个订单必须恰好分配到一个批次。

我们使用对 set 的可变参数 sum 算子和一个 lambda 函数 来计算一个批次中的总商品数量，该函数返回每个订单的商品数量。请注意，求和中的项数在搜索过程中会动态变化，set 的大小也会变化。然后我们可以将商品总数量约束为不超过拣货员的承载能力。

然后，对于每个批次，我们确定需要访问哪些通道。如果批次中至少有一个订单的商品存放在某条通道中，则该通道被访问。基于此，我们使用另一个 lambda 函数计算访问的通道数量以及所访问的最远通道的索引。

然后，按如下方式计算一个批次的 S 形遍历距离：拣货员从仓库起点出发，首先水平移动到达最远被访问的通道（距离与 gap x maxVisitedAisle 成比例），然后遍历每条被访问通道的全长（根据奇偶性调整，以便拣货员始终从前侧离开），最后返回仓库起点（见下文）。在批次为空的特殊情况下，其距离为零。

总体而言，目标是最小化所有批次的总行驶距离。

Python 实现¶

from pathlib import Path
import json

from optagent import OptModel, solve

def read_instance(input_file):
    with open(input_file) as f:
        data = json.load(f)

    nb_orders = data["nbOrders"]                # Number of orders
    capacity = data["capacity"]                 # Capacity of the picker
    nb_aisles = data["nbAisles"]                # Number of aisles
    max_nb_batches = data["maxNbBatches"]       # Upper bound on the number of batches
    aisle_traversal = data["aisleTraversal"]    # Length of each aisle
    depot_offset = data["depotOffset"]          # Distance from the depot to the first aisle
    gap = data["gap"]            # Distance between the center lines and two adjacent aisles

    nb_items = [0] * nb_orders                                  # Number of items in each order
    visits_aisle = [[0] * nb_orders for _ in range(nb_aisles)]  # Aisles visited by each order
    orders = data["orders"]
    for i in range(nb_orders):
        nb_items[i] = orders[i]["nbItems"]
        for a in orders[i]["aislesToVisit"]:
            visits_aisle[a][i] = 1

    return nb_orders, capacity, nb_aisles, max_nb_batches, aisle_traversal, gap, \
            depot_offset, nb_items, visits_aisle

def main(input_file, output_file=None, time_limit=20):
    (
        nb_orders, capacity, nb_aisles, max_nb_batches, aisle_traversal, gap,
        depot_offset, nb_items, visits_aisle,
    ) = read_instance(input_file)
    model = OptModel()

    batches = [
        model.set(nb_orders)
        for batch in range(max_nb_batches)
    ]
    model.constraint(model.partition(batches))
    items = model.array(nb_items)
    aisle_arrays = [model.array(visits_aisle[aisle]) for aisle in range(nb_aisles)]
    distances = []
    batch_orders = []
    batch_aisles = []
    for batch, orders in enumerate(batches):
        item_count = model.sum(
            orders, model.lambda_function(lambda order: items[order // 1])
        )
        model.constraint(item_count <= capacity)
        visited = []
        for aisle in range(nb_aisles):
            aisle_array = aisle_arrays[aisle]
            visited.append(
                model.sum(
                    orders,
                    model.lambda_function(lambda order: aisle_array[order // 1]),
                )
                >= 1
            )
        visited_count = model.sum(visited)
        farthest_aisle = model.max(visited[aisle] * aisle for aisle in range(nb_aisles))
        used = orders.count() > 0
        distances.append(
            used
            * (
                2 * depot_offset
                + 2 * gap * farthest_aisle
                + (visited_count + visited_count % 2) * aisle_traversal
            )
        )
        batch_orders.append(orders)
        batch_aisles.append(visited)

    total_distance = model.sum(distances)
    model.minimize(total_distance)
    solution = solve(model, time_limit_s=float(time_limit))
    values = {"total_distance": total_distance.value}
    lines = [
        f"Orders = {nb_orders}; Batches = {max_nb_batches}; "
        f"Total distance = {values['total_distance']}; Status = {solution.feasible}"
    ]
    for batch in range(max_nb_batches):
        orders = list(batch_orders[batch].value)
        if orders:
            aisles = [
                aisle for aisle in range(nb_aisles) if batch_aisles[batch][aisle].value
            ]
            lines.append(
                f"Batch {batch}: orders={orders}; aisles={aisles}; distance={distances[batch].value}"
            )
    result_text = "\n".join(lines)
    print(result_text)
    if output_file is not None:
        result = {
            "objective": int(total_distance.value),
            "batches": [
                {
                    "orders": list(batch_orders[batch].value),
                    "distance": int(distances[batch].value),
                    "visitedAisles": [
                        aisle for aisle in range(nb_aisles)
                        if batch_aisles[batch][aisle].value
                    ],
                }
                for batch in range(max_nb_batches)
                if batch_orders[batch].value
            ],
        }
        Path(output_file).write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
    return solution

本地运行¶

以下代码格演示如何调用 OptAgent 的订单分批模型。

INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)

solution_batching = main(INSTANCE_DIR / "21s-20-30-0.json", time_limit=1)