# Supervisor Builder > Construction d'agents superviseurs qui contrôlent, monitent et corrigent des sous-agents en temps réel. Se déclenche avec "supervisor agent", "agent superviseur", "contrôler sous-agents", "manager agent", "agent oversight", "superviser agents", "agent controller". Also triggers on "agent that monitors agents", "corrective supervision". - Skill: `khalilbenaz/supervisor-builder` (Agent Skill) - Install (CLI): `npx skillmds@latest add khalilbenaz/supervisor-builder` - Raw SKILL.md: https://api.skillmd.com/api/skills/khalilbenaz/supervisor-builder/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - Author: khalilbenaz (https://skillmd.com/u/khalilbenaz) - Updated: 2026-09-21 - Page: https://skillmd.com/skills/khalilbenaz/supervisor-builder --- # Agent Supervisor Builder ## Quand utiliser ce skill Utiliser ce skill quand une architecture multi-agents requiert une couche de contrôle active : garantie qualité dynamique, interruption d'un agent défaillant, redistribution de charge, escalade humaine. Indispensable dès que deux sous-agents ou plus s'exécutent en parallèle ou en chaîne avec des SLAs à respecter. --- ## Critères de décision — quel pattern choisir ? | Besoin | Pattern recommandé | |---|---| | Délégation simple, un agent à la fois | `create_supervisor` LangGraph (step 9) | | Équipe hiérarchique avec rôles fixes | CrewAI `Process.hierarchical` | | Intervenir en cours d'exécution | Monitor + CorrectionLoop (steps 3-5) | | Failover automatique + retry | InterventionRules + LoadBalancer (steps 4, 7) | | Escalade humaine obligatoire | EscalationManager (step 6) | --- ## Workflow ### 1. Définir les actions du supervisor Commencer par énumérer toutes les actions possibles avant d'écrire une ligne de logique : ```python from enum import Enum class SupervisorAction(Enum): DISPATCH = "dispatch" # Assigner une tâche à un sous-agent MONITOR = "monitor" # Surveiller la progression INTERVENE = "intervene" # Envoyer une correction en cours d'exécution REDIRECT = "redirect" # Réassigner vers un autre sous-agent TERMINATE = "terminate" # Arrêter un agent défaillant ESCALATE = "escalate" # Remonter à un humain ou orchestrateur supérieur APPROVE = "approve" # Valider un output avant livraison ``` ### 2. Routing intelligent — choisir le bon sous-agent Classifier l'intention, scorer les capacités, fallback vers le plus généraliste. ```python class IntelligentRouter: def __init__(self, agents: list): self.agents = {a.id: a for a in agents} def route(self, request: str) -> str: intent = self._classify_intent(request) best = max( self.agents.values(), key=lambda a: self._score(a, intent) ) return best.id def _classify_intent(self, request: str) -> dict: # Appel LLM léger (gpt-4o-mini suffit pour la classification) prompt = f"Classifie: '{request}'. JSON: task_type, domain, complexity(low|med|high)" return llm.invoke(prompt, response_format="json") def _score(self, agent, intent: dict) -> float: return ( 1.0 * (intent["task_type"] in agent.capabilities) + 0.5 * (intent["domain"] in agent.domains) + 0.3 * (agent.complexity_level >= intent["complexity"]) ) ``` **Décision** : si le score max < 0.5, logger un warning et router vers l'agent généraliste. Ne jamais bloquer. ### 3. Monitoring en temps réel Streamer les outputs des sous-agents et détecter les signaux d'alerte à la volée. ```python class AgentMonitor: def __init__(self): self.states: dict[str, dict] = {} async def watch(self, agent_id: str, agent): state = { "start_time": time.time(), "tokens_used": 0, "last_chunk": "", "alerts": [] } self.states[agent_id] = state async for chunk in agent.stream_output(): state["tokens_used"] += count_tokens(chunk) state["last_chunk"] = chunk[-300:] alert = self._check(state, chunk) if alert: state["alerts"].append(alert) yield alert # Le supervisor réagit immédiatement ``` Métriques à tracker obligatoirement : `elapsed_seconds`, `tokens_used`, `quality_score` (scoring LLM léger toutes les N secondes), `topic_drift_score`. ### 4. Règles d'intervention — seuils quantitatifs Tout seuil doit être un nombre. Aucun jugement subjectif non formalisé. ```python class InterventionRules: def __init__(self, config: dict): self.max_time = config.get("max_time", 120) # secondes self.max_tokens = config.get("max_tokens", 4000) self.quality_min = config.get("quality_min", 0.6) # 0.0–1.0 self.drift_max = config.get("drift_max", 0.5) def check(self, state: dict, task: dict) -> tuple[bool, str]: elapsed = time.time() - state["start_time"] if elapsed > self.max_time: return True, "timeout" if state["tokens_used"] > self.max_tokens: return True, "budget_exceeded" if state.get("quality_score", 1.0) < self.quality_min: return True, "quality_below_threshold" if self._drift(state["last_chunk"], task["objective"]) > self.drift_max: return True, "off_topic" return False, "" ``` Valeurs de départ recommandées (à calibrer sur vos données) : | Paramètre | Valeur initiale | Ajuster si... | |---|---|---| | `max_time` | 120 s | tâches longues → 300 s | | `max_tokens` | 4 000 | tâches volumineuses → 8 000 | | `quality_min` | 0.6 | domaine critique → 0.75 | | `drift_max` | 0.5 | sujets larges → 0.7 | ### 5. Correction loop — feedback et redirection Limiter `max_corrections` à 3. Au-delà → redirection automatique sans exception. ```python class CorrectionLoop: def __init__(self, max_corrections: int = 3): self.counts: dict[str, int] = {} self.max = max_corrections async def correct(self, agent_id: str, issue: str, ctx: dict) -> str: n = self.counts.get(agent_id, 0) if n >= self.max: return "redirect" # Déléguer à un autre agent prompt = self._prompt(issue, ctx) await agent_registry[agent_id].inject(prompt) self.counts[agent_id] = n + 1 return "corrected" def _prompt(self, issue: str, ctx: dict) -> str: return { "off_topic": f"CORRECTION: Recentre-toi sur l'objectif: {ctx['objective']}", "quality_below_threshold": f"CORRECTION: Ajoute des détails sur: {ctx.get('weak_points', 'la réponse')}", "timeout": "CORRECTION: Finalise avec ce que tu as déjà produit.", "budget_exceeded": "CORRECTION: Conclus en 3 phrases maximum." }.get(issue, f"CORRECTION: Améliore ta réponse pour mieux répondre à: {ctx.get('objective', 'l'objectif initial')}") ``` ### 6. Escalade — quand le supervisor est dépassé ```python class EscalationManager: def __init__(self, webhook_url: str, orchestrator=None): self.webhook = webhook_url self.orchestrator = orchestrator async def escalate(self, reason: str, ctx: dict, to: str = "human"): payload = { "reason": reason, "agent_id": ctx["agent_id"], "task": ctx["task"], "attempts": ctx["correction_count"], "last_output": ctx["last_output"][:500], # Tronquer pour le webhook "timestamp": datetime.utcnow().isoformat() } if to == "human": await httpx.AsyncClient().post(self.webhook, json=payload) elif to == "orchestrator" and self.orchestrator: await self.orchestrator.handle_escalation(payload) ``` Règle : toujours définir `webhook_url` en variable d'environnement (`SUPERVISOR_ESCALATION_WEBHOOK`), jamais en dur. ### 7. Load balancing — distribution équitable ```python import heapq class LoadBalancer: def __init__(self): self.loads: dict[str, int] = {} # agent_id → tâches actives self.queue: list = [] # min-heap (priority, task) def push(self, task, priority: int = 5): heapq.heappush(self.queue, (-priority, time.time(), task)) def least_loaded(self, capability: str) -> str: eligible = [ (load, aid) for aid, load in self.loads.items() if capability in agents[aid].capabilities ] if not eligible: raise ValueError(f"Aucun agent disponible pour: {capability}") return min(eligible)[1] def mark_done(self, agent_id: str): self.loads[agent_id] = max(0, self.loads.get(agent_id, 0) - 1) ``` ### 8. State management — persister l'état du supervisor ```python class SupervisorState: def __init__(self, backend: str = "memory"): self.backend = backend self.data = { "active_tasks": {}, "agent_statuses": {}, "metrics": {"dispatched": 0, "completed": 0, "failed": 0, "interventions": 0} } def update_agent(self, agent_id: str, status: str, task_id: str = None): entry = {"status": status, "task": task_id, "ts": time.time()} self.data["agent_statuses"][agent_id] = entry if self.backend == "redis": redis_client.hset("supervisor:agents", agent_id, json.dumps(entry)) def record_intervention(self): self.data["metrics"]["interventions"] += 1 ``` Backends supportés : `memory` (dev/test), `redis` (prod), `postgres` (audit long terme). ### 9. Implémentation concrète — LangGraph et CrewAI **LangGraph (recommandé 2026, package `langgraph-supervisor`) :** ```python from langgraph_supervisor import create_supervisor from langgraph.prebuilt import create_react_agent search_agent = create_react_agent( model, tools=[search_tool], name="search_agent", prompt="Tu es expert en recherche d'informations." ) code_agent = create_react_agent( model, tools=[code_tool], name="code_agent", prompt="Tu es expert en développement Python." ) supervisor = create_supervisor( agents=[search_agent, code_agent], model=model, prompt=( "Tu es le supervisor. Délègue:\n" "- Recherche → search_agent\n" "- Code → code_agent\n" "Valide la qualité avant de retourner le résultat." ), output_mode="last_message" # ou "full_history" pour audit ) result = supervisor.invoke({"messages": [{"role": "user", "content": "..."}]}) ``` **CrewAI (Process.hierarchical) :** ```python from crewai import Agent, Crew, Process from langchain_openai import ChatOpenAI manager = Agent( role="Supervisor Manager", goal="Coordonner les agents et garantir la qualité des outputs", backstory="Tu supervises une équipe d'agents spécialisés.", llm=ChatOpenAI(model="gpt-4o") ) researcher = Agent(role="Researcher", goal="Rechercher des informations précises", ...) writer = Agent(role="Writer", goal="Rédiger des contenus de qualité", ...) crew = Crew( agents=[researcher, writer], tasks=[research_task, writing_task], process=Process.hierarchical, manager_agent=manager, verbose=True ) result = crew.kickoff() ``` **AutoGen (pattern GroupChat avec admin) :** ```python import autogen supervisor_llm = {"config_list": [{"model": "gpt-4o"}]} supervisor = autogen.AssistantAgent("supervisor", llm_config=supervisor_llm, system_message="Tu coordonnes les agents. Valide chaque output avant de passer à l'étape suivante.") researcher = autogen.AssistantAgent("researcher", llm_config=supervisor_llm) user_proxy = autogen.UserProxyAgent("user", human_input_mode="NEVER", max_consecutive_auto_reply=10, code_execution_config={"use_docker": False}) chat = autogen.GroupChat(agents=[supervisor, researcher, user_proxy], messages=[], max_round=15) manager = autogen.GroupChatManager(groupchat=chat, llm_config=supervisor_llm) user_proxy.initiate_chat(manager, message="Lance la recherche sur X.") ``` ### 10. Reporting — métriques de session ```python class SupervisorReporter: def report(self, state: dict) -> dict: m = state["metrics"] total = max(m["dispatched"], 1) return { "success_rate": round(m["completed"] / total, 3), "intervention_rate": round(m["interventions"] / total, 3), "failure_rate": round(m["failed"] / total, 3), "top_failure_reasons": self._top_reasons(state) } def _top_reasons(self, state: dict) -> list[dict]: from collections import Counter reasons = [a["reason"] for a in state.get("intervention_log", [])] return [{"reason": r, "count": c} for r, c in Counter(reasons).most_common(5)] ``` --- ## Anti-patterns / Pièges | Anti-pattern | Conséquence | Correction | |---|---|---| | **Supervisor qui micro-manage** (correction toutes les 5 s) | Perturbe l'exécution, dégrade la qualité | Définir des seuils quantitatifs, laisser tourner entre les checks | | **Pas de `max_corrections`** | Boucle infinie, blocage total | Plafonner à 3, escalade automatique au-delà | | **Supervisor sans `TERMINATE`** | Impossible d'arrêter un agent bloqué en prod | `TERMINATE` + `REDIRECT` sont obligatoires avant toute mise en prod | | **Seuils subjectifs ("si la qualité est mauvaise")** | Imprévisible, impossible à déboguer | Chaque déclencheur est un nombre (score, temps, tokens) | | **État en mémoire seule en prod** | Perte totale au redémarrage | Redis ou Postgres dès que le supervisor tourne > 1 h | | **Escalade vers humain sans contexte** | L'opérateur ne comprend pas le problème | Toujours inclure `last_output`, `attempts`, `task` dans le payload | | **Supervisor qui exécute les tâches lui-même** | Couplage fort, testabilité nulle | Son rôle = coordonner uniquement ; mode dégradé explicitement documenté | --- ## Bonnes pratiques 2026 1. **Seuils numériques obligatoires** — aucune intervention sans déclencheur quantifiable. 2. **Dernier recours défini** — si `max_corrections` atteint et redirection impossible → escalade humaine, jamais de boucle silencieuse. 3. **Logs structurés pour chaque intervention** : `{agent_id, reason, action, result, timestamp}` — base de l'amélioration continue des règles. 4. **Health check du supervisor lui-même** — exposer `/health` ou équivalent ; un supervisor muet doit déclencher une alerte externe. 5. **Tester les chemins d'escalade en staging** avant la prod — simuler un agent qui timeout, un agent qui dérive, un agent qui dépasse le budget. 6. **Versioner la config des seuils** séparément du code — permet d'ajuster sans redéploiement.