Créer un système multi-agent

1. Introduction

Dans cet atelier, vous allez aller au-delà des simples chatbots et créer un système multi-agent distribué.

Bien qu'un seul LLM puisse répondre à des questions, la complexité du monde réel nécessite souvent des rôles spécialisés. Vous ne demandez pas à votre ingénieur backend de concevoir l'UI, ni à votre concepteur d'optimiser les requêtes de base de données. De même, nous pouvons créer des agents IA spécialisés qui se concentrent sur une tâche et se coordonnent entre eux pour résoudre des problèmes complexes.

Vous allez créer un système de création de cours composé des éléments suivants :

  • Agent de recherche : utilise google_search pour trouver des informations à jour.
  • Agent évaluateur : évalue la qualité et l'exhaustivité de la recherche.
  • Agent Content Builder : transformer les recherches en cours structuré.
  • Agent d'orchestration : il gère le workflow et la communication entre ces spécialistes.

Points abordés

  • Définissez un agent utilisant des outils (chercheur) capable d'effectuer des recherches sur le Web.
  • Implémentez une sortie structurée avec Pydantic pour le juge.
  • Connectez-vous à des agents distants à l'aide du protocole Agent-to-Agent (A2A).
  • Construisez un LoopAgent pour créer une boucle de rétroaction entre le chercheur et le juge.
  • Exécutez le système distribué en local à l'aide de l'ADK.
  • Déployez le système multi-agents sur Google Cloud Run.
  • Utilisez un modèle Gemma sur un GPU Cloud Run pour l'agent de création de contenu.

Prérequis

  • Un navigateur Web (par exemple, Chrome)
  • Un projet Google Cloud avec facturation activée

2. Principes d'architecture et d'orchestration

Commençons par comprendre comment ces agents fonctionnent ensemble. Nous allons créer un pipeline de création de cours.

Conception du système

Schéma de l'architecture

Orchestration avec des agents

Les agents standards (comme le chercheur) fonctionnent. Les agents d'orchestration (comme LoopAgent ou SequentialAgent) gèrent d'autres agents. Elles ne disposent pas de leurs propres outils. Leur "outil" est la délégation.

  1. LoopAgent : se comporte comme une boucle while dans le code. Il exécute une séquence d'agents de manière répétée jusqu'à ce qu'une condition soit remplie (ou que le nombre maximal d'itérations soit atteint). Nous l'utilisons pour la boucle de recherche :
    • Un chercheur trouve des informations.
    • Le juge le critique.
    • Si Judge indique "Échec", EscalationChecker permet à la boucle de se poursuivre.
    • Si Judge indique "Pass", EscalationChecker interrompt la boucle.
  2. SequentialAgent : se comporte comme une exécution de script standard. Il exécute les agents les uns après les autres. Nous l'utilisons pour le pipeline de haut niveau :
    • Tout d'abord, exécutez la boucle de recherche (jusqu'à ce qu'elle se termine avec des données de qualité).
    • Exécutez ensuite Content Builder (pour rédiger le cours).

En combinant ces éléments, nous créons un système robuste capable de s'autocorriger avant de générer le résultat final.

3. Configuration

Configuration du projet

Créer un projet Google Cloud

  1. Dans la console Google Cloud, sur la page de sélection du projet, sélectionnez ou créez un projet Google Cloud.
  2. Assurez-vous que la facturation est activée pour votre projet Cloud. Découvrez comment vérifier si la facturation est activée sur un projet.

Démarrer Cloud Shell

Cloud Shell est un environnement de ligne de commande exécuté dans Google Cloud et fourni avec les outils nécessaires.

  1. Cliquez sur Activer Cloud Shell en haut de la console Google Cloud.
  2. Une fois connecté à Cloud Shell, vérifiez votre authentification :
    gcloud auth list
  3. Vérifiez que votre projet est configuré :
    gcloud config get project
  4. Si votre projet n'est pas défini comme prévu, définissez-le :
    export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID

