spawner

Agent Spawner

Quand utiliser ce skill

Condition	Spawner requis ?
Nombre d'agents inconnu à l'avance	Oui
Agents identiques, volume variable	Oui (+ pool-manager)
Agents fixes et connus en design-time	Non — câbler statiquement
Besoin de parallélisme homogène	Préférer agent-pool-manager
Types d'agents différents selon le contexte	Oui

Workflow en étapes

1. Concevoir les templates d'agents

Chaque template encode une spécialisation. Définir au minimum :

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class AgentTemplate:
    name: str                          # identifiant du gabarit
    system_prompt: str                 # supporte les placeholders {domain}, {task}…
    tools: list[str]                   # liste des tools autorisés
    model: str = "claude-sonnet-4-5"   # modèle par défaut
    max_tokens: int = 4096
    timeout_seconds: int = 120

Exemples de templates courants :

• researcher — search_web, fetch_url, modèle léger (haiku)
• coder — bash, read_file, write_file, modèle puissant (sonnet/opus)
• reviewer — lecture seule, modèle sonnet
• summarizer — aucun tool, haiku suffit

2. Critères de décision : quel modèle choisir ?

Criticité / Complexité	Modèle recommandé
Analyse simple, résumé	claude-haiku-4
Tâche de code standard	claude-sonnet-4-5
Raisonnement multi-étapes, debugging	claude-opus-4
Réponses temps-réel < 2 s	claude-haiku-4

Règle : ne jamais utiliser opus pour les agents répétitifs à fort volume — le coût est 10–20× celui de haiku.

3. Implémenter la factory

import uuid
from datetime import datetime, timezone

@dataclass
class AgentInstance:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    template_name: str = ""
    status: str = "created"   # created | running | done | failed | terminated
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    parent_id: Optional[str] = None
    result: Optional[str] = None
    error: Optional[str] = None

class AgentFactory:
    _templates: dict[str, AgentTemplate] = {}
    _registry: dict[str, AgentInstance] = {}
    _max_concurrent: int = 10

    @classmethod
    def register_template(cls, t: AgentTemplate) -> None:
        cls._templates[t.name] = t

    @classmethod
    def spawn(cls, template_name: str, context: dict, parent_id: str | None = None) -> AgentInstance:
        running = sum(1 for a in cls._registry.values() if a.status == "running")
        if running >= cls._max_concurrent:
            raise RuntimeError(f"Limite {cls._max_concurrent} agents concurrents atteinte")

        tpl = cls._templates[template_name]
        agent = AgentInstance(template_name=template_name, parent_id=parent_id)
        cls._registry[agent.id] = agent

        enriched_prompt = tpl.system_prompt.format(**context)
        agent.status = "running"
        print(f"[SPAWN] {agent.id} ({template_name}) parent={parent_id} at {agent.created_at.isoformat()}")
        return agent

    @classmethod
    def terminate(cls, agent_id: str, result: str | None = None, error: str | None = None) -> None:
        if a := cls._registry.get(agent_id):
            a.status = "failed" if error else "terminated"
            a.result, a.error = result, error

    @classmethod
    def gc(cls, timeout_s: int = 300) -> list[str]:
        """Libère les agents bloqués en 'running' depuis trop longtemps."""
        now = datetime.now(timezone.utc)
        stale = [
            a.id for a in cls._registry.values()
            if a.status == "running" and (now - a.created_at).total_seconds() > timeout_s
        ]
        for aid in stale:
            cls.terminate(aid, error="GC timeout")
        return stale

    @classmethod
    def active(cls) -> list[AgentInstance]:
        return [a for a in cls._registry.values() if a.status == "running"]