Scaling et architectures multi-agents

Scaling et architectures multi-agents

Votre agent fonctionne pour 10 utilisateurs. Comment le faire tourner pour 10 000 ? Les difficultés ne sont plus les mêmes : ce ne sont pas les instructions qui cèdent, ce sont les limites de débit du fournisseur, la mémoire du serveur et votre facture mensuelle. Dans cette dernière leçon, vous apprendrez à scaler vos agents, à concevoir des architectures multi-agents performantes, et à gérer les contraintes de production à grande échelle.

Architecture de production

Le backend agent avec FastAPI

La première décision structurante concerne le cycle de vie des agents. Construire un Agent à chaque requête gaspille du temps et disperse la configuration ; le lifespan de FastAPI permet de les instancier une fois au démarrage et de les conserver dans un registre indexé par type. La route lit le type demandé, refuse proprement s’il n’existe pas, puis diffuse la réponse en flux : run_streamed émet les deltas au fil de la génération, ce qui fait apparaître le texte immédiatement à l’écran au lieu de laisser l’utilisateur devant un indicateur d’attente pendant huit secondes.

from fastapi import FastAPI, Request, Depends, HTTPException
from fastapi.responses import StreamingResponse
from agents import Agent, Runner, function_tool
from contextlib import asynccontextmanager
import asyncio

# Pool d'agents pré-configurés
agents_registry = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialiser les agents au démarrage
    agents_registry["commercial"] = Agent(
        name="Agent commercial",
        instructions="Vous assistez les clients pour leurs achats.",
        model="gpt-5.6-terra",
        tools=[rechercher_produit, calculer_prix],
    )
    agents_registry["support"] = Agent(
        name="Agent support",
        instructions="Vous faites du support technique.",
        model="gpt-5.6-terra",
        tools=[consulter_ticket, creer_ticket],
    )
    yield
    agents_registry.clear()

app = FastAPI(lifespan=lifespan)

@app.post("/api/chat/{agent_type}")
async def chat(agent_type: str, request: Request):
    agent = agents_registry.get(agent_type)
    if not agent:
        raise HTTPException(404, f"Agent '{agent_type}' non trouvé")

    body = await request.json()

    async def stream():
        result = Runner.run_streamed(agent, body["messages"])
        async for event in result.stream_events():
            if event.type == "raw_response_event":
                if hasattr(event.data, "delta") and event.data.delta:
                    yield f"data: {event.data.delta}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

Gestion de la concurrence

Partager ainsi les objets Agent entre toutes les requêtes est sans danger parce que chaque requête crée un Runner indépendant : l’agent porte la configuration, le Runner porte l’état de l’exécution. Le SDK gère nativement la concurrence, et traiter un lot de messages revient à empiler des coroutines dans asyncio.gather. Le détail qui compte ici est return_exceptions=True : sans lui, une seule requête en échec annule tout le lot. Avec lui, vous récupérez les erreurs comme des valeurs, vous les associez à leur message d’origine et vous traitez les quatre-vingt-dix-neuf résultats valides.

import asyncio
from agents import Agent, Runner

agent = Agent(
    name="Agent concurrent",
    instructions="Vous répondez aux questions.",
    model="gpt-5.6-terra",
)

async def traiter_requetes_paralleles(messages: list[str]):
    """Traite plusieurs requêtes en parallèle."""
    taches = [Runner.run(agent, msg) for msg in messages]
    resultats = await asyncio.gather(*taches, return_exceptions=True)

    for msg, res in zip(messages, resultats):
        if isinstance(res, Exception):
            print(f"Erreur pour '{msg}': {res}")
        else:
            print(f"'{msg}' -> {res.final_output[:50]}...")

    return resultats

Patterns multi-agents

Pattern superviseur

Passé une certaine ampleur, un agent unique doté de quinze tools devient imprévisible : ses instructions gonflent, ses choix se dégradent. Le pattern superviseur découpe le travail entre des agents spécialisés — recherche, analyse, rédaction — et confie à un agent de routage le soin de déléguer par handoff. Chaque spécialiste garde des instructions courtes et un modèle adapté à sa tâche : l’analyse hérite d’un modèle de raisonnement, tandis que le superviseur, qui ne fait que choisir un destinataire, tourne sur un modèle rapide puisque sa décision se joue sur quelques mots.

from agents import Agent, Runner, Handoff

agent_recherche = Agent(
    name="Recherche",
    instructions="Vous recherchez des informations sur le web et dans les bases de données.",
    tools=[WebSearchTool(), rechercher_dans_crm],
    model="gpt-5.6-terra",
)