Configuration de l'environnement

  1. Ouvrez Cloud Shell : cliquez sur l'icône Activer Cloud Shell en haut à droite de la console Google Cloud.

Obtenir le code de démarrage

  1. Clonez le dépôt de démarrage dans votre répertoire d'accueil : accédez à votre répertoire d'accueil.
      cd ~
    Clonez uniquement le code nécessaire pour cet atelier de programmation à partir du dossier de démonstrations Google Cloud DevRel.
    git clone --depth 1 --filter=blob:none --sparse https://github.com/GoogleCloudPlatform/devrel-demos.git temp-repo && cd temp-repo && git sparse-checkout set agents/multi-agent-system && cd .. && mv temp-repo/agents/multi-agent-system . && rm -rf temp-repo
    Accédez au dossier contenant le code de cet atelier de programmation.
    cd multi-agent-system
  2. Activez les API : exécutez la commande suivante pour activer les services Google Cloud nécessaires :
    gcloud services enable \
    run.googleapis.com \
    artifactregistry.googleapis.com \
    cloudbuild.googleapis.com \
    aiplatform.googleapis.com \
    compute.googleapis.com
  3. Ouvrez ce dossier dans votre éditeur.
    cloudshell edit .

Configurer l'environnement

  1. Configurez les variables d'environnement.Nous allons créer un fichier .env pour stocker ces variables afin que vous puissiez les recharger facilement si votre session est déconnectée.
    cat <<EOF > .env
export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project)
export GOOGLE_CLOUD_LOCATION=europe-west4
export GOOGLE_GENAI_USE_VERTEXAI=true
EOF
  2. Définissez les variables d'environnement :
    source .env

4. 🕵️ L'agent Researcher

Agent de recherche

L'assistant Recherche est un spécialiste. Son seul rôle est de trouver des informations. Pour ce faire, elle a besoin d'accéder à un outil : la recherche Google.

Pourquoi séparer le rôle de chercheur ?

En savoir plus : pourquoi ne pas confier toutes les tâches à un seul agent ?

Les petits agents ciblés sont plus faciles à évaluer et à déboguer. Si la recherche est mauvaise, vous itérez sur la requête du chercheur. Si la mise en forme du cours est incorrecte, vous devez itérer sur le générateur de contenu. Dans une requête monolithique "tout-en-un", corriger un élément en casse souvent un autre.

  1. Si vous travaillez dans Cloud Shell, exécutez la commande suivante pour ouvrir l'éditeur Cloud Shell :
    cloudshell workspace .
  2. Ouvrez agents/researcher/agent.py.
  3. Examinez le code suivant qui définit l'agent researcher :
    # ... existing imports ...

# Define the Researcher Agent
researcher = Agent(
    name="researcher",
    model=MODEL,
    description="Gathers information on a topic using Google Search.",
    instruction="""
    You are an expert researcher. Your goal is to find comprehensive and accurate information on the user's topic.
    Use the `google_search` tool to find relevant information.
    Summarize your findings clearly.
    If you receive feedback that your research is insufficient, use the feedback to refine your next search.
    """,
    tools=[google_search],
)

root_agent = researcher

Concept clé : utilisation d'outils

Notez que nous transmettons tools=[google_search]. L'ADK gère la complexité de la description de cet outil au LLM. Lorsque le modèle décide qu'il a besoin d'informations, il génère un appel d'outil structuré. L'ADK exécute la fonction Python google_search et renvoie le résultat au modèle.

5. ⚖️ L'agent Judge

Agent de jugement

Le chercheur travaille dur, mais les LLM peuvent être paresseux. Nous avons besoin d'un juge pour examiner le travail. Le juge accepte la recherche et renvoie une évaluation structurée (réussite/échec).

Sortie structurée

En savoir plus : Pour automatiser les workflows, nous avons besoin de résultats prévisibles. Il est difficile d'analyser un avis textuel décousu de manière programmatique. En appliquant un schéma JSON (à l'aide de Pydantic), nous nous assurons que le Judge renvoie une valeur booléenne pass ou fail sur laquelle notre code peut s'appuyer de manière fiable.

  1. Ouvrez agents/judge/agent.py.
  2. Examinez le code suivant qui définit le schéma JudgeFeedback et l'agent judge.
    # 1. Define the Schema
