Creazione di un sistema multi-agente

1. Introduzione

In questo lab, andrai oltre i semplici chatbot e creerai un sistema multi-agente distribuito.

Anche se un singolo LLM può rispondere alle domande, la complessità del mondo reale spesso richiede ruoli specializzati. Non chiedi al tuo ingegnere backend di progettare la UI e non chiedi al tuo designer di ottimizzare le query del database. Allo stesso modo, possiamo creare agenti AI specializzati che si concentrano su un'attività e si coordinano tra loro per risolvere problemi complessi.

Creerai un sistema di creazione di corsi composto da:

  • Agente ricercatore: utilizza google_search per trovare informazioni aggiornate.
  • Giudice dell'agente: valuta la qualità e la completezza della ricerca.
  • Agente Content Builder: trasforma la ricerca in un corso strutturato.
  • Agente orchestratore: gestione del flusso di lavoro e della comunicazione tra questi specialisti.

Obiettivi didattici

  • Definisci un agente che utilizza strumenti (ricercatore) in grado di effettuare ricerche sul web.
  • Implementa l'output strutturato con Pydantic per il giudice.
  • Connettiti ad agenti remoti utilizzando il protocollo A2A (Agent-to-Agent).
  • Crea un LoopAgent per creare un ciclo di feedback tra il ricercatore e il giudice.
  • Esegui il sistema distribuito localmente utilizzando l'ADK.
  • Esegui il deployment del sistema multi-agente in Google Cloud Run.
  • Utilizza un modello Gemma su una GPU Cloud Run per l'agente di creazione dei contenuti.

Che cosa ti serve

  • Un browser web come Chrome
  • Un progetto cloud Google Cloud con la fatturazione abilitata

2. Principi di architettura e orchestrazione

Innanzitutto, vediamo come funzionano insieme questi agenti. Stiamo creando una pipeline di creazione dei corsi.

Progettazione del sistema

Diagramma dell'architettura

Orchestrazione con gli agenti

Gli agenti standard (come il ricercatore) funzionano. Gli agenti orchestratori (come LoopAgent o SequentialAgent) gestiscono altri agenti. Non hanno strumenti propri, il loro "strumento" è la delega.

  1. LoopAgent: si comporta come un ciclo while nel codice. Esegue una sequenza di agenti ripetutamente finché non viene soddisfatta una condizione (o non viene raggiunto il numero massimo di iterazioni). Utilizziamo questa funzionalità per il ciclo di ricerca:
    • Ricercatore trova informazioni.
    • Judge lo critica.
    • Se Judge indica "Fail", EscalationChecker consente al ciclo di continuare.
    • Se Judge dice "Pass", EscalationChecker interrompe il ciclo.
  2. SequentialAgent: si comporta come una normale esecuzione di script. Esegue gli agenti uno dopo l'altro. Lo utilizziamo per la pipeline di alto livello:
    • Innanzitutto, esegui il ciclo di ricerca (finché non termina con dati validi).
    • Quindi, esegui Content Builder (per scrivere il corso).

Combinando questi elementi, creiamo un sistema solido in grado di autocorreggersi prima di generare l'output finale.

3. Configurazione

Configurazione del progetto

Crea un progetto Google Cloud

  1. Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud.
  2. Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata per un progetto.

Avvia Cloud Shell

Cloud Shell è un ambiente a riga di comando in esecuzione in Google Cloud che viene precaricato con gli strumenti necessari.

  1. Fai clic su Attiva Cloud Shell nella parte superiore della console Google Cloud.
  2. Una volta connesso a Cloud Shell, verifica l'autenticazione:
    gcloud auth list
  3. Verifica che il progetto sia configurato:
    gcloud config get project
  4. Se il progetto non è impostato come previsto, impostalo:
    export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID

Configurazione dell'ambiente

  1. Apri Cloud Shell: fai clic sull'icona Attiva Cloud Shell in alto a destra nella console Google Cloud.