agent_analyse = Agent(
    name="Analyse",
    instructions="Vous analysez les données et produisez des insights.",
    model="gpt-5.6-sol",
)

agent_redaction = Agent(
    name="Rédaction",
    instructions="Vous rédigez des rapports professionnels en français.",
    model="gpt-5.6-terra",
)

agent_superviseur = Agent(
    name="Superviseur",
    instructions="""Vous coordonnez une équipe d'agents spécialisés.
    Pour chaque demande :
    1. Analysez ce qui est nécessaire
    2. Déléguez aux agents spécialisés via handoff
    3. Le dernier agent produit la réponse finale

    Agents disponibles :
    - Recherche : pour collecter des données
    - Analyse : pour analyser et interpréter
    - Rédaction : pour produire le livrable final""",
    handoffs=[
        Handoff(agent=agent_recherche, description="Collecter des données"),
        Handoff(agent=agent_analyse, description="Analyser des données"),
        Handoff(agent=agent_redaction, description="Rédiger un rapport"),
    ],
    model="gpt-5.6-terra",  # Rapide pour le routage
)

Pattern pipeline

Quand la séquence est connue à l’avance, laisser un modèle décider de l’enchaînement est une dépense inutile. Le pattern pipeline câble les étapes dans votre code : collecte, analyse, rédaction. Une analyse concurrentielle commence ainsi par deux collectes lancées ensemble — le web et le CRM n’ont aucune raison de s’attendre —, puis passe le tout à l’analyse, dont la sortie alimente la rédaction. Vous obtenez un workflow reproductible, traçable étape par étape, et dont la durée totale est celle de la plus lente des collectes plus celle des deux étapes suivantes.

async def pipeline_analyse_concurrentielle(entreprise: str):
    # Étape 1 : Collecte parallèle de données
    tache_web = Runner.run(agent_recherche, f"Chercher des informations sur {entreprise}")
    tache_crm = Runner.run(agent_crm, f"Données CRM pour {entreprise}")
    resultats = await asyncio.gather(tache_web, tache_crm)

    donnees = f"""Données web : {resultats[0].final_output}
    Données CRM : {resultats[1].final_output}"""

    # Étape 2 : Analyse
    analyse = await Runner.run(agent_analyse, f"Analysez ces données : {donnees}")

    # Étape 3 : Rapport
    rapport = await Runner.run(
        agent_redaction,
        f"Rédigez un rapport d'analyse concurrentielle : {analyse.final_output}"
    )

    return rapport.final_output

Pattern map-reduce

Le troisième pattern répond à un problème de volume plutôt que de complexité. Deux mille verbatims clients ne tiennent pas dans une seule requête et ne s’analysent pas séquentiellement en un temps raisonnable. La phase MAP classe chaque feedback en parallèle avec un modèle rapide et une sortie typée ; la phase REDUCE agrège ces classifications, bien plus compactes que les textes d’origine, et confie la synthèse à un modèle plus capable. Le coût s’effondre par rapport à une analyse intégrale, car le gros du volume est traité par le modèle le moins cher.

async def analyser_feedbacks(feedbacks: list[str]):
    # MAP : analyser chaque feedback en parallèle
    agent_classificateur = Agent(
        name="Classificateur",
        instructions="Classifiez le feedback : positif, négatif ou neutre. Identifiez le thème principal.",
        model="gpt-5.6-terra",
        output_type=ClassificationFeedback,
    )

    taches = [Runner.run(agent_classificateur, fb) for fb in feedbacks]
    classifications = await asyncio.gather(*taches)

    # REDUCE : synthétiser les résultats
    resume_data = [
        {"feedback": fb, "class": c.final_output.dict()}
        for fb, c in zip(feedbacks, classifications)
    ]

    agent_synthetiseur = Agent(
        name="Synthétiseur",
        instructions="Synthétisez les classifications de feedbacks en un rapport avec statistiques et insights.",
        model="gpt-5.6-terra",
    )

    import json
    rapport = await Runner.run(
        agent_synthetiseur,
        f"Synthétisez ces {len(feedbacks)} feedbacks classifiés : {json.dumps(resume_data, ensure_ascii=False)}"
    )

    return rapport.final_output

Gestion des rate limits

