{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 订单分批问题\n", "\n", "**类别：** 路径优化\n", "\n", "使用 OptAgent 的 Python 接口描述变量、约束与目标。\n", "\n", "问题与原始示例来源：[Hexaly Code Templates](https://www.hexaly.com/templates/order-batching-problem)。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 问题描述\n", "\n", "订单分批问题出现在仓库物流场景中。在该问题中，一组订单（每个订单由位于具有平行通道的矩形仓库中特定位置的商品组成）需要被分组成批次。对于每个批次，一个拣货员在一次仓库巡回中收集所有商品，从仓库起点出发并返回。\n", "\n", "一个关键约束是拣货员有限的承载能力：每个批次只能包含一定数量的商品。完成一个批次所行驶的距离由 S 形路径策略确定，即拣货员沿每条通道全长移动，每次往返改变方向（从前到后，然后从后到前）。\n", "\n", "总体目标是以最小化拣货员总行驶距离的方式将订单组织成批次。\n", "\n", "### 建模要点\n", "\n", "- 使用 `json` 标准库 读取输入文件\n", "- 使用 [set 决策变量](https://optagent.pages.dev/guide/modeling/) 建模批次\n", "- 使用 [lambda 函数](https://optagent.pages.dev/guide/modeling/) 计算每个批次中的商品数量以及拣货员访问的通道\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 数据\n", "\n", "订单分批问题的实例采用 **JSON** 格式。\n", "\n", "每个文件包含以下字段：\n", "\n", "- “instanceName”：实例的名称\n", "\n", "- “nbOrders”：客户订单的总数\n", "\n", "- “capacity”：每个批次允许的最大商品数量\n", "\n", "- “nbAisles”：仓库中物理通道的数量\n", "\n", "- “maxNbBatches”：批次数量的上界\n", "\n", "- “aisleTraversal”：每条通道的长度（距离单位）\n", "\n", "- “gap”：相邻两条通道中心线之间的距离\n", "\n", "- “depotOffset”：从仓库起点到第一条通道的距离\n", "\n", "- “orders”：该字段包含一个客户订单列表。每个订单的定义如下：\n", "\n", "- “nbItems”：订单中的商品数量\n", "- “aislesToVisit”：为收集该订单需要访问的通道索引\n", "\n", "Python 实现使用标准库 `json` 读取实例。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 建模思路\n", "\n", "订单分批问题的 OptAgent 模型[使用 set 决策变量](https://optagent.pages.dev/guide/modeling/)，每个 set 表示分配给一个批次的订单集合。\n", "\n", "首先，这些 set 被约束为构成一个 [`partition`](https://optagent.pages.dev/api/model-reference/#optagent.OptModel.partition)，即每个订单必须恰好分配到一个批次。\n", "\n", "我们使用对 set 的可变参数 **sum** 算子和一个 [lambda 函数](https://optagent.pages.dev/guide/modeling/) 来计算一个批次中的总商品数量，该函数返回每个订单的商品数量。请注意，求和中的项数在搜索过程中会动态变化，set 的大小也会变化。然后我们可以将商品总数量约束为不超过拣货员的承载能力。\n", "\n", "然后，对于每个批次，我们确定需要访问哪些通道。如果批次中至少有一个订单的商品存放在某条通道中，则该通道被访问。基于此，我们使用另一个 lambda 函数计算访问的通道数量以及所访问的最远通道的索引。\n", "\n", "然后，按如下方式计算一个批次的 **S 形遍历距离**：拣货员从仓库起点出发，首先水平移动到达最远被访问的通道（距离与 *gap* x *maxVisitedAisle* 成比例），然后遍历每条被访问通道的全长（根据奇偶性调整，以便拣货员始终从前侧离开），最后返回仓库起点（见下文）。在批次为空的特殊情况下，其距离为零。\n", "\n", "总体而言，目标是最小化所有批次的总行驶距离。\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Python 实现\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from pathlib import Path\n", "import json\n", "\n", "from optagent import OptModel, solve\n", "\n", "def read_instance(input_file):\n", " with open(input_file) as f:\n", " data = json.load(f)\n", "\n", " nb_orders = data[\"nbOrders\"] # Number of orders\n", " capacity = data[\"capacity\"] # Capacity of the picker\n", " nb_aisles = data[\"nbAisles\"] # Number of aisles\n", " max_nb_batches = data[\"maxNbBatches\"] # Upper bound on the number of batches\n", " aisle_traversal = data[\"aisleTraversal\"] # Length of each aisle\n", " depot_offset = data[\"depotOffset\"] # Distance from the depot to the first aisle\n", " gap = data[\"gap\"] # Distance between the center lines and two adjacent aisles\n", "\n", " nb_items = [0] * nb_orders # Number of items in each order\n", " visits_aisle = [[0] * nb_orders for _ in range(nb_aisles)] # Aisles visited by each order\n", " orders = data[\"orders\"]\n", " for i in range(nb_orders):\n", " nb_items[i] = orders[i][\"nbItems\"]\n", " for a in orders[i][\"aislesToVisit\"]:\n", " visits_aisle[a][i] = 1\n", "\n", " return nb_orders, capacity, nb_aisles, max_nb_batches, aisle_traversal, gap, \\\n", " depot_offset, nb_items, visits_aisle\n", "\n", "def main(input_file, output_file=None, time_limit=20):\n", " (\n", " nb_orders, capacity, nb_aisles, max_nb_batches, aisle_traversal, gap,\n", " depot_offset, nb_items, visits_aisle,\n", " ) = read_instance(input_file)\n", " model = OptModel()\n", "\n", " batches = [\n", " model.set(nb_orders)\n", " for batch in range(max_nb_batches)\n", " ]\n", " model.constraint(model.partition(batches))\n", " items = model.array(nb_items)\n", " aisle_arrays = [model.array(visits_aisle[aisle]) for aisle in range(nb_aisles)]\n", " distances = []\n", " batch_orders = []\n", " batch_aisles = []\n", " for batch, orders in enumerate(batches):\n", " item_count = model.sum(\n", " orders, model.lambda_function(lambda order: items[order // 1])\n", " )\n", " model.constraint(item_count <= capacity)\n", " visited = []\n", " for aisle in range(nb_aisles):\n", " aisle_array = aisle_arrays[aisle]\n", " visited.append(\n", " model.sum(\n", " orders,\n", " model.lambda_function(lambda order: aisle_array[order // 1]),\n", " )\n", " >= 1\n", " )\n", " visited_count = model.sum(visited)\n", " farthest_aisle = model.max(visited[aisle] * aisle for aisle in range(nb_aisles))\n", " used = orders.count() > 0\n", " distances.append(\n", " used\n", " * (\n", " 2 * depot_offset\n", " + 2 * gap * farthest_aisle\n", " + (visited_count + visited_count % 2) * aisle_traversal\n", " )\n", " )\n", " batch_orders.append(orders)\n", " batch_aisles.append(visited)\n", "\n", " total_distance = model.sum(distances)\n", " model.minimize(total_distance)\n", " solution = solve(model, time_limit_s=float(time_limit))\n", " values = {\"total_distance\": total_distance.value}\n", " lines = [\n", " f\"Orders = {nb_orders}; Batches = {max_nb_batches}; \"\n", " f\"Total distance = {values['total_distance']}; Status = {solution.feasible}\"\n", " ]\n", " for batch in range(max_nb_batches):\n", " orders = list(batch_orders[batch].value)\n", " if orders:\n", " aisles = [\n", " aisle for aisle in range(nb_aisles) if batch_aisles[batch][aisle].value\n", " ]\n", " lines.append(\n", " f\"Batch {batch}: orders={orders}; aisles={aisles}; distance={distances[batch].value}\"\n", " )\n", " result_text = \"\\n\".join(lines)\n", " print(result_text)\n", " if output_file is not None:\n", " result = {\n", " \"objective\": int(total_distance.value),\n", " \"batches\": [\n", " {\n", " \"orders\": list(batch_orders[batch].value),\n", " \"distance\": int(distances[batch].value),\n", " \"visitedAisles\": [\n", " aisle for aisle in range(nb_aisles)\n", " if batch_aisles[batch][aisle].value\n", " ],\n", " }\n", " for batch in range(max_nb_batches)\n", " if batch_orders[batch].value\n", " ],\n", " }\n", " Path(output_file).write_text(json.dumps(result, indent=2) + \"\\n\", encoding=\"utf-8\")\n", " return solution\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 本地运行\n", "\n", "以下代码格演示如何调用 OptAgent 的订单分批模型。\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "INSTANCE_DIR = Path.cwd() / \"instances\"\n", "print(\"Instances:\", INSTANCE_DIR)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "solution_batching = main(INSTANCE_DIR / \"21s-20-30-0.json\", time_limit=1)\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3" } }, "nbformat": 4, "nbformat_minor": 5 }