Recuperare il codice iniziale

  1. Clona il repository iniziale nella tua home directory:sposta nella home directory
      cd ~
    Clona solo il codice necessario per questo codelab dalla cartella Google Cloud DevRel Demos.
    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
    Passa alla cartella contenente il codice per questo codelab
    cd multi-agent-system
  2. Abilita le API: esegui il seguente comando per abilitare i servizi Google Cloud necessari:
    gcloud services enable \
    run.googleapis.com \
    artifactregistry.googleapis.com \
    cloudbuild.googleapis.com \
    aiplatform.googleapis.com \
    compute.googleapis.com
  3. Apri questa cartella nell'editor.
    cloudshell edit .

Configura l'ambiente

  1. Configura le variabili di ambiente.Creeremo un file .env per archiviare queste variabili in modo che tu possa ricaricarle facilmente se la sessione si disconnette.
    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. Recupera le variabili di ambiente:
    source .env

4. 🕵️ L'agente di ricerca

Agente di ricerca

Il Ricercatore è uno specialista. Il suo unico compito è trovare informazioni. Per farlo, deve accedere a uno strumento: la Ricerca Google.

Perché separare il ruolo di Ricercatore?

Approfondimento:perché non avere un solo agente che faccia tutto?

Gli agenti piccoli e mirati sono più facili da valutare e debuggare. Se la ricerca non è buona, devi modificare il prompt del ricercatore. Se la formattazione del corso non è corretta, devi apportare modifiche al Content Builder. In un prompt monolitico "fai tutto", la correzione di un elemento spesso ne compromette un altro.

  1. Se lavori in Cloud Shell, esegui questo comando per aprire l'editor di Cloud Shell:
    cloudshell workspace .
  2. Apri agents/researcher/agent.py.
  3. Esamina il seguente codice che definisce l'agente 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

Concetto chiave: utilizzo dello strumento

Nota che passiamo tools=[google_search]. L'ADK gestisce la complessità della descrizione di questo strumento per l'LLM. Quando il modello decide di aver bisogno di informazioni, genera una chiamata di strumento strutturata, l'ADK esegue la funzione Python google_search e restituisce il risultato al modello.

5. ⚖️ L'agente giudice

Judge Agent

Il ricercatore lavora sodo, ma gli LLM possono essere pigri. Abbiamo bisogno di un Giudice per esaminare il lavoro. Il giudice accetta la ricerca e restituisce una valutazione strutturata di superamento/non superamento.

Output strutturato

Approfondimento:per automatizzare i flussi di lavoro, abbiamo bisogno di output prevedibili. Una recensione testuale prolissa è difficile da analizzare a livello di programmazione. Applicando uno schema JSON (utilizzando Pydantic), ci assicuriamo che Judge restituisca un valore booleano pass o fail su cui il nostro codice può agire in modo affidabile.

  1. Apri agents/judge/agent.py.
  2. Esamina il seguente codice che definisce lo schema JudgeFeedback e l'agente 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

Concetto chiave: limitare il comportamento dell'agente

Abbiamo impostato disallow_transfer_to_parent=True e disallow_transfer_to_peers=True. In questo modo, il giudice è obbligato a restituire solo il JudgeFeedback strutturato. Non può decidere di "chattare" con l'utente o delegare un altro agente. Ciò lo rende un componente deterministico nel nostro flusso logico.

6. ✍️ L'agente Content Builder

Content Builder

Content Builder è lo strumento di scrittura creativa. Prende la ricerca approvata e la trasforma in un corso. Utilizza un modello Gemma pubblicato da Cloud Run.

Diamo prima un'occhiata al servizio Cloud Run che ospita il modello

  1. Apri ollama_backend/Dockerfile
  2. Qui puoi vedere come il Dockerfile utilizza un'immagine Ollama, rimane in ascolto delle richieste sulla porta 8080 e archivia il modello richiesto in una cartella /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

⚙️ Durante il deployment, configurerai quanto segue:

  • GPU: NVIDIA L4 scelta per l'eccellente rapporto prezzo/prestazioni per i carichi di lavoro di inferenza. L4 fornisce 24 GB di memoria GPU e operazioni tensoriali ottimizzate, il che lo rende ideale per modelli con 270 milioni di parametri come Gemma
  • Memoria: 16 GB di memoria di sistema per gestire il caricamento dei modelli, le operazioni CUDA e la gestione della memoria di Ollama
  • CPU: 8 core per la gestione ottimale di I/O e attività di pre-elaborazione
  • Concurrency: 4 richieste per istanza bilanciano la velocità effettiva con la memoria utilizzata della GPU
  • Timeout: 600 secondi per il caricamento iniziale del modello e l'avvio del container

