物流における AI: ルートの最適化、倉庫の自動化、サプライ チェーン インテリジェンス
物流とサプライ チェーンは常に複雑さによって支配されてきました。 相互依存変数、狭い時間枠、利益を侵食する運用コスト、 そしてサービスの継続性を危険にさらす需要の予測不可能性。何十年もの間、 企業は、Excel シート、経験則、経験によってこの複雑さを抑えようとしてきました。 先輩プランナーの方々。今日、人工知能はゲームのルールを再設計しています。
の世界市場AIをサプライチェーンに適用 98億に達しました 2025 年にはドルが増加し、2030 年までに最大 320 億ドルの成長が見込まれる (CAGR 26.4%)。これは誇大広告ではありません：物流プロセスに AI を導入した企業 輸送コストの 10 ～ 15% の削減、納期の 15 ～ 20% の短縮を記録 配達が削減され、配達遅延が約 30% 減少します。 Amazonは52万台以上のロボットを運用している 倉庫で AI を活用し、フルフィルメント コストを 20% 削減し、商品の 40% を処理 今のところ追加注文。
この記事では、物流を変革する AI テクノロジーについて探ります。 配車ルートの問題 (VRP) ORツールと強化学習で解決し、 Temporal Fusion Transformerによる需要予測から倉庫自動化まで、 ラストマイルの最適化とインテリジェントな在庫管理まで。見てみましょう 動作する Python コードを使用した具体的な実装と、イタリアのコンテキストからの実際の使用例。
この記事で学べること
- Python の Google OR ツールを使用して配車経路問題 (VRP) を解決する方法
- Prophet、LightGBM、Temporal Fusion Transformer による需要予測
- 強化学習による在庫最適化 (PPO/DQN)
- 倉庫自動化: ロボティクス、ピッキング最適化、インテリジェント WMS システム
- ラストワンマイルの配送: 都市部における AI、ドローン、自動運転車
- サプライチェーンのリアルタイムの可視性とデジタルツイン
- 二酸化炭素排出量の最適化と持続可能な物流
- イタリアのユースケース: Amazon IT、Poste Italiane、GLS
データウェアハウス、AI、デジタルトランスフォーメーションシリーズにおける地位
| # | アイテム | Stato |
|---|---|---|
| 1 | データウェアハウスの進化 | 発行済み |
| 2 | データメッシュと分散型アーキテクチャ | 発行済み |
| 3 | ETL 対モダン ELT: dbt、Airbyte、Fivetran | 発行済み |
| 4 | パイプライン オーケストレーション: Airflow、Dagster、Prefect | 発行済み |
| 5 | 製造における AI: 予知保全 | 発行済み |
| 6 | 金融における AI: 不正行為の検出と信用スコアリング | 発行済み |
| 7 | 小売における AI: 需要予測と推奨 | 発行済み |
| 8 | ヘルスケアにおける AI: 診断と創薬 | 発行済み |
| 9 | 物流における AI (ここにいます) | 現在 |
| 10 | ビジネスにおける LLM: RAG Enterprise とガードレール | Prossimo |
車両ルートの問題: OR ツールを使用したルートの最適化
Il 配車ルートの問題 (VRP) 最も研究されている研究課題の 1 つ 運用: 特定の配送リクエストを持つ一連の顧客と車両群を想定 1 つ以上の拠点から出発して、顧客を車両に割り当て、ルートを計画する方法 総コスト（距離、時間、燃料）を最小限に抑えるには？
VRP は NP 困難です: 多項式時間で正確に解くアルゴリズムはありません 大きなインスタンス。このため、実際の解決策では次のことを組み合わせて使用します。 の メタヒューリスティックス (シミュレーテッドアニーリング、タブーサーチ、遺伝的アルゴリズム) 商用およびオープンソースのソルバー。 Google OR-Tools と今日のツール このタイプの問題で最もよく使用されるオープンソース: CVRP (容量付き VRP) をサポートします。 VRPTW (タイム ウィンドウ付き)、Multi-Depot VRP、および多くの現実的なバリアント。
UPS の ORION システムは、同様の技術に基づいて 30,000 のルート最適化を計算します 毎分3,800万リットルの燃料を節約し、 1億マイルの不必要な運転。それはわずかな優位性ではなく、競争上の優位性です 構造的には年間数千万ドルの節約につながります。
Google OR-Tools を使用した CVRP の実装
時間ウィンドウを備えた容量付き VRP (VRPTW) の完全な実装を見てみましょう。 顧客ごとに特定の営業時間が定められている実際の物流現場で最も一般的なタイプです。
"""
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")
製造現場では、移動時間マトリックスがリアルタイムで計算されます。 現在の交通状況を考慮して、Google Maps Distance Matrix または HERE Routing API を使用します。 顧客データは会社の ERP から取得され、1 時間ごとに更新されます。 OR ツール 最大 200 ～ 300 人の顧客のインスタンスに対して、数秒でソリューションを返します。たとえば 大規模なクラスター アプローチや、NVIDIA cuOpt などの GPU 高速化ソルバーが使用されます。
需要予測: ML を使用した需要予測
正確な需要予測はサプライチェーン全体の基礎です。知らずに 今後数週間でどれくらいの商品がリクエストされるか、最適化することは不可能です 購入、倉庫のサイズ設定、輸送の計画、レベルの保証 サービスの。何十年もの間、企業は ARIMA のような古典的な統計モデルを使用してきました。 SARIMA と指数平滑法。現在、機械学習モデルは一貫して優れたパフォーマンスを発揮します これらのベースライン。
2025 年における最も興味深い比較は、次の 3 つの異なるアプローチ間の比較です。
需要予測モデルの比較
| モデル | タイプ | 強み | 制限事項 | 代表的なMAPE |
|---|---|---|---|---|
| 預言者(メタ) | ベイジアン加法 | 複数の季節、休日、トレンドを管理 | 数千の SKU に簡単に拡張できない | 8-12% |
| ライトGBM | 勾配ブースティング | 高速かつ柔軟な機能エンジニアリング、生産 | 手動の特徴量エンジニアリングが必要 | 5～9% |
| 時間融合トランスフォーマー | ディープラーニング | マルチホライズン、解釈可能な外生変数 | トレーニングに時間がかかる、GPU が必要 | 4～7% |
| サリマ (ベースライン) | 統計的 | シンプルで解釈しやすい | 非線形性を捉えない | 12-20% |
LightGBM を使用したサプライチェーン向けの需要予測
LightGBM は多くの場合、実稼働デプロイメントに最適な選択肢です。つまり、高速トレーニング、 ミリ秒推論、欠損値のネイティブ サポート、優れたスケーラビリティ 数千のSKU。ここでは、物流固有の特徴エンジニアリングを使用した完全な実装を示します。
"""
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)
強化学習による在庫の最適化
在庫管理は逐次的な意思決定の問題です。毎日決定を下さなければなりません 在庫維持コストのバランスを考慮して、SKU ごとに注文するユニット数 (固定資本、物理的スペース、陳腐化リスク) と在庫切れコスト （売上の損失、契約上の罰金、風評被害）。のようなクラシックなモデル EOQ (経済的注文数量) モデル そして 再注文ポイントを修正 they do not adequately capture non-stationary demand, SKU dependencies, and disruptions サプライチェーンの。
Il 強化学習 (RL) より強力なアプローチを提供します: エージェント 環境のシミュレーションと対話することで、最適な並べ替えポリシーを学習します。 最近の研究 (2025 年) では、次のようなアプローチが有効であることが示されています。 近接政策 最適化（PPO） 再注文コストが 12.31% 削減され、在庫切れが減少します 2.21% で、従来の方法を大幅に上回っています。
"""
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
倉庫自動化: 最新の倉庫における AI
倉庫の自動化は物理的なロボットだけではありません。 AIは変革しつつある 商品を棚に置くことから倉庫業務のあらゆる側面まで （スロットの最適化）からピッキングルートの計画、品質管理まで 人員の動的な管理を自動化します。
自動化の主要テクノロジー
インテリジェント ウェアハウス テクノロジー スタック (2025)
| レベル | テクノロジー | 関数 | 一般的な ROI |
|---|---|---|---|
| 物理 | AMR（自律移動ロボット） | オペレーターへのビン/棚の輸送 | ピッキングの生産性が 30 ～ 40% 向上 |
| 物理 | コンピュータービジョンを備えたロボットアーム | ピックアンドプレイス、デパレタイズ | 24 時間年中無休の稼働、エラー 60% 未満 |
| ソフトウェア | AIを搭載したWMS（マンハッタン、ブルーヨンダー） | 運用オーケストレーション、タスクインターリーブ | 15～25%のスループット |
| ソフトウェア | スロッティング最適化ML | 回転率の高い商品を出口付近に配置する | ピッキング距離が 20% 短縮 |
| ソフトウェア | コンピュータビジョンQC | 寸法、ダメージ、ラベルを確認してください | 99.5% の精度 vs 96% 人間による精度 |
| データ | デジタルツイン倉庫 | レイアウトのシミュレーションと最適化 | 再描画時間を 70% 削減します |
TSP によるピック パスの最適化
倉庫内で 20 個のアイテムを収集する必要があるピッキング オペレーターは、平均で移動します。 最適化されていない命令に従った場合は、ミッションごとに 1.5 ～ 2.5 km。と 巡回セールスマン 問題 (TSP) ヒューリスティック、ルートは 20 ～ 30% 削減され、換算すると 時間と運用コストが大幅に節約されます。
"""
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})")
ラストマイル配送: ラストマイルの最適化
ラスト マイルは、サプライ チェーンの中で最も高価で複雑なフェーズです。これは 28 ～ 40% を占めます。 配送コストの合計のうち、最終顧客にとって最も目に見えるものです。コンテキスト内で イタリアの都市部では、ZTL、交通渋滞、困難な駐車、および 居住地の細分化。
AI テクノロジーにより、次のような新しいラストマイル モデルが可能になります。
2025 年のラストマイルに向けた AI テクノロジー
| テクノロジー | Stato | コスト削減 | 制限事項 |
|---|---|---|---|
| ルート最適化AI | 熟して広く普及 | 10-20% | データの品質に依存します |
| 動的再ルーティング | 成熟した | 5～10% | アプリドライバーとの統合 |
| ドローン（空からの配達） | パイロット、限定 | ポテンシャル 40% | ENAC 規制、積載量、天候 |
| 配送ロボット | 実験的 (IT) | ポテンシャル 60% | インフラ、規制 |
| マイクロフルフィルメントセンター | 成長する | 15-30% | 都市部の不動産コスト |
| クラウドソーシングによる配信 | ニッチ | 変数 | サービスの質 |
イタリアのユースケース: IT 企業が物流分野で AI をどのように活用しているか
イタリアの状況では、AI の導入を物流上で困難にする特有の課題が提示されています。 必要性が高まるほど複雑になる: 不均一な道路インフラ、強力なインフラ 規模が細分化され、季節性が顕著な中小企業の存在（観光、農業、ファッション）、 そして、計画システムに圧力をかける「直前注文」文化。
Amazon Italy: 最先端の自動化エコシステム
アマゾンがイタリアに巨額投資：カステル・サン・ジョバンニ配送センター (PC)、Vercelli、Passo Corese (RI)、Castelguglielmo (RO) および選別ハブは次のとおりです。 物流イノベーション研究所。主な特徴:
- キバ/スパローロボット: オペレータに向かって移動する可動棚で、歩行をほぼ完全に排除します。ピッキングの生産性が 200 ～ 300% 向上します。
- 予定出荷: ML アルゴリズムは、来週に注文される可能性が最も高い商品を、対象顧客に地理的に最も近い倉庫に事前に配置します。
- Amazon デリバリー サービス パートナー (DSP): トラフィック、気象条件、配送試行の失敗にリアルタイムで適応する動的ルーティング アルゴリズム。
- QC用のコンピュータビジョン: AI カメラがすべての発送荷物をチェックし、損傷や注文との不一致をミリ秒単位で検出します。
Poste Italiane: 歴史ある通信事業者のデジタル変革
Poste Italiane manages 60 million deliveries per year with a network of over 35,000 postmen そして13,000の郵便局。 Poste の物流のデジタル変革には、次の 3 つの主な軸があります。
- SDAエクスプレス宅配便: 宅配便のルートを最適化するための ML ベースのルーティング システム。リアルタイム追跡のために TomTom WEBFLEET ソリューションと統合されています。
- ピーク需要の管理: ブラック フライデーとクリスマス期間中の電子商取引の量を予測する予測アルゴリズムにより、スタッフと車両の積極的な削減が可能になります。
- PostePay と付加価値物流: 支払いデータと配送データを統合して、総需要の洞察を作成します。
- スマートロッカー: AI を備えた Punto Poste ネットワークにより、地理的分布を最適化し、使用率を予測します。
GLS Italy: B2B 向けのルート インテリジェンス
GLS グループ (イタリアに強い存在感) はインテリジェンス プラットフォームを導入しました 時間厳守が重要で契約が重要なB2Bセグメントに焦点を当てた物流 罰則付きのALSも含める。主な革新:
- 動的な日次ルーティング: ルートは固定されていませんが、実際の量に基づいて毎晩再計算され、収集ポイントに異常な量がある場合は日中に調整されます。
- 配信成功率予測: ML モデルは、住所/日ごとに配達の成功確率を予測するため、配達の試みをより効率的に計画できます。
- 顧客のERP統合: B2B 顧客が 48 時間前に正確な配送予測を受け取ることができる API により、最終顧客の満足度が向上します。
リアルタイムのサプライチェーンの可視化とデジタルツイン
リアルタイムの可視性は、あらゆる形式の AI 最適化の前提条件です。知らずに 商品はどこにあるのか、サプライヤーの注文状況はどうなっているか、利用可能な容量はどれくらいか 倉庫では、あらゆる予測モデルは暗闇の中で動作します。
La サプライチェーンの可視性 現代性は 3 つの技術の柱に基づいて構築されています。
サプライチェーンアーキテクチャのリアルタイム可視化
| レベル | テクノロジー | 収集されたデータ | レイテンシ |
|---|---|---|---|
| コレクション | IoT（GPS、RFID、温湿度センサー） | 場所、環境条件 | 1～30秒 |
| ストリーミング | Apache Kafka + フリンク | すべてのタッチポイントからのイベント ストリーム | < 1 秒 |
| 処理 | ML 異常検出 | ETAからの逸脱、プロアクティブなアラート | 1～5秒 |
| 視覚化 | コントロールタワー (Databricks/Snowflake) | 統合された運用ダッシュボード | 5～30秒 |
| シミュレーション | デジタルツイン | サプライチェーンの仮想レプリケーション | バッチ (毎晩) |
二酸化炭素排出量の最適化
締め切りが近づくにつれて CSRD指令（企業の持続可能性） 報告指令）、物流排出量の測定と削減 は単なる倫理的な問題ではなく、ビジネス上の優先事項となっています。 CSRDの対象となる企業は、 2025 年から始まるスコープ 3 排出量 (物流を含む) を報告します。
AI は、物流の二酸化炭素排出量削減に 3 つの具体的な方法で貢献します。
- 負荷の統合: ML アルゴリズムは車両の充填率を最大化し、イタリアの貨物輸送量の平均 20 ～ 25% を占める空旅行 (空マイル) の数を削減します。
- シフトモード: 納期が許せば鉄道と海軍のカボタージュを優先するマルチモーダルな最適化。
- エコルーティング: 高度プロファイルと交通状況を考慮して、距離だけではなく CO2 排出量を最小限に抑えるルートを計算します。
物流 AI のベスト プラクティスとアンチパターン
避けるべきアンチパターン
- サイロで最適化する: 倉庫の可用性を考慮せずにルーティングを最適化する、またはその逆は、局所的には最適でも、全体的には次善のソリューションにつながります。
- 実際の運用上の制約を無視する: 時間枠、顧客の営業時間、ZTL 制限、車両の最大軸重量。それらを知らないモデルは、使用できないソリューションを生成します。
- 季節性を修正していない過去のデータ: 適切な前処理を行わずに、異常な期間 (新型コロナウイルス、チップ危機、8 月 15 日) を含むデータで需要予測モデルをトレーニングすると、歪んだ予測が生成されます。
- 導入後のモニタリングの欠如: 需要パターンが変化し、道路網が変化し、顧客が変化します。監視されていないモデルは、静かに機能を低下させます。
- ビッグバン実装：すべての物流プロセスを一度にAIに置き換えないでください。 ROI の高いユースケースから始めて、価値を実証してから拡張します。
物流における AI 導入のベスト プラクティス
- データ品質第一: モデルをトレーニングする前に、顧客の所在地、車両のサイズ、倉庫の容量、需要履歴データがクリーンで一貫性があることを確認してください。
- ハイブリッドアプローチ: ビジネス ルール (プランナーの専門知識) と AI を組み合わせます。純粋な ML モデルは、人間の計画者が本能的に尊重するであろう制約に違反することがよくあります。
- 意思決定者にとっての説明可能性: 物流管理者は、システムがルートや再注文を提案する理由を理解する必要があります。 SHAP値と自然言語による説明を使用します。
- グレースフルフォールバック: モデルが不確実な場合 (信頼性が低い場合)、信頼性の低い予測を発行するのではなく、ヒューリスティック ルールに戻ります。
- 厳密なROI測定: 稼働前にベースライン指標 (km あたりのコスト、充填率、OTIF、在庫切れ率) を定義し、四半期ごとにデルタを測定します。
中小企業向け物流におけるAI導入ロードマップ
物流 AI の導入プロセスを開始したいイタリアの中小企業の場合、 we suggest a three-phase roadmap, with scalable investments and measurable ROI at each step:
物流における AI の 3 年間のロードマップ
| 段階 | タイムライン | 取り組み | 投資額（ユーロ） | 期待されるROI |
|---|---|---|---|---|
| 財団 | 1年目 | データ品質、最新の WMS、基本的なルート最適化、統計的需要予測 | 50K～200K | 15-25% |
| 知能 | 2年目 | ML 需要予測、高度な VRPTW、在庫最適化、リアルタイム追跡 | 150K～500K | 25～40% |
| オートメーション | 3年目 | AMR ウェアハウス、自律計画、デジタルツイン、炭素報告 AI | 300K～2M | 40-60% |
シリーズの他の記事とのつながり
- ビジネス向け MLOps: MLflow と CI/CD パイプラインを使用して、需要予測とルーティング モデルを運用環境に導入する方法。
- ビジネスにおけるLLM: 大規模言語モデルを使用して会話型の管制塔と自動サプライ チェーン レポートを作成する方法。
- ベクトル データベース エンタープライズ: サプライヤー文書および物流監査証跡のセマンティック検索に pgvector と Pinecone を使用する方法。
- データガバナンス: 物流におけるスコープ 3 排出量の報告に関する CSRD 準拠。
結論
物流における AI はもはや実験室実験ではなく、実際に運用されるものです 最も競争力の高い企業はすでにこれを利用して構造上の優位性を獲得しています。 OR-Tools で解決された配車ルートの問題、LightGBM と TFT による需要予測、 強化学習による在庫の最適化、物理的な自動化 AMR とコンピューター ビジョンを備えた倉庫: このパズルの各ピースは供給に貢献します より効率的で、より持続可能で、より弾力性のあるチェーン。
イタリアの中小企業にとって良いニュースは、すべてを一緒に取り組む必要がないことです。 この記事で紹介する 3 段階のロードマップにより、投資を始めることができます。 コンテンツ (初年度は 50 ～ 200,000 ユーロ) を提供し、拡張する前に具体的な ROI を実証します。の PNRR Transition 5.0、127億ユーロが割り当てられる（そのうちわずか17億ユーロ） 2026 年の初めに使用）、への投資に対して大幅な税制上の優遇措置を提供します。 デジタル化と自動化: イタリアの物流会社にはないチャンス 彼らは無視する余裕があります。
シリーズの次の記事では、 ビジネスにおけるLLM: 構築方法 内部文書化、独自データの微調整用の RAG エンタープライズ システム 重要なビジネスの場面で安全かつコンプライアンスに準拠した対応を確保するためのガードレール。