Lancer deux mille appels simultanés ne les fait pas aboutir plus vite : le fournisseur vous refuse et vous perdez le lot. Un sémaphore borne le nombre d’exécutions concurrentes, et le gestionnaire ci-dessous ajoute un backoff exponentiel plafonné à soixante secondes lorsqu’une erreur de limite survient. Le compteur d’erreurs est remis à zéro dès qu’un appel réussit, ce qui évite de pénaliser durablement le service pour un incident passager.

import asyncio
from agents import Runner

class GestionnaireRateLimits:
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.compteur_erreurs = 0

    async def executer(self, agent, message, **kwargs):
        async with self.semaphore:
            try:
                result = await Runner.run(agent, message, **kwargs)
                self.compteur_erreurs = 0
                return result
            except Exception as e:
                if "rate_limit" in str(e).lower():
                    self.compteur_erreurs += 1
                    attente = min(2 ** self.compteur_erreurs, 60)
                    await asyncio.sleep(attente)
                    return await self.executer(agent, message, **kwargs)
                raise

gestionnaire = GestionnaireRateLimits(max_concurrent=20)

Optimisation des coûts à grande échelle

Reste la facture. Deux leviers la réduisent sans dégrader le service. Le premier consiste à ne pas payer un modèle de raisonnement pour répondre « quels sont vos horaires » : une table de correspondance associe un niveau de complexité au modèle qui convient, du plus économique pour les questions simples au plus capable pour les cas critiques, avec une valeur par défaut pour tout ce qui n’est pas classé. Le second exploite la répétition — sur un support client, les mêmes questions reviennent des centaines de fois par jour, et un cache indexé sur une empreinte du message rend la réponse instantanément et gratuitement. Réservez ce cache aux questions dont la réponse ne dépend ni de l’utilisateur ni de l’heure : une politique de retour se met en cache, un statut de commande jamais.

# Stratégie de modèles par complexité
def choisir_modele(complexite: str) -> str:
    modeles = {
        "simple": "gpt-5.6-terra",      # Questions simples : pas cher
        "moyen": "gpt-5.6-terra",       # La plupart des cas
        "complexe": "gpt-5.6-terra",     # Raisonnement rapide
        "critique": "gpt-5.6-sol",      # Cas critiques : qualité maximale
    }
    return modeles.get(complexite, "gpt-5.6-terra")

# Cache des réponses fréquentes
from functools import lru_cache
import hashlib

reponses_cache = {}

async def executer_avec_cache(agent, message: str):
    cle = hashlib.md5(message.encode()).hexdigest()
    if cle in reponses_cache:
        return reponses_cache[cle]

    result = await Runner.run(agent, message)
    reponses_cache[cle] = result.final_output
    return result.final_output

Points clés à retenir

• Pré-configurez vos agents au démarrage et réutilisez-les entre les requêtes
• Le SDK gère nativement la concurrence via asyncio.gather()
• Le pattern superviseur coordonne des agents spécialisés via handoffs
• Le pattern map-reduce parallélise le traitement de gros volumes
• Gérez les rate limits avec un sémaphore et un backoff exponentiel
• Optimisez les coûts avec le choix de modèle par complexité et le caching

Testez vos connaissances

Avant de passer à l’échelle, un contrôle de l’architecture d’agents.

1. Quelles sont les briques de l'Agents SDK ?

Réponse : Agent (le rôle et ses instructions), Runner (la boucle d’exécution), Tool (les capacités) et Guardrail (les validations d’entrées/sorties) — l’anatomie de tout agent du SDK.

2. À quoi servent les Guardrails ?

Réponse : À valider ce qui entre et sort de l’agent : filtrer les demandes hors périmètre, contrôler les sorties avant qu’elles n’agissent — la sécurité déclarée dans l’architecture, pas espérée du prompt.

3. Qu'est-ce qu'un handoff ?

Réponse : Le transfert d’une conversation d’un agent à un autre, plus spécialisé : l’orchestration multi-agents où chaque agent fait ce qu’il sait faire, avec le contexte qui suit.

4. Pourquoi le tracing est-il indispensable en production ?

Réponse : Parce qu’un agent décide dynamiquement : sans trace des étapes, outils appelés et décisions, impossible de déboguer ou d’évaluer — le trace grading note précisément ces exécutions.

5. Quand monter vers une architecture multi-agents ?

Réponse : Quand un agent unique accumule trop de rôles et d’outils : on spécialise (triage, expertise, exécution) et on orchestre par handoffs — chaque agent redevient simple et testable.

SDK, guardrails, handoffs, tracing : l’échafaudage est complet — le scaling ci-dessus vous dit comment le déployer sans le fragiliser.