Оптимізація маршрутів

Оптимізація маршрутів використовує ML для планування ефективних шляхів для транспортних засобів, водіїв доставки та логістичних мереж. Класичне дослідження операцій надає точні алгоритми, але ML додає навчання з історичних паттернів, обробку невизначеності та адаптацію в реальному часі, що чиста оптимізація не може досягти сама.

Де використовується

Оптимізація маршрутів забезпечує сучасну логістику:

• Last-mile доставка: Amazon, FedEx, UPS планування маршрутів
• Райдшерінг: Uber, Lyft диспетчеризація водіїв
• Доставка їжі: DoorDash, Deliveroo групування замовлень
• Польові сервіси: Планування техніків, домашні візити
• Ланцюг постачання: Маршрутизація вантажівок, збір на складі
• Громадський транспорт: Оптимізація маршрутів автобусів
• Робототехніка: Складські роботи, автономні транспортні засоби

Варіанти проблем

Що ви робите на практиці

1. Визначте обмеження - Місткість, часові вікна, години водіїв
2. Змоделюйте проблему - Матриця відстаней, час обслуговування
3. Почніть з евристик - Nearest neighbour, sweep algorithm
4. Покращте локальним пошуком - 2-opt, 3-opt, LKH
5. Додайте ML прогнози - Час подорожі, тривалість обслуговування
6. Обробіть динаміку - Re-routing в реальному часі
7. Валідуйте рішення - Задоволення обмежень
8. Моніторте виконання - Фактичне vs заплановане

Спочатку базова лінія

Почніть з класичних OR алгоритмів перед ML-enhanced підходами:

import numpy as np
from typing import List, Tuple, Dict
from scipy.spatial.distance import cdist

class NearestNeighbourTSP:
    """Класична евристика nearest neighbour."""

    def solve(self, distances: np.ndarray, start: int = 0) -> List[int]:
        """
        Nearest neighbour рішення TSP.

        Args:
            distances: N x N матриця відстаней
            start: Початкова вершина

        Returns:
            Тур як список індексів вершин
        """
        n = len(distances)
        visited = [False] * n
        tour = [start]
        visited[start] = True

        for _ in range(n - 1):
            current = tour[-1]
            nearest = None
            nearest_dist = float('inf')

            for j in range(n):
                if not visited[j] and distances[current][j] < nearest_dist:
                    nearest = j
                    nearest_dist = distances[current][j]

            tour.append(nearest)
            visited[nearest] = True

        return tour

    def tour_length(self, tour: List[int], distances: np.ndarray) -> float:
        """Розрахунок загальної довжини туру."""
        total = sum(
            distances[tour[i]][tour[i + 1]] 
            for i in range(len(tour) - 1)
        )
        total += distances[tour[-1]][tour[0]]  # Повернення на старт
        return total

class TwoOptImprover:
    """2-opt покращення локальним пошуком."""

    def improve(self, tour: List[int], distances: np.ndarray, 
                max_iterations: int = 1000) -> List[int]:
        """Покращення туру використовуючи 2-opt swaps."""
        improved = True
        iteration = 0

        while improved and iteration < max_iterations:
            improved = False
            iteration += 1

            for i in range(len(tour) - 1):
                for j in range(i + 2, len(tour)):
                    if j == len(tour) - 1 and i == 0:
                        continue

                    # Розрахунок покращення
                    delta = self._swap_delta(tour, i, j, distances)

                    if delta < -1e-10:  # Знайдено покращення
                        # Виконання 2-opt swap (реверс сегменту)
                        tour = tour[:i+1] + tour[i+1:j+1][::-1] + tour[j+1:]
                        improved = True
                        break

                if improved:
                    break

        return tour

    def _swap_delta(self, tour: List[int], i: int, j: int, 
                    distances: np.ndarray) -> float:
        """Розрахунок зміни довжини туру від 2-opt swap."""
        n = len(tour)

        # Поточні ребра
        a, b = tour[i], tour[i + 1]
        c, d = tour[j], tour[(j + 1) % n]

        # Нові ребра після swap
        old_cost = distances[a][b] + distances[c][d]
        new_cost = distances[a][c] + distances[b][d]

        return new_cost - old_cost

