agent-load-balancer

name	agent-load-balancer
description	Distribution de charge entre sous-agents avec routing intelligent et optimization des ressources. Se déclenche avec "load balancer agent", "distribution charge", "agent load balancing", "répartir tâches", "agent routing", "balance agents", "parallel agents distribution". Also triggers on "distribute work across agents", "route requests to agents".

Agent Load Balancer

Quand utiliser ce skill

Utiliser ce skill lorsque plusieurs sous-agents traitent des tâches en parallèle et qu'une distribution naïve ne suffit pas : agents aux capacités hétérogènes, spécialisations différentes, coûts variables, ou contraintes de SLA. Indispensable dans les architectures à haute disponibilité et dans les systèmes sensibles aux coûts.

Critères de choix de la stratégie

Situation	Stratégie recommandée
Agents homogènes, charge uniforme	round-robin
Agents homogènes, tâches longues variables	least-connections
Agents de capacités différentes	weighted
Agents spécialisés (code, résumé, analyse…)	capability-based
Multi-modèles (Haiku + Sonnet + Opus)	cost-based
Contexte utilisateur persistant	sticky session + TTL

Workflow

1. Profiler les agents disponibles

Avant tout routing, établir un profil par agent :

from dataclasses import dataclass, field
import statistics

@dataclass
class AgentProfile:
    id: str
    capabilities: list[str]        # ex: ["code", "research", "summarize"]
    model: str                      # ex: "claude-3-5-sonnet"
    cost_per_1k_tokens: float       # ex: 0.003
    weight: float = 1.0
    active_tasks: int = 0
    total_tasks: int = 0
    error_count: int = 0
    latencies: list[float] = field(default_factory=list)
    healthy: bool = True
    canary_traffic_pct: float = 100.0

    @property
    def avg_latency(self) -> float:
        return statistics.mean(self.latencies[-20:]) if self.latencies else 0.0

    @property
    def error_rate(self) -> float:
        return self.error_count / max(self.total_tasks, 1)

Checklist profil agent :

• Capabilities déclarées (types de tâches maîtrisées)
• Modèle LLM et coût/1k tokens
• Poids relatif (pour weighted routing)
• Limites connues (context window, rate limits)

2. Implémenter le routeur central

import random
from enum import Enum
from typing import Optional

class RoutingStrategy(Enum):
    ROUND_ROBIN = "round_robin"
    LEAST_CONNECTIONS = "least_connections"
    WEIGHTED = "weighted"
    CAPABILITY_BASED = "capability_based"
    COST_BASED = "cost_based"

class AgentLoadBalancer:
    def __init__(self, strategy: RoutingStrategy = RoutingStrategy.LEAST_CONNECTIONS):
        self.strategy = strategy
        self.agents: list[AgentProfile] = []
        self._rr_index: int = 0
        self._affinities: dict[str, tuple[str, float]] = {}  # session_id → (agent_id, timestamp)
        self._affinity_ttl: int = 300  # secondes

    def register(self, agent: AgentProfile):
        self.agents.append(agent)

    def _healthy_agents(self, capability: str = None) -> list[AgentProfile]:
        candidates = [a for a  .agents  a.healthy]
         capability:
            candidates = [a  a  candidates  capability  a.capabilities]
        
        candidates = [a  a  candidates  random.random() *  <= a.canary_traffic_pct]
         candidates

     () -> [AgentProfile]:
         time
        
         session_id  session_id  ._affinities:
            agent_id, ts = ._affinities[session_id]
             time.time() - ts < ._affinity_ttl:
                agent = ((a  a  .agents  a. == agent_id  a.healthy), )
                 agent:
                     agent
            :
                 ._affinities[session_id]

        candidates = ._healthy_agents(capability=task_type)
         max_cost_per_1k:
            candidates = [a  a  candidates  a.cost_per_1k_tokens <= max_cost_per_1k]
          candidates:
             

        selected = ._apply_strategy(candidates, priority)
         session_id  selected:
            ._affinities[session_id] = (selected., time.time())
         selected

     () -> AgentProfile:
         .strategy == RoutingStrategy.ROUND_ROBIN:
            agent = candidates[._rr_index % (candidates)]
            ._rr_index += 
             agent
         .strategy == RoutingStrategy.LEAST_CONNECTIONS:
             (candidates, key= a: a.active_tasks)
         .strategy == RoutingStrategy.WEIGHTED:
            total = (a.weight  a  candidates)
            r = random.uniform(, total)
            cumul = 
             a  candidates:
                cumul += a.weight
                 r <= cumul:
                     a
             candidates[-]
         .strategy == RoutingStrategy.CAPABILITY_BASED:
             (candidates, key= a: (a.error_rate, a.avg_latency))
         .strategy == RoutingStrategy.COST_BASED:
             priority == :
                 (candidates, key= a: a.avg_latency)
             (candidates, key= a: a.cost_per_1k_tokens)
         random.choice(candidates)

