AI nella Logistica: Route Optimization, Warehouse Automation e Supply Chain Intelligence
La logistica e la catena di approvvigionamento sono da sempre dominate dalla complessità: milioni di variabili interdipendenti, finestre temporali strette, costi operativi che erodono i margini, e l'imprevedibilita della domanda che mette a rischio la continuita del servizio. Per decenni, le aziende hanno cercato di contenere questa complessità con fogli Excel, regole empiriche e l'esperienza dei pianificatori più anziani. Oggi, l'intelligenza artificiale sta ridisegnando le regole del gioco.
Il mercato globale dell'AI applicata alla supply chain ha raggiunto i 9,8 miliardi di dollari nel 2025, con proiezioni che indicano una crescita fino a 32 miliardi entro il 2030 (CAGR 26.4%). Non si tratta di hype: le aziende che hanno adottato l'AI nei processi logistici registrano riduzioni del 10-15% nei costi di trasporto, 15-20% di accelerazione nei tempi di consegna e circa il 30% in meno di consegne in ritardo. Amazon gestisce oltre 520.000 robot AI-powered nei suoi magazzini, riducendo i costi di evasione del 20% e processando il 40% di ordini in più per ora.
In questo articolo esploriamo le tecnologie AI che stanno trasformando la logistica: dal Vehicle Routing Problem (VRP) risolto con OR-Tools e reinforcement learning, al demand forecasting con Temporal Fusion Transformer, fino all'automazione dei magazzini, all'ottimizzazione dell'ultimo miglio e alla gestione intelligente dell'inventario. Vedremo implementazioni concrete con codice Python funzionante e casi d'uso reali dal contesto italiano.
Cosa Imparerai in Questo Articolo
- Come risolvere il Vehicle Routing Problem (VRP) con Google OR-Tools in Python
- Demand forecasting con Prophet, LightGBM e Temporal Fusion Transformer
- Ottimizzazione dell'inventario con Reinforcement Learning (PPO/DQN)
- Automazione magazzino: robotica, pick optimization e sistemi WMS intelligenti
- Last-mile delivery: AI, droni e veicoli autonomi nel contesto urbano
- Real-time visibility e digital twin della supply chain
- Carbon footprint optimization e logistica sostenibile
- Casi d'uso italiani: Amazon IT, Poste Italiane, GLS
Posizione nella Serie Data Warehouse, AI e Trasformazione Digitale
| # | Articolo | Stato |
|---|---|---|
| 1 | Evoluzione del Data Warehouse | Pubblicato |
| 2 | Data Mesh e Architettura Decentralizzata | Pubblicato |
| 3 | ETL vs ELT Moderno: dbt, Airbyte e Fivetran | Pubblicato |
| 4 | Orchestrazione Pipeline: Airflow, Dagster e Prefect | Pubblicato |
| 5 | AI nella Manifattura: Predictive Maintenance | Pubblicato |
| 6 | AI nel Finance: Fraud Detection e Credit Scoring | Pubblicato |
| 7 | AI nel Retail: Demand Forecasting e Recommendation | Pubblicato |
| 8 | AI in Healthcare: Diagnostica e Drug Discovery | Pubblicato |
| 9 | AI nella Logistica (Sei qui) | Corrente |
| 10 | LLM in Azienda: RAG Enterprise e Guardrails | Prossimo |
Vehicle Routing Problem: L'Ottimizzazione dei Percorsi con OR-Tools
Il Vehicle Routing Problem (VRP) e uno dei problemi più studiati della ricerca operativa: dato un insieme di clienti con specifiche richieste di consegna e un parco veicoli partente da uno o più depositi, come assegnare i clienti ai veicoli e pianificare i percorsi per minimizzare il costo totale (distanza, tempo, carburante)?
Il VRP e NP-hard: non esiste un algoritmo che lo risolva esattamente in tempo polinomiale per istanze di grandi dimensioni. Per questo motivo, le soluzioni pratiche usano una combinazione di metaeuristiche (Simulated Annealing, Tabu Search, algoritmi genetici) e solver commerciali e open source. Google OR-Tools e oggi lo strumento open-source più utilizzato per questo tipo di problemi: supporta CVRP (capacitàted VRP), VRPTW (con finestre temporali), Multi-Depot VRP e molte varianti realistiche.
Il sistema ORION di UPS, basato su tecniche simili, calcola 30.000 ottimizzazioni di percorso al minuto e ha permesso di risparmiare 38 milioni di litri di carburante all'anno, evitando 100 milioni di miglia di guida superflua. Non e un vantaggio marginale: e un vantaggio competitivo strutturale che si traduce in decine di milioni di dollari di risparmio annuo.
Implementazione CVRP con Google OR-Tools
Vediamo un'implementazione completa del capacitàted VRP con finestre temporali (VRPTW), il tipo più comune in contesti logistici reali dove ogni cliente ha orari di apertura precisi.
"""
VRPTW - Vehicle Routing Problem with Time Windows
Risolto con Google OR-Tools
Scenario: consegne B2B in area metropolitana italiana
"""
from ortools.constraint_solver import routing_enums_pb2
from ortools.constraint_solver import pywrapcp
import numpy as np
from typing import List, Dict, Tuple
import json
# ============================================================
# DEFINIZIONE DEI DATI DEL PROBLEMA
# ============================================================
def create_data_model() -> Dict:
"""
Crea il modello dati per il VRPTW.
In produzione questi dati vengono da:
- Database ordini (PostgreSQL/DWH)
- API di geocoding per le coordinate
- API Google Maps Distance Matrix per le distanze
"""
data = {}
# Matrice delle distanze in secondi (tempo di viaggio)
# Indice 0 = deposito, indici 1-N = clienti
data['time_matrix'] = [
[0, 548, 776, 696, 582, 274, 502, 194, 308, 194, 536, 502, 388, 354],
[548, 0, 684, 308, 194, 502, 730, 354, 696, 742, 1084, 594, 480, 514],
[776, 684, 0, 992, 878, 502, 274, 810, 468, 742, 400, 1278, 1164, 1130],
[696, 308, 992, 0, 114, 650, 878, 502, 844, 890, 1232, 514, 628, 822],
[582, 194, 878, 114, 0, 536, 764, 388, 730, 776, 1118, 400, 514, 708],
[274, 502, 502, 650, 536, 0, 228, 308, 194, 240, 582, 776, 662, 628],
[502, 730, 274, 878, 764, 228, 0, 536, 194, 468, 354, 1004, 890, 856],
[194, 354, 810, 502, 388, 308, 536, 0, 342, 388, 730, 468, 354, 320],
[308, 696, 468, 844, 730, 194, 194, 342, 0, 274, 388, 810, 696, 662],
[194, 742, 742, 890, 776, 240, 468, 388, 274, 0, 342, 536, 422, 388],
[536, 1084, 400, 1232, 1118, 582, 354, 730, 388, 342, 0, 878, 764, 730],
[502, 594, 1278, 514, 400, 776, 1004, 468, 810, 536, 878, 0, 114, 308],
[388, 480, 1164, 628, 514, 662, 890, 354, 696, 422, 764, 114, 0, 194],
[354, 514, 1130, 822, 708, 628, 856, 320, 662, 388, 730, 308, 194, 0],
]
# Finestre temporali [inizio, fine] in secondi dall'apertura deposito
# 0 = 08:00, 3600 = 09:00, 28800 = 16:00
data['time_windows'] = [
(0, 28800), # Deposito: aperto tutto il giorno
(7200, 14400), # Cliente 1: 10:00-12:00
(10800, 18000), # Cliente 2: 11:00-13:00
(3600, 14400), # Cliente 3: 09:00-12:00
(0, 10800), # Cliente 4: 08:00-11:00
(14400, 21600), # Cliente 5: 12:00-14:00
(0, 14400), # Cliente 6: 08:00-12:00
(7200, 18000), # Cliente 7: 10:00-13:00
(0, 21600), # Cliente 8: 08:00-14:00
(3600, 10800), # Cliente 9: 09:00-11:00
(18000, 25200), # Cliente 10: 13:00-15:00
(0, 14400), # Cliente 11: 08:00-12:00
(3600, 18000), # Cliente 12: 09:00-13:00
(7200, 21600), # Cliente 13: 10:00-14:00
]
# capacità di ciascun veicolo (in kg)
data['vehicle_capacities'] = [1000, 1000, 800, 800]
data['num_vehicles'] = 4
# Indice del deposito
data['depot'] = 0
# Domanda di ogni cliente (in kg)
data['demands'] = [0, 120, 80, 200, 150, 90, 110, 60, 180, 70, 200, 130, 95, 85]
return data
# ============================================================
# CALLBACK FUNCTIONS PER OR-TOOLS
# ============================================================
def create_time_callback(data: Dict, manager):
"""Ritorna una callback per il tempo di percorrenza."""
time_matrix = data['time_matrix']
def time_callback(from_index, to_index):
from_node = manager.IndexToNode(from_index)
to_node = manager.IndexToNode(to_index)
return time_matrix[from_node][to_node]
return time_callback
def create_demand_callback(data: Dict, manager):
"""Ritorna una callback per le domande dei clienti."""
demands = data['demands']
def demand_callback(from_index):
from_node = manager.IndexToNode(from_index)
return demands[from_node]
return demand_callback
# ============================================================
# RISOLUZIONE DEL PROBLEMA
# ============================================================
def solve_vrptw(data: Dict) -> Dict:
"""
Risolve il VRPTW con OR-Tools.
Returns:
Dict con i percorsi ottimizzati e le metriche
"""
# Crea il gestore dell'indice dei nodi
manager = pywrapcp.RoutingIndexManager(
len(data['time_matrix']),
data['num_vehicles'],
data['depot']
)
# Crea il modello di routing
routing = pywrapcp.RoutingModel(manager)
# Registra le callback
time_callback = create_time_callback(data, manager)
transit_callback_index = routing.RegisterTransitCallback(time_callback)
demand_callback = create_demand_callback(data, manager)
demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)
# Imposta il costo dell'arco (tempo di percorrenza)
routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)
# Aggiunge il vincolo di capacità
routing.AddDimensionWithVehicleCapacity(
demand_callback_index,
0, # Slack iniziale
data['vehicle_capacities'], # capacità massima per veicolo
True, # Start cumul to zero
'Capacity'
)
# Aggiunge la dimensione temporale con finestre
routing.AddDimension(
transit_callback_index,
30, # Slack max (attesa max in secondi)
28800, # Orizzonte temporale massimo (8 ore)
False, # Non forzare start a zero
'Time'
)
time_dimension = routing.GetDimensionOrDie('Time')
# Imposta le finestre temporali
for location_idx, time_window in enumerate(data['time_windows']):
if location_idx == data['depot']:
continue
index = manager.NodeToIndex(location_idx)
time_dimension.CumulVar(index).SetRange(time_window[0], time_window[1])
# Imposta le finestre temporali del deposito per ogni veicolo
depot_idx = data['depot']
for vehicle_id in range(data['num_vehicles']):
index = routing.Start(vehicle_id)
time_dimension.CumulVar(index).SetRange(
data['time_windows'][depot_idx][0],
data['time_windows'][depot_idx][1]
)
# Minimizza il tempo totale di percorrenza
for i in range(data['num_vehicles']):
routing.AddVariableMinimizedByFinalizer(
time_dimension.CumulVar(routing.Start(i))
)
routing.AddVariableMinimizedByFinalizer(
time_dimension.CumulVar(routing.End(i))
)
# Parametri di ricerca: PATH_CHEAPEST_ARC come prima soluzione,
# poi miglioramento con GUIDED_LOCAL_SEARCH
search_parameters = pywrapcp.DefaultRoutingSearchParameters()
search_parameters.first_solution_strategy = (
routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
)
search_parameters.local_search_metaheuristic = (
routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
)
search_parameters.time_limit.FromSeconds(30) # 30 secondi di ottimizzazione
# Risoluzione
solution = routing.SolveWithParameters(search_parameters)
if not solution:
return {"status": "INFEASIBLE", "routes": []}
# Estrai i risultati
return extract_solution(data, manager, routing, solution)
def extract_solution(data, manager, routing, solution) -> Dict:
"""Estrae la soluzione in formato leggibile."""
time_dimension = routing.GetDimensionOrDie('Time')
results = {
"status": "OPTIMAL",
"total_time": 0,
"routes": []
}
for vehicle_id in range(data['num_vehicles']):
index = routing.Start(vehicle_id)
route = {
"vehicle_id": vehicle_id,
"stops": [],
"total_time": 0,
"total_load": 0
}
while not routing.IsEnd(index):
node_index = manager.IndexToNode(index)
time_var = time_dimension.CumulVar(index)
route["stops"].append({
"node": node_index,
"arrival": solution.Min(time_var),
"departure": solution.Max(time_var)
})
route["total_load"] += data['demands'][node_index]
index = solution.Value(routing.NextVar(index))
# Nodo finale (deposito)
time_var = time_dimension.CumulVar(index)
route["total_time"] = solution.Min(time_var)
results["routes"].append(route)
results["total_time"] += route["total_time"]
return results
# ============================================================
# MAIN
# ============================================================
if __name__ == "__main__":
data = create_data_model()
result = solve_vrptw(data)
print(f"Status: {result['status']}")
print(f"Tempo totale di percorrenza: {result['total_time']} secondi")
for route in result["routes"]:
print(f"\nVeicolo {route['vehicle_id']}:")
print(f" Carico totale: {route['total_load']} kg")
stops_str = " -> ".join(
[f"Cliente{s['node']}({s['arrival']//3600}:{(s['arrival']%3600)//60:02d})"
for s in route["stops"] if s["node"] != 0]
)
print(f" Percorso: Deposito -> {stops_str} -> Deposito")
In un contesto produttivo, la matrice dei tempi di percorrenza viene calcolata in tempo reale tramite le API Google Maps Distance Matrix o HERE Routing, tenendo conto del traffico attuale. I dati dei clienti provengono dall'ERP aziendale e vengono aggiornati ogni ora. OR-Tools restituisce la soluzione in pochi secondi per istanze fino a 200-300 clienti; per istanze più grandi si usano approcci a cluster o solver GPU-accelerati come NVIDIA cuOpt.
Demand Forecasting: Previsione della Domanda con ML
La previsione accurata della domanda e il fondamento di tutta la supply chain. Senza sapere quanti prodotti verranno richiesti nelle prossime settimane, e impossibile ottimizzare gli acquisti, dimensionare il magazzino, pianificare i trasporti e garantire i livelli di servizio. Per decenni, le aziende hanno usato modelli statistici classici come ARIMA, SARIMA e smoothing esponenziale. Oggi, i modelli di machine learning superano sistematicamente queste baseline.
Il confronto più interessante nel 2025 e tra tre approcci distinti:
Confronto Modelli di Demand Forecasting
| Modello | Tipo | Punti di Forza | Limitazioni | MAPE Tipico |
|---|---|---|---|---|
| Prophet (Meta) | Additivo bayesiano | Gestisce stagionalita multipla, festivi, trend | Non scala su migliaia di SKU facilmente | 8-12% |
| LightGBM | Gradient Boosting | Veloce, feature engineering flessibile, produzione | Richiede feature engineering manuale | 5-9% |
| Temporal Fusion Transformer | Deep Learning | Multi-horizon, interpretabile, variabili esogene | Più lento da addestrare, GPU necessaria | 4-7% |
| SARIMA (baseline) | Statistico | Semplice, interpretabile | Non cattura non-linearita | 12-20% |
Demand Forecasting con LightGBM per Supply Chain
LightGBM e spesso la scelta migliore per il deployment in produzione: addestramento veloce, inferenza in millisecondi, supporto nativo per valori mancanti e ottima scalabilità su migliaia di SKU. Ecco un'implementazione completa con feature engineering specifico per logistica.
"""
Demand Forecasting per Supply Chain con LightGBM
Feature engineering avanzato per serie temporali logistiche
"""
import pandas as pd
import numpy as np
import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_absolute_percentage_error
from typing import List, Tuple
import warnings
warnings.filterwarnings('ignore')
# ============================================================
# FEATURE ENGINEERING PER SERIE TEMPORALI LOGISTICHE
# ============================================================
def create_lag_features(df: pd.DataFrame, target_col: str,
lags: List[int]) -> pd.DataFrame:
"""Crea feature di lag per catturare la dipendenza temporale."""
df = df.copy()
for lag in lags:
df[f'lag_{lag}'] = df.groupby('sku_id')[target_col].shift(lag)
return df
def create_rolling_features(df: pd.DataFrame, target_col: str,
windows: List[int]) -> pd.DataFrame:
"""Media e deviazione standard mobile per catturare trend e variabilità."""
df = df.copy()
for window in windows:
df[f'rolling_mean_{window}'] = (
df.groupby('sku_id')[target_col]
.transform(lambda x: x.shift(1).rolling(window).mean())
)
df[f'rolling_std_{window}'] = (
df.groupby('sku_id')[target_col]
.transform(lambda x: x.shift(1).rolling(window).std())
)
return df
def create_calendar_features(df: pd.DataFrame, date_col: str) -> pd.DataFrame:
"""Feature calendario: stagionalita, festivi italiani, weekend."""
df = df.copy()
df['date'] = pd.to_datetime(df[date_col])
# Feature temporali base
df['day_of_week'] = df['date'].dt.dayofweek
df['day_of_month'] = df['date'].dt.day
df['week_of_year'] = df['date'].dt.isocalendar().week.astype(int)
df['month'] = df['date'].dt.month
df['quarter'] = df['date'].dt.quarter
df['is_weekend'] = (df['day_of_week'] >= 5).astype(int)
# Festivi italiani (principali)
italian_holidays = [
'01-01', # Capodanno
'04-25', # Festa della Liberazione
'05-01', # Festa del Lavoro
'06-02', # Festa della Repubblica
'08-15', # Ferragosto
'11-01', # Ognissanti
'12-08', # Immacolata
'12-25', # Natale
'12-26', # Santo Stefano
]
df['is_holiday'] = df['date'].apply(
lambda d: 1 if f'{d.month:02d}-{d.day:02d}' in italian_holidays else 0
)
# Proximity ai festivi (effetti anticipazione/posticipazione)
df['days_to_holiday'] = df.apply(
lambda row: min(
abs((row['date'] - pd.Timestamp(f"{row['date'].year}-{h}")).days)
for h in italian_holidays
), axis=1
).clip(upper=7)
# Effetto Ferragosto (agosto): domanda compressa
df['is_august'] = (df['month'] == 8).astype(int)
# Peak season (Q4: ottobre-dicembre, Black Friday, Natale)
df['is_peak_season'] = df['month'].isin([10, 11, 12]).astype(int)
return df
def engineer_features(df: pd.DataFrame) -> pd.DataFrame:
"""
Pipeline completa di feature engineering.
Input DataFrame deve avere: sku_id, date, quantity, price,
promotions, stock_level, supplier_lead_time
"""
# Ordina per SKU e data
df = df.sort_values(['sku_id', 'date']).reset_index(drop=True)
# Lag features: 1, 7, 14, 28, 56 giorni
df = create_lag_features(df, 'quantity', lags=[1, 7, 14, 28, 56])
# Rolling features: 7, 14, 28 giorni
df = create_rolling_features(df, 'quantity', windows=[7, 14, 28])
# Feature calendario
df = create_calendar_features(df, 'date')
# Rapporto tra prezzo corrente e media storica (effetto promo)
df['price_ratio'] = df.groupby('sku_id')['price'].transform(
lambda x: x / x.expanding().mean()
)
# Indicatore di stockout recente (qualità del dato)
df['recent_stockout'] = (
df.groupby('sku_id')['stock_level']
.transform(lambda x: x.shift(1).rolling(7).min()) == 0
).astype(int)
return df
# ============================================================
# TRAINING E VALIDAZIONE
# ============================================================
def train_lgbm_forecaster(df: pd.DataFrame) -> Tuple[lgb.Booster, List[str]]:
"""
Addestra LightGBM con validazione time-series (walk-forward).
Returns:
Modello addestrato e lista delle feature usate
"""
FEATURE_COLS = [
# Lag features
'lag_1', 'lag_7', 'lag_14', 'lag_28', 'lag_56',
# Rolling features
'rolling_mean_7', 'rolling_mean_14', 'rolling_mean_28',
'rolling_std_7', 'rolling_std_14', 'rolling_std_28',
# Calendar
'day_of_week', 'day_of_month', 'week_of_year', 'month', 'quarter',
'is_weekend', 'is_holiday', 'days_to_holiday',
'is_august', 'is_peak_season',
# Business features
'price_ratio', 'promotions', 'supplier_lead_time', 'recent_stockout'
]
# Rimuovi righe con NaN dalle feature lag
df_train = df.dropna(subset=FEATURE_COLS).copy()
X = df_train[FEATURE_COLS]
y = df_train['quantity']
# Validazione time-series: non shuffle!
tscv = TimeSeriesSplit(n_splits=5)
lgb_params = {
'objective': 'regression_l1', # MAE loss, robusta agli outlier
'metric': 'mape',
'num_leaves': 127,
'learning_rate': 0.05,
'feature_fraction': 0.8,
'bagging_fraction': 0.8,
'bagging_freq': 5,
'min_data_in_leaf': 50,
'lambda_l1': 0.1,
'lambda_l2': 0.1,
'verbose': -1,
'n_jobs': -1
}
mape_scores = []
for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):
X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]
dtrain = lgb.Dataset(X_train, label=y_train)
dval = lgb.Dataset(X_val, label=y_val, reference=dtrain)
model = lgb.train(
lgb_params,
dtrain,
num_boost_round=1000,
valid_sets=[dval],
callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)]
)
y_pred = model.predict(X_val)
mape = mean_absolute_percentage_error(y_val, np.maximum(y_pred, 0))
mape_scores.append(mape)
print(f"Fold {fold+1} MAPE: {mape:.2%}")
print(f"\nMAPE medio: {np.mean(mape_scores):.2%} (+/-{np.std(mape_scores):.2%})")
# Addestramento finale su tutti i dati
dtrain_full = lgb.Dataset(X, label=y)
final_model = lgb.train(lgb_params, dtrain_full, num_boost_round=model.best_iteration)
return final_model, FEATURE_COLS
# ============================================================
# PREVISIONE MULTI-STEP (28 GIORNI)
# ============================================================
def forecast_next_28_days(model: lgb.Booster, history: pd.DataFrame,
sku_id: str, feature_cols: List[str]) -> pd.DataFrame:
"""
Genera previsioni per i prossimi 28 giorni per uno SKU specifico.
Usa previsioni iterative (ogni giorno usa i valori previsti precedenti).
"""
sku_history = history[history['sku_id'] == sku_id].copy()
last_date = sku_history['date'].max()
forecasts = []
for day in range(1, 29):
next_date = last_date + pd.Timedelta(days=day)
# Costruisce il vettore di feature per questa data
row = pd.DataFrame([{
'sku_id': sku_id,
'date': next_date,
# Usa le ultime previsioni come lag (previsione iterativa)
'lag_1': sku_history['quantity'].iloc[-1],
'lag_7': sku_history['quantity'].iloc[-7] if len(sku_history) >= 7 else 0,
'lag_14': sku_history['quantity'].iloc[-14] if len(sku_history) >= 14 else 0,
'lag_28': sku_history['quantity'].iloc[-28] if len(sku_history) >= 28 else 0,
'lag_56': sku_history['quantity'].iloc[-56] if len(sku_history) >= 56 else 0,
'rolling_mean_7': sku_history['quantity'].iloc[-7:].mean(),
'rolling_mean_14': sku_history['quantity'].iloc[-14:].mean(),
'rolling_mean_28': sku_history['quantity'].iloc[-28:].mean(),
'rolling_std_7': sku_history['quantity'].iloc[-7:].std(),
'rolling_std_14': sku_history['quantity'].iloc[-14:].std(),
'rolling_std_28': sku_history['quantity'].iloc[-28:].std(),
# Valori business (assume stabili nel breve periodo)
'price_ratio': 1.0,
'promotions': 0,
'supplier_lead_time': sku_history['supplier_lead_time'].iloc[-1],
'recent_stockout': 0
}])
# Aggiunge feature calendario
row = create_calendar_features(row, 'date')
# Previsione
X_pred = row[feature_cols]
quantity_pred = max(0, model.predict(X_pred)[0])
forecasts.append({
'date': next_date,
'sku_id': sku_id,
'forecast': round(quantity_pred, 1),
'lower_bound': round(quantity_pred * 0.85, 1), # 15% di incertezza
'upper_bound': round(quantity_pred * 1.15, 1)
})
# Aggiunge la previsione alla history per il passo successivo
new_row = sku_history.iloc[-1:].copy()
new_row['date'] = next_date
new_row['quantity'] = quantity_pred
sku_history = pd.concat([sku_history, new_row], ignore_index=True)
return pd.DataFrame(forecasts)
Ottimizzazione Inventario con Reinforcement Learning
La gestione dell'inventario e un problema di decisione sequenziale: ogni giorno bisogna decidere quante unita ordinare per ciascun SKU, bilanciando il costo di mantenimento delle scorte (capitale immobilizzato, spazio fisico, rischio obsolescenza) con il costo di stockout (vendite perse, penali contrattuali, danni alla reputazione). I modelli classici come il modello EOQ (Economic Order Quantity) e il punto di riordino fisso non catturano adeguatamente la domanda non stazionaria, le dipendenze tra SKU e le perturbazioni della supply chain.
Il Reinforcement Learning (RL) offre un approccio più potente: un agente impara una politica di riordino ottimale interagendo con una simulazione dell'ambiente. Ricerche recenti (2025) dimostrano che l'approccio basato su Proximal Policy Optimization (PPO) riduce i costi di riordino del 12,31% e abbatte gli stockout al 2,21%, superando significativamente i metodi tradizionali.
"""
Inventory Optimization con Reinforcement Learning (PPO)
Usando Gymnasium (ex OpenAI Gym) e Stable-Baselines3
"""
import gymnasium as gym
import numpy as np
from stable_baselines3 import PPO
from stable_baselines3.common.env_checker import check_env
from stable_baselines3.common.callbacks import EvalCallback
import pandas as pd
from typing import Optional
class InventoryEnv(gym.Env):
"""
Ambiente custom per ottimizzazione inventario.
Stato: [stock_corrente, domanda_media_7g, lead_time_atteso,
giorni_alla_scadenza (se deperibile), prezzo_corrente]
Azione: quantità da ordinare (discreta, 0-10 volte il MOQ)
Reward: -costo_holding - costo_stockout - costo_ordine
"""
metadata = {"render_modes": ["human"]}
def __init__(self, demand_data: np.ndarray, config: dict):
super().__init__()
self.demand_data = demand_data
self.n_steps = len(demand_data)
# Parametri del problema
self.holding_cost = config.get('holding_cost', 0.5) # Euro/unita/giorno
self.stockout_cost = config.get('stockout_cost', 5.0) # Euro/unita/mancante
self.order_cost = config.get('order_cost', 50.0) # Euro per ordine
self.lead_time = config.get('lead_time', 3) # Giorni di lead time
self.max_stock = config.get('max_stock', 1000) # capacità max
self.moq = config.get('moq', 10) # Minimum Order Quantity
# Spazio delle azioni: 0 (non ordinare) fino a 10 MOQ
self.action_space = gym.spaces.Discrete(11)
# Spazio degli stati: 5 variabili normalizzate
self.observation_space = gym.spaces.Box(
low=np.float32([0, 0, 0, 0, 0]),
high=np.float32([1, 1, 1, 1, 1]),
dtype=np.float32
)
self.reset()
def reset(self, seed: Optional[int] = None, options=None):
super().reset(seed=seed)
self.current_step = 0
self.stock = self.max_stock // 2 # Inizia a meta capacità
self.pending_orders = [] # (quantità, giorno_arrivo)
self.total_cost = 0.0
self.stockouts = 0
return self._get_observation(), {}
def _get_observation(self) -> np.ndarray:
"""Osservazione normalizzata dell'ambiente."""
demand_window = self.demand_data[
self.current_step:self.current_step + 7
]
avg_demand_7d = np.mean(demand_window) if len(demand_window) > 0 else 0
return np.float32([
self.stock / self.max_stock, # Stock attuale
avg_demand_7d / 100, # Domanda media 7 giorni
self.lead_time / 14, # Lead time normalizzato
len(self.pending_orders) / 5, # Ordini in transito
min(1.0, self.total_cost / 10000) # Costo accumulato (reward signal)
])
def step(self, action: int):
"""Esegue un passo: riceve ordini, soddisfa domanda, emette nuovi ordini."""
# 1. Ricevi ordini in arrivo
arrived = [qty for qty, arrive_day in self.pending_orders
if arrive_day <= self.current_step]
self.pending_orders = [(qty, day) for qty, day in self.pending_orders
if day > self.current_step]
for qty in arrived:
self.stock = min(self.max_stock, self.stock + qty)
# 2. Emetti nuovo ordine (azione)
order_qty = action * self.moq
order_cost = 0
if order_qty > 0:
order_cost = self.order_cost
arrive_day = self.current_step + self.lead_time
self.pending_orders.append((order_qty, arrive_day))
# 3. Soddisfa la domanda
demand = self.demand_data[min(self.current_step, self.n_steps - 1)]
if demand <= self.stock:
# Domanda soddisfatta
self.stock -= demand
stockout_cost = 0
else:
# Stockout parziale
unsatisfied = demand - self.stock
self.stock = 0
stockout_cost = unsatisfied * self.stockout_cost
self.stockouts += 1
# 4. Calcola costi
holding_cost = self.stock * self.holding_cost
step_cost = holding_cost + stockout_cost + order_cost
self.total_cost += step_cost
# Reward negativo (minimizziamo i costi)
reward = -step_cost / 100 # Scala il reward
# 5. Avanza
self.current_step += 1
terminated = self.current_step >= self.n_steps
return self._get_observation(), reward, terminated, False, {
"step_cost": step_cost,
"holding_cost": holding_cost,
"stockout_cost": stockout_cost,
"stock": self.stock,
"stockouts": self.stockouts
}
def train_inventory_agent(demand_data: np.ndarray, config: dict) -> PPO:
"""
Addestra un agente PPO per l'ottimizzazione dell'inventario.
PPO (Proximal Policy Optimization) e la scelta standard per:
- Ambienti con azioni discrete o continue
- Necessità di stabilità nell'addestramento
- Deployment in produzione
"""
# Crea l'ambiente
env = InventoryEnv(demand_data, config)
check_env(env, warn=True) # Valida la compatibilità Gymnasium
# Callback di valutazione: salva il modello migliore
eval_env = InventoryEnv(demand_data, config)
eval_callback = EvalCallback(
eval_env,
best_model_save_path="./inventory_agent/",
log_path="./inventory_logs/",
eval_freq=5000,
deterministic=True,
render=False
)
# Configura e addestra l'agente PPO
model = PPO(
"MlpPolicy",
env,
verbose=1,
learning_rate=3e-4,
n_steps=2048,
batch_size=64,
n_epochs=10,
gamma=0.99, # Fattore di sconto: 0.99 per problemi a lungo termine
gae_lambda=0.95,
clip_range=0.2, # Clip del ratio PPO
ent_coef=0.01, # Coefficiente di entropia per esplorazione
tensorboard_log="./inventory_tensorboard/"
)
model.learn(
total_timesteps=500_000,
callback=eval_callback,
progress_bar=True
)
return model
Warehouse Automation: AI nei Magazzini Moderni
L'automazione dei magazzini non riguarda solo i robot fisici. L'AI sta trasformando ogni aspetto delle operazioni di magazzino, dal posizionamento degli articoli sugli scaffali (slotting optimization) alla pianificazione dei percorsi di picking, dal controllo qualità automatizzato alla gestione dinamica del personale.
Le Tecnologie Chiave dell'Automazione
Stack Tecnologico del Magazzino Intelligente (2025)
| Livello | Tecnologia | Funzione | ROI Tipico |
|---|---|---|---|
| Fisica | AMR (Autonomous Mobile Robots) | Trasporto bin/scaffali verso operatori | 30-40% produttività picking |
| Fisica | Bracci robotici con computer vision | Pick-and-place, depalletizzazione | Operativita 24/7, -60% errori |
| Software | WMS con AI (Manhattan, Blue Yonder) | Orchestrazione operazioni, task interleaving | 15-25% throughput |
| Software | Slotting Optimization ML | Posiziona articoli ad alta rotazione vicino alle uscite | 20% riduzione distanza picking |
| Software | Computer Vision QC | Verifica dimensioni, danni, etichette | 99.5% accuratezza vs 96% umano |
| Dati | Digital Twin magazzino | Simulazione e ottimizzazione layout | Riduce tempi di ridisegno del 70% |
Pick Path Optimization con TSP
Un operatore di picking che deve raccogliere 20 articoli in un magazzino percorre in media 1.5-2.5 km per missione se segue un ordine non ottimizzato. Con la Traveling Salesman Problem (TSP) heuristic, il percorso si riduce del 20-30%, traducendosi in significativi risparmi di tempo e costi operativi.
"""
Pick Path Optimization per magazzino con layout a corridoi.
Algoritmo: S-shape + nearest neighbor heuristic
"""
from dataclasses import dataclass
from typing import List, Tuple, Dict
import math
@dataclass
class Location:
"""Posizione di uno slot nel magazzino."""
aisle: int # Numero corridoio (1-N)
bay: int # Posizione nel corridoio (1-M)
level: int # Piano (0=pavimento, 1=primo ripiano, ecc.)
@dataclass
class PickItem:
"""Articolo da prelevare."""
sku_id: str
location: Location
quantity: int
def manhattan_distance(loc1: Location, loc2: Location,
aisle_width: float = 3.0,
bay_depth: float = 1.2) -> float:
"""
Distanza di Manhattan tra due posizioni nel magazzino.
Considera la necessità di uscire e rientrare nei corridoi.
"""
# Se stesso corridoio: percorso diretto
if loc1.aisle == loc2.aisle:
return abs(loc1.bay - loc2.bay) * bay_depth
# Corridoi diversi: esce dal corridoio 1, percorre il main aisle, entra nel 2
aisle_distance = abs(loc1.aisle - loc2.aisle) * aisle_width
# Scegli l'uscita più vicina (testa o coda del corridoio)
max_bay = max(loc1.bay, loc2.bay)
exit_distance = min(loc1.bay, max_bay - loc1.bay + 1) * bay_depth
entry_distance = min(loc2.bay, max_bay - loc2.bay + 1) * bay_depth
return aisle_distance + exit_distance + entry_distance
def s_shape_routing(items: List[PickItem]) -> List[PickItem]:
"""
S-Shape routing: percorre i corridoi in senso alternato
(avanti-indietro) - ottimale per missioni con molti articoli.
"""
# Raggruppa per corridoio
by_aisle: Dict[int, List[PickItem]] = {}
for item in items:
aisle = item.location.aisle
if aisle not in by_aisle:
by_aisle[aisle] = []
by_aisle[aisle].append(item)
sorted_aisles = sorted(by_aisle.keys())
route = []
for i, aisle in enumerate(sorted_aisles):
aisle_items = sorted(by_aisle[aisle], key=lambda x: x.location.bay)
# Corridoi pari: percorri in avanti; dispari: in senso inverso
if i % 2 == 0:
route.extend(aisle_items)
else:
route.extend(reversed(aisle_items))
return route
def nearest_neighbor_routing(items: List[PickItem],
start: Location = None) -> List[PickItem]:
"""
Nearest Neighbor heuristic: scegli sempre l'articolo più vicino.
Ottimale per missioni con pochi articoli dispersi.
"""
if not items:
return []
if start is None:
start = Location(aisle=1, bay=1, level=0)
remaining = list(items)
route = []
current = start
while remaining:
# Trova l'articolo più vicino dalla posizione corrente
nearest = min(remaining,
key=lambda x: manhattan_distance(current, x.location))
route.append(nearest)
current = nearest.location
remaining.remove(nearest)
return route
def optimize_pick_mission(items: List[PickItem]) -> Tuple[List[PickItem], float]:
"""
Sceglie la strategia di routing migliore in base alla missione.
- Pochi item (<= 10): Nearest Neighbor
- Molti item (> 10): S-Shape
Returns:
(route_ottimizzato, distanza_totale_metri)
"""
if len(items) <= 10:
route = nearest_neighbor_routing(items)
else:
route = s_shape_routing(items)
# Calcola distanza totale
total_distance = 0.0
start = Location(aisle=1, bay=1, level=0) # Punto di partenza (ingresso)
current = start
for item in route:
total_distance += manhattan_distance(current, item.location)
current = item.location
# Ritorno al punto di deposito
total_distance += manhattan_distance(current, start)
return route, total_distance
# Esempio d'uso
if __name__ == "__main__":
# Missione di picking con 15 articoli sparsi nel magazzino
mission_items = [
PickItem("SKU-001", Location(2, 5, 0), 3),
PickItem("SKU-002", Location(5, 12, 1), 1),
PickItem("SKU-003", Location(1, 3, 0), 2),
PickItem("SKU-004", Location(7, 8, 0), 5),
PickItem("SKU-005", Location(3, 15, 1), 1),
PickItem("SKU-006", Location(4, 2, 0), 2),
PickItem("SKU-007", Location(6, 10, 0), 3),
PickItem("SKU-008", Location(2, 18, 1), 1),
PickItem("SKU-009", Location(8, 4, 0), 4),
PickItem("SKU-010", Location(1, 20, 0), 2),
PickItem("SKU-011", Location(9, 7, 1), 1),
PickItem("SKU-012", Location(3, 11, 0), 3),
PickItem("SKU-013", Location(5, 16, 0), 2),
PickItem("SKU-014", Location(7, 3, 1), 1),
PickItem("SKU-015", Location(6, 14, 0), 2),
]
route, distance = optimize_pick_mission(mission_items)
print(f"Missione ottimizzata: {len(route)} articoli")
print(f"Distanza totale: {distance:.1f} metri")
print("\nSequenza di picking:")
for i, item in enumerate(route, 1):
print(f" {i:2d}. {item.sku_id} - "
f"Corridoio {item.location.aisle}, "
f"Posizione {item.location.bay}, "
f"Livello {item.location.level} "
f"(qty: {item.quantity})")
Last-Mile Delivery: L'Ottimizzazione dell'Ultimo Miglio
L'ultimo miglio e la fase più costosa e complessa della supply chain: rappresenta il 28-40% del costo totale di consegna, eppure e quella più visibile al cliente finale. In contesti urbani italiani, la sfida e amplificata da ZTL, traffico, parcheggio difficile e la frammentazione delle destinazioni residenziali.
Le tecnologie AI stanno abilitando nuovi modelli di last-mile:
Tecnologie AI per il Last-Mile nel 2025
| Tecnologia | Stato | Riduzione Costi | Limitazioni |
|---|---|---|---|
| Route optimization AI | Matura, diffusa | 10-20% | Dipende da qualità dati |
| Dynamic re-routing | Matura | 5-10% | Integrazione con driver app |
| Droni (consegne aeree) | Pilota, limitata | Potenziale 40% | Normativa ENAC, payload, meteo |
| Robot da consegna | Sperimentale (IT) | Potenziale 60% | Infrastruttura, regolamentazione |
| Micro-fulfillment center | In crescita | 15-30% | Costi immobiliari urbani |
| Crowd-sourced delivery | Nicchia | Variabile | qualità del servizio |
Casi d'Uso Italiani: Come le Aziende IT Usano l'AI in Logistica
Il contesto italiano presenta sfide specifiche che rendono l'adozione dell'AI logistica tanto più necessaria quanto più complessa: infrastruttura stradale non uniforme, una forte presenza di PMI con volumi frammentati, stagionalita marcata (turismo, agricoltura, moda), e una cultura del "last-minute order" che mette sotto pressione i sistemi di planning.
Amazon Italia: L'Ecosistema di Automazione più Avanzato
Amazon ha investito massicciamente in Italia: i centri di distribuzione di Castel San Giovanni (PC), Vercelli, Passo Corese (RI), Castelguglielmo (RO) e gli hub di smistamento sono laboratori di innovazione logistica. Caratteristiche principali:
- Kiva/Sparrow robots: scaffalature mobili che si spostano verso gli operatori, eliminando quasi interamente il walking. La produttività di picking aumenta del 200-300%.
- Anticipatory shipping: algoritmi ML pre-posizionano gli articoli più probabilmente ordinati nella prossima settimana nel magazzino geograficamente più vicino ai clienti target.
- Amazon Delivery Service Partners (DSP): algoritmi di routing dinamico che si adattano in tempo reale al traffico, alle condizioni meteo e ai tentativi di consegna falliti.
- Computer vision per QC: telecamere AI verificano ogni pacco in uscita, rilevando danni e incongruenze con l'ordine in millisecondi.
Poste Italiane: Trasformazione Digitale di un Operatore Storico
Poste Italiane gestisce 60 milioni di recapiti all'anno con una rete di oltre 35.000 portalettere e 13.000 uffici postali. La trasformazione digitale della logistica di Poste ha tre assi principali:
- SDA Express Courier: sistema di routing basato su ML per l'ottimizzazione dei giri dei corrieri, integrato con la soluzione TomTom WEBFLEET per il tracking in tempo reale.
- Gestione dei picchi di domanda: algoritmi predittivi che anticipano i volumi di e-commerce durante il Black Friday e il periodo natalizio, permettendo il ridimensionamento proattivo del personale e dei mezzi.
- PostePay e logistica value-added: integrazione tra dati di pagamento e dati di spedizione per creare insight di domanda aggregata.
- Locker intelligenti: rete di Punto Poste con AI per ottimizzare la distribuzione geografica e prevedere i tassi di utilizzo.
GLS Italy: Route Intelligence per il B2B
GLS Group (con forte presenza in Italia) ha implementato una piattaforma di intelligenza logistica focalizzata sul segmento B2B, dove la puntualita e critica e i contratti includono SLA con penali. Le innovazioni chiave:
- Dynamic routing giornaliero: i percorsi non sono fissi ma vengono ricalcolati ogni notte in base al volume effettivo, con aggiustamenti intraday se un punto di raccolta ha volumi anomali.
- Previsione delivery success rate: modelli ML predicono la probabilità di successo della consegna per ogni indirizzo/giorno, permettendo di organizzare i tentativi in modo più efficiente.
- Integrazione ERP clienti: API che permettono ai clienti B2B di ricevere previsioni di consegna accurate 48h prima, migliorando la soddisfazione del cliente finale.
Real-Time Supply Chain Visibility e Digital Twin
La visibilità in tempo reale e il prerequisito per ogni forma di ottimizzazione AI. Senza sapere dove si trovano le merci, qual è lo stato degli ordini fornitori e quale e la capacità disponibile nei magazzini, qualsiasi modello predictivo opera nel buio.
La supply chain visibility moderna si costruisce su tre pilastri tecnologici:
Architettura della Supply Chain Visibility in Tempo Reale
| Livello | Tecnologia | Dati Raccolti | Latenza |
|---|---|---|---|
| Raccolta | IoT (GPS, RFID, sensori temperatura/umidita) | Posizione, condizioni ambientali | 1-30 secondi |
| Streaming | Apache Kafka + Flink | Event stream da tutti i touchpoint | < 1 secondo |
| Processing | ML anomaly detection | Deviazioni da ETA, alert proattivi | 1-5 secondi |
| Visualizzazione | Control tower (Databricks/Snowflake) | Dashboard operativa unificata | 5-30 secondi |
| Simulazione | Digital Twin | Replica virtuale della supply chain | Batch (notturno) |
Carbon Footprint Optimization
Con l'avvicinarsi delle scadenze della Direttiva CSRD (Corporate Sustainability Reporting Directive), la misurazione e riduzione delle emissioni logistiche e diventata una priorità business, non solo etica. Le aziende soggette alla CSRD devono rendicontare le emissioni Scope 3 (che includono la logistica) a partire dal 2025.
L'AI contribuisce in tre modi concreti alla riduzione del carbon footprint logistico:
- Consolidamento carichi: algoritmi ML massimizzano il Fill Rate dei veicoli, riducendo il numero di viaggi a vuoto (Empty Miles), che rappresentano mediamente il 20-25% del traffico merci in Italia.
- Modalità shift: ottimizzazione multi-modale che preferisce ferrovia e cabotaggio navale quando i tempi di consegna lo consentono.
- Eco-routing: calcolo di percorsi che minimizzano le emissioni CO2 invece della sola distanza, tenendo conto del profilo altimetrico e delle condizioni di traffico.
Best Practices e Anti-Pattern nell'AI Logistica
Anti-Pattern da Evitare
- Ottimizzare in silos: ottimizzare il routing senza considerare la disponibilità del magazzino, o viceversa, porta a soluzioni localmente ottime ma globalmente subottimali.
- Ignorare i constraint operativi reali: finestre temporali, orari di apertura dei clienti, restrizioni ZTL, peso assiale massimo dei veicoli. Un modello che non li conosce genera soluzioni inutilizzabili.
- Dati storici senza stagionalita corretta: addestrare un modello di demand forecasting su dati che includono periodi anomali (COVID, crisi chip, Ferragosto) senza adeguato preprocessing produce previsioni distorte.
- Mancanza di monitoring post-deployment: i pattern di domanda cambiano, le reti stradali cambiano, i clienti cambiano. Un modello non monitorato degrada silenziosamente.
- Big bang implementation: non sostituire tutti i processi logistici con AI in una sola volta. Inizia con un caso d'uso ad alto ROI, dimostra il valore, poi scala.
Best Practices per l'Implementazione AI in Logistica
- Data quality first: prima di addestrare qualsiasi modello, assicurati che i dati di posizione clienti, dimensioni veicoli, capacità magazzino e storico domanda siano puliti e consistenti.
- Hybrid approach: combina regole business (expertise dei pianificatori) con AI. I modelli puri ML spesso violano vincoli che il pianificatore umano rispetterebbe istintivamente.
- Explainability per i decision maker: i responsabili logistici devono capire perchè il sistema suggerisce un percorso o un riordino. Usa SHAP values e spiegazioni in linguaggio naturale.
- Fallback graceful: quando il modello e incerto (confidence bassa), torna a regole euristiche invece di emettere previsioni inaffidabili.
- ROI measurement rigoroso: definisci metriche baseline prima del go-live (costi per km, fill rate, OTIF, stockout rate) e misura il delta ogni trimestre.
Roadmap di Adozione AI in Logistica per le PMI
Per le PMI italiane che vogliono iniziare il percorso di adozione dell'AI logistica, suggeriamo una roadmap a tre fasi, con investimenti scalabili e ROI misurabili a ogni step:
Roadmap Triennale AI in Logistica
| Fase | Timeline | Iniziative | Investimento (EUR) | ROI Atteso |
|---|---|---|---|---|
| Foundation | Anno 1 | Data quality, WMS moderno, route optimization base, demand forecasting statistico | 50K - 200K | 15-25% |
| Intelligence | Anno 2 | ML demand forecasting, VRPTW avanzato, inventory optimization, real-time tracking | 150K - 500K | 25-40% |
| Automation | Anno 3 | AMR warehouse, autonomous planning, digital twin, carbon reporting AI | 300K - 2M | 40-60% |
Connessioni con Altri Articoli della Serie
- MLOps per Business: come portare in produzione i modelli di demand forecasting e routing con MLflow e CI/CD pipeline.
- LLM in Azienda: come usare i Large Language Model per creare control tower conversazionali e report automatici sulla supply chain.
- Vector Database Enterprise: come usare pgvector e Pinecone per ricerca semantica su documentazione fornitori e audit trail logistico.
- Data Governance: compliance CSRD per il reporting delle emissioni Scope 3 nella logistica.
Conclusioni
L'AI nella logistica non e più una sperimentazione da laboratorio: e una realta operativa che le aziende più competitive stanno già sfruttando per guadagnare vantaggi strutturali. Il Vehicle Routing Problem risolto con OR-Tools, il demand forecasting con LightGBM e TFT, l'ottimizzazione dell'inventario con Reinforcement Learning, l'automazione fisica dei magazzini con AMR e computer vision: ogni pezzo di questo puzzle contribuisce a una supply chain più efficiente, più sostenibile e più resiliente.
Per le PMI italiane, la buona notizia e che non e necessario affrontare tutto insieme. La roadmap a tre fasi presentata in questo articolo permette di partire con investimenti contenuti (50-200K EUR nel primo anno) e dimostrare ROI concreto prima di scalare. Il PNRR Transizione 5.0, con i suoi 12,7 miliardi di euro allocati (di cui solo 1,7 miliardi utilizzati a inizio 2026), offre incentivi fiscali significativi per gli investimenti in digitalizzazione e automazione: un'opportunità che le aziende logistiche italiane non possono permettersi di ignorare.
Nel prossimo articolo della serie esploriamo i LLM in Azienda: come costruire sistemi RAG enterprise per la documentazione interna, il fine-tuning su dati proprietari e i guardrails per garantire risposte sicure e conformi in contesti business critici.