class CapacitatedVRP:
    """Базовий CVRP з savings algorithm."""

    def __init__(self, depot: int = 0):
        self.depot = depot

    def solve(self, distances: np.ndarray, demands: np.ndarray,
              vehicle_capacity: float, num_vehicles: int) -> List[List[int]]:
        """
        Clarke-Wright savings algorithm для CVRP.

        Returns:
            Список маршрутів, кожен маршрут - список індексів клієнтів
        """
        n = len(distances)
        customers = [i for i in range(n) if i != self.depot]

        # Розрахунок savings
        savings = []
        for i in customers:
            for j in customers:
                if i < j:
                    s = (distances[self.depot][i] + distances[self.depot][j] 
                         - distances[i][j])
                    savings.append((s, i, j))

        # Сортування за savings (за спаданням)
        savings.sort(reverse=True)

        # Ініціалізація: кожен клієнт у власному маршруті
        routes = {i: [i] for i in customers}
        route_demands = {i: demands[i] for i in customers}

        # Об'єднання маршрутів на основі savings
        for s, i, j in savings:
            if i not in routes or j not in routes:
                continue

            route_i = routes[i]
            route_j = routes[j]

            if route_i is route_j:
                continue

            # Перевірка місткості
            combined_demand = route_demands[i] + route_demands[j]
            if combined_demand > vehicle_capacity:
                continue

            # Перевірка чи i та j на кінцях маршрутів
            if route_i[-1] == i and route_j[0] == j:
                # Об'єднання маршрутів
                new_route = route_i + route_j
                new_demand = combined_demand

                # Оновлення посилань на маршрути
                for node in new_route:
                    routes[node] = new_route
                    route_demands[node] = new_demand

        # Екстракція унікальних маршрутів
        unique_routes = []
        seen = set()
        for route in routes.values():
            route_key = tuple(route)
            if route_key not in seen:
                seen.add(route_key)
                unique_routes.append(route)

        return unique_routes[:num_vehicles]

ML-Enhanced маршрутизація

import numpy as np
import torch
import torch.nn as nn
from torch.distributions import Categorical

class TravelTimePredictor:
    """Прогнозування часу подорожі з фіч."""

    def __init__(self, feature_dim: int):
        self.model = nn.Sequential(
            nn.Linear(feature_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 32),
            nn.ReLU(),
            nn.Linear(32, 1),
            nn.Softplus()  # Забезпечення позитивних часів
        )

    def predict(self, origin: np.ndarray, destination: np.ndarray,
                time_features: np.ndarray) -> float:
        """Прогнозування часу подорожі з origin, destination та контексту."""
        features = np.concatenate([origin, destination, time_features])
        with torch.no_grad():
            x = torch.FloatTensor(features)
            return self.model(x).item()

    def build_time_matrix(self, locations: np.ndarray, 
                          time_features: np.ndarray) -> np.ndarray:
        """Побудова прогнозованої матриці часу подорожі."""
        n = len(locations)
        times = np.zeros((n, n))

        for i in range(n):
            for j in range(n):
                if i != j:
                    times[i][j] = self.predict(
                        locations[i], locations[j], time_features
                    )

        return times

class AttentionVRPModel(nn.Module):
    """Attention-based модель для VRP (натхнена Attention, Learn to Solve)."""

    def __init__(self, input_dim: int = 2, embed_dim: int = 128, n_heads: int = 8):
        super().__init__()

        # Ембедінг вершин
        self.node_embed = nn.Linear(input_dim, embed_dim)

        # Transformer encoder
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embed_dim, nhead=n_heads, batch_first=True
        )
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=3)

        # Контекст для декодування
        self.context_embed = nn.Linear(embed_dim * 2, embed_dim)

        # Pointer network attention
        self.pointer = nn.Linear(embed_dim, embed_dim)

    def forward(self, nodes: torch.Tensor) -> torch.Tensor:
        """
        Кодування вершин та обчислення attention weights.

        Args:
            nodes: (batch, n_nodes, input_dim)

        Returns:
            Node embeddings
        """
        embedded = self.node_embed(nodes)
        encoded = self.encoder(embedded)
        return encoded

    def get_action_probs(self, encoded: torch.Tensor, 
                         current_node: int,
                         visited_mask: torch.Tensor) -> torch.Tensor:
        """Отримання ймовірностей для вибору наступної вершини."""
        batch_size, n_nodes, embed_dim = encoded.shape

        # Ембедінг поточної вершини
        current_embed = encoded[:, current_node, :]

        # Глобальний контекст
        graph_embed = encoded.mean(dim=1)

        # Комбінований контекст
        context = torch.cat([current_embed, graph_embed], dim=-1)
        context = self.context_embed(context)

        # Обчислення attention scores
        query = self.pointer(context).unsqueeze(1)
        scores = torch.bmm(query, encoded.transpose(1, 2)).squeeze(1)

        # Маскування відвіданих вершин
        scores = scores.masked_fill(visited_mask, float('-inf'))

        # Softmax для ймовірностей
        probs = torch.softmax(scores, dim=-1)

        return probs

    def sample_tour(self, nodes: torch.Tensor, greedy: bool = False) -> List[int]:
        """Семплювання туру використовуючи модель."""
        batch_size, n_nodes, _ = nodes.shape

        encoded = self.forward(nodes)

        tour = [0]  # Старт з депо
        visited = torch.zeros(batch_size, n_nodes, dtype=torch.bool)
        visited[:, 0] = True

        current = 0
        for _ in range(n_nodes - 1):
            probs = self.get_action_probs(encoded, current, visited)

            if greedy:
                next_node = probs.argmax(dim=-1).item()
            else:
                dist = Categorical(probs)
                next_node = dist.sample().item()

            tour.append(next_node)
            visited[:, next_node] = True
            current = next_node

        return tour