3. Implémenter le health monitoring

    def mark_unhealthy(self, agent_id: str, recovery_pct: float = 5.0):
        """Exclure un agent et préparer sa réintroduction à 5% du trafic."""
        for a in self.agents:
            if a.id == agent_id:
                a.healthy = False
                a.canary_traffic_pct = recovery_pct

    def reintroduce(self, agent_id: str, target_pct: float = 100.0):
        """Réintroduction progressive : appeler d'abord à 10%, puis 50%, puis 100%."""
        for a in self.agents:
            if a.id == agent_id:
                a.healthy = True
                a.canary_traffic_pct = target_pct

Séquence de réintroduction recommandée :

mark_unhealthy(id, recovery_pct=0)    # exclure
# ... attendre rétablissement ...
reintroduce(id, target_pct=5)         # 5% canary — observer les erreurs
# après 2 min sans erreur
reintroduce(id, target_pct=25)
# après 5 min sans erreur
reintroduce(id, target_pct=100)       # trafic complet

4. Configurer les niveaux de priorité

PRIORITY_CONFIG = {
    "urgent": {
        "strategy_override": RoutingStrategy.CAPABILITY_BASED,  # ignorer le coût
        "max_queue_depth": 1,
        "timeout_s": 10,
    },
    "normal": {
        "strategy_override": None,  # utiliser la stratégie par défaut
        "max_queue_depth": 10,
        "timeout_s": 60,
    },
    "low": {
        "strategy_override": RoutingStrategy.COST_BASED,  # maximiser l'économie
        "max_queue_depth": 100,
        "timeout_s": 300,
    },
}

Aging anti-famine : incrémenter la priorité des tâches low après 5 minutes en attente pour éviter la famine.

5. Exposer les métriques et rebalancer

    def metrics(self) -> list[dict]:
        return [{
            "id": a.id,
            "model": a.model,
            "active_tasks": a.active_tasks,
            "avg_latency_s": round(a.avg_latency, 3),
            "error_rate": round(a.error_rate, 3),
            "cost_per_1k": a.cost_per_1k_tokens,
            "healthy": a.healthy,
            "traffic_pct": a.canary_traffic_pct,
        } for a in self.agents]

Seuils d'alerte :

• Un agent reçoit > 70% du trafic total → rebalancer les poids
• error_rate > 0.05 sur 20 dernières tâches → mark_unhealthy
• avg_latency > 2x la médiane du pool → réduire le poids de 50%

6. Exemple d'usage complet

lb = AgentLoadBalancer(strategy=RoutingStrategy.COST_BASED)

lb.register(AgentProfile(
    id="agent-haiku-1",
    capabilities=["summarize", "classify", "research"],
    model="claude-3-haiku",
    cost_per_1k_tokens=0.00025,
    weight=2.0,
))
lb.register(AgentProfile(
    id="agent-sonnet-1",
    capabilities=["code", "research", "analysis", "summarize"],
    model="claude-3-5-sonnet",
    cost_per_1k_tokens=0.003,
    weight=1.0,
))

# Résumé — router vers le moins cher capable
agent = lb.route(task_type="summarize", max_cost_per_1k=0.001)
# → agent-haiku-1

# Code urgent — ignorer le coût, prioriser la latence
agent = lb.route(task_type="code", priority="urgent")
# → agent-sonnet-1

