Survivre à une forte incertitude dans la logistique avec MARL

fait partie de la série sur l’optimisation des plannings en logistique avec l’apprentissage par renforcement multi-agents (MARL). Ici, je me concentre davantage sur la manière dont la généralisation a été réalisée. Je recommande la lecture Partie 1 d’abord si vous souhaitez avoir une idée du contexte architectural et commercial.

L’objectif était que le modèle généralise les processus à mi-kilomètre et survive même dans des conditions changeantes. J’ai réalisé cette vision à travers trois concepts fondamentaux :

1. L’architecture hybride fait abstraction de la complexité physique
2. Les observations invariantes d’échelle créent une entrée de modèle universelle
3. MARL rend les agents adaptables

Alerte spoiler: Les deux premiers concepts nous permettent de transférer facilement des agents entre les tâches, tandis que le troisième rend l’agent adaptatif au sein d’une même tâche et au-delà. Regardons chacun.

Architecture hybride

Comment concevoir un système capable de fournir des solutions robustes, même lorsqu’il est déplacé dans des contextes entièrement nouveaux ? Il vous suffit de lui faire résoudre non pas un cas particulier spécifique, mais quelque chose de plus généralisé — un problème à un niveau d’abstraction plus élevé.

Mais comment donner vie à cela ? Divisons le problème en couches et résolvons-le en utilisant un hybride : RL commande la stratégie de haut niveau et LP vit l’exécution de bas niveau. Ce faisant, nous permettons à RL de synthétiser des connaissances dans un domaine plus large, tandis que LP résout des cas d’empaquetage spécifiques et individuels.

action = [num_vehicles_1, .. , num_vehicles_n]

Voir Partie 1 pour plus de détails sur l’approche hybride et les versions action

Grâce à cette « séparation des tâches », la composante RL est soulagée à la minute près des trivialités techniques relatives à l’endroit où les colis sont envoyés ou à la manière dont ils sont emballés. Comme un manager détaché des détails d’exécution.

En fin de compte, l’agent RL affecte indirectement l’environnement : ses grandes actions sont traitées via le solveur LP, qui actualise ensuite l’état de l’environnement.

Voici comment nous traitons l’action de l’agent RL et la transmettons au solveur LP.

def decide_send_LP(self, action: np.array):
    # Parse the RL agent's action array into a dictionary of active destinations
    neighb_action = {v_id: num_v for v_id, num_v in enumerate(action) if num_v > 0}
   
    if not neighb_action:
        return 0, 0 # No vehicles dispatched
    # Get warehouse inventory for parcels that can actually go to the chosen destinations
    available_parcels = self.get_available_parcels(destinations=neighb_action.keys())
   
    if available_parcels.empty:
        return 0, 0 # No packages to send
       
    # The LP decides which parcels go into the vehicles to maximize volume/profit
    av_vehicles = self.get_available_vehicles()
    parcels_result, edges_result = send_veh(neighb_action, available_parcels, av_vehicles)
   
    # Update the environment state based on the LP's physical execution
    self.process_sent(parcels_result)
   
    # Return costs to the environment (for reward calculation)
    shipment_cost = sum(edges_result.c_cost * edges_result.v_varr_value)
    num_vehicles_sent = edges_result.v_varr_value.sum()
   
    return shipment_cost, num_vehicles_sent

Que se passe-t-il ici ? Dans un premier temps, nous devons traduire les actions de l’agent dans un format compréhensible, garantissant que l’agent a effectivement demandé au moins une expédition. Ensuite, nous vérifions si des colis présents dans l’entrepôt peuvent être envoyés.

Ensuite, nous exécutons une programmation linéaire, qui regroupe les colis disponibles dans les véhicules disponibles, en choisissant non seulement la classe de transport mais aussi le véhicule spécifique, ainsi que la destination de ce colis.

Et enfin, nous mettons à jour l’état de l’environnement en fonction de l’exécution du LP, calculons les frais d’expédition et le renvoyons pour calculer la récompense.

Ainsi, nous avons obtenu le portabilité — tant que la structure de la tâche est la même, le système peut s’adapter à n’importe quel problème au sein d’une même classe.

