load-balancer

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 in self.agents if a.healthy]
        if capability:
            candidates = [a for a in candidates if capability in a.capabilities]
        # Filtrer selon le pourcentage canary
        candidates = [a for a in candidates if random.random() * 100 <= a.canary_traffic_pct]
        return candidates

    def route(
        self,
        task_type: str = None,
        session_id: str = None,
        priority: str = "normal",
        max_cost_per_1k: float = None,
    ) -> Optional[AgentProfile]:
        import time
        # Sticky session — vérifier TTL
        if session_id and session_id in self._affinities:
            agent_id, ts = self._affinities[session_id]
            if time.time() - ts < self._affinity_ttl:
                agent = next((a for a in self.agents if a.id == agent_id and a.healthy), None)
                if agent:
                    return agent
            else:
                del self._affinities[session_id]

        candidates = self._healthy_agents(capability=task_type)
        if max_cost_per_1k:
            candidates = [a for a in candidates if a.cost_per_1k_tokens <= max_cost_per_1k]
        if not candidates:
            return None

        selected = self._apply_strategy(candidates, priority)
        if session_id and selected:
            self._affinities[session_id] = (selected.id, time.time())
        return selected

    def _apply_strategy(self, candidates: list[AgentProfile], priority: str) -> AgentProfile:
        if self.strategy == RoutingStrategy.ROUND_ROBIN:
            agent = candidates[self._rr_index % len(candidates)]
            self._rr_index += 1
            return agent
        elif self.strategy == RoutingStrategy.LEAST_CONNECTIONS:
            return min(candidates, key=lambda a: a.active_tasks)
        elif self.strategy == RoutingStrategy.WEIGHTED:
            total = sum(a.weight for a in candidates)
            r = random.uniform(0, total)
            cumul = 0
            for a in candidates:
                cumul += a.weight
                if r <= cumul:
                    return a
            return candidates[-1]
        elif self.strategy == RoutingStrategy.CAPABILITY_BASED:
            return min(candidates, key=lambda a: (a.error_rate, a.avg_latency))
        elif self.strategy == RoutingStrategy.COST_BASED:
            if priority == "urgent":
                return min(candidates, key=lambda a: a.avg_latency)
            return min(candidates, key=lambda a: a.cost_per_1k_tokens)
        return random.choice(candidates)