class JudgeFeedback(BaseModel):
    """Structured feedback from the Judge agent."""
    status: Literal["pass", "fail"] = Field(
        description="Whether the research is sufficient ('pass') or needs more work ('fail')."
    )
    feedback: str = Field(
        description="Detailed feedback on what is missing. If 'pass', a brief confirmation."
    )

# 2. Define the Agent
judge = Agent(
    name="judge",
    model=MODEL,
    description="Evaluates research findings for completeness and accuracy.",
    instruction="""
    You are a strict editor.
    Evaluate the 'research_findings' against the user's original request.
    If the findings are missing key info, return status='fail'.
    If they are comprehensive, return status='pass'.
    """,
    output_schema=JudgeFeedback,
    # Disallow delegation because it should only output the schema
    disallow_transfer_to_parent=True,
    disallow_transfer_to_peers=True,
)

root_agent = judge

Concept clé : restreindre le comportement de l'agent

Nous avons défini disallow_transfer_to_parent=True et disallow_transfer_to_peers=True. Cela oblige le juge à renvoyer uniquement le JudgeFeedback structuré. Il ne peut pas décider de "discuter" avec l'utilisateur ni déléguer à un autre agent. Il s'agit donc d'un composant déterministe dans notre flux logique.

6. ✍️ L'agent Content Builder

Générateur de contenu

Le Générateur de contenu est le rédacteur créatif. Il transforme les recherches approuvées en cours. Il utilise un modèle Gemma diffusé par Cloud Run.

Commençons par examiner le service Cloud Run qui héberge le modèle.

  1. Ouvrir ollama_backend/Dockerfile
  2. Vous pouvez voir ici comment le Dockerfile utilise une image Ollama, écoute les requêtes sur le port 8080 et stocke le modèle demandé dans un dossier /model.
FROM ollama/ollama:latest

# Listen on all interfaces, port 8080 (Cloud Run default)
ENV OLLAMA_HOST 0.0.0.0:8080

# Store model weight files in /models
ENV OLLAMA_MODELS /models

⚙️ Lorsque vous déployez, vous définissez les configurations suivantes :

  • GPU : NVIDIA L4 choisi pour son excellent rapport prix/performances pour les charges de travail d'inférence. Le L4 fournit 24 Go de mémoire GPU et des opérations de tenseur optimisées, ce qui le rend idéal pour les modèles à 270 millions de paramètres comme Gemma.
  • Mémoire : 16 Go de mémoire système pour gérer le chargement des modèles, les opérations CUDA et la gestion de la mémoire d'Ollama
  • Processeur : 8 cœurs pour une gestion optimale des E/S et des tâches de prétraitement
  • Simultanéité : quatre requêtes par instance permettent d'équilibrer le débit et l'utilisation de la mémoire GPU.
  • Délai avant expiration : 600 secondes pour permettre le chargement initial du modèle et le démarrage du conteneur

Examinons maintenant l'agent Content Builder qui utilise le modèle Gemma.

  1. Ouvrez agents/content_builder/agent.py.
  2. Examinez le code suivant qui définit l'agent content_builder.
# the `ollama-gemma-gpu` Cloud Run service URL which hosts the Gemma model
target_url = os.environ.get("OLLAMA_API_BASE")

# ... existing code ...

# (Note: We use 'ollama/gemma3:270m' to align with ADK's expected prefix)
gemma_model_name = os.environ.get("GEMMA_MODEL_NAME", "gemma3:270m")
model = LiteLlm(
    model=f"ollama_chat/{gemma_model_name}",
    api_base=target_url
)