Ora esaminiamo l'agente Builder di contenuti che utilizza il modello Gemma.

  1. Apri agents/content_builder/agent.py.
  2. Esamina il seguente codice che definisce l'agente 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

Concetto chiave: propagazione del contesto

Ti starai chiedendo: "Come fa Content Builder a sapere cosa ha trovato il ricercatore?" Nell'ADK, gli agenti in una pipeline condividono un session.state. In un secondo momento, in Orchestrator configureremo Researcher e Judge in modo che salvino i loro output in questo stato condiviso. Il prompt di Content Builder ha effettivamente accesso a questa cronologia.

7. 🎻 L'organizzatore

Agente di orchestrazione

L'Orchestratore è il gestore del nostro team multi-agente. A differenza degli agenti specializzati (Ricercatore, Giudice, Generatore di contenuti) che svolgono attività specifiche, il compito dell'orchestratore è coordinare il workflow e garantire che le informazioni fluiscano correttamente tra loro.

🌐 L'architettura: da agente ad agente (A2A)

Architettura A2A

In questo lab, creeremo un sistema distribuito. Anziché eseguire tutti gli agenti in un unico processo Python, li implementiamo come microservizi indipendenti. In questo modo, ogni agente può scalare in modo indipendente e non funzionare senza arrestare in modo anomalo l'intero sistema.

Per rendere possibile questa operazione, utilizziamo il protocollo Agent-to-Agent (A2A).

Il protocollo A2A

Approfondimento:in un sistema di produzione, gli agenti vengono eseguiti su server diversi (o anche su cloud diversi). Il protocollo A2A crea un modo standard per la scoperta e la comunicazione reciproca tramite HTTP. RemoteA2aAgent è il client ADK per questo protocollo.

  1. Apri agents/orchestrator/agent.py.
  2. Esamina il seguente codice che definisce le connessioni.
    # ... 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. 🛑 Il controllo dell'escalation

Un ciclo deve avere un modo per fermarsi. Se il giudice dice "Passa", vogliamo uscire immediatamente dal ciclo e passare al Content Builder.

Logica personalizzata con BaseAgent

Approfondimento:non tutti gli agenti utilizzano gli LLM. A volte hai bisogno di una semplice logica Python. BaseAgent ti consente di definire un agente che esegue solo codice. In questo caso, controlliamo lo stato della sessione e utilizziamo EventActions(escalate=True) per segnalare a LoopAgent di interrompersi.

  1. Ancora in agents/orchestrator/agent.py.
  2. Rivedi il seguente codice, che esamina il feedback del giudice e procede al passaggio successivo quando è pronto
    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")

Concetto chiave: controllo del flusso tramite eventi

Gli agenti comunicano non solo con il testo, ma anche con gli eventi. Generando un evento con escalate=True, questo agente invia un segnale al relativo elemento principale (LoopAgent). LoopAgent è programmato per intercettare questo segnale e terminare il loop.

9. 🔁 Il ciclo di ricerca

Loop di ricerca

Abbiamo bisogno di un ciclo di feedback: Ricerca -> Giudizio -> (Errore) -> Ricerca -> ...

  1. In agents/orchestrator/agent.py.
  2. Esamina in che modo il seguente codice definisce la definizione 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,
)

Concetto chiave: LoopAgent

LoopAgent scorre in ordine le sub_agents.

  1. researcher: trova i dati.
  2. judge: valuta i dati.
  3. escalation_checker: decide se yield Event(escalate=True). Se si verifica escalate=True, il ciclo si interrompe in anticipo. In caso contrario, riprende dalla ricercatrice (fino a max_iterations).

10. 🔗 La pipeline finale

Pipeline finale

Riepilogo…

  1. In agents/orchestrator/agent.py.
  2. Esamina la definizione di root_agent in fondo al file.
    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],
)

Concetto chiave: composizione gerarchica

