1. A missão
Você está à deriva no espaço silencioso e desconhecido. Um enorme Pulso Solar rasgou sua nave por uma fenda dimensional, deixando você preso em um canto do universo sem nenhum mapa estelar.
Depois de dias de reparos exaustivos, o zumbido familiar dos motores finalmente volta. Sua nave está operacional. Você até conseguiu estabelecer um uplink de longo alcance com a nave-mãe. A partida é iminente. Você está pronto para ir para casa.
Mas, enquanto você se prepara para engatar o drive de salto, um sinal de socorro corta a estática. Seus sensores identificam um pedido de ajuda de um planeta chamado "Ozymandias". Os sobreviventes estão presos neste mundo moribundo, com a nave encalhada. Sua missão é crítica: resgate-os antes que a atmosfera do planeta entre em colapso.
A única maneira de escapar é um foguete antigo e abandonado construído com tecnologia alienígena. Embora funcional, o propulsor de dobra está destruído. Para salvar os sobreviventes, você precisa se conectar remotamente ao Volatile Workbench deles e montar manualmente uma unidade de substituição.
O desafio
Você não tem experiência com essa tecnologia alienígena, que é notoriamente frágil. Um componente desestabilizado pode se tornar um risco radioativo em segundos. Você tem uma tentativa para operar o Volatile Workbench. Seu assistente de IA atual tem dificuldade para processar dados visuais e manuais técnicos simultaneamente, o que leva a instruções alucinatórias e avisos de risco perdidos.
Para ter sucesso, você precisa fazer upgrade da sua IA de uma entidade monolítica para um sistema multiagente colaborativo.
Objetivos da missão:
Siga as instruções especializadas e em tempo real do seu novo sistema multiagente para montar o Warp Drive.
O que você criará
- Um sistema de IA multiagente bidirecional em tempo real com um Agente de Despacho central que gerencia a interação do usuário e coordena com agentes especializados.
- Um agente de arquitetura que se conecta a um banco de dados Redis para recuperar e disponibilizar dados esquemáticos.
- Um Monitor de segurança proativo que usa ferramentas de streaming para analisar um feed de vídeo ao vivo em busca de riscos visuais e acionar alertas em tempo real.
- Um front-end baseado em React que fornece uma interface do usuário para interagir com o sistema, transmitindo vídeo e áudio para os agentes de back-end.
O que você vai aprender
Tecnologia / conceito | Descrição |
Kit de Desenvolvimento de Agente (ADK) do Google | Você vai usar o ADK para criar, testar e gerenciar os agentes, aproveitando a estrutura dele para lidar com comunicação em tempo real, integração de ferramentas e ciclo de vida do agente. |
Streaming bidirecional (Bidi) | Você vai implementar um agente de streaming bidirecional que permite uma comunicação natural, de baixa latência e bidirecional, permitindo que humanos e IA interrompam e respondam em tempo real. |
Sistemas multiagente | Você vai aprender a projetar um sistema de IA distribuído em que um agente principal delega tarefas a agentes especializados, permitindo uma separação de responsabilidades e uma arquitetura mais escalonável. |
Protocolo Agent-to-Agent (A2A) | Você vai usar o protocolo A2A para ativar a comunicação entre o agente de encaminhamento e o agente de arquitetura, permitindo que eles descubram os recursos um do outro e troquem dados. |
Ferramentas de streaming | Você vai implementar uma ferramenta de streaming que funciona como um processo em segundo plano, analisando continuamente um feed de vídeo para monitorar mudanças de estado (riscos) e gerar resultados de forma proativa. |
Google Cloud Run e Memorystore | Você vai implantar todo o aplicativo multiagente em um ambiente de produção usando o Cloud Run para hospedar os serviços de agente e o Memorystore (Redis) como o banco de dados permanente. |
FastAPI e WebSockets | O back-end é criado usando FastAPI e WebSockets para lidar com a comunicação de alta performance e em tempo real necessária para streaming de áudio, vídeo e respostas do agente. |
Front-end do React | Você vai trabalhar com um front-end baseado em React que captura e transmite mídia do usuário (áudio/vídeo) e mostra as respostas em tempo real dos agentes de IA. |
2. Configuração de seu ambiente
Acessar o Cloud Shell
👉Clique em "Ativar o Cloud Shell" na parte de cima do console do Google Cloud. É o ícone em formato de terminal na parte de cima do painel do Cloud Shell. 
👉Clique no botão "Abrir editor" (parece uma pasta aberta com um lápis). Isso vai abrir o editor de código do Cloud Shell na janela. Um explorador de arquivos vai aparecer no lado esquerdo. 
👉Abra o terminal no IDE da nuvem.
👉💻 No terminal, verifique se você já está autenticado e se o projeto está definido como seu ID do projeto usando o seguinte comando:
gcloud auth list
Sua conta vai aparecer como (ACTIVE).
Pré-requisitos
ℹ️ O nível 0 é opcional (mas recomendado)
É possível concluir essa missão sem o nível 0, mas terminar primeiro oferece uma experiência mais imersiva, permitindo que você veja seu farol acender no mapa global à medida que avança.
Configurar o ambiente do projeto
De volta ao terminal, conclua a configuração definindo o projeto ativo e ativando os serviços necessários do Google Cloud (Cloud Run, Vertex AI etc.).
👉💻 No terminal, defina o ID do projeto:
gcloud config set project $(cat ~/project_id.txt) --quiet
👉💻 Ative os serviços obrigatórios:
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
Instalar dependências
👉💻 Acesse o nível 4 e instale os pacotes Python necessários:
cd $HOME/way-back-home/level_4
uv sync
As principais dependências são:
Pacote | Finalidade |
| Framework da Web de alto desempenho para a estação de satélite e streaming SSE. |
| Servidor ASGI necessário para executar o aplicativo FastAPI |
| O Kit de Desenvolvimento de Agente usado para criar o agente de formação |
| Biblioteca de protocolo Agent-to-Agent para comunicação padronizada. |
| Cliente nativo para acessar modelos do Gemini. |
| Cliente Python para se conectar ao Schematic Vault (Memorystore). |
| Suporte para comunicação bidirecional em tempo real |
| Gerencia variáveis de ambiente e secrets de configuração. |
| Validação de dados e gerenciamento de configurações |
Verificar configuração
Antes de começar a codificar, vamos verificar se todos os sistemas estão funcionando. Execute o script de verificação para auditar seu projeto do Google Cloud, APIs e dependências do Python.
👉💻 Execute o script de verificação:
cd $HOME/way-back-home/level_4/scripts
chmod +x verify_setup.sh
. verify_setup.sh
👀 Uma série de marcas de seleção verdes (✅) vai aparecer.
- Se você vir Cruzes vermelhas (❌), siga os comandos de correção sugeridos na saída (por exemplo,
gcloud services enable ...oupip install ...). - Observação:um aviso amarelo para
.envé aceitável por enquanto. Vamos criar esse arquivo na próxima etapa.
🚀 Verifying Mission Bravo (Level 4) Infrastructure... ✅ Google Cloud Project: xxxxxxx ✅ Cloud APIs: Active ✅ Python Environment: Ready 🎉 SYSTEMS ONLINE. READY FOR MISSION.
3. Como criar o Schematic Vault no Redis e o agente bidirecional com o ADK
Você localizou o repositório esquemático planetário que contém os blueprints do foguete abandonado. Para recuperar esses dados com precisão, é necessário interagir com a interface de gerenciamento dedicada do repositório: o agente do Architect.
Provisionamento do cofre do Schematic (Redis)
Antes que o arquiteto possa nos ajudar, precisamos garantir que os dados estejam hospedados em um ambiente seguro e de alta disponibilidade. Vamos usar o Redis como um repositório de dados rápido para nossos esquemas alienígenas. Para facilitar o desenvolvimento, vamos criar uma instância local do Redis, mas as instruções sobre como implantar em um ambiente de produção com o Google Cloud Memorystore serão fornecidas mais tarde.
👉💻 Execute os seguintes comandos no terminal para provisionar a instância do Redis. Isso pode levar de 2 a 3 minutos:
docker run -d --name ozymandias-vault -p 6379:6379 redis:8.6-rc1-alpine
👉💻 Para carregar os dados preliminares, execute o seguinte comando para entrar no Redis Shell:
docker exec -it ozymandias-vault redis-cli
(Seu comando vai mudar para 127.0.0.1:6379)
👉💻 Cole estes 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"
👉💻 Digite exit para voltar ao shell normal.
👉💻 Para verificar se os dados existem consultando um navio específico diretamente do seu terminal, execute:
# Check 'TITAN-PRIME'
docker exec ozymandias-vault redis-cli LRANGE "TITAN-PRIME" 0 -1
👀 Esta é a saída esperada:
1) "Ion Thruster" 2) "Quantum Cell" 3) "Warp Core"
Como implementar o Architect Agent
O Architect Agent é um agente especializado responsável por recuperar plantas esquemáticas do nosso cofre do Redis. Ele atua como uma interface de dados dedicada, garantindo que o agente de expedição principal receba informações precisas e estruturadas sem precisar conhecer a lógica do banco de dados subjacente.
O Kit de Desenvolvimento de Agente (ADK) do Google é o framework modular que torna possível essa configuração multiagente. Ele processa duas camadas críticas:
- Ciclo de vida da conexão e da sessão:interagir com APIs em tempo real exige um gerenciamento complexo de protocolos, incluindo processamento de handshakes, autenticação e sinais de atividade.
- A chamada de função:é a "viagem de ida e volta modelo-código-modelo". Quando o LLM decide que precisa de dados, ele gera uma chamada de função estruturada. O ADK intercepta isso, executa seu código Python (
lookup_schematic_tool) e envia o resultado de volta ao contexto do modelo em milissegundos.
Agora vamos criar o Architect. Este agente não tem acesso à câmera. Ele existe apenas para receber um "Nome da unidade" e retornar a "Lista de peças" do banco de dados.
👉💻 Vamos usar o comando "adk create". Essa é uma ferramenta do Kit de Desenvolvimento de Agente (ADK) que gera automaticamente o código boilerplate e a estrutura de arquivos de um novo agente, economizando tempo de configuração.
cd $HOME/way-back-home/level_4/backend/
uv run adk create architect_agent
Configurar o agente
A CLI vai iniciar um assistente de configuração interativo. Use as respostas a seguir para configurar seu agente:
- Escolha um modelo: selecione Opção 1 (Gemini Flash).
- Observação: a versão específica (por exemplo, 2.5, 3.0) pode variar de acordo com a disponibilidade. Sempre escolha a variante "Flash" para velocidade.
- Escolha um back-end: selecione Opção 2 (Vertex AI).
- Inserir ID do projeto do Google Cloud: pressione Enter para aceitar o padrão (detectado no seu ambiente).
- Insira a região do Google Cloud: pressione Enter para aceitar o padrão (
us-central1).
👀 Sua interação com o terminal vai ficar parecida com esta:
(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
Agora você vai ver uma mensagem de sucesso Agent created. Isso gera o código de estrutura que vamos modificar na próxima etapa.
👉✏️ Navegue até o arquivo $HOME/way-back-home/level_4/backend/architect_agent/agent.py recém-criado e abra-o no editor. Adicione o snippet da ferramenta ao arquivo depois da primeira linha de importação:
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
👉✏️ Substitua toda a linha instruction na definição root_agent pelo seguinte e adicione também a ferramenta que definimos anteriormente:
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],
A vantagem do ADK
Com o Architect on-line, agora temos uma fonte de verdade. Antes de conectar isso ao agente principal,o Kit de Desenvolvimento de Agente (ADK) oferece uma vantagem significativa ao simplificar as complexidades da criação e do teste de agentes de IA. Com o console do desenvolvedor adk web integrado, podemos isolar e verificar a funcionalidade do nosso Architect Agent, especificamente os recursos de chamada de função, antes de integrá-lo ao sistema multiagente maior. Essa abordagem modular para desenvolvimento e testes é crucial para criar aplicativos de IA robustos e confiáveis.
👉💻 No terminal, execute:
cd $HOME/way-back-home/level_4/
. scripts/check_redis.sh
cd $HOME/way-back-home/level_4/backend/
uv run adk web
👀 Aguarde até ver:
+-----------------------------------------------------------------------------+ | 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)
- Clique no ícone Visualização da Web na barra de ferramentas do Cloud Shell. Selecione Alterar porta, defina como 8000 e clique em Alterar e visualizar.
- Selecione architect_agent.
- Acionar a ferramenta:na interface de chat, digite:
CHRONOS-ALPHA(ou qualquer ID do Drive do banco de dados esquemático). - Observe o comportamento:
- O arquiteto precisa acionar imediatamente o
lookup_schematic_tool. - Devido às nossas instruções estritas do sistema, ele deve retornar apenas a lista de peças (por exemplo,
['Shield Emitter', 'Data Crystal', 'Quantum Cell']) sem palavras desnecessárias.
- O arquiteto precisa acionar imediatamente o
- Verifique os registros:consulte a janela do terminal. Você vai ver o registro de execução bem-sucedida:
[ARCHITECT] Returning schematic for CHRONOS-ALPHA: ['Shield Emitter', 'Data Crystal', 'Quantum Cell']!(architect_agent adk)[img/03-02-adkweb.png]
Se você vir o registro de execução da ferramenta e a resposta de dados limpos, o agente especialista está funcionando conforme o esperado. Ele pode processar solicitações, consultar o cofre e retornar dados estruturados.
👉💻 Pressione Ctrl+C para sair.
Inicializar o servidor A2A
Para conectar o agente de encaminhamento ao Architect, usamos o protocolo de agente para agente (A2A).
Enquanto protocolos como o MCP (Protocolo de Contexto de Modelo) se concentram em conectar agentes a ferramentas, o A2A se concentra em conectar agentes a outros agentes. É o padrão que permite ao Dispatcher "descobrir" o Architect e entender a capacidade dele de pesquisar esquemas.
O fluxo A2A:nesta missão, usamos um modelo cliente-servidor:
- Servidor (arquiteto): hospeda as ferramentas de banco de dados e "anuncia" suas habilidades com um card do agente.
- Cliente (Dispatch): lê o card do Architect, entende a API e envia uma solicitação esquemática.
O que é um card de agente?
Pense no card do agente como um cartão de visita digital ou uma "carteira de habilitação" para uma IA. Quando um servidor A2A é iniciado, ele publica este objeto JSON contendo:
- Identidade:o nome (
architect_agent) e o ID do agente. - Descrição:um resumo legível para humanos e máquinas do que ele faz ("Função do sistema: API de banco de dados...").
- Interface:as chaves de entrada (
drive_name) e os formatos de saída específicos esperados.
Sem esse card, o agente de Dispatch operaria sem informações, tentando adivinhar como se comunicar com o Architect.
Criar o código do servidor
👉✏️ No editor, em $HOME/way-back-home/level_4/backend/architect_agent, crie um arquivo chamado server.py e cole o seguinte 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)
👉💻 De volta ao terminal, navegue até a pasta e inicie o 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
👀 Confirme se o servidor A2A foi iniciado:
INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)
Verificar o cartão do agente
Abra uma nova guia do terminal (clique no ícone +). Vamos verificar se o Architect está transmitindo a identidade corretamente buscando o card do agente manualmente.
👉💻 Execute o seguinte comando:
curl -s http://localhost:8081/.well-known/agent.json | jq .
👀 Você vai receber uma resposta JSON. Procure o campo description na saída. Ele precisa corresponder à instrução que você deu ao 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"
}
Se você vir esse JSON, o Architect estará ativo, o protocolo A2A estará ativo e o card do agente estará pronto para ser descoberto pelo Dispatcher.
Agora que o Architect está pronto para servir como um recurso remoto, podemos conectá-lo ao Agente de envio.
👉💻 Pressione Ctrl+C para sair do servidor A2A.
4. Como conectar o agente de fluxos BIDI ao agente remoto e às ferramentas de streaming
Agora você vai configurar o hub de comunicação principal para diminuir a distância entre os dados em tempo real e o Architect remoto. Essa conexão exige um pipeline de alta largura de banda e baixa latência para garantir que a bancada de montagem permaneça estável durante a operação.
Entender agentes de streaming bidirecional (ao vivo)
O streaming bidirecional (Bidi) no ADK adiciona aos agentes de IA a capacidade de interação de voz e vídeo bidirecional de baixa latência da API Gemini Live. Ela representa uma mudança fundamental nas interações tradicionais de IA. Em vez do padrão rígido de "perguntar e esperar", ele permite uma comunicação bidirecional em tempo real em que humanos e IA podem falar, ouvir e responder simultaneamente.
Pense na diferença entre enviar e-mails e ter uma conversa por telefone. As interações tradicionais com o agente são como e-mails: você envia uma mensagem completa, aguarda uma resposta completa e envia outra. O streaming bidirecional é como uma conversa por telefone: fluida, natural, com a capacidade de interromper, esclarecer e responder em tempo real.
Principais características:
- Comunicação bidirecional:troca contínua de dados sem esperar por respostas completas. A IA responde assim que detecta que o usuário terminou de falar.
- Interrupção responsiva:os usuários podem interromper o agente no meio da resposta com uma nova entrada, assim como em uma conversa humana. Se uma IA estiver explicando uma etapa complexa e você disser "Espere, repita isso", ela vai parar imediatamente e responder à sua interrupção.
- Otimizado para multimodalidade:o Bidi-streaming é excelente para processar diferentes tipos de entrada simultaneamente. Você pode falar com o agente enquanto mostra as peças do alienígena por vídeo, e ele processa os dois fluxos em uma única conexão unificada.
👀 Antes de implementar a lógica do cliente, vamos examinar o esqueleto pré-gerado do agente de envio. Esse agente vai se comunicar com o usuário por voz e vídeo e delegar consultas ao agente de arquitetura.
__init__.py agent.py hazard_db.py
agent.py: é o "cérebro". No momento, ele contém uma configuração básica de streaming Bidi. Vamos modificar esse arquivo para adicionar a lógica do cliente A2A para que ele possa se comunicar com o arquiteto.hazard_db.py: é uma ferramenta local específica do agente de despacho que contém protocolos de segurança. Ele é separado do banco de dados esquemático do Architect.
Implementar o cliente A2A
Para permitir que o agente de encaminhamento se comunique com nosso arquiteto remoto, precisamos definir um agente A2A remoto. Isso informa ao agente de encaminhamento onde encontrar o Architect e como é o "card do agente" dele.
👉✏️ Substitua #REPLACE-REMOTEA2AAGENT em $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py pelo seguinte:
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,
)
Como as ferramentas de streaming funcionam
Com o agente anterior, as ferramentas seguiam um padrão "Solicitação-Resposta". O agente fazia uma pergunta, a ferramenta dava uma resposta e a interação terminava. No entanto, em Ozymandias, os perigos não esperam que você pergunte se eles estão presentes. Para isso, você precisa de uma ferramenta de streaming.
As ferramentas de streaming permitem que as funções transmitam resultados intermediários de volta ao agente em tempo real, permitindo que ele reaja às mudanças à medida que elas acontecem. Os casos de uso comuns incluem monitorar preços de ações flutuantes ou, no nosso caso, monitorar um stream de vídeo ao vivo para mudanças de estado.
Ao contrário das ferramentas padrão, uma ferramenta de streaming é uma função assíncrona que atua como um AsyncGenerator. Isso significa que, em vez de return um único valor, ele yield várias atualizações ao longo do tempo.
Para definir uma ferramenta de streaming no ADK, é necessário obedecer a estes requisitos técnicos:
- Função assíncrona:a ferramenta precisa ser definida com
async def. - Tipo de retorno AsyncGenerator:a função precisa ser digitada para retornar um
AsyncGenerator. O primeiro parâmetro é o tipo de dados que está sendo gerado (por exemplo,str) e o segundo geralmente éNone. - Streams de entrada:usamos ferramentas de streaming de vídeo. Nesse modo, o fluxo de vídeo/stream de áudio real (o
LiveRequestQueue) é transmitido diretamente para a função, permitindo que a ferramenta "veja" os mesmos frames que o agente.
Pense em uma ferramenta de streaming como um Sentinel. Enquanto você e o agente da Dispatch discutem plantas, o sentinela é executado em segundo plano, processando silenciosamente cada frame de vídeo para garantir sua segurança.
Implementar a ferramenta de monitoramento em segundo plano
Agora vamos implementar a ferramenta monitor_for_hazard. Essa ferramenta vai ingerir os input_stream (frames de vídeo), analisá-los usando uma chamada de visão separada e leve e yield um aviso somente quando um risco for detectado.
👉✏️ Em $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py, substitua #REPLACE_MONITOR_HAZARD pela seguinte 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)
Implementar o agente de expedição
O Agente de encaminhamento é sua interface principal e o orquestrador. Como ele gerencia o link de streaming bidirecional (sua voz e vídeo ao vivo), ele precisa manter o controle da conversa em todos os momentos. Para isso, vamos usar um recurso específico do ADK: Agent-as-a-Tool.
Conceito: agente como ferramenta x subagentes
Ao criar sistemas multiagente, você precisa decidir como a responsabilidade é compartilhada. Na nossa missão de resgate, a distinção é fundamental:
- Agent-as-a-Tool::essa é a abordagem recomendada para nosso hub de streaming bidirecional. Quando o agente de atendimento (agente A) chama o agente de arquitetura (agente B) como uma ferramenta, os dados do agente B são transmitidos de volta para o agente A. Em seguida, o Dispatch interpreta esses dados e gera uma resposta para você. O envio permanece no controle e continua processando todas as entradas subsequentes do usuário.
- Subagente:em uma relação de subagente, a responsabilidade é totalmente transferida. Se o Dispatch transferisse você para o Architect como um subagente, você estaria falando diretamente com uma API de banco de dados sem "visão" e sem habilidades de conversa. O agente principal (Dispatch) ficaria de fora.
Ao usar o Agent-as-a-Tool, aproveitamos o conhecimento especializado do Architect e mantemos a interação fluida e humana do agente de streaming bidirecional.
Como programar a lógica de roteamento
Agora vamos incluir nosso architect_agent em um AgentTool e fornecer ao agente de envio um "Mapa lógico". Esse mapa informa ao agente exatamente quando buscar dados do cofre e quando informar descobertas do sentinel em segundo plano.
Para dar ao Dispatch "olhos" que nunca piscam, precisamos conceder acesso à ferramenta de streaming que criamos na etapa anterior.
No ADK, quando você adiciona uma função AsyncGenerator (como monitor_for_hazard) à lista tools, o agente a trata como um processo em segundo plano persistente. Em vez de uma execução única, o agente "se inscreve" na saída da ferramenta. Isso permite que o Dispatch continue a conversa principal enquanto o Sentinel gera alertas de risco silenciosamente em segundo plano.
👉✏️ Substitua #REPLACE_AGENT_TOOLS em $HOME/way-back-home/level_4/backend/dispatch_agent/agent.py pelo seguinte:
tools=[AgentTool(agent=architect_agent), monitor_for_hazard],
Verificação
👉💻 Com os dois agentes configurados, podemos testar a interação multiagente ao vivo.
- No terminal A, inicie o Architect Agent:
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
- Em um novo terminal (terminal B), execute o agente de despacho:
cd $HOME/way-back-home/level_4/backend/
cp architect_agent/.env .env
uv run adk web
Testar um sistema multiagente que usa um modelo multimodal em tempo real, como o gemini-live, no simulador adk web envolve um fluxo de trabalho específico. O simulador é excelente para inspecionar chamadas de ferramentas, mas tem uma incompatibilidade conhecida ao processar imagens pela primeira vez com esse tipo de modelo.
- Clique no ícone Visualização da Web na barra de ferramentas do Cloud Shell. Selecione Alterar porta, defina como 8000 e clique em Alterar e visualizar.
👉Selecione dispatch_agent e faça upload do Blueprint e do tratamento do erro esperado
Essa é a etapa mais importante. Precisamos fornecer o contexto da imagem ao agente.
- Quando a interface carregar, permita o acesso ao microfone quando solicitado.
- Faça o download desta imagem do blueprint para seu computador:
- Na interface
adk web, clique no ícone de clipe de papel e faça upload da imagem do projeto que você acabou de baixar.
⚠️⚠️Você vai receber um erro 400 INVALID_ARGUMENT. Isso é esperado.⚠️⚠️
Esse erro ocorre porque o gerenciador de imagens adk web não é totalmente compatível com a API do modelo gemini-live para um upload único. No entanto, a imagem foi adicionada ao contexto da sessão.
- 👉 Para corrigir o erro, basta recarregar a página do navegador.
Acionar o processo de montagem
👉 Depois de recarregar, o erro vai desaparecer, e você verá a imagem do projeto no histórico de chat. Agora o agente tem o contexto visual necessário.
- Clique no ícone do microfone para ativar. A interface vai mostrar "Ouvindo...".
- Diga o comando de voz: "começar a montar".
- O agente vai processar sua solicitação, e a interface vai mudar para "Falando...". Você vai ouvir uma resposta apenas de áudio com as peças necessárias.
4. Verificar as chamadas de ferramentas de agente para agente
👉 A resposta de áudio inicial confirma que o sistema está funcionando, mas a verdadeira mágica está no rastreamento de comunicação multiagente.
- Desative o microfone.
- Atualize a página mais uma vez.
O painel "Rastreamento" à esquerda será preenchido. Você pode conferir o fluxo de execução completo e bem-sucedido:
- O
dispatch_agentprimeiro chamamonitor_for_hazard. - Em seguida, ele faz várias chamadas
execute_architectpara oarchitect_agente extrai os dados esquemáticos.
Essa sequência confirma que todo o fluxo de trabalho multiagente está funcionando corretamente: o dispatch_agent recebeu a solicitação, delegou a tarefa de recuperação de dados ao architect_agent por uma chamada de ferramenta e recebeu os dados de volta para atender ao comando do usuário.
Seu link de streaming bidirecional agora é capaz de fazer monitoramento em segundo plano e colaboração multiagentes. Em seguida, vamos aprender a analisar essas respostas complexas no front-end.
👉💻 Pressione Ctrl+c nos dois terminais para sair.
5. Uma análise detalhada sobre fluxos de eventos multimodais ao vivo
Na etapa anterior, verificamos com sucesso nosso sistema multiagente usando o servidor de desenvolvimento integrado, o adk web. Essa utilidade usa um executor padrão do ADK para gerenciar automaticamente a sessão, os fluxos e o ciclo de vida do agente. No entanto, para criar um aplicativo independente pronto para produção, como nosso serviço FastAPI (main.py), precisamos de controle explícito. Precisamos criar e gerenciar manualmente o ADK Runner para processar sessões de usuários ativas, já que ele é o componente principal que processa os fluxos bidirecionais de áudio, vídeo e texto.
O loop modelo-código-modelo
Para entender como o sistema opera em tempo real, vamos acompanhar o ciclo de vida de uma única sessão de missão. Esse loop representa a troca contínua de objetos LlmRequest e LlmResponse.
- O link visual:você inicia a conexão e compartilha sua webcam/tela. Os frames JPEG de alta fidelidade começam a fluir upstream via
realtimeInput(usando oLiveRequestQueue). - Ativação do Sentinel:o sistema envia um estímulo inicial "Olá". De acordo com as instruções, o agente de encaminhamento aciona imediatamente a
monitor_for_hazardferramenta de streaming. Isso inicia um loop em segundo plano que monitora silenciosamente todos os frames recebidos. - Comando do piloto:fale no intercomunicador: "Comece a montagem".
- Vocal Upstream:sua voz é capturada como áudio de 16 kHz e enviada Upstream junto com os frames de vídeo.
- Delegação (A2A): o Dispatch "ouve" sua intenção. Ele percebe que não tem os esquemas e chama o Architect Agent usando o protocolo
AgentTool(agente como ferramenta). - Recuperação de fatos:o Architect consulta o banco de dados do Redis e retorna a lista de peças para o Dispatch. O Dispatch continua sendo o "Master of the Session" (Mestre da sessão), recebendo os dados sem transferir você.
- Downstream informativo:o Dispatch envia um
modelTurn(Downstream) com texto e áudio nativo: "Arquiteto confirmado. O subconjunto necessário é: Warp Core, Flux Pipe, Ion Thruster." - A crise:de repente, uma peça na bancada se desestabiliza e começa a brilhar em branco.
- Detecção autônoma:o loop
monitor_for_hazardem segundo plano (o Sentinel) detecta o frame JPEG específico que contém o brilho. Ele processa o frame chamando o Gemini e identifica o risco. - Segurança downstream:a ferramenta de streaming
yieldsum resultado. Como esse é um agente de streaming bidirecional, o Dispatch pode interromper o estado atual para enviar imediatamente um aviso de segurança crítico downstream: "Perigo detectado! Neutralizando o Cristal de Dados agora. Mova para a caixa VERMELHA."
Definir a configuração do ambiente de execução do agente
O RunConfig no ADK permite a configuração detalhada do comportamento de um agente, incluindo como ele processa dados de streaming e interage com várias modalidades.
O streaming_mode é definido como BIDI para comunicação bidirecional em tempo real, permitindo que o usuário e o agente falem e ouçam simultaneamente. O parâmetro response_modalities define os tipos de saída que o agente pode produzir, como voz e texto. O input_audio_transcription configura como o agente processa e transcreve a fala do usuário. Para criar uma experiência mais resiliente, o session_resumption permite que o agente se lembre do contexto da conversa e retome se a conexão for perdida. Por fim, o proactivity permite que o agente inicie ações ou falas sem um comando direto do usuário, como emitir um aviso de risco espontâneo, enquanto o enable_affective_dialog permite que o agente gere respostas mais naturais e empáticas. Saiba mais sobre o RunConfig do ADK aqui.
👉✏️ Localize o marcador #REPLACE_RUN_CONFIG no arquivo $HOME/way-back-home/level_4/backend/main.py e substitua-o pela seguinte lógica de dissecção:
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,
)
Implementar o Request to Agent
Em seguida, vamos implementar o uplink de comunicações principal que transmite dados multimodais em tempo real da área de trabalho volátil do usuário para o agente de despacho via WebSocket. O agente "vê" (frames de vídeo) e "ouve" (comandos de voz) continuamente. A lógica recebe continuamente o fluxo de dados, distingue entre os blocos de áudio binários recebidos e os pacotes de texto/imagem encapsulados em JSON, encapsulando-os em objetos Blob (para multimídia) ou Content (para texto) e enviando-os para o LiveRequestQueue para alimentar a sessão do agente bidirecional.
Localize o marcador #PROCESS_AGENT_REQUEST no arquivo $HOME/way-back-home/level_4/backend/main.py e substitua-o pela seguinte lógica de dissecção:
# 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
Os dados multimodais agora estão sendo enviados ao agente.
Implementação da resposta: a estrutura de dados de eventos downstream
Ao executar um agente bidirecional (ao vivo) com o ADK, os dados retornados pelo agente são agrupados em um tipo específico de Evento que herda das estruturas principais do SDK de IA generativa. O objeto Event que você recebe no loop async for event in runner.run_live(...) é um único objeto que contém vários campos opcionais, cada um para um tipo diferente de informação:
Como o conteúdo é estruturado:
- Quando o agente fala (via
.server_content): o campo não é apenas texto simples. Ele contém uma lista deParts. CadaParté um contêiner para um tipo de dado: uma string de texto (como"The part is stable.") ou um blob de áudio bruto (a voz). - Quando o agente age (via
.tool_call): o campo contém uma lista de objetosFunctionCall. CadaFunctionCallé um objeto simples e estruturado que especifica o nome da ferramenta e os argumentos de entrada em um formato limpo que seu código de back-end pode ler e executar facilmente.
👀 Se você analisasse um único Event gerado pelo loop run_live, o JSON (produzido por event.model_dump(by_alias=True)) seria assim, seguindo estritamente os formatos do SDK da 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" }
}
]
}
}
👉✏️ Agora vamos atualizar o downstream_task em main.py para encaminhar os dados de eventos completos. Essa lógica garante que cada "pensamento" da IA seja registrado no terminal de diagnóstico da nave e enviado como um único objeto JSON para a interface do usuário de front-end.
Localize o marcador #PROCESS_AGENT_RESPONSE no arquivo $HOME/way-back-home/level_4/backend/main.py e substitua-o pela seguinte lógica de dissecção:
# 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)
Execução da missão
Com o cofre de back-end conectado e os dois agentes configurados, todos os sistemas estão prontos para a missão. As etapas a seguir vão iniciar o aplicativo completo, permitindo que você interaja com o sistema de dois agentes que acabou de criar.
Objetivo:monte o propulsor de dobra atribuído aleatoriamente que aparece na sua bancada. Protocolo:siga as orientações vocais do agente de despacho, principalmente os avisos de risco para componentes específicos.
Ativar o especialista (o arquiteto)
👉💻 Na primeira janela de terminal, inicie o agente Architect. Esse serviço de back-end se conecta ao cofre do Redis e aguarda solicitações esquemáticas do 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
Mantenha esse terminal em execução. Agora ele é seu "agente de banco de dados" ativo.)
Iniciar o Cockpit (The Dispatcher)
👉💻 Em uma nova janela de terminal (Terminal B), vamos criar a interface do usuário do front-end e iniciar o agente principal do Dispatch, que serve a interface do usuário e lida com toda a comunicação em tempo real.
# 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
Isso inicia o servidor principal na porta 8080.
Executar o cenário de teste
O sistema já está disponível. Seu objetivo é seguir as instruções do agente para concluir a montagem.
- 👉 Acesse o Workbench:
- Clique no ícone Visualização da Web na barra de ferramentas do Cloud Shell.
- Selecione Alterar porta, defina como 8080 e clique em Alterar e visualizar.
- 👉 Inicie a missão:
- Quando a interface carregar, permita o acesso à tela e ao microfone.
- Você vai precisar selecionar uma guia ou janela para compartilhar. Se você estiver compartilhando a janela, para evitar problemas, verifique se essa é a ÚNICA guia na janela.
- Uma unidade com um nome aleatório (por exemplo, "NOVA-V", "OMEGA-9").
- Quando a interface carregar, permita o acesso à tela e ao microfone.
- 👉 O loop de montagem:
- Comando:para começar a montar o drive, diga: "Começar a montar".
- Resposta do arquiteto:o agente vai fornecer as peças corretas para montar o drive.
- Verificação de risco:quando uma peça parece ser perigosa na bancada:
- A ferramenta
monitor_for_hazarddo agente Dispatch vai identificar visualmente o agente. - Isso vai gerar um "ALERTA DE RISCO VISUAL". Isso vai levar cerca de 30 segundos.
- Ele vai verificar qual compartimento usar para desengatar o risco.
- A ferramenta
- Ação:o agente de despacho vai dar um comando direto: "Perigo confirmado. Coloque XXX na lixeira vermelha imediatamente." Siga esta instrução para continuar.
- Comando:para começar a montar o drive, diga: "Começar a montar".
Missão cumprida. Você criou um sistema multiagente interativo. Os sobreviventes estão seguros, o foguete saiu da atmosfera e sua "Volta para casa" continua.
👉💻 Pressione Ctrl+c nos dois terminais para sair.
6. Implantar na produção (opcional)
Você testou o agente localmente. Agora, precisamos fazer upload do núcleo neural do arquiteto para os mainframes da nave (Cloud Run). Isso permite que ele opere como um serviço permanente e independente que o agente de encaminhamento pode consultar de qualquer lugar.
Provisionar o Cofre seguro (infraestrutura)
Antes de implantar o agente, precisamos criar a memória permanente dele (Memorystore) e o canal seguro para acessá-la (conector VPC).
👉💻 Crie a instância do Memorystore (Redis Vault):
export REGION="us-central1"
gcloud redis instances create ozymandias-vault-prod --size=1 --tier=basic --region=${REGION}
👉💻 Recupere o endereço de rede do Vault: execute este comando e copie o endereço IP host. Esse é o endereço particular da nova instância do Redis.
gcloud redis instances describe ozymandias-vault-prod --region=us-central1
👉💻 Crie o conector de acesso à VPC (ponte segura): esse conector funciona como uma ponte particular, permitindo que o Cloud Run acesse a instância do Redis na sua 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
👉💻 Carregue os dados:
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
Implantar o aplicativo de agente
Compilar e criar imagem do agente
👉💻 Navegue até o diretório de back-end e crie o 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
👉💻 Empacote o aplicativo em uma imagem de contêiner.
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}
Implantar no Cloud Run
👉💻 Implante o agente no Cloud Run. Vamos injetar o IP do Redis e vincular o conector da VPC diretamente ao comando de inicialização. Isso garante que o agente comece com uma conexão segura e privada ao banco de dados.
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"
👉💻 Verifique se o servidor A2A está em execução.
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
Quando o comando for concluído, um URL do serviço vai aparecer. O Architect Agent agora está ativo na nuvem, conectado permanentemente ao cofre e pronto para fornecer dados esquemáticos a outros agentes.
Implantar o Dispatch Hub no mainframe de produção
Com o Architect Agent operacional na nuvem, agora precisamos implantar o Dispatch Hub. Esse agente vai servir como a principal interface do usuário, processando streams de voz/vídeo ao vivo e delegando consultas de banco de dados ao endpoint seguro do Architect.
👉💻 Execute o comando a seguir no terminal do Cloud Shell. Isso vai criar o Dockerfile completo de vários estágios no diretório de back-end.
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
Compilar e criar imagem do agente/front-end
👉💻 Navegue até o diretório de back-end que contém o código do agente Dispatch (main.py) e empacote-o em uma imagem de contêiner.
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}
Implantar no Cloud Run
👉💻 Implante o Dispatch Hub no Cloud Run. Vamos injetar o URL do arquiteto como uma variável de ambiente, criando o link essencial entre nossos dois agentes nativos da nuvem.
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}"
Quando o comando for concluído, você verá um URL de serviço (por exemplo, https://mission-bravo-...run.app). O aplicativo agora está ativo na nuvem.
👉 Acesse a página Google Cloud Run e selecione o serviço biometric-scout na lista. 
Localize o URL público exibido na parte de cima da página de detalhes do serviço. 
Verificação final do sistema (teste de ponta a ponta)
👉 Agora você vai interagir com o sistema ativo.
- Extraia o URL:copie o URL do serviço da saída do último comando de implantação. Ele deve terminar com
run.app. - Abra o Cockpit:cole o URL no navegador da Web.
- Iniciar contato:quando a interface carregar, permita o acesso à tela e ao microfone.
- Solicitar dados:quando uma unidade é atribuída, peça para começar a montagem. Por exemplo: "Começar a montar"
Agora você está interagindo com um sistema multiagente totalmente implantado que é executado inteiramente no Google Cloud.
O sistema multiagente trava o anel de contenção final no lugar, e a radiação errática se transforma em um zumbido constante.
"WARP DRIVE: ESTABILIZADO. Rescue Craft: ENGINES IGNITED."
No monitor, a nave alienígena sobe em alta velocidade, escapando por pouco da superfície em ruínas de Ozymandias enquanto a atmosfera entra em colapso. Ele se estabelece em uma órbita segura ao lado da sua nave, e as comunicações se enchem das vozes dos sobreviventes, abalados, mas vivos. Com o resgate concluído e o caminho para casa livre, o link remoto é desconectado.
Graças a você, os sobreviventes foram resgatados.
Se você participou do Nível 0, não se esqueça de verificar seu progresso na missão "A caminho de casa".