Часові вікна та обмеження

from dataclasses import dataclass
from typing import Optional

@dataclass
class Stop:
    """Зупинка доставки з обмеженнями."""
    id: int
    location: np.ndarray
    demand: float
    time_window_start: float
    time_window_end: float
    service_time: float
    priority: int = 1

class VRPTWSolver:
    """VRP з часовими вікнами."""

    def __init__(self, depot_location: np.ndarray):
        self.depot = depot_location

    def is_feasible(self, route: List[Stop], arrival_times: List[float],
                    vehicle_capacity: float) -> bool:
        """Перевірка чи маршрут задовольняє всі обмеження."""
        # Обмеження місткості
        total_demand = sum(stop.demand for stop in route)
        if total_demand > vehicle_capacity:
            return False

        # Обмеження часових вікон
        for stop, arrival in zip(route, arrival_times):
            if arrival > stop.time_window_end:
                return False

        return True

    def compute_arrival_times(self, route: List[Stop], 
                              travel_times: callable,
                              start_time: float = 0) -> List[float]:
        """Обчислення часів прибуття для кожної зупинки."""
        arrivals = []
        current_time = start_time
        current_loc = self.depot

        for stop in route:
            travel = travel_times(current_loc, stop.location)
            arrival = current_time + travel

            # Очікування якщо прибуття до часового вікна
            if arrival < stop.time_window_start:
                arrival = stop.time_window_start

            arrivals.append(arrival)
            current_time = arrival + stop.service_time
            current_loc = stop.location

        return arrivals

    def greedy_insertion(self, stops: List[Stop], 
                         num_vehicles: int,
                         vehicle_capacity: float,
                         travel_times: callable) -> List[List[Stop]]:
        """Побудова маршрутів жадібною вставкою."""
        remaining = sorted(stops, key=lambda s: s.time_window_end)
        routes = [[] for _ in range(num_vehicles)]
        route_loads = [0] * num_vehicles

        for stop in remaining:
            best_route = None
            best_position = None
            best_cost = float('inf')

            for r_idx, route in enumerate(routes):
                if route_loads[r_idx] + stop.demand > vehicle_capacity:
                    continue

                for pos in range(len(route) + 1):
                    cost = self.insertion_cost(route, stop, pos, travel_times)
                    if cost is not None and cost < best_cost:
                        best_cost = cost
                        best_route = r_idx
                        best_position = pos

            if best_route is not None:
                routes[best_route].insert(best_position, stop)
                route_loads[best_route] += stop.demand

        return routes

Метрики оцінки

Типові помилки та чек-лист

• [ ] Недопустимі рішення: Маршрути порушують обмеження
• [ ] Субоптимальні результати: Застрягання в локальних мінімумах
• [ ] Помилки прогнозування часу: Фактичний != прогнозований час
• [ ] Динамічні зміни: Не адаптація до оновлень в реальному часі
• [ ] Масштабованість: Занадто повільно для розміру проблеми
• [ ] Відповідність водіїв: Маршрути не дотримуються на практиці

Чек-лист для продакшену

Перед деплоєм: - [ ] Валідація на історичних маршрутах - [ ] Тестування обробки обмежень - [ ] Бенчмарк часу розв'язання - [ ] Налаштування fallback рішень - [ ] Інтерфейс водія/навігація

Після деплою: - [ ] Моніторинг фактичних vs запланованих маршрутів - [ ] Відстеження on-time delivery rate - [ ] Вимірювання задоволеності водіїв - [ ] Оновлення моделей часу подорожі - [ ] A/B тест покращень

Інструменти та стек

Приклад міні-кейсу

Сценарій: Оптимізація last-mile доставки для e-commerce.

Підхід: - ML прогнозування часу подорожі (traffic-aware) - VRPTW з м'якими часовими вікнами - Real-time re-routing для затримок - Навчання вподобань водіїв

Результати: - 15% зменшення загальної відстані - 23% покращення on-time доставки - 8% менше транспортних засобів - Задоволеність водіїв покращена на 12%

Пов'язані сторінки

• Індекс: ML Field Guide
• Розподіл ресурсів
• Reinforcement Learning
• Прогнозування часових рядів

Додаткові матеріали

• Google OR-Tools Routing
• Attention, Learn to Solve Routing Problems
• Vehicle Routing: Problems, Methods, and Applications

Спробуйте наживо

Дослідіть пов'язану модель на практиці: відкрийте AI Route Optimization Lab — A Delivery Fleet Case Study і змінюйте параметри під час роботи. Усе лишається у вашому браузері.