菜单
Explorer les techniques d'apprentissage par renforcement multi-agents (MARL), incluant la communication dynamique, la coordination, la décomposition des tâches et les protocoles d'entraînement décentralisés.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
基于 SOC 职业分类
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/JohnNuwan/EVA_CORE --skill multi-agent-reinforcement-learning命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
正在显示 SKILL.md
Concevoir et maintenir un watchdog auto-correcteur pour services HTTP — checks de santé, auto-restart, état persistant, rapports et pièges bash.
Serveur de messagerie sécurisé auto-hébergé (Signal-like) avec Flask + WebSocket + AES-256-GCM + pont EVA
ADAM-SENTINEL — Veilleur technologique 24h/24h. Scanne 10 domaines, cree des rapports, met a jour les skills, alerte sur les CVE et breaking changes.
| name | multi-agent-reinforcement-learning |
| description | Explorer les techniques d'apprentissage par renforcement multi-agents (MARL), incluant la communication dynamique, la coordination, la décomposition des tâches et les protocoles d'entraînement décentralisés. |
| version | 2.0.0 |
| author | EVA Agent |
| license | Privée EVA St-Étienne |
| platforms | ["linux","macos","windows"] |
| tags | ["MARL","apprentissage-renforcement","multi-agents","coordination","communication","CTDE"] |
| keywords | ["Multi-Agent Reinforcement Learning","CTDE","communication protocols","cooperative AI","decentralized training"] |
L'apprentissage par renforcement multi-agents (MARL) étend le cadre classique du reinforcement learning à des environnements où plusieurs agents interagissent simultanément. Chaque agent observe l'état de l'environnement, prend des décisions, et reçoit des récompenses — mais la présence d'autres agents rend l'environnement non-stationnaire du point de vue de chaque apprenant.
Cette compétence couvre les aspects fondamentaux du MARL :
| Scénario | Pertinence |
|---|---|
| Coordination de flotte (drones, robots, véhicules autonomes) | Élevée |
| Optimisation de trafic urbain (feux, flux, itinéraires) | Élevée |
| Gestion de ressources partagées (énergie, bande passante, entrepôt) | Élevée |
| Jeux compétitifs multi-agents (stratégie, trading) | Moyenne |
| Problème mono-agent standard (ex. jeu Atari classique) | Faible (préférer RL simple) |
Un jeu stochastique à N agents est défini par le tuple $(S, \mathcal{A}^1, \dots, \mathcal{A}^N, P, R^1, \dots, R^N, \gamma)$ où :
L'objectif de chaque agent $i$ est de maximiser son espérance de récompense cumulée : $\mathbb{E} \left[ \sum_{t=0}^{\infty} \gamma^t R^i_t \right]$.
| Régime |
|---|
| Type de récompense |
|---|
| Exemple |
|---|
| Coopératif | $R^1 = R^2 = \dots = R^N$ | Robots collaborant à une tâche commune |
| Compétitif | $R^1 = -R^2$ (jeu à somme nulle) | Jeu d'échecs, trading opposé |
| Mixte | Récompenses indépendantes ou partiellement alignées | Trafic routier, enchères |
Dans MARL, la politique $\pi^i(a^i \mid s)$ de l'agent $i$ dépend de l'état $s$, mais la transition vers $s'$ dépend des actions de tous les agents :
$$ s_{t+1} \sim P(\cdot \mid s_t, a^1_t, \dots, a^N_t) $$
Quand un autre agent met à jour sa politique, la distribution des transitions change du point de vue de l'agent $i$ — c'est la non-stationnarité, le défi central du MARL.
CTDE est le paradigme dominant : les agents sont entraînés avec accès à l'information globale (états et actions de tous), mais exécutent de manière décentralisée (chaque agent n'utilise que ses observations locales).
┌──────────────────┐
│ Entraînement │ ← Accès global (tous états × toutes actions)
│ Centralisé │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Exécution │ ← Chaque agent n'utilise que ses observations locales
│ Décentralisée │
└──────────────────┘
QMIX factorise la fonction de valeur globale $Q_\text{tot}$ comme une combinaison monotone des fonctions de valeur individuelles $Q_i$ :
import torch
import torch.nn as nn
import torch.nn.functional as F
class ReseauQMIX(nn.Module):
"""Réseau de mélange QMIX pour MARL coopératif."""
def __init__(self, n_agents: int, dim_etat: int, dim_action: int, dim_embedding: int = 64):
super().__init__()
self.n_agents = n_agents
# Réseaux individuels (agents partagent les poids ou non)
self.reseaux_agents = nn.ModuleList([
nn.Sequential(
nn.Linear(dim_etat, dim_embedding),
nn.ReLU(),
nn.Linear(dim_embedding, dim_embedding),
nn.ReLU(),
nn.Linear(dim_embedding, dim_action),
)
for _ in range(n_agents)
])
# Hyper-réseaux pour les poids du mélangeur
self.hyper_w1 = nn.Linear(dim_etat, dim_embedding * n_agents)
self.hyper_w2 = nn.Linear(dim_etat, dim_embedding)
self.hyper_b1 = nn.Linear(dim_etat, dim_embedding)
self.hyper_b2 = nn.Linear(dim_etat, 1)
def forward(self, etats: torch.Tensor) -> torch.Tensor:
"""Calcule Q_tot à partir des états et actions.
Args:
etats: Tenseur (batch, n_agents, dim_etat).
Returns:
Q_tot valeurs (batch, n_actions_tot).
"""
batch_size = etats.shape[0]
# Q_i pour chaque agent
q_i = torch.stack([
reseau(etats[:, i]) for i, reseau in enumerate(self.reseaux_agents)
], dim=1) # (batch, n_agents, n_actions)
# État global (concaténation)
etat_global = etats.reshape(batch_size, -1)
# Poids du mélangeur
w1 = torch.abs(self.hyper_w1(etat_global)).reshape(
batch_size, self.n_agents, -1
)
b1 = self.hyper_b1(etat_global).reshape(batch_size, -1, 1)
w2 = torch.abs(self.hyper_w2(etat_global)).reshape(
batch_size, -1, 1
)
b2 = self.hyper_b2(etat_global)
# Q_tot = mélange monotone
q_hidden = F.elu(torch.bmm(q_i, w1) + b1)
q_tot = (torch.bmm(q_hidden, w2) + b2).squeeze(-1)
return q_tot
Independent Q-Learning (IQL) traite chaque agent comme un apprenant RL indépendant, ignorant les autres agents :
class AgentIQL:
"""Agent Q-Learning indépendant pour MARL décentralisé."""
def __init__(self, dim_etat: int, dim_action: int, lr: float = 1e-3,
gamma: float = 0.99, epsilon: float = 0.1):
self.q_network = nn.Sequential(
nn.Linear(dim_etat, 128),
nn.ReLU(),
nn.Linear(128, dim_action),
)
self.optimiseur = torch.optim.Adam(self.q_network.parameters(), lr=lr)
self.gamma = gamma
self.epsilon = epsilon
def selectionner_action(self, etat: np.ndarray) -> int:
"""Sélection epsilon-gloutonne (décentralisée : n'utilise que l'état local).
Args:
etat: Observation locale de l'agent.
Returns:
Action sélectionnée.
"""
if np.random.random() < self.epsilon:
return np.random.randint(self.q_network[-1].out_features)
with torch.no_grad():
q_vals = self.q_network(torch.FloatTensor(etat))
return int(q_vals.argmax().item())
def mettre_a_jour(self, etat, action, recompense, etat_suivant, termine):
"""Met à jour Q via l'équation de Bellman (indépendamment des autres agents)."""
etat_t = torch.FloatTensor(etat)
etat_suivant_t = torch.FloatTensor(etat_suivant)
q_courant = self.q_network(etat_t)[action]
q_suivant = self.q_network(etat_suivant_t).max() if not termine else 0.0
cible = recompense + self.gamma * q_suivant
perte = F.mse_loss(q_courant, torch.tensor(cible))
self.optimiseur.zero_grad()
perte.backward()
self.optimiseur.step()
| Bénéfice | Description |
|---|---|
| Partage d'observation | Un agent voit ce qu'un autre ne voit pas |
| Coordination | Éviter les conflits d'actions (ex. deux robots vers le même objet) |
| Apprentissage accéléré | Transmission de politiques apprises (knowledge transfer) |
import torch.nn as nn
class CommunicateurGNN(nn.Module):
"""Couche de communication entre agents basée sur un GNN (message passing)."""
def __init__(self, dim_etat: int, dim_message: int = 32):
super().__init__()
self.encoder_message = nn.Linear(dim_etat, dim_message)
self.aggregateur = nn.Linear(dim_message * 2, dim_message)
self.decodeur = nn.Linear(dim_message + dim_etat, dim_etat)
def forward(self, etats: torch.Tensor, matrice_adjacence: torch.Tensor) -> torch.Tensor:
"""Fait passer des messages entre agents connectés.
Args:
etats: Tenseur (n_agents, dim_etat).
matrice_adjacence: Tenseur booléen (n_agents, n_agents) — qui parle à qui.
Returns:
États mis à jour après communication (n_agents, dim_etat).
"""
n_agents = etats.shape[0]
messages = self.encoder_message(etats) # (n_agents, dim_message)
# Agrégation des messages entrants
messages_aggreges = []
for i in range(n_agents):
voisins = matrice_adjacence[i].nonzero(as_tuple=True)[0]
if len(voisins) > 0:
msg_voisins = messages[voisins].mean(dim=0)
else:
msg_voisins = torch.zeros_like(messages[0])
messages_aggreges.append(msg_voisins)
messages_aggreges = torch.stack(messages_aggreges)
# Mise à jour des états avec les messages reçus
entrees = torch.cat([messages_aggreges, etats], dim=-1)
return self.decodeur(entrees)
Les agents peuvent apprendre quand et quoi communiquer :
class AgentAvecCommunication:
"""Agent MARL avec protocole de communication appris."""
def __init__(self, dim_etat: int, dim_action: int, dim_message: int = 16):
# Politique de communication : génère un message
self.reseau_emission = nn.Sequential(
nn.Linear(dim_etat, 32),
nn.ReLU(),
nn.Linear(32, dim_message),
)
# Gating : décide si le message est envoyé
self.gating = nn.Sequential(
nn.Linear(dim_etat, 1),
nn.Sigmoid(),
)
# Politique d'action : combine état et messages reçus
self.reseau_action = nn.Sequential(
nn.Linear(dim_etat + dim_message, 64),
nn.ReLU(),
nn.Linear(64, dim_action),
)
def emettre_message(self, etat: torch.Tensor) -> tuple[torch.Tensor, float]:
"""Génère un message et décide s'il est envoyé.
Args:
etat: État local de l'agent.
Returns:
Tuple (message, probabilité d'envoi).
"""
message = self.reseau_emission(etat)
proba_envoi = self.gating(etat)
# Pendant l'entraînement, échantillonner depuis Bernoulli(proba_envoi)
envoyer = torch.bernoulli(proba_envoi) > 0.5
return message if envoyer else torch.zeros_like(message), proba_envoi
def agir(self, etat: torch.Tensor, messages_recus: torch.Tensor) -> int:
"""Sélectionne une action basée sur l'état local et les messages reçus.
Args:
etat: État local de l'agent.
messages_recus: Messages reçus des autres agents (agrégés).
Returns:
Action sélectionnée.
"""
entrees = torch.cat([etat, messages_recus], dim=-1)
return self.reseau_action(entrees).argmax().item()
from marl_library.simulation import configure_agents, initialize_environment, train_agents
def entrainer_flotte_drones(n_drones: int = 5, n_epochs: int = 100) -> dict:
"""Entraîne une flotte de drones pour une mission de couverture coopérative.
Args:
n_drones: Nombre de drones dans la flotte.
n_epochs: Nombre d'épisodes d'entraînement.
Returns:
Métriques de performance (récompense moyenne, taux de couverture).
"""
environnement = initialize_environment("couverture_cooperative", n_agents=n_drones)
agents = configure_agents(n_agents=n_drones, algorithme="qmix")
resultats = train_agents(
agents=agents,
env=environnement,
n_episodes=n_epochs,
evaluer_tous=10, # Évaluer tous les 10 épisodes
)
return analyser_performance(resultats)
| Agent | Action | Récompense |
|---|---|---|
| Feu de signalisation (intersection) | Durée du feu vert/rouge | Réduction du temps d'attente moyen |
| Véhicule autonome | Choix d'itinéraire | Temps de trajet individuel |
| Centre de régulation | Modification des priorités | Flux global maximisé |
Problème : Traiter MARL comme N problèmes RL indépendants (IQL) ignore la non-stationnarité, conduisant à des politiques instables.
Solution : Utilisez le paradigme CTDE. Si IQL est nécessaire, stabilisez avec :
Problème : Quand la récompense est globale (tous les agents reçoivent la même récompense), comment attribuer le crédit à chaque agent ?
Solution : Utilisez des mécanismes de factorisation de la valeur :
Problème : Avec N agents, la dimension de l'espace d'actions conjoint est $|\mathcal{A}|^N$, explosive.
Solution : Utilisez des méthodes mean-field (approximation du champ moyen) où chaque agent interagit avec la distribution moyenne des autres agents plutôt qu'avec chaque agent individuellement.
Problème : Dans les espaces d'états-actions conjoints, l'exploration aléatoire est extrêmement inefficace.
Solution : Implémentez une exploration structurée :
| Framework | Particularité | Langage |
|---|---|---|
| PettingZoo | API Gym pour multi-agents | Python |
| SMAC | StarCraft Multi-Agent Challenge | Python |
| RLlib | MARL distribué (Ray) | Python |
| EPyMARL | Implémentations de référence | Python |
research-multi-agent-complex-tasks : pour la planification de tâches complexes dans un cadre multi-agents.agentic-research-and-arxiv : pour la veille sur les publications MARL récentes.references/marl_algorithms_comparison.md : comparaison détaillée des algorithmes MARL (IQL, VDN, QMIX, MADDPG, MAPPO).scripts/marl_training_loop.py : boucle d'entraînement MARL générique avec support CTDE.scripts/marl_communication_benchmark.py : benchmark des protocoles de communication.Documentation maintenue par EVA Agent — Dernière mise à jour : 2025