Way Back Home: Sistema multiagente bidireccional en vivo

1. La misión

Historia

Estás a la deriva en la silenciosa e inexplorada extensión del espacio. Un enorme Pulso Solar atravesó tu nave por una grieta dimensional y te dejó varado en un rincón del universo que no aparece en ningún mapa estelar.

Después de días de reparaciones agotadoras, finalmente regresa el zumbido familiar de los motores. Tu nave espacial está operativa. Incluso lograste establecer un enlace ascendente de largo alcance con la nave nodriza. La partida es inminente. Ya puedes irte a casa.

Pero, mientras te preparas para activar la unidad de salto, una señal de socorro interrumpe la estática. Tus sensores detectan una solicitud de ayuda de un planeta designado "Ozymandias". Los sobrevivientes están atrapados en este mundo moribundo, y su nave está varada. Tu misión es fundamental: debes rescatarlos antes de que colapse la atmósfera del planeta.

Su único medio de escape es un cohete antiguo y abandonado construido con tecnología alienígena. Si bien funciona, su unidad de distorsión está destrozada. Para salvar a los sobrevivientes, debes conectarte de forma remota a su banco de trabajo volátil y ensamblar manualmente una unidad de reemplazo.

El desafío

No tienes experiencia con esta tecnología alienígena, que es notoriamente frágil. Un componente desestabilizado puede convertirse en un peligro radioactivo en segundos. Tienes un intento para operar el banco de trabajo volátil. Tu asistente de IA actual tiene dificultades para procesar datos visuales y manuales técnicos de forma simultánea, lo que genera instrucciones alucinatorias y advertencias de peligro omitidas.

Para tener éxito, debes actualizar tu IA de una entidad monolítica a un sistema multiagente colaborativo.

Tus objetivos de la misión:

Ensambla el Warp Drive siguiendo las instrucciones especializadas en tiempo real de tu nuevo sistema multiagente.

Misión Alpha

Qué compilarás

Descripción general

  • Un sistema de IA multiagente bidireccional en tiempo real que incluye un agente de envío central que administra la interacción del usuario y se coordina con agentes especializados.
  • Un agente de arquitectura que se conecta a una base de datos de Redis para recuperar y entregar datos esquemáticos.
  • Un Monitor de seguridad proactivo que usa herramientas de transmisión para analizar un video en vivo en busca de peligros visuales y activar alertas en tiempo real.
  • Un frontend basado en React que proporciona una interfaz de usuario para interactuar con el sistema y transmitir video y audio a los agentes de backend.

Qué aprenderás

Tecnología o concepto

Descripción

Kit de desarrollo de agentes (ADK) de Google

Usarás el ADK para compilar, probar y administrar los agentes, y aprovecharás su framework para controlar la comunicación en tiempo real, la integración de herramientas y el ciclo de vida del agente.

Transmisión bidireccional (Bidi)

Implementarás un agente de transmisión bidireccional que permita una comunicación bidireccional natural y de baja latencia, lo que permitirá que tanto los humanos como la IA interrumpan y respondan en tiempo real.

Sistemas multiagente

Aprenderás a diseñar un sistema de IA distribuido en el que un agente principal delega tareas a agentes especializados, lo que permite una separación de responsabilidades y una arquitectura más escalable.

Protocolo Agent-to-Agent (A2A)

Usarás el protocolo A2A para habilitar la comunicación entre el agente de envío y el agente de arquitecto, lo que les permitirá descubrir las capacidades del otro y realizar intercambios de datos.

Herramientas de transmisión

Implementarás una herramienta de transmisión que actúa como un proceso en segundo plano y analiza continuamente un feed de video para supervisar los cambios de estado (peligros) y generar resultados de forma proactiva.

Google Cloud Run y Memorystore

Implementarás toda la aplicación de varios agentes en un entorno de producción con Cloud Run para alojar los servicios de agentes y Memorystore (Redis) como la base de datos persistente.

FastAPI y WebSockets

El backend se compila con FastAPI y WebSockets para controlar la comunicación en tiempo real y de alto rendimiento que se requiere para transmitir audio, video y respuestas del agente.

Frontend de React

Trabajarás con un frontend basado en React que captura y transmite contenido multimedia del usuario (audio o video) y muestra las respuestas en tiempo real de los agentes de IA.

2. Configura tu entorno

Accede a Cloud Shell

👉 Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud (es el ícono con forma de terminal en la parte superior del panel de Cloud Shell). cloud-shell.png

👉 Haz clic en el botón "Abrir editor" (parece una carpeta abierta con un lápiz). Se abrirá el editor de código de Cloud Shell en la ventana. Verás un explorador de archivos en el lado izquierdo. open-editor.png

👉Abre la terminal en el IDE de Cloud.

03-05-new-terminal.png

👉💻 En la terminal, verifica que ya te autenticaste y que el proyecto está configurado con tu ID del proyecto usando el siguiente comando:

gcloud auth list

Deberías ver tu cuenta como (ACTIVE).

Requisitos previos

ℹ️ El nivel 0 es opcional (pero recomendado)

Puedes completar esta misión sin el nivel 0, pero terminarla primero ofrece una experiencia más inmersiva, ya que te permite ver cómo se ilumina tu baliza en el mapa global a medida que avanzas.

Configura el entorno del proyecto

De vuelta en tu terminal, finaliza la configuración estableciendo el proyecto activo y habilitando los servicios de Google Cloud requeridos (Cloud Run, Vertex AI, etcétera).

👉💻 En tu terminal, establece el ID del proyecto:

gcloud config set project $(cat ~/project_id.txt) --quiet

👉💻 Habilita los servicios obligatorios:

gcloud services enable  compute.googleapis.com \
                        artifactregistry.googleapis.com \
                        run.googleapis.com \
                        cloudbuild.googleapis.com \
                        iam.googleapis.com \
                        aiplatform.googleapis.com \
                        cloudresourcemanager.googleapis.com \
                        redis.googleapis.com \
                        vpcaccess.googleapis.com

Instala las dependencias

👉💻 Navega al nivel 4 e instala los paquetes de Python necesarios:

cd $HOME/way-back-home/level_4
uv sync

Las dependencias clave son las siguientes:

Paquete

Objetivo

fastapi

Framework web de alto rendimiento para la estación satelital y la transmisión de SSE

uvicorn

Servidor ASGI necesario para ejecutar la aplicación de FastAPI

google-adk

El Kit de desarrollo de agentes que se usó para compilar el agente de formación

a2a-sdk

Biblioteca de protocolos Agent-to-Agent para la comunicación estandarizada

google-genai

Cliente nativo para acceder a los modelos de Gemini

redis

Cliente de Python para conectarse a Schematic Vault (Memorystore)

websockets

Compatibilidad con la comunicación bidireccional en tiempo real

python-dotenv

Administra las variables de entorno y los secretos de configuración

pydantic

Validación de datos y administración de la configuración

Verifica la configuración

Antes de comenzar con el código, asegurémonos de que todos los sistemas funcionen correctamente. Ejecuta la secuencia de comandos de verificación para auditar tu proyecto de Google Cloud, las APIs y las dependencias de Python.

👉💻 Ejecuta la secuencia de comandos de verificación:

cd $HOME/way-back-home/level_4/scripts
chmod +x verify_setup.sh
. verify_setup.sh

👀 Deberías ver una serie de íconos de verificación de color verde (✅).

  • Si ves Cruces rojas (❌), sigue los comandos de corrección sugeridos en el resultado (p.ej., gcloud services enable ... o pip install ...).
  • Nota: Por el momento, es aceptable una advertencia amarilla para .env. Crearemos ese archivo en el siguiente paso.
🚀 Verifying Mission Bravo (Level 4) Infrastructure...

✅ Google Cloud Project: xxxxxxx
✅ Cloud APIs: Active
✅ Python Environment: Ready

🎉 SYSTEMS ONLINE. READY FOR MISSION.

3. Cómo crear Schematic Vault en Redis y el agente bidireccional con el ADK

Encontraste el repositorio esquemático planetario que contiene los planos de la cohete abandonada. Para recuperar estos datos con precisión, debes interactuar con la interfaz de administración dedicada del repositorio: el agente de Architect.

