1.2.2

下载 Notebook 下载完整案例

库存路径问题 (IRP)¶

类别： 路径优化

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

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

问题描述¶

库存路径问题 (IRP) 是一个配送问题，其中需要在给定的时间范围内将产品从一个供应商运送到多个客户。每个客户都有一个最大库存水平。供应商监控每个客户的库存并确定其补货策略，保证客户不会出现缺货（供应商管理库存策略）。从供应商到客户的运输由具有给定容量的车辆完成。

学习要点¶

• 添加 列表决策变量 来建模卡车的客户序列
• 定义 符号 Lambda 函数 来计算每个时间槽的行驶距离
• 递归地定义表达式来计算库存水平

数据¶

我们提供的库存路径问题 (IRP) 实例来自 Archetti et al. instances。数据文件的格式如下：

• 第一行：客户数、规划时间范围内的离散时间槽数以及车辆的运输容量。

• 第二行：对于供应商：

• 供应商的索引

• x 坐标

• y 坐标

• 供应商库存的起始水平

• 每个时间槽可获得的产品数量

• 单位库存成本

• 接下来的每一行：对每个客户

• 客户的索引

• x 坐标

• y 坐标

• 库存的起始水平

• 最大库存水平

• 最小库存水平

• 每个时间槽消耗的产品数量

• 单位库存成本

建模思路¶

库存路径问题 (IRP) 的 OptAgent 模型使用 列表决策变量。对每个离散时间槽，我们定义一个 list variable，表示卡车在该时间槽访问的客户。每条路线的成本取决于卡车行驶的距离。使用 符号 Lambda 函数，我们将沿途从一个客户到下一个客户的距离累加起来。

我们递归地计算供应商的库存水平。事实上，时间 t 的库存水平等于时间 t-1 的库存水平，加上时间 t-1 可获得的产品数量，再减去时间 t-1 运送给客户的产品总量。供应商处的缺货约束确保供应商的库存水平始终足以运送给定时间槽内交付给客户的总数量。

类似地，我们随后计算客户的库存水平。对于每个客户，时间 t 的库存水平等于时间 t-1 的库存水平，加上时间 t-1 从供应商运送给该客户的产品数量，再减去时间 t-1 消耗的产品数量。客户处的缺货约束保证每个客户的库存水平始终为正。

容量约束确保车辆装载的产品总量不超过其容量。最大水平约束保证从供应商运送给每个客户的产品数量不超过该客户的最大库存水平。

目标是最小化所有成本之和：供应商的总库存成本、客户的总库存成本以及总运输成本。

Python 实现¶

import math
from pathlib import Path

from optagent import OptModel, solve




def read_elem(filename):
    return Path(filename).read_text(encoding="utf-8").split()


def main(instance_file, output_file=None, time_limit=5):
    (
        nb_customers,
        horizon_length,
        capacity,
        start_level_supplier,
        production_rate_supplier,
        holding_cost_supplier,
        start_level,
        max_level,
        demand_rate,
        holding_cost,
        dist_matrix_data,
        dist_supplier_data,
    ) = read_input_irp(instance_file)

    model = OptModel()
    # Quantity of product delivered at each discrete time instant of
    # the planning time horizon to each customer
    delivery = [
        [model.float(0, capacity) for _ in range(nb_customers)]
        for _ in range(horizon_length)
    ]
    # Sequence of customers visited at each discrete time instant of
    # the planning time horizon
    route = [model.list(nb_customers) for _ in range(horizon_length)]

    # Customers receive products only if they are visited at least once
    is_delivered = [
        [model.contains(route[t], customer) for customer in range(nb_customers)] for t in range(horizon_length)
    ]
    # Create Hexaly arrays to be able to access them with an "at" operator
    dist_matrix = model.array(dist_matrix_data)
    dist_supplier = model.array(dist_supplier_data)
    dist_routes = []
    for t in range(horizon_length):
        sequence = route[t]
        count = model.count(sequence)
        # Distance traveled at instant t
        dist_lambda = model.lambda_function(lambda position: dist_matrix[sequence[position - 1], sequence[position]])
        dist_routes.append(
            model.iif(
                count > 0,
                dist_supplier[sequence[0]]
                + model.sum(model.range(1, count), dist_lambda)
                + dist_supplier[sequence[count - 1]],
                0,
            )
        )

    # Stockout constraints at the supplier
    inventory_supplier = [start_level_supplier]
    for t in range(horizon_length):
        total_delivery = model.sum(delivery[t])
        inventory_supplier.append(inventory_supplier[t] - total_delivery + production_rate_supplier)
        model.constraint(
            inventory_supplier[t] >= total_delivery,
        )

    # Stockout constraints at the customers
    inventory = [[start_level[customer]] for customer in range(nb_customers)]
    for customer in range(nb_customers):
        for t in range(horizon_length):
            next_inventory = inventory[customer][t] + delivery[t][customer] - demand_rate[customer]
            inventory[customer].append(next_inventory)
            model.constraint(
                next_inventory >= 0,
            )

    for t in range(horizon_length):
        model.constraint(
            model.sum(delivery[t]) <= capacity,
        )
        for customer in range(nb_customers):
            model.constraint(
                delivery[t][customer] <= max_level[customer] - inventory[customer][t],
            )
            model.constraint(
                delivery[t][customer] <= max_level[customer] * is_delivered[t][customer],
            )

    total_cost_inventory_supplier = holding_cost_supplier * model.sum(*inventory_supplier)
    total_cost_inventory = model.sum(
        *(
            holding_cost[customer] * inventory[customer][t]
            for customer in range(nb_customers)
            for t in range(horizon_length + 1)
        )
    )
    total_cost_route = model.sum(dist_routes)
    objective = total_cost_inventory_supplier + total_cost_inventory + total_cost_route
    model.minimize(objective)

    solution = solve(model, time_limit_s=float(time_limit))
    if not solution.feasible:
        # With the original zero-delivery/empty-route defaults, the current
        # OptAgent search may not find a feasible coupled list/float solution
        # within a short run. Keep the model unchanged and report the status.
        print(f"Status = {solution.feasible}; no feasible solution found")
        return solution

    values = {'objective': objective.value, 'route_cost': total_cost_route.value, **{f'route_{t}': route[t].value for t in range(horizon_length)}}
    lines = [
        f"Status = {solution.feasible}; Objective = {values['objective']}; Route cost = {values['route_cost']}"
    ]
    for t in range(horizon_length):
        customers = " ".join(str(customer + 1) for customer in values[f"route_{t}"])
        lines.append(f"Period {t + 1}: {customers}")
    result_text = "\n".join(lines)
    print(result_text)

    if output_file is not None:
        output_lines = [str(int(round(values["route_cost"])))]
        output_lines.extend(
            " ".join(str(customer + 1) for customer in values[f"route_{t}"]) for t in range(horizon_length)
        )
        Path(output_file).write_text("\n".join(output_lines) + "\n", encoding="utf-8")
    return solution