# 5. Define the Agent
content_builder = Agent(
    name="content_builder",
    model=model,
    description="Transforms research findings into a structured course.",
    instruction="""
    You are an expert course creator.
    Take the approved 'research_findings' and transform them into a well-structured, engaging course module.

    **Formatting Rules:**
    1. Start with a main title using a single `#` (H1).
    2. Use `##` (H2) for main section headings. These will be used for the Table of Contents.
    3. Use `###` (H3) for sub-sections within main sections.
    4. Use bullet points and clear paragraphs.
    5. Maintain a professional but engaging tone.

    **Structure Requirements:**
    - Begin with a brief Introduction section explaining what the learner will gain.
    - Organize content into 3-5 main sections with clear headings.
    - Include Key Takeaways at the end as a bulleted summary.
    - Keep each section focused and concise.

    Ensure the content directly addresses the user's original request.
    Do not include any preamble or explanation outside the course content itself.
    """,
)

root_agent = content_builder

Concept clé : propagation du contexte

Vous vous demandez peut-être : "Comment le générateur de contenu sait-il ce que le chercheur a trouvé ?" Dans l'ADK, les agents d'un pipeline partagent un session.state. Plus tard, dans l'orchestrateur, nous configurerons le chercheur et le juge pour qu'ils enregistrent leurs résultats dans cet état partagé. Le prompt du générateur de contenu a effectivement accès à cet historique.

7. 🎻 L'outil d'orchestration

Agent Orchestrator

L'orchestrateur est le gestionnaire de notre équipe multi-agents. Contrairement aux agents spécialisés (Chercheur, Juge, Créateur de contenu) qui effectuent des tâches spécifiques, le rôle de l'Orchestrateur est de coordonner le workflow et de s'assurer que les informations circulent correctement entre eux.

🌐 Architecture : Agent-to-Agent (A2A)

Architecture A2A

Dans cet atelier, nous allons créer un système distribué. Au lieu d'exécuter tous les agents dans un seul processus Python, nous les déployons en tant que microservices indépendants. Chaque agent peut ainsi évoluer indépendamment et échouer sans faire planter l'ensemble du système.

Pour ce faire, nous utilisons le protocole Agent-to-Agent (A2A).

Protocole A2A

En savoir plus : dans un système de production, les agents s'exécutent sur différents serveurs (voire différents clouds). Le protocole A2A leur permet de se découvrir et de communiquer entre eux via HTTP de manière standardisée. RemoteA2aAgent est le client ADK pour ce protocole.

  1. Ouvrez agents/orchestrator/agent.py.
  2. Examinez le code suivant qui définit les connexions.
    # ... existing code ...

# Connect to the Researcher (Localhost port 8001)
researcher_url = os.environ.get("RESEARCHER_AGENT_CARD_URL", "http://localhost:8001/a2a/agent/.well-known/agent-card.json")
researcher = RemoteA2aAgent(
    name="researcher",
    agent_card=researcher_url,
    description="Gathers information using Google Search.",
    # IMPORTANT: Save the output to state for the Judge to see
    after_agent_callback=create_save_output_callback("research_findings"),
    # IMPORTANT: Use authenticated client for communication
    httpx_client=create_authenticated_client(researcher_url)
)

# Connect to the Judge (Localhost port 8002)
judge_url = os.environ.get("JUDGE_AGENT_CARD_URL", "http://localhost:8002/a2a/agent/.well-known/agent-card.json")
judge = RemoteA2aAgent(
    name="judge",
    agent_card=judge_url,
    description="Evaluates research.",
    after_agent_callback=create_save_output_callback("judge_feedback"),
    httpx_client=create_authenticated_client(judge_url)
)

# Content Builder (Localhost port 8003)
content_builder_url = os.environ.get("CONTENT_BUILDER_AGENT_CARD_URL", "http://localhost:8003/a2a/agent/.well-known/agent-card.json")
content_builder = RemoteA2aAgent(
    name="content_builder",
    agent_card=content_builder_url,
    description="Builds the course.",
    httpx_client=create_authenticated_client(content_builder_url)
)

8. 🛑 Vérificateur d'escalade