Observations invariantes à l’échelle

Disons que nous avons l’architecture hybride. Mais comment le faire survivre dans des contextes variés si les espaces d’observation et d’action des agents RL sont techniquement figés à l’initialisation ?

J’y suis parvenu en transformant les observations – je normalisé l’espace d’observation pour le faire invariant à l’échelle. Au lieu de suivre décomptes bruts (par exemple, « combien de colis ont été envoyés »), nous suivons ratios (par exemple, « quel pourcentage de l’arriéré total a été envoyé »).

Il s’agit d’une astuce technique spécifique qui vous permet de transférer « librement » un agent d’une tâche à une autre en permettant à l’agent d’opérer à un niveau d’abstraction plus élevé où les nombres absolus n’ont pas d’importance.

Discutons de quelques exemples.

Observations

Inventaire local perc_piles_wh— La quantité de colis dans chaque entrepôt.

def upd_perc_piles_wh(env):
    piles_wh = env.metrics['piles_wh']
    return np.array([piles_wh / env.num_piles])

Ici, pour rendre l’échelle d’observation invariante, je divise l’inventaire actuel de l’entrepôt piles_wh par le nombre absolu de colis qui passeront par la simulation env.num_piles. Ce faisant, l’agent apprend à établir des priorités en fonction du pourcentage de la charge de travail quotidienne qu’il détient actuellement.

Inventaire local par directions — Indique exactement où doit aller la charge actuelle. C’est le fondement de la décision de routage.

def upd_warehouse_loading_level_by_directions(env):
    # Get the current physical inventory at this specific node
    parcels = env.get_current_warehouse_parcels()
    if parcels.empty:
        return np.zeros(env.num_vertices)
   
    # Prepare the destinations array
    destinations = parcels['destination'].values.astype(int)
 
    # Get the counts for the destinations
    counts = np.bincount(destinations, minlength=env.num_vertices)
    return counts / len(parcels)

Tout d’abord, nous extrayons le stock actuel de colis dans cet entrepôt spécifique et vérifions qu’il n’est pas vide. Ensuite, nous extrayons la colonne « destination » sous la forme d’un tableau d’entiers, qui représentent les identifiants de l’entrepôt cible. Enfin, np.bincount calcule la répartition des colis sur toutes les destinations. En divisant ces chiffres par le nombre total de colis actuellement dans cet entrepôt local, nous convertissons un volume absolu en part. Le résultat est un vecteur de flotteurs invariant à l’échelle, où chaque indice représente le pourcentage exact du stock local dirigé vers ce sommet spécifique.

Date limite la plus proche par direction (deadlines_min_dist) — Répartition des délais les plus proches pour le stock actuel.

def upd_deadlines_min_dist(env):
    parcels = env.get_current_warehouse_parcels()
    deadlines = np.ones(env.num_vertices) # 1.0 means no urgency or no parcels
    if not parcels.empty:
        # Group by destination and find the actual minimum time left
        min_times = parcels.groupby('destination')['time_left'].min() / env.max_time_left
       
        # Assign the calculated minimums to their respective destination indices
        deadlines[min_times.index.astype(int)] = min_times.values
    return np.clip(deadlines, env.config.OBS_BOX_LOW, env.config.OBS_BOX_HIGH)

Ici, nous extrayons à nouveau l’inventaire local actuel. Nous initialisons un vecteur de délais pour qu’il ait la taille du graphique et le remplissons avec des unités (où 1,0 signifie aucune urgence et les valeurs proches de 0,0 indiquent un délai qui est arrivé).

Ensuite, nous regroupons les colis par destination et trouvons le time_left minimum pour chaque itinéraire. Et nous divisons cela par le temps maximum possible restant pour convertir le temps absolu en un rapport relatif (même approche ici).

Étant donné que ce vecteur résultant ne contient que des données sur les destinations actives, il est clairsemé et non aligné avec notre espace d’action. Nous mappons ces délais urgents à leurs identifiants de position topologique corrects en utilisant les destinations comme indices entiers.

Pour finir, nous découpons le tableau pour qu’il reste strictement entre 0 et 1. Il s’agit d’une mesure de sécurité essentielle, car les paquets en retard généreront des valeurs de temps négatives, ce qui briserait les limites d’observation du réseau neuronal.