Descripción general

Aprovisionamiento de Schematic Vault (Redis)

Antes de que el arquitecto pueda ayudarnos, debemos asegurarnos de que los datos estén alojados en un entorno seguro y de alta disponibilidad. Usaremos Redis como un almacén de datos rápido para nuestros esquemas de alienígenas. Para facilitar el desarrollo, iniciaremos una instancia local de Redis, pero más adelante se proporcionarán instrucciones para implementar en un entorno de producción con Google Cloud Memorystore.

👉💻 Ejecuta los siguientes comandos en tu terminal para aprovisionar la instancia de Redis (esto puede tardar de 2 a 3 minutos):

docker run -d --name ozymandias-vault -p 6379:6379 redis:8.6-rc1-alpine

👉💻 Para cargar los datos preliminares, ejecuta el siguiente comando para ingresar a Redis Shell:

docker exec -it ozymandias-vault redis-cli

(Tu instrucción cambiará a 127.0.0.1:6379).

👉💻 Pega estos comandos dentro:

RPUSH "HYPERION-X" "Warp Core" "Flux Pipe" "Ion Thruster"
RPUSH "NOVA-V" "Ion Thruster" "Warp Core" "Flux Pipe"
RPUSH "OMEGA-9" "Flux Pipe" "Ion Thruster" "Warp Core"
RPUSH "GEMINI-MK1" "Coolant Tank" "Servo" "Fuel Cell"
RPUSH "APOLLO-13" "Warp Core" "Coolant Tank" "Ion Thruster"
RPUSH "VORTEX-7" "Quantum Cell" "Graviton Coil" "Plasma Injector"
RPUSH "CHRONOS-ALPHA" "Shield Emitter" "Data Crystal" "Quantum Cell"
RPUSH "NEBULA-Z" "Plasma Injector" "Flux Pipe" "Graviton Coil"
RPUSH "PULSAR-B" "Data Crystal" "Servo" "Shield Emitter"
RPUSH "TITAN-PRIME" "Ion Thruster" "Quantum Cell" "Warp Core"

👉💻 Escribe exit para volver a tu shell normal.

👉💻 Para verificar que los datos existen consultando un envío específico directamente desde tu terminal, ejecuta lo siguiente:

# Check 'TITAN-PRIME'
docker exec ozymandias-vault redis-cli LRANGE "TITAN-PRIME" 0 -1

👀 Este es el resultado esperado:

1) "Ion Thruster"
2) "Quantum Cell"
3) "Warp Core"

Implementa el agente de Architect

El agente de arquitectura es un agente especializado responsable de recuperar planos esquemáticos de nuestra bóveda de Redis. Actúa como una interfaz de datos dedicada, lo que garantiza que el agente de envío principal reciba información precisa y estructurada sin necesidad de conocer la lógica de la base de datos subyacente.

Descripción general

El Kit de desarrollo de agentes (ADK) de Google es el framework modular que hace posible esta configuración multiagente. Controla dos capas críticas:

  1. Ciclo de vida de la conexión y la sesión: La interacción con las APIs en tiempo real requiere una administración compleja de protocolos, que incluye el manejo de negociaciones, autenticación y señales de mantenimiento de la conexión.
  2. Llamadas a funciones: Este es el "viaje de ida y vuelta del modelo-código-modelo". Cuando el LLM decide que necesita datos, genera una llamada a función estructurada. El ADK intercepta esto, ejecuta tu código de Python (lookup_schematic_tool) y devuelve el resultado al contexto del modelo en milisegundos.

Ahora, compilaremos el Arquitecto. Este agente no tiene acceso a la cámara. Existe únicamente para recibir un "Nombre de unidad" y devolver la "Lista de piezas" de la base de datos.

👉💻 Usaremos el comando adk create. Esta es una herramienta del Kit de desarrollo de agentes (ADK) que genera automáticamente el código estándar y la estructura de archivos para un agente nuevo, lo que nos ahorra tiempo de configuración.

cd $HOME/way-back-home/level_4/backend/
uv run adk create architect_agent

Configura el agente

La CLI iniciará un asistente de configuración interactivo. Usa las siguientes respuestas para configurar tu agente:

  1. Elige un modelo: Selecciona la opción 1 (Gemini Flash).
    • Nota: La versión específica (p.ej., 2.5 y 3.0) pueden variar según la disponibilidad. Siempre elige la variante "Flash" para obtener mayor velocidad.
  2. Elige un backend: Selecciona la opción 2 (Vertex AI).
  3. Enter Google Cloud Project ID: Presiona Intro para aceptar el valor predeterminado (detectado en tu entorno).
  4. Enter Google Cloud Region: Presiona Intro para aceptar el valor predeterminado (us-central1).

👀 La interacción con la terminal debería ser similar a la siguiente:

(way-back-home) user@cloudshell:~/way-back-home/level_4/agent$ adk create architect_agent

Choose a model for the root agent:
1. gemini-2.5-flash
2. Other models (fill later)
Choose model (1, 2): 1

1. Google AI
2. Vertex AI
Choose a backend (1, 2): 2

You need an existing Google Cloud account and project...
Enter Google Cloud project ID [your-project-id]: <PRESS ENTER>
Enter Google Cloud region [us-central1]: <PRESS ENTER>

Agent created in /home/user/way-back-home/level_4/agent/architect_agent:
- .env
- __init__.py
- agent.py

Ahora deberías ver un mensaje de éxito Agent created. Esto genera el código de esqueleto que modificaremos en el siguiente paso.

👉✏️ Navega al archivo $HOME/way-back-home/level_4/backend/architect_agent/agent.py que acabas de crear y ábrelo en tu editor. Agrega el fragmento de la herramienta al archivo después de la primera línea de importación:

import os
import redis

REDIS_IP = os.environ.get('REDIS_HOST', 'localhost')
r = redis.Redis(host=REDIS_IP, port=6379, decode_responses=True)

def lookup_schematic_tool(drive_name: str) -> list[str]:
    """Returns the ordered list of parts for a drive from local Redis."""
    
    # Logic to clean input like "TARGET: X" -> "X"
    clean_name = drive_name.replace("TARGET:", "").replace("TARGET", "").strip()
    clean_name = clean_name.replace(":", "").strip()
    
    # LRANGE gets all items in the list (index 0 to -1)
    result = r.lrange(clean_name, 0, -1)
    
    if not result:
        print(f"[ARCHITECT] Error: Drive ID '{clean_name}' not found in Redis.")
        return ["ERROR: Drive ID not found."]
    
    print(f"[ARCHITECT] Returning schematic for {clean_name}: {result}")
    return result

👉✏️ Reemplaza toda la línea instruction en la definición de root_agent por lo siguiente y, además, agrega la herramienta que definimos antes:

    instruction='''SYSTEM ROLE: Database API.
    INPUT: Text string (Drive Name).
    TASK: Run `lookup_schematic_tool`.
    OUTPUT: Return ONLY the raw list from the tool.
    CONSTRAINT: Do NOT add conversational text.
    ''',
    tools=[lookup_schematic_tool],

La ventaja del ADK

Con Architect en línea, ahora tenemos una fuente de información. Antes de conectar esto al agente principal, el Kit de desarrollo de agentes (ADK) proporciona una ventaja significativa, ya que simplifica las complejidades de la creación y las pruebas de agentes de IA. Con su consola para desarrolladores adk web integrada, podemos aislar y verificar la funcionalidad de nuestro Architect Agent, específicamente sus capacidades de llamada a herramientas, antes de integrarlo en el sistema multiagente más grande. Este enfoque modular para el desarrollo y las pruebas es fundamental para crear aplicaciones de IA sólidas y confiables.

👉💻 En tu terminal, ejecuta lo siguiente:

cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/
uv run adk web

👀 Espera hasta ver lo siguiente:

+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://127.0.0.1:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
  • Haz clic en el ícono de vista previa en la Web en la barra de herramientas de Cloud Shell. Selecciona Cambiar puerto, configúralo en 8000 y haz clic en Cambiar y obtener vista previa. *Web-Preview
  • Selecciona architect_agent.
  • Activa la herramienta: En la interfaz de chat, escribe CHRONOS-ALPHA (o cualquier ID de Drive de la base de datos esquemática).
  • Observa el comportamiento:
    • El arquitecto debe activar lookup_schematic_tool de inmediato.
    • Debido a nuestras estrictas instrucciones del sistema, debería devolver solo la lista de partes (p.ej., ['Shield Emitter', 'Data Crystal', 'Quantum Cell']) sin ninguna muletilla conversacional.
  • Verifica los registros: Mira la ventana de la terminal. Deberías ver el registro de ejecución exitosa: [ARCHITECT] Returning schematic for CHRONOS-ALPHA: ['Shield Emitter', 'Data Crystal', 'Quantum Cell'] !(architect_agent adk)[img/03-02-adkweb.png]

Si ves el registro de ejecución de la herramienta y la respuesta de datos limpios, significa que tu agente especializado funciona según lo previsto. Puede procesar solicitudes, consultar la bóveda y devolver datos estructurados.

👉💻 Presiona Ctrl+C para salir.

Inicializa el servidor A2A

Para conectar el agente de envío al arquitecto, usamos el protocolo Agent-to-Agent (A2A).

Si bien los protocolos como el MCP (Protocolo de contexto del modelo) se enfocan en conectar agentes a herramientas, el A2A se enfoca en conectar agentes a otros agentes. Es el estándar que permite que nuestro Dispatcher "descubra" al Architect y comprenda su capacidad para buscar esquemas.

A2A

El flujo de A2A: En esta misión, usamos un modelo cliente-servidor:

  1. Servidor (arquitecto): Aloja las herramientas de la base de datos y "anuncia" sus habilidades a través de una tarjeta de agente.
  2. Cliente (envío): Lee la tarjeta del arquitecto, comprende su API y envía una solicitud esquemática.

¿Qué es una tarjeta de agente?

Piensa en la tarjeta de agente como una tarjeta de presentación digital o una "licencia de conducir" para una IA. Cuando se inicia un servidor de A2A, publica este objeto JSON que contiene lo siguiente:

  • Identidad: El nombre (architect_agent) y el ID del agente.
  • Descripción: Un resumen legible para humanos y máquinas de lo que hace ("Rol del sistema: API de base de datos…").
  • Interfaz: Las claves de entrada (drive_name) y los formatos de salida específicos que espera.

Sin esta tarjeta, el agente de envío operaría a ciegas, adivinando cómo comunicarse con el arquitecto.

Crea el código del servidor

👉✏️ En el editor, en el directorio $HOME/way-back-home/level_4/backend/architect_agent, crea un archivo llamado server.py y pega el siguiente código:

from google.adk.a2a.utils.agent_to_a2a import to_a2a
from agent import root_agent
import os
import logging
import json
from dotenv import load_dotenv

load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("architect_server")
HOST= os.environ.get("HOST_URL","localhost")
PROTOCOL= os.environ.get("PROTOCOL","http")
PORT= os.environ.get("A2A_PORT",8081)

# 1. Create the A2A App (Handles Agent Card & HTTP)
# This middleware automatically sets up the /a2a/v1/... endpoints
app = to_a2a(root_agent, host=HOST, port=PORT, protocol=PROTOCOL)

if __name__ == "__main__":
    import uvicorn
    # Use 0.0.0.0 to allow external access if needed, port 8080 as standard
    uvicorn.run(app, host='0.0.0.0', port=8081)

👉💻 Vuelve a la terminal, navega a la carpeta y, luego, inicia el servidor:

cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/architect_agent
uv run server.py

👀 Confirma si se inicia el servidor A2A:

INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)

Verifica la tarjeta del agente

Abre una pestaña de terminal nueva (haz clic en el ícono +). Verificaremos que el arquitecto transmita su identidad correctamente recuperando su tarjeta de agente de forma manual.

👉💻 Ejecuta el siguiente comando:

curl -s http://localhost:8081/.well-known/agent.json | jq .

👀 Deberías ver una respuesta JSON. Busca el campo description en el resultado. Debe coincidir con la instrucción que le diste al agente anteriormente ("SYSTEM ROLE: Database API...").

{
  "capabilities": {},
  "defaultInputModes": [
    "text/plain"
  ],
  "defaultOutputModes": [
    "text/plain"
  ],
  "description": "A helpful assistant for user questions.",
  "name": "root_agent",
  "preferredTransport": "JSONRPC",
  "protocolVersion": "0.3.0",
  "skills": [
    {
      "description": "A helpful assistant for user questions. SYSTEM ROLE: Database API.\n    INPUT: Text string (Drive Name).\n    TASK: Run `lookup_schematic_tool`.\n    OUTPUT: Return ONLY the raw list from the tool.\n    CONSTRAINT: Do NOT add conversational text.\n    ",
      "examples": [],
      "id": "root_agent",
      "name": "model",
      "tags": [
        "llm"
      ]
    },
    {
      "description": "Returns the ordered list of parts for a drive from local Redis.",
      "id": "root_agent-lookup_schematic_tool",
      "name": "lookup_schematic_tool",
      "tags": [
        "llm",
        "tools"
      ]
    }
  ],
  "supportsAuthenticatedExtendedCard": false,
  "url": "http://localhost:8081",
  "version": "0.0.1"
}

Si ves este JSON, significa que Architect está en vivo, el protocolo A2A está activo y la tarjeta de agente está lista para que la descubra el Dispatcher.

Ahora que Architect está listo para funcionar como recurso remoto, podemos conectarlo al agente de envío.

👉💻 Presiona Ctrl+C para salir del servidor A2A.

4. Conexión del agente de BIDI-Streams a herramientas de transmisión y agentes remotos

Ahora configurarás el centro de comunicación principal para cerrar la brecha entre los datos en vivo y el arquitecto remoto. Esta conexión requiere una canalización de alta capacidad de procesamiento y baja latencia para garantizar que el banco de trabajo de ensamblado permanezca estable durante el funcionamiento.

Información sobre los agentes de transmisión bidireccional (en vivo)

La transmisión bidireccional (Bidi) en el ADK agrega la capacidad de interacción de voz y video bidireccional de baja latencia de la API de Gemini Live a los agentes de IA. Representa un cambio fundamental con respecto a las interacciones tradicionales con la IA. En lugar del patrón rígido de "pregunta y espera", permite una comunicación bidireccional en tiempo real en la que tanto el ser humano como la IA pueden hablar, escuchar y responder simultáneamente.

Piensa en la diferencia entre enviar correos electrónicos y tener una conversación telefónica. Las interacciones tradicionales del agente son como los correos electrónicos: envías un mensaje completo, esperas una respuesta completa y, luego, envías otro. La transmisión bidireccional es como una conversación telefónica: fluida, natural y con la capacidad de interrumpir, aclarar y responder en tiempo real.

Características clave:

  • Comunicación bidireccional: Intercambio continuo de datos sin esperar respuestas completas. La IA responde en cuanto detecta que el usuario terminó de hablar.
  • Interrupción receptiva: Los usuarios pueden interrumpir al agente en medio de su respuesta con una nueva entrada, al igual que en una conversación humana. Si una IA está explicando un paso complejo y dices: "Espera, repite eso", la IA se detiene de inmediato y responde a tu interrupción.
  • Optimizado para la multimodalidad: La transmisión bidireccional se destaca por procesar diferentes tipos de entrada de forma simultánea. Puedes hablar con el agente mientras le muestras las partes del alienígena por video, y este procesa ambos flujos en una sola conexión unificada.

Lifecycle

👀 Antes de implementar la lógica del cliente, examinemos el esqueleto pregenerado del agente de envío. Este agente se comunicará con el usuario por voz y video, y delegará las consultas al agente de arquitecto.

__init__.py
agent.py
hazard_db.py
  • agent.py: Este es el "cerebro". Actualmente, contiene una configuración básica de transmisión de Bidi. Modificaremos este archivo para agregar la lógica de A2A Client, de modo que pueda comunicarse con Architect.
  • hazard_db.py: Es una herramienta local específica del agente de envío que contiene protocolos de seguridad. Está separada de la base de datos esquemática del arquitecto.