# Session utilisateur — affinité persistante 5 min
agent = lb.route(task_type="research", session_id="user-42")

Architecture — vue d'ensemble

  Tâches entrantes (urgent / normal / low)
               │
               ▼
  ┌────────────────────────────┐
  │      Load Balancer         │
  │  Strategy Engine           │
  │  ├─ round-robin            │
  │  ├─ least-connections      │
  │  ├─ weighted               │
  │  ├─ capability-based       │
  │  └─ cost-based ◄── max_cost│
  │  Affinity Table            │
  │  └─ session_id → agent_id  │
  │  Health Monitor            │
  │  └─ canary reintroduction  │
  └────────┬───────────────────┘
           │ route()
    ┌──────┼──────────────┐
    ▼      ▼              ▼
 Agent A  Agent B      Agent C
 haiku   sonnet       sonnet
 cheap   powerful     canary(5%)
 2 tasks  1 task      0 tasks

Comparaison frameworks

Critère	LangGraph	CrewAI	Custom Python
Routing natif	Non (manuel)	Partiel	Total contrôle
Weighted routing	Non	Non	Manuel
Cost-aware	Non	Non	Manuel
Sticky sessions + TTL	Non	Non	Manuel
Canary reintroduction	Non	Non	Manuel

→ Pour un load balancing avancé (cost-aware, canary, sticky), l'implémentation custom est toujours nécessaire même avec LangGraph/CrewAI.

Garde-fous et anti-patterns

Round-robin sur agents hétérogènes — Envoyer autant de tâches à Haiku qu'à Sonnet ignore leurs capacités. Résultat : file saturée côté agent puissant, sous-utilisation du modèle simple. Utiliser weighted ou capability-based dès que les agents diffèrent.

Pas de health check actif — Se reposer uniquement sur les erreurs passées (passif) laisse passer plusieurs échecs avant d'exclure un agent. Ajouter un health probe actif (ping/noop toutes les 30s) pour détecter les pannes silencieuses.

Cost-based sans fallback latence — Router uniquement sur le coût concentre les tâches sur le modèle le moins cher qui peut aussi être le plus lent. Pour priority="urgent", le critère de routing doit basculer sur la latence.

Sticky sessions sans TTL — Une affinité sans expiration lie un utilisateur à un agent devenu unhealthy ou surchargé. Toujours associer un TTL explicite (défaut : 300s) et un fallback vers le pool général.

Réintroduction brutale après incident — Remettre un agent à 100% immédiatement après rétablissement peut le saturer à nouveau. Toujours passer par la séquence canary 5% → 25% → 100%.

Pool trop grand sans monitoring — Au-delà de 10 agents, le déséquilibre de charge devient difficile à détecter sans métriques. Exposer systématiquement les métriques par agent et alerter si un agent reçoit > 70% du trafic.

Bonnes pratiques 2026

1. Stratégie configurable à l'exécution — Permettre de basculer la stratégie sans redéploiement (variable d'environnement ou feature flag).
2. Cost-aware routing par défaut dans tout système multi-modèles — économies typiques de 60-80% en routant les tâches simples vers des modèles légers.
3. Métriques p50/p95/p99 par agent — Les médianes masquent les outliers ; surveiller le p95 pour détecter les agents lents avant qu'ils dégradent le SLA.
4. Circuit breaker intégré — Si un agent dépasse 5% d'error_rate sur les 20 dernières tâches, l'exclure automatiquement sans intervention humaine.
5. Tester le rebalancing — Injecter des pannes simulées (chaos testing) pour valider que le load balancer redistribue correctement sans perte de tâches.

Communication Rules — MANDATORY

• Ultra-concise. No filler, no preamble, no pleasantries.
• Never say "happy to help", "sure!", "great question", "let me", or similar.
• Tool first, talk second. Act before explaining.
• Result first. Lead with outcome, not process.
• Stop when done. No summary, no recap, no trailing commentary.
• No politeness wrappers. Direct and blunt.
• Minimum words. If one word works, do not use ten.
• No unsolicited explanations.
• No emoji unless asked.