Recherche
Rechercher dans le radar
3 résultats pour "Agent Failure"
Tutoriels
Intermediate · 2026-06-26
# Construire un agent de réapprovisionnement autonome gouverné pour SAP S/4HANA avec Claude et MCP
## Introduction
La gestion du réapprovisionnement dans les environnements ERP d'entreprise nécessite à la fois une automatisation intelligente et une gouvernance stricte. Ce guide démontre comment construire un agent autonome pour SAP S/4HANA qui utilise Claude via le Model Context Protocol (MCP) pour prendre des décisions de réapprovisionnement tout en maintenant des garde-fous d'entreprise.
## Architecture de l'agent
### Composants principaux
L'agent se compose de quatre couches :
1. **Couche MCP** : Expose les fonctions SAP et les données de réapprovisionnement à Claude
2. **Moteur de décision** : Claude analyse les niveaux de stock, la demande et les contraintes
3. **Cadre de gouvernance** : Valide les décisions selon les règles métier
4. **Couche d'exécution** : Exécute les commandes d'achat approuvées dans S/4HANA
### Flux de données
```
SAP S/4HANA → Serveur MCP → Claude → Validateur de gouvernance → SAP S/4HANA
```
## Mise en œuvre du serveur MCP
### Définir les outils de réapprovisionnement
Créez des outils MCP qui exposent les fonctions de réapprovisionnement critiques :
```python
from mcp.server import Server
from mcp.types import Tool, TextContent
import sapnwrfc
app = Server("sap-replenishment-mcp")
@app.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="get_inventory_status",
description="Récupère les niveaux de stock actuels, le stock de sécurité et le délai pour un SKU",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"plant": {"type": "string"}
},
"required": ["material_number", "plant"]
}
),
Tool(
name="get_demand_forecast",
description="Récupère les prévisions de demande pour un SKU sur un horizon temporel",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"plant": {"type": "string"},
"horizon_days": {"type": "integer"}
},
"required": ["material_number", "plant"]
}
),
Tool(
name="get_supplier_info",
description="Récupère les informations fournisseur incluant MOQ, prix et performance de livraison",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"vendor": {"type": "string"}
},
"required": ["material_number"]
}
),
Tool(
name="create_purchase_requisition",
description="Crée une demande d'achat pour approbation (n'exécute pas)",
inputSchema={
"type": "object",
"properties": {
"material_number": {"type": "string"},
"quantity": {"type": "number"},
"plant": {"type": "string"},
"vendor": {"type": "string"},
"delivery_date": {"type": "string"},
"justification": {"type": "string"}
},
"required": ["material_number", "quantity", "plant", "justification"]
}
)
]
```
### Implémenter les appels SAP
Connectez les outils MCP aux fonctions SAP via RFC :
```python
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
conn = sapnwrfc.Connection(
user=SAP_USER,
passwd=SAP_PASS,
ashost=SAP_HOST,
sysnr=SAP_SYSNR,
client=SAP_CLIENT
)
if name == "get_inventory_status":
result = conn.call('BAPI_MATERIAL_STOCK_REQ_LIST', {
'MATERIAL': arguments['material_number'],
'PLANT': arguments['plant']
})
inventory_data = {
'current_stock': result['STOCK_QUANTITY'],
'safety_stock': result['SAFETY_STOCK'],
'reorder_point': result['REORDER_POINT'],
'lead_time_days': result['LEAD_TIME'],
'in_transit': result['IN_TRANSIT_QTY']
}
return [TextContent(
type="text",
text=json.dumps(inventory_data, indent=2)
)]
elif name == "get_demand_forecast":
result = conn.call('BAPI_DEMAND_FORECAST_GET', {
'MATERIAL': arguments['material_number'],
'PLANT': arguments['plant'],
'HORIZON': arguments.get('horizon_days', 90)
})
return [TextContent(
type="text",
text=json.dumps(result['FORECAST_DATA'], indent=2)
)]
# Implémenter d'autres outils de manière similaire...
```
## Invites d'agent pour Claude
### Invite système
Définissez la personnalité et les contraintes de l'agent :
```python
SYSTEM_PROMPT = """Vous êtes un agent de réapprovisionnement autonome pour SAP S/4HANA. Votre objectif est de maintenir des niveaux de stock optimaux tout en minimisant les coûts et en prévenant les ruptures de stock.
Responsabilités :
- Surveiller les niveaux de stock à travers les usines
- Analyser les prévisions de demande et les tendances
- Recommander des quantités de réapprovisionnement en tenant compte de MOQ, délais et contraintes budgétaires
- Fournir une justification claire pour chaque décision
Contraintes obligatoires :
1. Ne jamais dépasser les limites d'approbation d'achat (50 000€ par réquisition)
2. Respecter les contrats fournisseurs et MOQ
3. Maintenir le stock de sécurité au-dessus des niveaux minimum
4. Tenir compte de la saisonnalité et des tendances de demande
5. Vérifier le stock en transit avant de commander
6. Expliquer toutes les décisions avec des données
Vous avez accès aux outils SAP via MCP pour interroger l'inventaire, les prévisions et les informations fournisseurs. Utilisez toujours des données pour soutenir vos recommandations.
"""
```
### Invite de tâche
Créez des invites structurées pour les tâches de réapprovisionnement :
```python
def create_replenishment_task(material_number: str, plant: str) -> str:
return f"""Analyser les besoins de réapprovisionnement pour :
- Numéro de matériel : {material_number}
- Usine : {plant}
Processus :
1. Récupérer le statut actuel de l'inventaire
2. Obtenir les prévisions de demande pour les 90 prochains jours
3. Vérifier le stock en transit
4. Examiner les informations et performance des fournisseurs
5. Calculer la quantité de commande optimale
6. Vérifier si le réapprovisionnement est nécessaire
Si le réapprovisionnement est nécessaire :
- Déterminer la quantité optimale (considérer MOQ, remises par paliers)
- Sélectionner le meilleur fournisseur
- Calculer la date de livraison requise
- Créer une demande d'achat avec justification détaillée
Fournir le raisonnement complet pour votre décision.
"""
```
## Cadre de gouvernance
### Validateur de règles
Implémentez des vérifications de validation avant l'exécution :
```python
class GovernanceValidator:
def __init__(self, config: dict):
self.max_purchase_value = config['max_purchase_value']
self.approved_vendors = config['approved_vendors']
self.required_approvers = config['required_approvers']
def validate_requisition(self, requisition: dict) -> tuple[bool, list[str]]:
"""Valide la demande d'achat selon les règles de gouvernance"""
errors = []
# Vérifier les limites de valeur
total_value = requisition['quantity'] * requisition['unit_price']
if total_value > self.max_purchase_value:
errors.append(
f"Valeur de réquisition {total_value}€ dépasse la limite {self.max_purchase_value}€. "
"Approbation de niveau supérieur requise."
)
# Vérifier les fournisseurs approuvés
if requisition['vendor'] not in self.approved_vendors:
errors.append(
f"Fournisseur {requisition['vendor']} non dans la liste approuvée. "
"Approbation d'achat requise."
)
# Vérifier le stock de sécurité
if requisition['resulting_stock'] < requisition['safety_stock']:
errors.append(
"La quantité commandée ne maintiendra pas le stock de sécurité. "
"Augmenter la quantité de commande."
)
# Vérifier le MOQ
if requisition['quantity'] < requisition['moq']:
errors.append(
f"Quantité {requisition['quantity']} inférieure au MOQ {requisition['moq']}. "
"Ajuster à la quantité minimum."
)
# Vérifier les contraintes budgétaires
budget_available = self.get_budget_available(
requisition['plant'],
requisition['cost_center']
)
if total_value > budget_available:
errors.append(
f"Fonds insuffisants. Requis : {total_value}€, Disponible : {budget_available}€"
)
return len(errors) == 0, errors
def get_budget_available(self, plant: str, cost_center: str) -> float:
"""Interroge SAP pour le budget disponible"""
# Implémenter l'appel SAP pour vérifier le budget
pass
```
### Flux d'approbation
Implémentez des approbations multi-niveaux basées sur la valeur :
```python
class ApprovalWorkflow:
def __init__(self):
self.approval_levels = {
'L1': {'max_value': 10000, 'approvers': ['procurement_specialist']},
'L2': {'max_value': 50000, 'approvers': ['procurement_manager']},
'L3': {'max_value': float('inf'), 'approvers': ['cfo', 'procurement_director']}
}
def get_required_approvers(self, requisition_value: float) -> list[str]:
"""Détermine les approbateurs requis en fonction de la valeur"""
for level, config in self.approval_levels.items():
if requisition_value <= config['max_value']:
return config['approvers']
return self.approval_levels['L3']['approvers']
def submit_for_approval(self, requisition: dict, approvers: list[str]):
"""Soumet la réquisition pour approbation humaine"""
approval_request = {
'requisition_id': requisition['id'],
'material': requisition['material_number'],
'quantity': requisition['quantity'],
'value': requisition['total_value'],
'vendor': requisition['vendor'],
'justification': requisition['justification'],
'agent_reasoning': requisition['agent_reasoning'],
'approvers': approvers,
'created_at': datetime.now().isoformat()
}
# Envoyer aux approbateurs via système de workflow
self.send_approval_notification(approval_request)
return approval_request['requisition_id']
```
## Orchestrateur de l'agent
### Boucle d'exécution principale
Orchestrez le cycle complet de décision de l'agent :
```python
from anthropic import Anthropic
class ReplenishmentAgent:
def __init__(self, mcp_server_url: str, governance_config: dict):
self.client = Anthropic()
self.mcp_url = mcp_server_url
self.validator = GovernanceValidator(governance_config)
self.workflow = ApprovalWorkflow()
async def analyze_and_replenish(self, material_number: str, plant: str):
"""Exécute l'analyse de réapprovisionnement et la génération de décision"""
# Créer la tâche pour Claude
task_prompt = create_replenishment_task(material_number, plant)
# Exécuter la conversation Claude avec accès MCP
response = self.client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=4000,
system=SYSTEM_PROMPT,
messages=[{
"role": "user",
"content": task_prompt
}],
tools=await self.get_mcp_tools()
)
# Traiter la décision de l'agent
decision = self.extract_decision(response)
if decision['action'] == 'replenish':
# Valider selon les règles de gouvernance
is_valid, errors = self.validator.validate_requisition(
decision['requisition']
)
if not is_valid:
self.log_governance_violation(decision, errors)
return {
'status': 'rejected',
'reason': errors,
'decision': decision
}
# Déterminer le flux d'approbation
approvers = self.workflow.get_required_approvers(
decision['requisition']['total_value']
)
# Soumettre pour approbation
approval_id = self.workflow.submit_for_approval(
decision['requisition'],
approvers
)
return {
'status': 'pending_approval',
'approval_id': approval_id,
'approvers': approvers,
'decision': decision
}
else:
return {
'status': 'no_action_needed',
'reasoning': decision['reasoning']
}
def extract_decision(self, response) -> dict:
"""Extrait la décision structurée de la réponse de Claude"""
# Parcourir les appels d'outils et analyser la réponse finale
# Extraire les détails de la réquisition si créée
pass
async def get_mcp_tools(self) -> list:
"""Récupère les définitions d'outils du serveur MCP"""
# Se connecter au serveur MCP et lister les outils disponibles
pass
```
## Capacités de surveillance et d'audit
### Journal des décisions
Enregistrez toutes les décisions de l'agent pour audit :
```python
class DecisionLogger:
def __init__(self, db_connection):
self.db = db_connection
def log_decision(self, decision: dict, context: dict):
"""Enregistre les décisions de l'agent avec le contexte complet"""
log_entry = {
'timestamp': datetime.now().isoformat(),
'material_number': context['material_number'],
'plant': context['plant'],
'decision_type': decision['action'],
'reasoning': decision['reasoning'],
'data_used': decision['data_sources'],
'requisition_details': decision.get('requisition'),
'governance_checks': decision['governance_validation'],
'agent_model': 'claude-3-7-sonnet-20250219',
'approval_status': decision.get('approval_status')
}
self.db.insert('agent_decisions', log_entry)
def get_decision_history(self, material_number: str, days: int = 90):
"""Récupère l'historique des décisions pour analyse"""
return self.db.query(
"""
SELECT * FROM agent_decisions
WHERE material_number = ?
AND timestamp > datetime('now', '-' || ? || ' days')
ORDER BY timestamp DESC
""",
(material_number, days)
)
```
### Tableaux de bord de performance
Suivez les KPI de l'agent :
```python
class AgentMetrics:
def __init__(self):
self.metrics = {
'decisions_made': 0,
'requisitions_created': 0,
'governance_violations': 0,
'approval_rate': 0.0,
'avg_decision_time': 0.0,
'stockout_prevention': 0,
'cost_savings': 0.0
}
def calculate_performance(self, time_period: str) -> dict:
"""Calcule les métriques de performance de l'agent"""
decisions = self.get_decisions(time_period)
return {
'total_decisions': len(decisions),
'auto_approved': len([d for d in decisions if d['auto_approved']]),
'human_reviewed': len([d for d in decisions if not d['auto_approved']]),
'rejection_rate': self.calculate_rejection_rate(decisions),
'avg_order_accuracy': self.calculate_order_accuracy(decisions),
'inventory_turns_improvement': self.calculate_turns_improvement(decisions),
'cost_per_decision': self.calculate_cost_efficiency(decisions)
}
```
## Déploiement en production
### Configuration de l'environnement
```yaml
# config.yaml
sap_connection:
host: "sap-prod.company.com"
client: "100"
system_number: "00"
mcp_server:
host: "localhost"
port: 8080
agent:
model: "claude-3-7-sonnet-20250219"
max_tokens: 4000
temperature: 0.0 # Déterministe pour les décisions de production
governance:
max_purchase_value: 50000
approved_vendors_file: "vendors.json"
budget_check_enabled: true
require_human_approval_over: 10000
monitoring:
log_level: "INFO"
metrics_export: true
audit_retention_days: 2555
```
### Considérations de sécurité
```python
class SecurityLayer:
def __init__(self):
self.encryption = AES256Encryption()
self.audit_log = AuditLogger()
def secure_sap_credentials(self, credentials: dict) -> dict:
"""Chiffre les identifiants SAP au repos"""
return {
'user': self.encryption.encrypt(credentials['user']),
'password': self.encryption.encrypt(credentials['password']),
'encrypted': True
}
def validate_tool_access(self, tool_name: str, user_context: dict) -> bool:
"""Valide que l'agent peut accéder à l'outil SAP"""
self.audit_log.log_access_attempt(tool_name, user_context)
# Vérifier les permissions basées sur les rôles
if tool_name == "create_purchase_requisition":
return user_context['role'] in ['procurement_agent', 'admin']
return True
```
## Gestion des cas limites
### Gestion des données manquantes
```python
def handle_missing_data(material_number: str, plant: str) -> dict:
"""Gestion gracieuse lorsque les données SAP sont incomplètes"""
try:
inventory = get_inventory_status(material_number, plant)
except DataNotFoundError:
# Revenir aux paramètres par défaut ou escalader
return {
'action': 'escalate',
'reason': 'Données d'inventaire manquantes pour décision',
'requires_human': True
}
if inventory['forecast_confidence'] < 0.6:
# Faible confiance - utiliser une approche conservatrice
return {
'action': 'conservative_replenish',
'quantity': inventory['safety_stock'] * 1.5,
'reason': 'Faible confiance de prévision - stock de sécurité augmenté'
}
return {'action': 'proceed'}
```
### Scénarios de récupération
```python
class RecoveryHandler:
def handle_sap_connection_failure(self, error: Exception):
"""Gérer les pannes de connexion SAP"""
self.logger.error(f"Échec de connexion SAP : {error}")
# Essayer le système de secours
if self.failover_available():
return self.connect_to_failover()
# Passer en mode lecture seule
self.switch_to_readonly_mode()
self.notify_operations_team()
def handle_governance_system_failure(self):
"""Lorsque le système de gouvernance est en panne"""
# Arrêter toutes les créations de réquisitions automatiques
self.pause_autonomous_actions()
# Passer à l'approbation manuelle uniquement
self.require_human_approval_all()
# Alerter l'équipe de gouvernance
self.send_alert("Système de gouvernance en panne - approbation manuelle requise")
```
## Conclusion
Cet agent de réapprovisionnement autonome démontre comment combiner les capacités de raisonnement de Claude avec les garde-fous de gouvernance d'entreprise via MCP. Les éléments clés incluent :
1. **Exposition structurée des données** via les outils MCP pour SAP
2. **Prise de décision guidée** par des invites système claires
3. **Validation multi-couches** via les cadres de gouvernance
4. **Supervision humaine** pour les décisions à enjeux élevés
5. **Traçabilité complète** via les journaux de décision et d'audit
Cette architecture peut s'étendre à d'autres processus ERP autonomes tout en maintenant le contrôle et la conformité d'entreprise.
Intermediate · 2026-04-21
# Construire un Agent AI Simulation-First pour Surveiller les Échecs de Livraison de la Supply Chain
Les échecs de livraison de la supply chain peuvent coûter des millions en revenus perdus, pénalités et érosion de la confiance client. Les tableaux de bord traditionnels vous indiquent *ce qui* a échoué, mais pas *pourquoi* ni *quoi faire ensuite*. Un agent AI simulation-first surveille en continu les performances de livraison, exécute des simulations de scénarios contrefactuels et recommande des actions correctives avant que les problèmes ne s'aggravent.
## Pourquoi Simulation-First pour la Surveillance des Échecs de Livraison
Les approches de surveillance standard détectent les écarts après coup. Une approche simulation-first modélise le système de livraison en tant qu'entités dynamiques interagissantes — transporteurs, entrepôts, commandes, capacité — et teste comment les changements d'une variable affectent les résultats. Cela permet une détection précoce des problèmes, une attribution des causes racines et une planification proactive de la récupération.
**Capacités clés :**
- **Détection d'anomalies en temps réel** : identifier les écarts par rapport aux taux de livraison attendus par transporteur, voie ou région
- **Attribution de cause racine** : simuler des scénarios contrefactuels pour isoler si les retards proviennent de la capacité transporteur, du débit entrepôt ou de pics de demande
- **Optimisation de récupération** : tester des routages alternatifs, des changements de transporteur ou des réallocations d'inventaire pour minimiser l'impact
- **Analyse prédictive** : prévoir les échecs de livraison probables avant qu'ils ne se produisent en fonction des contraintes de capacité et des tendances de demande
## Architecture de l'Agent AI
### 1. Couche d'Ingestion de Données
Connectez-vous aux sources de données de livraison :
- **Systèmes de gestion de transport (TMS)** : statuts d'expédition, données de tracking transporteur, heures de livraison promises vs réelles
- **Systèmes de gestion d'entrepôt (WMS)** : temps de traitement des commandes, niveaux d'inventaire, débit
- **Données de commande** : volumes de commande, priorités, SLA clients
- **Facteurs externes** : météo, congestion trafic, jours fériés, perturbations transporteur
Normalisez les données dans un schéma commun incluant : ID commande, transporteur, origine, destination, heure promise, heure réelle, statut, raison d'échec.
### 2. Moteur de Simulation
Construisez un modèle de simulation à événements discrets de votre réseau de livraison :
**Entités :**
- Commandes (avec priorité, destination, fenêtres de temps)
- Transporteurs (avec capacité, zones de service, taux de fiabilité)
- Entrepôts (avec capacité de traitement, inventaire)
- Véhicules/routes (avec contraintes de capacité)
**Processus :**
- Arrivée de commande → File d'attente → Traitement entrepôt → Attribution transporteur → Transit → Livraison
- Modélisez la variabilité : temps de traitement, durées de transit, taux d'échec transporteur
**Outils :**
- **SimPy** (Python) pour simulation à événements discrets
- **AnyLogic** pour modélisation multi-méthode (events discrets + dynamique)
- **Simulation personnalisée** utilisant des méthodes Monte Carlo pour la variabilité
### 3. Module de Détection d'Anomalies
Surveillez les métriques de livraison en temps réel vs. résultats de simulation :
**Métriques à tracker :**
- Taux de livraison à temps (OTDR) par transporteur, voie, région
- Temps de cycle moyen : commande → expédition → livraison
- Taux d'échec et catégories de raison (retardé, manqué, retourné)
- Utilisation de capacité transporteur
**Méthodes de détection :**
- **Contrôle de processus statistique** : détecter quand les métriques dépassent les limites de contrôle
- **ML basé sur anomalie** : isolation forest, autoencodeurs pour identifier les schémas inhabituels
- **Détection de dérive** : comparer la distribution actuelle vs. baseline de simulation
```python
import numpy as np
from sklearn.ensemble import IsolationForest
# Exemple : détecter les taux d'échec anormaux de transporteur
def detect_carrier_anomalies(delivery_data):
features = delivery_data[['avg_delay', 'failure_rate', 'capacity_utilization']]
model = IsolationForest(contamination=0.05, random_state=42)
anomalies = model.fit_predict(features)
delivery_data['anomaly'] = anomalies
flagged_carriers = delivery_data[delivery_data['anomaly'] == -1]
return flagged_carriers
```
### 4. Moteur d'Attribution de Cause Racine
Lorsqu'un échec est détecté, exécutez des simulations contrefactuelles pour identifier la cause :
**Approche :**
1. **Baseline** : simuler le scénario réel avec les paramètres actuels
2. **Contrefactuels** : modifier une variable à la fois et re-simuler
- Augmenter la capacité transporteur de 20%
- Réduire le temps de traitement entrepôt de 15%
- Rerouter les commandes vers un transporteur alternatif
- Augmenter les niveaux d'inventaire de 10%
3. **Comparer les résultats** : quelle modification élimine ou réduit le plus l'échec ?
```python
# Simulation contrefactuelle simplifiée
def run_counterfactual(scenario, parameter, adjustment):
baseline_otdr = simulate_deliveries(scenario)
# Modifier le paramètre
scenario[parameter] *= adjustment
counterfactual_otdr = simulate_deliveries(scenario)
impact = counterfactual_otdr - baseline_otdr
return impact
# Tester plusieurs hypothèses
causes = {
'carrier_capacity': run_counterfactual(scenario, 'carrier_capacity', 1.2),
'warehouse_throughput': run_counterfactual(scenario, 'processing_time', 0.85),
'demand_spike': run_counterfactual(scenario, 'order_volume', 0.9)
}
root_cause = max(causes, key=causes.get)
```
### 5. Optimiseur de Récupération
Une fois la cause racine identifiée, simulez des actions de récupération :
**Options de récupération :**
- **Changement de transporteur** : déplacer le volume vers des transporteurs avec capacité disponible
- **Rérouting** : utiliser des chemins alternatifs ou des hubs
- **Expédition accélérée** : surclasser les commandes critiques vers un service plus rapide
- **Réallocation d'inventaire** : expédier depuis des sites alternatifs plus proches de la destination
**Méthode d'optimisation :**
- Formuler comme un problème OR : minimiser le coût total (pénalités retard + coûts transport + coûts opérationnels)
- Contraintes : capacité transporteur, disponibilité inventaire, SLA clients
- Résoudre en utilisant des solveurs OR (OR-Tools, Gurobi) ou des méthodes heuristiques
```python
from ortools.linear_solver import pywraplp
def optimize_recovery(orders, carriers, warehouses):
solver = pywraplp.Solver.CreateSolver('SCIP')
# Variables : x[i,j,k] = 1 si commande i est assignée au transporteur j depuis l'entrepôt k
x = {}
for i in orders:
for j in carriers:
for k in warehouses:
x[i,j,k] = solver.BoolVar(f'x_{i}_{j}_{k}')
# Fonction objectif : minimiser le coût total (transport + pénalités)
objective = solver.Objective()
for i in orders:
for j in carriers:
for k in warehouses:
cost = carriers[j]['cost'] + warehouses[k]['distance'] * 0.1
if carriers[j]['estimated_delay'] > 0:
cost += orders[i]['delay_penalty']
objective.SetCoefficient(x[i,j,k], cost)
objective.SetMinimization()
# Contraintes : chaque commande doit être assignée une fois
for i in orders:
solver.Add(solver.Sum([x[i,j,k] for j in carriers for k in warehouses]) == 1)
# Contraintes de capacité
for j in carriers:
solver.Add(solver.Sum([x[i,j,k] for i in orders for k in warehouses]) <= carriers[j]['capacity'])
status = solver.Solve()
if status == pywraplp.Solver.OPTIMAL:
return extract_solution(x, orders, carriers, warehouses)
else:
return None
```
### 6. Module de Prédiction
Entraînez des modèles ML pour prédire les échecs de livraison probables :
**Features :**
- Historique de performance transporteur (taux de livraison à temps, temps de transit moyen)
- Conditions de commande (priorité, destination, taille)
- Facteurs contextuels (jour de la semaine, saison, météo)
- Métriques de capacité (utilisation transporteur, charge d'entrepôt)
**Modèles :**
- **Classification binaire** : livraison à temps vs. échec
- **Régression** : temps de livraison prédit
- **Analyse de survie** : probabilité d'échec au fil du temps
```python
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
# Entraîner le prédicteur d'échec de livraison
def train_failure_predictor(historical_data):
features = ['carrier_id', 'origin', 'destination', 'order_priority',
'carrier_utilization', 'day_of_week', 'weather_severity']
target = 'delivery_failed'
X = historical_data[features]
y = historical_data[target]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
model = GradientBoostingClassifier(n_estimators=100, max_depth=5)
model.fit(X_train, y_train)
accuracy = model.score(X_test, y_test)
print(f"Précision modèle : {accuracy:.2f}")
return model
# Prédire les échecs pour les commandes entrantes
def predict_failures(model, incoming_orders):
predictions = model.predict_proba(incoming_orders)[:, 1]
incoming_orders['failure_probability'] = predictions
high_risk = incoming_orders[incoming_orders['failure_probability'] > 0.3]
return high_risk
```
### 7. Interface Agent et Actions
L'agent surveille en continu et agit :
**Workflow d'agent :**
1. **Surveiller** : ingérer des données de livraison en temps réel
2. **Détecter** : exécuter la détection d'anomalies sur les métriques clés
3. **Analyser** : déclencher des simulations contrefactuelles pour l'attribution de cause racine
4. **Optimiser** : générer des plans de récupération en utilisant l'optimisation OR
5. **Prédire** : signaler les commandes à haut risque pour intervention proactive
6. **Agir** : recommander ou exécuter automatiquement des actions (avec approbation humaine pour les décisions coûteuses)
**Exemples d'actions :**
- Alerte : notifier la supply chain manager de l'augmentation des échecs transporteur
- Recommander : suggérer le changement de transporteur pour les commandes à haut risque
- Automatiser : réassigner automatiquement les commandes si l'impact projeté < seuil
- Escalader : alerter les gestionnaires si le plan de récupération dépasse le budget
```python
class DeliveryMonitorAgent:
def __init__(self, simulation_engine, anomaly_detector, optimizer):
self.simulation = simulation_engine
self.detector = anomaly_detector
self.optimizer = optimizer
self.alert_threshold = 0.15 # Alerte si OTDR baisse > 15%
def monitor(self, delivery_data):
# Détecter les anomalies
anomalies = self.detector.detect(delivery_data)
if not anomalies.empty:
for _, anomaly in anomalies.iterrows():
# Exécuter l'analyse de cause racine
root_cause = self.diagnose(anomaly)
# Générer le plan de récupération
recovery_plan = self.optimizer.optimize(
orders=anomaly['affected_orders'],
root_cause=root_cause
)
# Décider de l'action
if recovery_plan['impact'] > self.alert_threshold:
self.alert_manager(anomaly, root_cause, recovery_plan)
else:
self.execute_recovery(recovery_plan)
def diagnose(self, anomaly):
# Exécuter des simulations contrefactuelles
scenarios = {
'carrier_capacity': self.simulation.run({'carrier_capacity': 1.2}),
'warehouse_delay': self.simulation.run({'processing_time': 0.85}),
'demand_spike': self.simulation.run({'order_volume': 0.9})
}
root_cause = max(scenarios, key=lambda k: scenarios[k]['improvement'])
return root_cause
def alert_manager(self, anomaly, root_cause, recovery_plan):
message = f"""
Échec de livraison détecté :
- Transporteur : {anomaly['carrier']}
- OTDR impact : -{anomaly['otdr_drop']*100:.1f}%
- Cause racine : {root_cause}
- Plan de récupération : {recovery_plan['description']}
- Impact coût estimé : ${recovery_plan['cost']:,.0f}
"""
send_alert(message, priority='high')
def execute_recovery(self, recovery_plan):
# Exécuter automatiquement pour les problèmes à faible impact
execute_plan(recovery_plan)
log_action(recovery_plan)
```
## Métriques d'Implémentation et de Validation
**Démarrer petit :**
1. **Pilote** : sélectionner une région ou un transporteur avec des données historiques d'échec élevées
2. **Construire baseline** : exécuter la simulation sur 3-6 mois de données historiques
3. **Calibrer** : ajuster les paramètres de simulation pour correspondre aux résultats réels
4. **Valider** : comparer les prédictions d'agent avec les échecs réels sur des données non vues
**Métriques de succès :**
- **Taux de détection** : % d'échecs réels détectés par l'agent
- **Faux positifs** : % d'alertes qui n'ont pas conduit à de véritables échecs
- **Temps de réponse** : temps moyen de détection jusqu'à la récupération
- **Amélioration OTDR** : augmentation du taux de livraison à temps après intervention de l'agent
- **Économies de coûts** : réduction des pénalités, expéditions accélérées, escalades clients
**Benchmarks cibles :**
- Détection 80%+ des échecs avant livraison manquée
- Réduction de 30-50% des pénalités de livraison tardive
- Réduction de 20%+ des coûts d'expédition accélérée réactive
- ROI 3-6 mois via la réduction des pénalités et l'amélioration de la satisfaction client
## Pièges Courants
**Qualité des données :** les données de livraison sont souvent fragmentées entre TMS, WMS, portails transporteur. Investir dans l'intégration de données et la normalisation d'abord.
**Complexité de simulation :** commencer simple. Un modèle basique de file d'attente avec variabilité bat les tableaux de bord statiques. Ajouter de la complexité de manière itérative.
**Sur-automatisation :** ne pas automatiser les décisions coûteuses sans approbation humaine. Utiliser l'agent pour recommander ; les humains décident des actions majeures.
**Ignorer les facteurs externes :** la météo, les jours fériés, les grèves transporteur ont un impact majeur sur les livraisons. Incorporer des sources de données externes.
**Fatigue des alertes :** trop d'alertes conduisent à l'ignorance. Calibrer les seuils pour signaler seulement les anomalies exploitables.
## Résultat Pratique
Un agent AI simulation-first transforme la surveillance de livraison de réactive à proactive. Vous détectez les problèmes plus tôt, comprenez leurs causes racines et exécutez des récupérations optimisées — souvent avant que les clients ne remarquent. Cela nécessite un investissement dans l'intégration de données et la construction de simulation, mais les gains sont mesurables : moins de pénalités, des clients plus satisfaits, et une équipe supply chain qui résout les problèmes au lieu de les combattre.
Commencez par une région pilote, prouvez la valeur, puis étendez à votre réseau. Les gains de fiabilité de livraison paient eux-mêmes l'investissement.