# The input files follow the "Archetti" format
def read_input_irp(filename):
    file_it = iter(read_elem(filename))

    nb_customers = int(next(file_it)) - 1
    horizon_length = int(next(file_it))
    capacity = int(next(file_it))

    x_coord = [None] * nb_customers
    y_coord = [None] * nb_customers
    start_level = [None] * nb_customers
    max_level = [None] * nb_customers
    min_level = [None] * nb_customers
    demand_rate = [None] * nb_customers
    holding_cost = [None] * nb_customers

    next(file_it)
    x_coord_supplier = float(next(file_it))
    y_coord_supplier = float(next(file_it))
    start_level_supplier = int(next(file_it))
    production_rate_supplier = int(next(file_it))
    holding_cost_supplier = float(next(file_it))
    for i in range(nb_customers):
        next(file_it)
        x_coord[i] = float(next(file_it))
        y_coord[i] = float(next(file_it))
        start_level[i] = int(next(file_it))
        max_level[i] = int(next(file_it))
        min_level[i] = int(next(file_it))
        demand_rate[i] = int(next(file_it))
        holding_cost[i] = float(next(file_it))

    distance_matrix = compute_distance_matrix(x_coord, y_coord)
    distance_supplier = compute_distance_supplier(x_coord_supplier, y_coord_supplier, x_coord, y_coord)

    return (
        nb_customers,
        horizon_length,
        capacity,
        start_level_supplier,
        production_rate_supplier,
        holding_cost_supplier,
        start_level,
        max_level,
        demand_rate,
        holding_cost,
        distance_matrix,
        distance_supplier,
    )


# Compute the distance matrix
def compute_distance_matrix(x_coord, y_coord):
    nb_customers = len(x_coord)
    distance_matrix = [[None for _ in range(nb_customers)] for _ in range(nb_customers)]
    for i in range(nb_customers):
        distance_matrix[i][i] = 0
        for j in range(nb_customers):
            dist = compute_dist(x_coord[i], x_coord[j], y_coord[i], y_coord[j])
            distance_matrix[i][j] = dist
            distance_matrix[j][i] = dist
    return distance_matrix


# Compute the distances to the supplier
def compute_distance_supplier(x_coord_supplier, y_coord_supplier, x_coord, y_coord):
    nb_customers = len(x_coord)
    distance_supplier = [None] * nb_customers
    for i in range(nb_customers):
        dist = compute_dist(x_coord_supplier, x_coord[i], y_coord_supplier, y_coord[i])
        distance_supplier[i] = dist
    return distance_supplier


def compute_dist(xi, xj, yi, yj):
    return round(math.sqrt(math.pow(xi - xj, 2) + math.pow(yi - yj, 2)))

INSTANCE_DIR = Path.cwd() / "instances"

solution = main(INSTANCE_DIR / "abs1n5.dat", time_limit=5)