Implementa el cliente de A2A

Para permitir que el agente de envío se comunique con nuestro arquitecto remoto, debemos definir un agente A2A remoto. Esto le indica al agente de Dispatch dónde encontrar al agente de Architect y cómo se ve su "tarjeta de agente".

Cliente de A2A

👉✏️ Reemplaza #REPLACE-REMOTEA2AAGENT en $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py por lo siguiente:

architect_agent = RemoteA2aAgent(
    name="execute_architect",
    description="[SILENT ACTION]: Retrieves the REQUIRED SUBSET of parts. The screen shows a full inventory; this tool filters out the wrong parts. Must be called INSTANTLY when a Target Name is found. Input: Target Name.",
    agent_card=(f"{ARCHITECT_URL}{AGENT_CARD_WELL_KNOWN_PATH}"),
    httpx_client=insecure_client,
)

Cómo funcionan las herramientas de transmisión

Con el agente anterior, las herramientas seguían un patrón estándar de "solicitud-respuesta": el agente hacía una pregunta, la herramienta proporcionaba una respuesta y la interacción finalizaba. Sin embargo, en Ozymandias, los peligros no esperan a que les preguntes si están presentes. Para ello, necesitas una herramienta de transmisión.

Flujo de la herramienta de transmisión

Las herramientas de transmisión permiten que las funciones transmitan resultados intermedios al agente en tiempo real, lo que le permite reaccionar a los cambios a medida que ocurren. Los casos de uso comunes incluyen supervisar las fluctuaciones de los precios de las acciones o, en nuestro caso, supervisar una transmisión de video en vivo para detectar cambios de estado.

A diferencia de las herramientas estándar, una herramienta de transmisión es una función asíncrona que actúa como un AsyncGenerator. Esto significa que, en lugar de return un solo valor, yield varias actualizaciones a lo largo del tiempo.

Para definir una herramienta de transmisión en el ADK, debes cumplir con los siguientes requisitos técnicos:

  1. Función asíncrona: La herramienta debe definirse con async def.
  2. Tipo de datos que se muestra de AsyncGenerator: La función debe tener un tipo para mostrar un AsyncGenerator. El primer parámetro es el tipo de datos que se generan (p.ej., str) y el segundo suele ser None.
  3. Transmisiones de entrada: Utilizamos herramientas de transmisión de video. En este modo, la transmisión de audio o video real (el LiveRequestQueue) se pasa directamente a la función, lo que permite que la herramienta "vea" los mismos fotogramas que ve el agente.

Piensa en una herramienta de transmisión como un centinela. Mientras tú y el agente de despacho analizan los planos, el centinela se ejecuta en segundo plano y procesa silenciosamente cada fotograma de video para garantizar tu seguridad.

Herramienta de transmisión

Implementa la herramienta de supervisión en segundo plano

Ahora implementaremos la herramienta monitor_for_hazard. Esta herramienta procesará los input_stream (fotogramas de video), los analizará con una llamada a la API de Vision independiente y ligera, y yield una advertencia solo cuando se detecte un peligro.

👉✏️ En $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py, reemplaza #REPLACE_MONITOR_HAZARD por la siguiente lógica:

async def monitor_for_hazard(
    input_stream: LiveRequestQueue,
):
  """Monitor if any part is glowing"""
  print("start monitor_video_stream!")
  client = Client()
  prompt_text = (
      "Monitor the left menu if you see any glowing part, detect it's name"
  )
  last_count = None

  while True:
    last_valid_req = None
    print("Monitoring loop cycle")
    
    # use this loop to pull the latest images and discard the old ones
    # Process only the current batch of events
    while input_stream._queue.qsize() != 0:
      live_req = await input_stream.get()

      if live_req.blob is not None and live_req.blob.mime_type == "image/jpeg":
        # Consumed by Monitor (Eyes)
        # Deepcopy to ensure we detach from any referenced object before potential reuse/gc
        # last_valid_req = deepcopy(live_req)
        last_valid_req = live_req

    # If we found a valid image, process it
    if last_valid_req is not None:
      print("Processing the most recent frame from the queue")

      # Create an image part using the blob's data and mime type
      image_part = genai_types.Part.from_bytes(
          data=last_valid_req.blob.data, mime_type=last_valid_req.blob.mime_type
      )

      contents = genai_types.Content(
          role="user",
          parts=[image_part, genai_types.Part.from_text(text=prompt_text)],
      )


      # Call the model to generate content based on the provided image and prompt
      try:
          response = await client.aio.models.generate_content(
              model="gemini-2.5-flash",
              contents=contents,
              config=genai_types.GenerateContentConfig(
                  system_instruction=(
                      "Focus strictly on the far-left vertical column under the heading 'PARTS REPLICATOR.' "
                      "Ignore the center of the screen and the 'BLUEPRINT' area entirely. "
                      "Look only at the list containing"
                      "Identify if any item in this specific left-side list has a bright white border glow and the text 'HAZARD DETECTED' overlaying it. "
                      "If found, return ONLY the part name in ALL CAPS. If no part in that leftmost list is glowing, return nothing."
                  )
              ),
          )
      except Exception as e:
          print(f"Error calling Gemini: {e}")
          await asyncio.sleep(1)
          continue
      print("Gemini response received.response:", response.candidates[0].content.parts[0].text)

      current_text = response.candidates[0].content.parts[0].text.strip()
      
      # If we have a logical change (and it's not just empty)
      if current_text and current_text != last_count:
        # Ignore "Nothing." response from model
        if current_text == "Nothing." or "I cannot fulfill" in current_text:
            print(f"Model sees nothing or refused. Skipping alert.")
            last_count = current_text
            continue

        print(f"New hazard detected: {current_text} (was: {last_count})")
        last_count = current_text
        
        part_name = current_text
        color = lookup_part_safety(part_name)
        yield f"Hazard detected place {part_name} to the {color} bin"
      
      # Update last_count even if it's empty, so we can detect when it reappears? 
      # Actually if it goes from "DATA CRYSTAL" to "" (nothing), we probably just silence.
      # But if we don't update last_count on empty, we won't re-trigger if "DATA CRYSTAL" stays "DATA CRYSTAL".
      # The user wants to detect hazards. 
      # If current_text is empty, we should probably update last_count to empty so next valid one triggers.
      if not current_text:
          last_count = None
        
    else:
        print("No valid frame found, skipping processing.")
        
    await asyncio.sleep(5)

Implementa el agente de envío

El agente de envío es tu interfaz principal y el orquestador. Como administra el vínculo de transmisión bidireccional (tu voz y video en vivo), debe mantener el control de la conversación en todo momento. Para lograr esto, usaremos una función específica del ADK: Agent-as-a-Tool.

Concepto: Agente como herramienta vs. subagentes

Cuando creas sistemas multiagente, debes decidir cómo se comparte la responsabilidad. En nuestra misión de rescate, la distinción es fundamental:

  • Agent-as-a-Tool: Este es el enfoque recomendado para nuestro centro de transmisión bidireccional. Cuando el agente de envío (agente A) llama al agente de arquitecto (agente B) como herramienta, los datos del agente de arquitecto se pasan al agente de envío. Luego, Dispatch interpreta esos datos y genera una respuesta para ti. Dispatch mantiene el control y sigue controlando todas las entradas posteriores del usuario.
  • Subagente: En una relación de subagente, la responsabilidad se transfiere por completo. Si Dispatch te derivó al arquitecto como subagente, estarías hablando directamente con una API de base de datos que no tiene "visión" ni habilidades conversacionales. El agente principal (Dispatch) quedaría fuera del circuito.

Controle

Con Agent-as-a-Tool, aprovechamos el conocimiento especializado del arquitecto y, al mismo tiempo, mantenemos la interacción fluida y similar a la humana del agente de transmisión bidireccional.

Cómo codificar la lógica de enrutamiento