Ainsi, généralement, une nouvelle tâche implique un espace d’observation complètement nouveau. Cependant, dans mon approche hybride, ce n’est pas le cas : les agents peuvent être transférés d’un entrepôt à l’autre par conception, quel que soit le nombre de colis, de véhicules ou de nœuds voisins.

Remplissage nul ou remplissage maximal de nœuds

Dans la version actuelle, la seule exception est le nombre total d’entrepôts du réseau (l’ordre du graphique). Ceci doit être connu à l’avance, car le transfert n’est possible que vers un graphe de même taille maximale.

Nous gérons cette limitation en utilisant la norme zéro remplissage. Nous définissons un taille maximale du graphique (par exemple, 100 sommets), et pour tout graphique plus petit, nous masquons les nœuds inexistants avec des valeurs nulles. Si la taille maximale de votre graphique est de 100 sommets, il vous suffit de déployer l’agent sur les sommets actifs existants et de masquer le reste avec des zéros. La même logique s’applique à l’observation des voisins : la taille du vecteur est toujours égale à l’ordre du graphe logistique, mais seuls les voisins disponibles (observables) ont des valeurs non nulles.

MARNE

De bonnes solutions dans un contexte changeant

Abordons maintenant un autre problème : la réalité est volatile.

Une tempête de neige soudaine frappe, les tarifs 3PL triplent ou il y a une augmentation massive des commandes juste avant les vacances. Une entreprise doit être adaptable sur le plan opérationnel pour survivre à cette situation. A noter que les règles physiques du jeu (taille des véhicules, carte) restent les mêmes, mais le contexte change complètement.

Les heuristiques statiques (par exemple, une règle codée en dur pour « expédier à 85 % de sa capacité ») commenceront immédiatement à générer des pertes colossales dans ces scénarios. Un avantage majeur de l’approche MARL est qu’elle généralise la situation compte tenu des observations. Il modifie dynamiquement son seuil de décision « à la volée » en réponse à ces observations changeantes.

Un autre grand avantage de MARL est que le problème est divisé en parties plus petites, qui sont résolues indépendamment par les agents. Les architectures multi-agents nous évitent d’être obligés de résoudre l’ensemble du problème du réseau avec un seul « méga-agent ». Cependant, j’aborderai cela plus en détail dans mon prochain article sur la réduction de dimensionnalité.

Implémentation MARL

Quelques mots sur la façon dont nous avons spécifiquement implémenté l’aspect multi-agent. J’ai été confronté à deux défis distincts :

1. Les actions des agents étant interdépendantes, ils peuvent facilement s’adapter à les comportements sous-optimaux de chacun. Par conséquent, au début de la formation, le MARL traditionnel peut être très instable.
2. Je voulais rester dans la pile OpenAI Gym + Stable-baselines, qui ne prend pas explicitement en charge la formation MARL native.

Dans le même temps, le retour à une solution à agent unique était impossible en raison du grand nombre d’entrepôts, et l’approche « un méga-agent » a été abandonnée au stade architectural (les détails dans Architecture de la première partie).

En conséquence, j’ai conçu le pipeline de formation suivant :

• Au lieu de former tous les agents simultanément, nous n’en formons qu’un seul : l’agent « actuel » par épisode.
• Pendant que l’agent « actuel » s’entraîne, les autres fonctionnent uniquement en mode d’inférence figée.
• Une « étape » environnementale globale consiste en une exécution séquentielle de tous les agents: l’agent « formation » entre en action, suivi des agents « inférence ».

Voici à quoi cela ressemble dans le code :

# Initialize environment and load the current best weights for all agents
env.env_method('prepare_env', best_agent_paths)
for i in range(NUM_MARL_LOOPS):
  for training_ag_id in agents.keys():
      # Shift the environment's perspective to the current active agent
      env.env_method('set_cur_training_agent', training_ag_id)
      # Fetch the active agent's policy model
      agent_obj = agents.get(training_ag_id)
      # Train ONLY this agent
      # (This will call env.step() under the hood
      # and will run the other agents in frozen inference mode)
      agent_obj = agent_obj.learn(
          TS_PER_AGENT,
          reset_num_timesteps=False,
          tb_log_name=f"Agent_{training_ag_id}",
          callback=callbacks,
          )
      # Save the updated weights and push them to the live models cache
      agent_obj.save(last_agent_paths[training_ag_id])
      agents[training_ag_id] = agent_obj

