VRP 實戰:物流平台 API
建立配送最佳化 API
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional, Tuple
import numpy as np
from ortools.constraint_solver import routing_enums_pb2, pywrapcp
app = FastAPI(title="物流配送最佳化 API")
class DeliveryRequest(BaseModel):
depot: Tuple[float, float]
customers: List[Tuple[float, float]]
demands: List[int]
vehicle_capacity: int
num_vehicles: int
class RouteInfo(BaseModel):
vehicle_id: int
stops: List[int]
total_distance: float
total_demand: int
class DeliveryResponse(BaseModel):
success: bool
routes: List[RouteInfo]
total_distance: float
@app.post("/optimize", response_model=DeliveryResponse)
def optimize_delivery(req: DeliveryRequest):
# 建立距離矩陣
all_points = [req.depot] + req.customers
n = len(all_points)
dist_matrix = [[0] * n for _ in range(n)]
for i in range(n):
for j in range(n):
dx = all_points[i][0] - all_points[j][0]
dy = all_points[i][1] - all_points[j][1]
dist_matrix[i][j] = int(np.sqrt(dx*dx + dy*dy))
# OR-Tools 求解
manager = pywrapcp.RoutingIndexManager(n, req.num_vehicles, 0)
routing = pywrapcp.RoutingModel(manager)
def distance_cb(from_idx, to_idx):
return dist_matrix[manager.IndexToNode(from_idx)][manager.IndexToNode(to_idx)]
transit_idx = routing.RegisterTransitCallback(distance_cb)
routing.SetArcCostEvaluatorOfAllVehicles(transit_idx)
demands = [0] + req.demands
def demand_cb(from_idx):
return demands[manager.IndexToNode(from_idx)]
demand_idx = routing.RegisterUnaryTransitCallback(demand_cb)
routing.AddDimensionWithVehicleCapacity(demand_idx, 0, [req.vehicle_capacity]*req.num_vehicles, True, "Capacity")
params = pywrapcp.DefaultRoutingSearchParameters()
params.first_solution_strategy = routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
params.time_limit.seconds = 5
solution = routing.SolveWithParameters(params)
if not solution:
return DeliveryResponse(success=False, routes=[], total_distance=0)
routes = []
total_dist = 0
for v in range(req.num_vehicles):
idx = routing.Start(v)
stops = []
dist = 0
demand = 0
while not routing.IsEnd(idx):
node = manager.IndexToNode(idx)
stops.append(node)
demand += demands[node]
prev = idx
idx = solution.Value(routing.NextVar(idx))
dist += dist_matrix[node][manager.IndexToNode(idx)]
stops.append(0)
routes.append(RouteInfo(vehicle_id=v, stops=stops, total_distance=float(dist), total_demand=demand))
total_dist += dist
return DeliveryResponse(success=True, routes=routes, total_distance=float(total_dist))
這套 API 可以直接讓物流平台的前端調用,實現即時路線規劃!