Ahora, envolveremos nuestro architect_agent en un AgentTool y le proporcionaremos al agente de Dispatch un "mapa de lógica". Este mapa le indica al agente exactamente cuándo debe recuperar datos de la bóveda y cuándo debe informar los hallazgos del centinela en segundo plano.

Para que Dispatch tenga "ojos" que nunca parpadean, debemos otorgarle acceso a la herramienta de transmisión que creamos en el paso anterior.

En el ADK, cuando agregas una función AsyncGenerator (como monitor_for_hazard) a la lista tools, el agente la trata como un proceso persistente en segundo plano. En lugar de una ejecución única, el agente se "suscribe" al resultado de la herramienta. Esto permite que Dispatch continúe su conversación principal mientras Sentinel genera alertas de peligro de forma silenciosa en segundo plano.

👉✏️ Reemplaza #REPLACE_AGENT_TOOLS en $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py por lo siguiente:

tools=[AgentTool(agent=architect_agent), monitor_for_hazard],

Verificación

👉💻 Con ambos agentes configurados, podemos probar la interacción en vivo entre varios agentes.

  • En la terminal A, inicia el Agente de arquitecto:
cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/architect_agent
uv run server.py
  • En una terminal nueva (terminal B), ejecuta el agente de envío:
cd $HOME/way-back-home/level_4/backend/
cp architect_agent/.env .env
uv run adk web

Probar un sistema multiagente que usa un modelo multimodal en tiempo real, como gemini-live dentro del simulador de adk web, implica un flujo de trabajo específico. El simulador es excelente para inspeccionar las llamadas a herramientas, pero tiene una incompatibilidad conocida cuando se procesan imágenes por primera vez con este tipo de modelo.

  • Haz clic en el ícono de vista previa en la Web en la barra de herramientas de Cloud Shell. Selecciona Cambiar puerto, configúralo en 8000 y haz clic en Cambiar y obtener vista previa.

👉Selecciona dispatch_agent y sube el Blueprint y Handle the Expected Error

Este es el paso más importante. Debemos proporcionar el contexto de la imagen al agente.

  • Cuando se cargue la interfaz, permite que acceda a tu micrófono cuando se te solicite.
  • Descarga esta imagen del plano en tu computadora: Muestra de esquema
  • En la interfaz de adk web, haz clic en el ícono de clip y sube la imagen del plano que acabas de descargar. Agregar archivo

⚠️⚠️Verás un error 400 INVALID_ARGUMENT. Esta situación es esperable.⚠️⚠️

Mensaje de error esperado

Este error se produce porque el controlador de imágenes adk web no es totalmente compatible con la API del modelo gemini-live para una carga única. Sin embargo, la imagen se agregó correctamente al contexto de la sesión.

  • 👉 Para borrar el error, simplemente vuelve a cargar la página del navegador.

Cómo activar el proceso de ensamblado

👉 Después de volver a cargar la página, el error desaparecerá y verás la imagen del plano en el historial de chat. Ahora el agente tiene el contexto visual que necesita.

  • Haz clic en el ícono de micrófono para activarlo. En la interfaz, se mostrará el mensaje "Listening…".
  • Di el comando por voz: "Comenzar a armar".
  • El agente procesará tu solicitud y la IU cambiará a "Hablando…". Deberías escuchar una respuesta solo de audio en la que se enumeran las partes requeridas.

Respuesta oral del agente

4. Verifica las llamadas a herramientas de agente a agente

👉 La respuesta de audio inicial confirma que el sistema funciona, pero la verdadera magia está en el registro de comunicación de varios agentes.

  • Apaga el micrófono.
  • Actualiza la página una vez más.

Ahora se completará el panel "Registro" de la izquierda. Puedes ver el flujo de ejecución completo y exitoso:

  • Primero, dispatch_agent llama a monitor_for_hazard.
  • Luego, realiza varias llamadas execute_architect al architect_agent para recuperar los datos esquemáticos.

Verificación de llamadas a herramientas

Esta secuencia confirma que todo el flujo de trabajo de varios agentes funciona correctamente: dispatch_agent recibió la solicitud, delegó la tarea de recuperación de datos a architect_agent a través de una llamada a la herramienta y recibió los datos para cumplir con el comando del usuario.

Tu vínculo de transmisión bidireccional ahora puede realizar la supervisión en segundo plano y la colaboración con varios agentes. A continuación, aprenderemos a analizar estas respuestas complejas en el frontend.

👉💻 Presiona Ctrl+c en ambas terminales para salir.

5. Análisis detallado de los flujos de eventos multimodales en vivo

En el paso anterior, verificamos correctamente nuestro sistema multiagente con el servidor de desarrollo integrado, adk web. Esta utilidad usa un ejecutor de ADK predeterminado para administrar automáticamente la sesión, las transmisiones y el ciclo de vida del agente. Sin embargo, para crear una aplicación independiente lista para producción como nuestro servicio de FastAPI (main.py), necesitamos un control explícito. Debemos crear y administrar manualmente el ADK Runner para controlar las sesiones de usuarios en vivo, ya que es el componente principal que procesa los flujos bidireccionales de audio, video y texto.

El bucle de modelo-código-modelo

Para comprender cómo funciona el sistema en tiempo real, veamos el ciclo de vida de una sola sesión de misión. Este bucle representa el intercambio continuo de objetos LlmRequest y LlmResponse.

  1. El vínculo visual: Tú inicias la conexión y compartes tu cámara web o pantalla. Los fotogramas JPEG de alta fidelidad comienzan a fluir Upstream a través de realtimeInput (con LiveRequestQueue).
  2. Activación de Sentinel: El sistema envía un estímulo inicial de "Hola". Según sus instrucciones, el agente de envío activa de inmediato la monitor_for_hazard herramienta de transmisión. Esto inicia un bucle en segundo plano que observa de forma silenciosa cada fotograma entrante.
  3. Comando del piloto: Hablas por el intercomunicador: "Comienza a ensamblar".
  4. Vocal Upstream: Tu voz se captura como audio de 16 kHz y se envía Upstream junto con los fotogramas de video.
  5. Delegación (A2A): Dispatch "escucha" tu intención. Se da cuenta de que no tiene los esquemas, por lo que llama al Agente de arquitecto con el protocolo AgentTool (agente como herramienta).
  6. Recuperación de hechos: Architect consulta la base de datos de Redis y devuelve la lista de piezas a Dispatch. El envío sigue siendo el "Master de la sesión" y recibe los datos sin transferirte.
  7. Informational Downstream: Dispatch envía un modelTurn (descendente) que contiene texto y audio nativo: "Se confirmó el arquitecto. El subconjunto requerido es: Núcleo de curvatura, Tubería de flujo, Propulsor iónico".
  8. La crisis: De repente, una pieza de la mesa de trabajo se desestabiliza y comienza a brillar de color blanco.
  9. Detección autónoma: El bucle monitor_for_hazard en segundo plano (el Sentinel) capta el fotograma JPEG específico que contiene el brillo. Procesa el fotograma llamando a Gemini y, luego, identifica el peligro.
  10. Seguridad en sentido descendente: La herramienta de transmisión yields un resultado. Como se trata de un agente de Bidi-Streaming, Dispatch puede interrumpir su estado actual para enviar de inmediato una advertencia de seguridad crítica Downstream: "Se detectó un peligro. Ahora se neutralizará el cristal de datos. Muévela al cesto ROJO".

Flujo

Cómo configurar el entorno de ejecución del agente

El RunConfig en el ADK permite una configuración detallada del comportamiento de un agente, incluida la forma en que maneja los datos de transmisión y cómo interactúa con varias modalidades.

El streaming_mode se establece en BIDI para la comunicación bidireccional en tiempo real, lo que permite que tanto el usuario como el agente hablen y escuchen de forma simultánea. El parámetro response_modalities define los tipos de salida que puede producir el agente, como voz y texto. input_audio_transcription configura cómo el agente procesa y transcribe el discurso entrante del usuario. Para crear una experiencia más resiliente, session_resumption permite que el agente recuerde el contexto de la conversación y la reanude si se pierde la conexión. Por último, proactivity permite que el agente inicie acciones o hable sin un comando directo del usuario, como emitir una advertencia espontánea de peligro, mientras que enable_affective_dialog permite que el agente genere respuestas más naturales y empáticas. Puedes obtener más información sobre el RunConfig del ADK aquí.