Une boucle doit pouvoir s'arrêter. Si le Juge dit "Pass", nous voulons sortir immédiatement de la boucle et passer au Content Builder.

Logique personnalisée avec BaseAgent

En savoir plus : Tous les agents n'utilisent pas de LLM. Parfois, vous avez besoin d'une logique Python simple. BaseAgent vous permet de définir un agent qui exécute uniquement du code. Dans ce cas, nous vérifions l'état de la session et utilisons EventActions(escalate=True) pour signaler l'arrêt de LoopAgent.

  1. Toujours dans agents/orchestrator/agent.py.
  2. Examinez le code suivant, qui passe en revue les commentaires du juge et passe à l'étape suivante lorsque vous êtes prêt.
    class EscalationChecker(BaseAgent):
    """Checks the judge's feedback and escalates (breaks the loop) if it passed."""

    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        # Retrieve the feedback saved by the Judge
        feedback = ctx.session.state.get("judge_feedback")
        print(f"[EscalationChecker] Feedback: {feedback}")

        # Check for 'pass' status
        is_pass = False
        if isinstance(feedback, dict) and feedback.get("status") == "pass":
            is_pass = True
        # Handle string fallback if JSON parsing failed
        elif isinstance(feedback, str) and '"status": "pass"' in feedback:
            is_pass = True

        if is_pass:
            # 'escalate=True' tells the parent LoopAgent to stop looping
            yield Event(author=self.name, actions=EventActions(escalate=True))
        else:
            # Continue the loop
            yield Event(author=self.name)

escalation_checker = EscalationChecker(name="escalation_checker")

Concept clé : flux de contrôle via les événements

Les agents communiquent non seulement par texte, mais aussi par le biais d'événements. En générant un événement avec escalate=True, cet agent envoie un signal à son parent (le LoopAgent). Le LoopAgent est programmé pour intercepter ce signal et mettre fin à la boucle.

9. 🔁 La boucle de recherche

Boucle de recherche

Nous avons besoin d'une boucle de rétroaction : Recherche > Évaluation > (Échec) > Recherche > …

  1. Dans agents/orchestrator/agent.py.
  2. Examinez comment le code suivant définit la définition research_loop.
    research_loop = LoopAgent(
    name="research_loop",
    description="Iteratively researches and judges until quality standards are met.",
    sub_agents=[researcher, judge, escalation_checker],
    max_iterations=3,
)

Concept clé : LoopAgent