Tieni presente che research_loop è a sua volta un agente (un LoopAgent). Lo trattiamo come qualsiasi altro sub-agente in SequentialAgent. Questa componibilità ti consente di creare logiche complesse nidificando pattern semplici (cicli all'interno di sequenze, sequenze all'interno di router e così via).

11. 🚀 Esegui il deployment in Cloud Run

Eseguiremo il deployment di ogni agente come servizio separato su Cloud Run, incluso un servizio Cloud Run per la UI del creatore del corso e un servizio Cloud Run che utilizza le GPU per il modello Gemma.

Informazioni sulla configurazione del deployment

Quando esegui il deployment degli agenti in Cloud Run, passiamo diverse variabili di ambiente per configurare il loro comportamento e la loro connettività:

  • GOOGLE_CLOUD_PROJECT: garantisce che l'agente utilizzi il progetto Google Cloud corretto per il logging e le chiamate Vertex AI.
  • GOOGLE_GENAI_USE_VERTEXAI: indica al framework dell'agente (ADK) di utilizzare Vertex AI per l'inferenza del modello anziché chiamare direttamente le API Gemini.
  • [AGENT]_AGENT_CARD_URL: questo è fondamentale per l'orchestratore. Indica all'agente di orchestrazione dove trovare gli agenti remoti. Se imposti questo valore sull'URL Cloud Run di cui è stato eseguito il deployment (in particolare il percorso della scheda dell'agente), consenti a Orchestrator di rilevare e comunicare con Researcher, Judge e Content Builder su internet.

Per eseguire il deployment di tutti gli agenti nei servizi Cloud Run, esegui questo script.

Innanzitutto, assicurati che lo script sia eseguibile.

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

Nota: l'esecuzione richiederà diversi minuti, poiché il deployment di ogni servizio viene eseguito in sequenza.

~/multi-agent-system/deploy.sh

12. Crea un corso.

Apri il sito web di Course Creator. Il servizio Cloud Run Course Creator è l'ultimo servizio di cui è stato eseguito il deployment dallo script. Puoi identificare l'URL del creatore del corso come https://course-creator-..run.app, che deve essere l'ultima riga di output dello script di deployment.

e digita un'idea per un corso, ad esempio "algebra lineare".

Gli agenti inizieranno a lavorare al tuo corso.

Pipeline finale

13. Elimina

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo codelab, segui questi passaggi per eliminare i servizi e le immagini container.

1. Elimina i servizi Cloud Run

Il modo più efficiente per liberare spazio è eliminare i servizi di cui hai eseguito il deployment su 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. Elimina le immagini di Artifact Registry

Quando hai utilizzato il flag --source per eseguire il deployment, Google Cloud ha creato un repository in Artifact Registry per archiviare le immagini container. Per rimuovere questi elementi e risparmiare sui costi di archiviazione, elimina il repository:

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

3. Rimuovere i file locali e l'ambiente

Per mantenere pulito l'ambiente Cloud Shell, rimuovi la cartella del progetto e qualsiasi configurazione locale:

cd ~
rm -rf multi-agent-system

4. (Facoltativo) Elimina il progetto

Se hai creato un progetto solo per questo codelab, puoi assicurarti che non venga addebitato alcun costo chiudendo il progetto stesso tramite la pagina Gestisci risorse.

14. Complimenti!

Hai creato e implementato correttamente un sistema multi-agente distribuito pronto per la produzione.

Cosa hai realizzato

  • Decomposizione di un'attività complessa: anziché un unico prompt gigante, abbiamo suddiviso il lavoro in ruoli specializzati (Ricercatore, Giudice, Generatore di contenuti).
  • Implementazione del controllo qualità: abbiamo utilizzato un LoopAgent e un Judge strutturato per garantire che solo le informazioni di alta qualità raggiungano la fase finale.
  • Creato per la produzione: utilizzando il protocollo Agent-to-Agent (A2A) e Cloud Run, abbiamo creato un sistema in cui ogni agente è un microservizio indipendente e scalabile. Questo approccio è molto più solido rispetto all'esecuzione di tutto in un unico script Python.
  • Orchestrazione: abbiamo utilizzato SequentialAgent e LoopAgent per definire pattern di flusso di controllo chiari. *. GPU Cloud Run: è stato eseguito il deployment di un modello Gemma su una GPU Cloud Run