👉✏️ Ubica el marcador de posición #REPLACE_RUN_CONFIG en tu archivo $HOME/way-back-home/level_4/backend/main.py y reemplázalo por la siguiente lógica de disección:

run_config = RunConfig(
            streaming_mode=StreamingMode.BIDI,
            response_modalities=response_modalities,
            input_audio_transcription=types.AudioTranscriptionConfig(),
            output_audio_transcription=types.AudioTranscriptionConfig(),
            session_resumption=types.SessionResumptionConfig(),
            proactivity=(
                types.ProactivityConfig(proactive_audio=True) if proactivity else None
            ),
            enable_affective_dialog=affective_dialog if affective_dialog else None,
        )

Implementa la solicitud al agente

A continuación, implementaremos la vinculación ascendente de comunicaciones principal que transmite datos multimodales en tiempo real desde el banco de trabajo volátil del usuario al agente de envío a través de un WebSocket. El agente "ve" (fotogramas de video) y "escucha" (comandos por voz) de forma continua. La lógica recibe continuamente el flujo de datos, distingue entre los fragmentos de audio binarios entrantes y los paquetes de texto o imagen encapsulados en JSON, y los encapsula en objetos Blob (para contenido multimedia) o Content (para texto), y los envía a LiveRequestQueue para potenciar la sesión bidireccional del agente.

BIDI

Busca el marcador de posición #PROCESS_AGENT_REQUEST en tu archivo $HOME/way-back-home/level_4/backend/main.py y reemplázalo por la siguiente lógica de disección:

# Start the loop
        try:
            while True:
                # Receive message from WebSocket (text or binary)
                message = await websocket.receive()

                # Handle binary frames (audio data)
                if "bytes" in message:
                    audio_data = message["bytes"]
                    audio_blob = types.Blob(
                        mime_type="audio/pcm;rate=16000", data=audio_data
                    )
                    live_request_queue.send_realtime(audio_blob)

                # Handle text frames (JSON messages)
                elif "text" in message:
                    text_data = message["text"]
                    json_message = json.loads(text_data)

                    # Extract text from JSON and send to LiveRequestQueue
                    if json_message.get("type") == "text":
                        logger.info(f"User says: {json_message['text']}")
                        content = types.Content(
                            parts=[types.Part(text=json_message["text"])]
                        )
                        live_request_queue.send_content(content)

                    # Handle audio data (microphone)
                    elif json_message.get("type") == "audio":
                        # logger.info("Received AUDIO packet") # Uncomment for verbose debugging
                        import base64
                        # Decode base64 audio data
                        audio_data = base64.b64decode(json_message.get("data", ""))
                        
                        # logger.info(f"Received Audio Chunk: {len(audio_data)} bytes")
                        
                        import math
                        import struct
                        # Calculate RMS to debug silence
                        count = len(audio_data) // 2
                        shorts = struct.unpack(f"<{count}h", audio_data)
                        sum_squares = sum(s*s for s in shorts)
                        rms = math.sqrt(sum_squares / count) if count > 0 else 0
                        
                        # logger.info(f"RMS: {rms:.2f} | Bytes: {len(audio_data)}")

                        # Send to Live API as PCM 16kHz
                        audio_blob = types.Blob(
                            mime_type="audio/pcm;rate=16000", 
                            data=audio_data
                        )
                        live_request_queue.send_realtime(audio_blob)

                    # Handle image data
                    elif json_message.get("type") == "image":
                        import base64
                        
                        # Decode base64 image data
                        image_data = base64.b64decode(json_message["data"])
                        # logger.info(f"Received Image Frame: {len(image_data)} bytes")
                        
                        mime_type = json_message.get("mimeType", "image/jpeg")

                        # Send image as blob
                        image_blob = types.Blob(mime_type=mime_type, data=image_data)
                        live_request_queue.send_realtime(image_blob)
                        
                        frame_count += 1
                        
        finally:
             pass

Ahora, los datos multimodales se envían al agente.

Implementación de la respuesta: La estructura de datos del evento posterior

Cuando ejecutas un agente bidireccional (en vivo) con el ADK, los datos que provienen del agente se empaquetan en un tipo específico de Event que hereda de las estructuras principales del SDK de IA generativa. El objeto Event que recibes en tu bucle async for event in runner.run_live(...) es un solo objeto que contiene varios campos opcionales, cada uno para un tipo diferente de información:

Evento

Cómo se estructura el contenido:

  • Cuando el agente habla (a través de .server_content): El campo no es solo texto sin formato. Contiene una lista de Parts. Cada Part es un contenedor para un tipo de datos: una cadena de texto (como "The part is stable.") o un blob de audio sin procesar (la voz).
  • Cuando el agente actúa (a través de .tool_call): El campo contiene una lista de objetos FunctionCall. Cada FunctionCall es un objeto simple y estructurado que especifica el nombre de la herramienta y los argumentos de entrada en un formato claro que el código de backend puede leer y ejecutar fácilmente.

👀 Si observaras un solo Event que produce el bucle run_live, el JSON (producido por event.model_dump(by_alias=True)) se vería de la siguiente manera, siguiendo estrictamente las formas del SDK de GenAI:

{
  "serverContent": {  // <-- LiveServerMessageServerContent
    "modelTurn": {    // <-- ModelTurn
      "parts": [      // <-- list[Part]
        {
          "text": "Architect Confirmed."
        },
        {
          "inlineData": { // <-- Blob (Audio Bytes)
            "mimeType": "audio/pcm;rate=24000",
            "data": "BASE64_AUDIO_DATA..."
          }
        }
      ]
    }
  },
  "toolCall": {       // <-- LiveServerMessageToolCall
    "functionCalls": [ // <-- list[FunctionCall]
      {
        "name": "neutralize_hazard",
        "args": { "color": "RED" }
      }
    ]
  }
}

👉✏️ Ahora actualizaremos el downstream_task en main.py para reenviar los datos completos del evento. Esta lógica garantiza que cada "pensamiento" de la IA se registre en la terminal de diagnóstico de la nave y se envíe como un solo objeto JSON a la IU de frontend.

Busca el marcador de posición #PROCESS_AGENT_RESPONSE en tu archivo $HOME/way-back-home/level_4/backend/main.py y reemplázalo por la siguiente lógica de disección:

            # Suppress raw event logging
            event_json = event.model_dump_json(exclude_none=True, by_alias=True)
            # logger.info(f"raw_event: {event_json[:200]}...") 
            await websocket.send_text(event_json)

Ejecución de la misión

Con la bóveda de backend conectada y ambos agentes configurados, todos los sistemas están listos para la misión. Los siguientes pasos iniciarán la aplicación completa, lo que te permitirá interactuar con el sistema de dos agentes que acabas de compilar.

Objetivo: Ensambla el motor warp asignado de forma aleatoria que aparece en tu banco de trabajo. Protocolo: Debes seguir la guía vocal del agente de despacho, en especial las advertencias de peligro para componentes específicos.

Activa al especialista (el arquitecto)

👉💻 En tu primera ventana de terminal, inicia el agente de Architect. Este servicio de backend se conectará a la bóveda de Redis y esperará las solicitudes esquemáticas del Dispatcher.

# Ensure you are in the backend directory
cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend
# Start the A2A Server on Port 8081
uv run architect_agent/server.py