Le LoopAgent parcourt ses sub_agents dans l'ordre.

  1. researcher : recherche des données.
  2. judge : évalue les données.
  3. escalation_checker : détermine s'il faut yield Event(escalate=True). Si escalate=True se produit, la boucle s'arrête prématurément. Sinon, il redémarre au niveau du chercheur (jusqu'à max_iterations).

10. 🔗 Pipeline final

Pipeline final

Pour résumer…

  1. Dans agents/orchestrator/agent.py.
  2. Examinez la façon dont root_agent est défini au bas du fichier.
    root_agent = SequentialAgent(
    name="course_creation_pipeline",
    description="A pipeline that researches a topic and then builds a course from it.",
    sub_agents=[research_loop, content_builder],
)

Concept clé : composition hiérarchique

Notez que research_loop est lui-même un agent (un LoopAgent). Nous le traitons comme n'importe quel autre sous-agent dans SequentialAgent. Cette composabilité vous permet de créer une logique complexe en imbriquant des modèles simples (boucles dans des séquences, séquences dans des routeurs, etc.).

11. 🚀 Déployer sur Cloud Run

Nous allons déployer chaque agent en tant que service distinct sur Cloud Run, y compris un service Cloud Run pour l'UI du créateur de cours et un service Cloud Run utilisant des GPU pour le modèle Gemma.

Comprendre la configuration du déploiement

Lorsque vous déployez des agents sur Cloud Run, nous transmettons plusieurs variables d'environnement pour configurer leur comportement et leur connectivité :

  • GOOGLE_CLOUD_PROJECT : garantit que l'agent utilise le bon projet Google Cloud pour la journalisation et les appels Vertex AI.
  • GOOGLE_GENAI_USE_VERTEXAI : indique au framework de l'agent (ADK) d'utiliser Vertex AI pour l'inférence du modèle au lieu d'appeler directement les API Gemini.
  • [AGENT]_AGENT_CARD_URL : cette valeur est essentielle pour l'orchestrateur. Il indique à l'orchestrateur où trouver les agents distants. En définissant cette valeur sur l'URL Cloud Run déployée (plus précisément le chemin d'accès à la fiche de l'agent), nous permettons à l'orchestrateur de découvrir le chercheur, le juge et le créateur de contenu, et de communiquer avec eux sur Internet.

Pour déployer tous les agents sur les services Cloud Run, exécutez le script suivant.

Tout d'abord, assurez-vous que le script est exécutable.

chmod u+x ~/multi-agent-system/deploy.sh

Remarque : L'exécution de cette commande prendra plusieurs minutes, car chaque service est déployé séquentiellement.

~/multi-agent-system/deploy.sh

12. Créez un cours !

Ouvrez le site Web Course Creator. Le service Course Creator Cloud Run est le dernier service déployé à partir du script. Vous pouvez identifier l'URL du créateur du cours comme https://course-creator-..run.app, qui devrait être la dernière ligne de sortie du script de déploiement.

Saisissez une idée de cours, par exemple "algèbre linéaire".

Vos agents commenceront à travailler sur votre cours.

Pipeline final

13. Effectuer un nettoyage

Pour éviter que les ressources utilisées dans cet atelier de programmation soient facturées sur votre compte Google Cloud, suivez ces étapes pour supprimer vos services et vos images de conteneurs.

1. Supprimer des services Cloud Run

Le moyen le plus efficace de nettoyer votre environnement consiste à supprimer les services que vous avez déployés sur Cloud Run.

# Delete the main agent and app services
gcloud run services delete researcher content-builder judge orchestrator course-creator \
    --region $REGION --quiet

# Delete the GPU backend (Ollama)
gcloud run services delete ollama-gemma-gpu \
    --region $OLLAMA_REGION --quiet

2. Supprimer des images Artifact Registry

Lorsque vous avez utilisé l'indicateur --source pour déployer, Google Cloud a créé un dépôt dans Artifact Registry pour stocker vos images de conteneurs. Pour les supprimer et économiser sur les coûts de stockage, supprimez le dépôt :

gcloud artifacts repositories delete cloud-run-source-deploy --location us-east4 --quiet

3. Supprimer les fichiers et l'environnement locaux

Pour que votre environnement Cloud Shell reste propre, supprimez le dossier du projet et toute configuration locale :

cd ~
rm -rf multi-agent-system

4. (Facultatif) Supprimer le projet

Si vous avez créé un projet uniquement pour cet atelier de programmation, vous pouvez vous assurer qu'aucun autre frais ne sera facturé en arrêtant le projet lui-même sur la page Gérer les ressources.

14. Félicitations !

Vous avez réussi à créer et à déployer un système multi-agents distribué prêt pour la production.

Ce que vous avez accompli

  • Décomposition d'une tâche complexe : au lieu d'un seul prompt géant, nous avons réparti le travail en rôles spécialisés (chercheur, juge, créateur de contenu).
  • Contrôle qualité mis en place : nous avons utilisé un LoopAgent et un Judge structuré pour nous assurer que seules des informations de haute qualité atteignent l'étape finale.
  • Conçu pour la production : en utilisant le protocole Agent-to-Agent (A2A) et Cloud Run, nous avons créé un système dans lequel chaque agent est un microservice indépendant et évolutif. Cette approche est beaucoup plus robuste que d'exécuter l'ensemble du processus dans un seul script Python.
  • Orchestration : nous avons utilisé SequentialAgent et LoopAgent pour définir des modèles de flux de contrôle clairs. *. GPU Cloud Run : déploiement d'un modèle Gemma sur un GPU Cloud Run