Tout d’abord, prepare_env() est exécuté, qui définit les valeurs par défaut et les chemins d’enregistrement des agents. Ensuite, nous lançons la boucle principale, qui dicte le nombre de passes de formation NUM_MARL_LOOPS sur l’ensemble du réseau.

À l’intérieur de cela, nous nous occupons de la formation d’un seul agent « actuel ». Le agents est un dictionnaire : les clés sont des identifiants, les valeurs sont les objets du modèle. La méthode set_cur_training_agent() change la perspective de l’environnement. Ensuite, nous prenons le modèle de l’agent actuel et déclenchons .learn(). Après cela, c’est assez simple : nous sauvegardons le modèle et mettons à jour le agents dictionnaire.

Voyons maintenant brièvement comment cette étape s’exécute réellement dans l’environnement :

def step(self, action) -> tuple[dict, float, bool, dict]:
    # Training Agent executes its action
    reward = self.process_packages(action)
    self.process_inflow() # Localized to the active agent's node
    self.update_state_and_metrics(reward)
    self.save_current_act_agent()

    # Inference Loop: Other agents take their turns sequentially
    for ag_id in self.inference_agents.keys():
        if ag_id == self.cur_training_agent:
            continue # Skip the training agent (it already acted)
            
        # Switch environment context to the current inference agent
        self.current_origin = ag_id
        self.load_act_agent()
       
        # Load model and get masked prediction
        agent_obj = self.inference_agents.get(ag_id)
        action_mask = self.valid_action_mask()
        ag_action, _ = agent_obj.predict(self.state, action_masks=action_mask)
       
        # Execute inference agent's action
        sub_reward = self.process_packages(ag_action)
        self.update_state_and_metrics(sub_reward)              
        self.save_current_act_agent()
   
    # Restore environment state to the Training Agent's perspective
    self.current_origin = self.cur_training_agent
    self.load_act_agent()
   
    # Check terminal conditions
    done = self.check_if_done()
   
    self.step_n += 1
    return self.state, reward, done, self.info

Tout d’abord, nous exécutons l’action pour l’agent de formation « actuel ». Nous commençons par traiter les parcelles actuellement dans le système via self.process_packages(action), où l’action de l’agent est appliquée à la logique de l’environnement. En d’autres termes, si l’agent décide d’envoyer des camions vers certains entrepôts, le solveur LP l’exécute ici.

Après cela, nous recevons les nouveaux packages entrants dans self.process_inflow(), mettons à jour l’état et les métriques dans self.update_state_and_metrics() et enregistrons le contexte de l’agent dans save_current_act_agent().

Maintenant, la partie amusante commence. Puisque l’agent de formation actuel a déjà entrepris son action, nous devons en déduire les actions pour le reste du réseau. Nous démarrons donc une boucle for sur nos agents disponibles, en sautant celle de formation. Dans cette boucle, nous changeons le contexte d’agent « actuel », chargeons son modèle et générons une inférence en introduisant l’état actuel et le masque d’action dans agent_obj.predict().

A partir de là, le flux est identique à celui de l’agent de formation : nous traitons l’action générée (cette fois, par un agent d’inférence), et mettons à jour l’environnement. Enfin, à la fin de la boucle, nous revenir au contexte à l’agent de formation actuel et transmettre les résultats finaux à la boucle.

Dans les prochains épisodes

Nous disposons donc désormais d’une boucle de formation entièrement fonctionnelle. Le code s’exécute et l’environnement MARL s’initialise, mais comment pouvons-nous réellement garantir ce processus de formation :

• Termine dans un délai raisonnable ?
• Est-ce que les modèles convergent ?
• Produit des stratégies de routage « assez bonnes » ?

C’est ce que je détaillerai dans les prochains articles. Restez à l’écoute!

LinkedIn | E-mail