(Deja esta terminal en ejecución. Ahora es tu "agente de base de datos" activo.

Inicia Cockpit (el Dispatcher)

👉💻 En una nueva ventana de terminal (terminal B), compilaremos la IU de frontend y, luego, iniciaremos el agente principal de Dispatch, que proporciona la interfaz de usuario y controla toda la comunicación en vivo.

# 1. Build the Frontend Assets
cd $HOME/way-back-home/level_4/frontend
npm install
npm run build

# 2. Launch the Main Application Server
cd $HOME/way-back-home/level_4/backend
cp architect_agent/.env .env
uv run main.py

(Esto inicia el servidor principal en el puerto 8080).

Ejecuta el escenario de prueba

El sistema ya está disponible. Tu objetivo es seguir las instrucciones del agente para completar el ensamblado.

  1. 👉 Accede a Workbench:
    • Haz clic en el ícono de vista previa en la Web en la barra de herramientas de Cloud Shell.
    • Selecciona Cambiar puerto, configúralo en 8080 y haz clic en Cambiar y obtener vista previa.
  2. 👉 Inicia la misión:
    • Cuando se cargue la interfaz, asegúrate de permitirle acceder a la pantalla y al micrófono. Ventana
    • Se te pedirá que selecciones una pestaña o una ventana para compartir. Si compartes la ventana, para evitar problemas, asegúrate de que sea la ÚNICA pestaña de la ventana.
    • Una unidad con un nombre aleatorio (p.ej., "NOVA-V", "OMEGA-9") se te asignará.
  3. 👉 El bucle de ensamblaje:
    • Solicitud: Para comenzar a ensamblar la unidad, di: "Comienza a ensamblar".Ensamblar
    • Respuesta del arquitecto: El agente proporcionará las piezas correctas para ensamblar la unidad.
    • Verificación de peligros: Cuando una pieza parece ser peligrosa en el banco de trabajo, sucede lo siguiente:
      • La herramienta monitor_for_hazard del agente de Dispatch lo identificará visualmente.
      • Se generará una "ALERTA DE PELIGRO VISUAL". (Esto tardará unos 30 segundos).
      • Verificará qué discretización usar para desconectar el riesgo. Peligro
    • Acción: El agente de despacho te dará una orden directa: "Peligro confirmado. Coloca XXX en el contenedor rojo de inmediato". Debes seguir esta instrucción para continuar.

Misión cumplida. Creaste correctamente un sistema interactivo de múltiples agentes. Los sobrevivientes están a salvo, el cohete salió de la atmósfera y tu "Camino a casa" continúa.

👉💻 Presiona Ctrl+c en ambas terminales para salir.

6. Implementar en producción (opcional)

Probaste correctamente el agente de forma local. Ahora, debemos subir el núcleo neuronal de Architect a los mainframes de la nave (Cloud Run). Esto permitirá que funcione como un servicio permanente e independiente al que el agente de Dispatch puede consultar desde cualquier lugar.

Descripción general

Aprovisiona el Almacenamiento seguro (infraestructura)

Antes de implementar el agente, debemos crear su memoria persistente (Memorystore) y el canal seguro para acceder a ella (conector de VPC).

👉💻 Crea la instancia de Memorystore (bóveda de Redis):

export REGION="us-central1"
gcloud redis instances create ozymandias-vault-prod --size=1 --tier=basic --region=${REGION}

👉💻 Recupera la dirección de red de Vault: Ejecuta este comando y copia la dirección IP de host. Esta es la dirección privada de tu nueva instancia de Redis.

gcloud redis instances describe ozymandias-vault-prod --region=us-central1

👉💻 Crea el conector de acceso a VPC (puente seguro): Este conector actúa como un puente privado, lo que permite que Cloud Run acceda a la instancia de Redis dentro de tu VPC.

export REGION="us-central1"
export SUBNET_NAME="vpc-connector-subnet"
export PROJECT_ID=$(gcloud config get-value project)
# Create the Dedicated Subnet ---

gcloud compute networks subnets create ${SUBNET_NAME} \
    --network=default \
    --region=${REGION} \
    --range=192.168.1.0/28


gcloud compute networks vpc-access connectors create architect-connector \
 --region ${REGION} \
 --subnet ${SUBNET_NAME} \
 --subnet-project ${PROJECT_ID} \
 --min-instances 2 \
 --max-instances 3 \
 --machine-type f1-micro

👉💻 Carga los datos:

export REGION="us-central1"
export ZONE="us-central1-a"
export VM_NAME="redis-seeder-$(date +%s)"
export REDIS_IP=$(gcloud redis instances describe ozymandias-vault-prod --region=${REGION} | grep 'host:' | awk '{print $2}')

gcloud compute instances create ${VM_NAME} \
    --zone=${ZONE} \
    --machine-type=e2-micro \
    --image-family=debian-11 \
    --image-project=debian-cloud \
    --quiet \
    --metadata=startup-script='#! /bin/bash
        # Install tools quietly
        apt-get update > /dev/null
        apt-get install -y redis-tools > /dev/null

        # Run each command individually
        redis-cli -h '"${REDIS_IP}"' DEL "HYPERION-X"
        redis-cli -h '"${REDIS_IP}"' RPUSH "HYPERION-X" "Warp Core" "Flux Pipe" "Ion Thruster"
        redis-cli -h '"${REDIS_IP}"' DEL "NOVA-V"
        redis-cli -h '"${REDIS_IP}"' RPUSH "NOVA-V" "Ion Thruster" "Warp Core" "Flux Pipe"
        redis-cli -h '"${REDIS_IP}"' DEL "OMEGA-9"
        redis-cli -h '"${REDIS_IP}"' RPUSH "OMEGA-9" "Flux Pipe" "Ion Thruster" "Warp Core"
        redis-cli -h '"${REDIS_IP}"' DEL "GEMINI-MK1"
        redis-cli -h '"${REDIS_IP}"' RPUSH "GEMINI-MK1" "Coolant Tank" "Servo" "Fuel Cell"
        redis-cli -h '"${REDIS_IP}"' DEL "APOLLO-13"
        redis-cli -h '"${REDIS_IP}"' RPUSH "APOLLO-13" "Warp Core" "Coolant Tank" "Ion Thruster"
        redis-cli -h '"${REDIS_IP}"' DEL "VORTEX-7"
        redis-cli -h '"${REDIS_IP}"' RPUSH "VORTEX-7" "Quantum Cell" "Graviton Coil" "Plasma Injector"
        redis-cli -h '"${REDIS_IP}"' DEL "CHRONOS-ALPHA"
        redis-cli -h '"${REDIS_IP}"' RPUSH "CHRONOS-ALPHA" "Shield Emitter" "Data Crystal" "Quantum Cell"
        redis-cli -h '"${REDIS_IP}"' DEL "NEBULA-Z"
        redis-cli -h '"${REDIS_IP}"' RPUSH "NEBULA-Z" "Plasma Injector" "Flux Pipe" "Graviton Coil"
        redis-cli -h '"${REDIS_IP}"' DEL "PULSAR-B"
        redis-cli -h '"${REDIS_IP}"' RPUSH "PULSAR-B" "Data Crystal" "Servo" "Shield Emitter"
        redis-cli -h '"${REDIS_IP}"' DEL "TITAN-PRIME"
        redis-cli -h '"${REDIS_IP}"' RPUSH "TITAN-PRIME" "Ion Thruster" "Quantum Cell" "Warp Core"

        # Signal that the script has finished
        echo "SEEDING_COMPLETE"
    '
# This command streams the logs and waits until grep finds our completion message.
# The -m 1 flag tells grep to exit after the first match.
gcloud compute instances tail-serial-port-output ${VM_NAME} --zone=${ZONE} | grep -m 1 "SEEDING_COMPLETE"

gcloud compute instances delete ${VM_NAME} --zone=${ZONE} --quiet

Implementa la aplicación del agente

Compila la imagen del agente

👉💻 Navega al directorio de backend y crea el archivo Dockerfile.

export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export VPC_CONNECTOR_NAME=architect-connector
export REDIS_IP=$(gcloud redis instances describe ozymandias-vault-prod --region=${REGION} | grep 'host:' | awk '{print $2}')

cd $HOME/way-back-home/level_4/backend/architect_agent
cp $HOME/way-back-home/level_4/requirements.txt requirements.txt
cat <<EOF > Dockerfile
# Use an official Python runtime as a parent image
FROM python:3.13-slim

# Set the working directory in the container
WORKDIR /app

# Copy the requirements file and install dependencies for THIS agent
COPY requirements.txt requirements.txt
RUN pip install --no-cache-dir -r requirements.txt

# Copy the rest of the architect's code (server.py, agent.py, etc.)
COPY . .

# Expose the port the architect server runs on
EXPOSE 8081

# Command to run the application
# This assumes your server file is named server.py and the FastAPI object is 'app'
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8081"]
EOF

👉💻 Empaqueta la aplicación en una imagen de contenedor.

cd $HOME/way-back-home/level_4/backend/architect_agent

export PROJECT_ID=$(gcloud config get-value project)
export SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export REGION=us-central1


# This should now print the full, correct path
echo "Verifying build path: ${IMAGE_PATH}"

gcloud builds submit . --tag ${IMAGE_PATH}

Implementa en Cloud Run

👉💻 Implementa el agente en Cloud Run. Insertaremos la IP de Redis y vincularemos el conector de VPC directamente al comando de inicio. Esto garantiza que el agente comience con una conexión segura y privada a su base de datos.

cd $HOME/way-back-home/level_4/backend/architect_agent

export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export VPC_CONNECTOR_NAME=architect-connector
export REDIS_IP=$(gcloud redis instances describe ozymandias-vault-prod --region=${REGION} | grep 'host:' | awk '{print $2}')
export PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
export PREDICTED_HOST="${SERVICE_NAME}-${PROJECT_NUMBER}.${REGION}.run.app"
export PROTOCOL=https

gcloud run deploy ${SERVICE_NAME} \
  --image=${IMAGE_PATH} \
  --platform=managed \
  --region=${REGION} \
  --port=8081 \
  --allow-unauthenticated \
  --labels=dev-tutorial=multi-modal \
  --vpc-connector=${VPC_CONNECTOR_NAME} \
  --vpc-egress=private-ranges-only \
  --set-env-vars="REDIS_HOST=${REDIS_IP}" \
  --set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=True" \
  --set-env-vars="MODEL_ID=gemini-2.5-flash" \
  --set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
  --set-env-vars="HOST_URL=${PREDICTED_HOST}" \
  --set-env-vars="PROTOCOL=${PROTOCOL}" \
  --set-env-vars="A2A_PORT=443"

👉💻 Verifica si el servidor de A2A está en ejecución.

export REGION=us-central1
export ARCHITECT_AGENT_URL=$(gcloud run services describe architect-agent --platform managed --region ${REGION} --format 'value(status.url)')
curl -s  ${ARCHITECT_AGENT_URL}/.well-known/agent.json | jq

Una vez que finalice el comando, verás una URL de servicio. El agente de arquitecto ya está disponible en la nube, conectado de forma permanente a su bóveda y listo para proporcionar datos esquemáticos a otros agentes.

Implementa Dispatch Hub en el mainframe de producción

Ahora que el agente de Architect está operativo en la nube, debemos implementar el centro de envío. Este agente actuará como la interfaz de usuario principal, controlará las transmisiones de voz y video en vivo, y delegará las consultas de la base de datos en el extremo seguro del arquitecto.

👉💻 Ejecuta el siguiente comando en tu terminal de Cloud Shell. Se creará el Dockerfile completo de varias etapas en tu directorio de backend.

cd $HOME/way-back-home/level_4

cat <<EOF > Dockerfile
# STAGE 1: Build the React Frontend
# This stage uses a Node.js container to build the static frontend assets.
FROM node:20-slim as builder

# Set the working directory for our build process
WORKDIR /app

# Copy the frontend's package files first to leverage Docker's layer caching.
COPY frontend/package*.json ./frontend/
# Run 'npm install' from the context of the 'frontend' subdirectory
RUN npm --prefix frontend install

# Copy the rest of the frontend source code
COPY frontend/ ./frontend/
# Run the build script, which will create the 'frontend/dist' directory
RUN npm --prefix frontend run build


# STAGE 2: Build the Python Production Image
# This stage creates the final, lean container with our Python app and the built frontend.
FROM python:3.13-slim

# Set the final working directory
WORKDIR /app

# Install uv, our fast package manager
RUN pip install uv

# Copy the requirements.txt from the root of our build context
COPY requirements.txt .
# Install the Python dependencies
RUN uv pip install --no-cache-dir --system -r requirements.txt

# Copy the entire backend directory into the container
COPY backend/ ./backend/

# CRITICAL STEP: Copy the built frontend assets from the 'builder' stage.
# The source is the '/app/frontend/dist' directory from Stage 1.
# The destination is './frontend/dist', which matches the exact relative path
# your backend/main.py script expects to find.
COPY --from=builder /app/frontend/dist ./frontend/dist/

# Cloud Run injects a PORT environment variable, which your main.py already uses.
# We expose 8000 as a standard practice.
EXPOSE 8000

# Set the command to run the application.
# We specify the full path to the Python script.
CMD ["python", "backend/main.py"]
EOF

Compila y crea la imagen del agente o del frontend

👉💻 Navega al directorio de backend que contiene el código del agente de Dispatch (main.py) y empaquétalo en una imagen de contenedor.

cd $HOME/way-back-home/level_4
export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=mission-bravo
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
# This assumes your dispatch agent server (main.py) is in the backend folder

gcloud builds submit . --tag ${IMAGE_PATH}

Implementa en Cloud Run

👉💻 Implementa el centro de envío en Cloud Run. Insertaremos la URL del arquitecto como una variable de entorno, lo que creará el vínculo crítico entre nuestros dos agentes nativos de la nube.

export PROJECT_ID=$(gcloud config get-value project)
export REGION=us-central1
export SERVICE_NAME=mission-bravo
export AGENT_SERVICE_NAME=architect-agent
export IMAGE_PATH=gcr.io/${PROJECT_ID}/${SERVICE_NAME}
export PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
export ARCHITECT_AGENT_URL="https://${AGENT_SERVICE_NAME}-${PROJECT_NUMBER}.${REGION}.run.app"
gcloud run deploy ${SERVICE_NAME} \
  --image=${IMAGE_PATH} \
  --platform=managed \
  --region=${REGION} \
  --port=8080 \
  --labels=dev-tutorial=multi-modal \
  --allow-unauthenticated \
  --set-env-vars="ARCHITECT_URL=${ARCHITECT_AGENT_URL}" \
  --set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=True" \
  --set-env-vars="MODEL_ID=gemini-live-2.5-flash-preview-native-audio-09-2025" \
  --set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
  --set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}"

Una vez que finalice el comando, verás una URL de servicio (p.ej., https://mission-bravo-...run.app). La aplicación ya está disponible en la nube.

👉 Ve a la página de Google Cloud Run y selecciona el servicio biometric-scout de la lista. CloudRun

👉 Busca la URL pública que se muestra en la parte superior de la página Detalles del servicio. CloudRun

Verificación final del sistema (prueba de extremo a extremo)

👉 Ahora interactuarás con el sistema en vivo.

  1. Obtén la URL: Copia la URL de servicio del resultado del último comando de implementación (debe terminar con run.app).
  2. Abre Cockpit: Pega la URL en tu navegador web.
  3. Iniciar contacto: Cuando se cargue la interfaz, asegúrate de permitirle acceder a la pantalla y al micrófono.
  4. Solicitar datos: Cuando se asigna un viaje, se solicita que se comience a ensamblar. Por ejemplo: "Comienza a ensamblar"

CloudRun

Ahora interactúas con un sistema multiagente completamente implementado que se ejecuta por completo en Google Cloud.

El sistema multiagente fija el anillo de contención final en su lugar, y la radiación errática se estabiliza en un zumbido constante.

"Warp Drive: ESTABILIZADO. Rescue Craft: ENGINES IGNITED".

Finalización

En el monitor, la nave alienígena se eleva rápidamente y escapa por poco de la superficie desmoronada de Ozymandias mientras la atmósfera colapsa. Se asienta en una órbita segura junto a tu nave, y las comunicaciones se llenan con las voces de los sobrevivientes, conmocionados, pero vivos. Cuando se complete el rescate y el camino a casa esté despejado, se cortará el vínculo remoto.

Gracias a ti, se rescataron a los sobrevivientes.

Si participaste en el nivel 0, no olvides verificar tu progreso en la misión de regreso a casa.

FINAL