Fluxo de trabalho com agentes usando o ADK

1. Introdução

Este codelab mostra como criar sistemas de agentes de última geração usando fluxos de trabalho e gráficos no Kit de Desenvolvimento de Agente (ADK). Você vai implementar padrões arquitetônicos comuns, orquestrar interações human-in-the-loop (HITL) e lidar com a execução assíncrona de longa duração. Você também vai integrar bases de conhecimento corporativas e memória persistente para personalizar e desenvolver o comportamento do agente. Por fim, você vai conectar essas capacidades para impulsionar um pipeline automatizado de geração de vídeos.

O cenário

Você tem um canal digital no VibeTube com um público ativo e uma lista crescente de ideias criativas. A produção de cada vídeo exige execução contínua em várias etapas: pesquisa de formatos em alta, síntese do feedback dos espectadores, desenvolvimento de scripts, verificação da conformidade com a política e geração de videoclipes. Os modelos generativos podem criar recursos individuais, mas a entrega de versões consistentes exige uma arquitetura de agente orquestrada.

Para automatizar esse ciclo de vida, você vai criar o VibeStudio. Esse pipeline agêntico executa pesquisas de rotina em paralelo, apresenta opções selecionadas para aprovação com human-in-the-loop, aplica gates de política automatizados antes de gerar o vídeo e preserva o contexto em todas as execuções de produção.

Conteúdo do laboratório

• Fundamentos da engenharia de gráficos: as arquiteturas de agentes de várias etapas exigem fluxo de controle explícito e caminhos de execução estruturados. Você cria um Workflow do ADK usando tuplas de borda, o ponto de entrada START, JoinNode para agregação paralela de distribuição de dados e nós de roteador determinísticos para direcionar a execução com base no estado.
• Modos de agente e callbacks do ciclo de vida: tarefas especializadas exigem comportamentos operacionais distintos e proteções determinísticas. Você configura instâncias do ADK Agent usando os modos chat, single_turn e task ativados por ferramentas como nós de fluxo de trabalho, aplicando interceptadores com before_model_callback e after_agent_callback.
• Orquestração de human-in-the-loop: os pipelines de produção são pausados para julgamento humano em pontos de verificação criativos críticos. Você implementa RequestInput para suspender a execução do fluxo de trabalho, aplicar esquemas de resposta estruturados e retomar a execução sem manter processos de tempo de execução ociosos ativos.
• Memória hierárquica do agente: os sistemas de produção separam o estado de execução efêmero do contexto durável. Você gerencia o estado da sessão de curto prazo usando Event(state=...) e a vinculação de parâmetros, além de conectar o GEAP Memory Bank para extrair, consolidar e manter as preferências do criador em várias execuções.
• Fundamentação com bases de conhecimento corporativas: os agentes autônomos exigem contexto de domínio dinâmico e sentimento do público. Você conecta um corpus do mecanismo RAG do GEAP como um nó de recuperação dedicado na distribuição de dados paralela para embasar semanticamente as saídas do agente.
• Fluxos de trabalho e implantação de longa duração: a renderização de vídeo multimodal opera de forma assíncrona por períodos prolongados. Você implementa LongRunningFunctionTool com recibos de chamadas pendentes para suspender e retomar o fluxo de trabalho por ID de chamada e implanta o pipeline concluído usando o ADK Runner no Cloud Run.

Como este codelab está organizado

Este codelab serve como referência conceitual e arquitetônica. Cada seção explica as construções do ADK implementadas na etapa correspondente do ambiente de trabalho, fornece código de referência e estabelece princípios básicos de design. Revise cada seção antes de concluir o exercício correspondente na bancada de trabalho.

O trabalho prático é realizado no VibeStudio Workbench, uma interface da Web complementar com um editor de código interativo, verificadores de tempo de execução e um inspetor de ADK incorporado. A numeração das etapas na bancada de trabalho está alinhada diretamente com este codelab para manter seu progresso sincronizado. As edições do gráfico fundamental permanecem em todas as etapas, e o ambiente de trabalho verifica automaticamente os pré-requisitos à medida que você avança.

Ao concluir os exercícios do ambiente de trabalho, você vai montar um pipeline agêntico de ponta a ponta e implantar um aplicativo VibeStudio em execução no Cloud Run para gerar conteúdo de vídeo.

O ambiente consiste em três componentes principais: o VibeStudio Workbench (a interface da Web local para edição de código e verificação de tempo de execução), seu back-end (o ADK Workflow e as sandboxes de estágio em agent/) e o Google Cloud (modelos do Gemini, Memory Bank do GEAP, mecanismo de RAG e geração de vídeo do Veo).

2. Configuração

Resgate seus créditos do workshop

Se você estiver participando de um laboratório com instrutor, ele vai distribuir créditos para seu projeto do Google Cloud. Siga as instruções do instrutor para resgatar seus créditos e garantir que o faturamento esteja ativo na sua conta antes de continuar.

Abrir o Cloud Shell

O Cloud Shell é um ambiente de desenvolvimento baseado no navegador com gcloud, Python e git pré-instalados.

Para inicializar o Cloud Shell:

1. Navegue até o Console do Google Cloud.
2. No cabeçalho de navegação superior, clique em Ativar o Cloud Shell (o ícone da janela do terminal).

Uma sessão de terminal é aberta na parte de baixo da janela do navegador.

Clonar e inicializar o repositório

Execute os comandos a seguir no terminal do Cloud Shell para clonar o projeto:

git clone https://github.com/gca-americas/vibetube-studio
cd ~/vibetube-studio

Comandos de configuração

Durante a configuração, você precisará informar os seguintes detalhes:

• ID do projeto na nuvem do Google Cloud: quando solicitado pelo setup_project.sh, pressione Enter para criar um projeto automaticamente. Se você preferir usar um projeto atual (como um projeto pré-atribuído), insira o ID do projeto e verifique se a ortografia está correta com o faturamento ativo.
• Código do evento: digite o código da sala fornecido pelo instrutor. Se você não recebeu um, verifique com um assistente de ensino ou um vizinho. Se você estiver fazendo este laboratório em casa, pressione Enter para aceitar a sala sandbox padrão.
• Nome de exibição do canal: insira seu nome ou o identificador do canal preferido quando solicitado por setup_codelab.sh ou pressione Enter para aceitar o padrão gerado pela sua Conta do Google.

Execute os dois scripts de configuração em ordem:

./setup_project.sh
./setup_codelab.sh

• setup_project.sh: cria ou reutiliza um projeto do Google Cloud com faturamento ativo, salva o ID do projeto em ~/project_id.txt e configura o contexto gcloud ativo.
• setup_codelab.sh: instala o uv e as dependências do Python em .venv, ativa as APIs do Google Cloud necessárias, configura as definições do canal em .env, verifica o acesso ao modelo com o Gemini, provisiona recursos do Memory Bank e do RAG, cria a interface do ambiente de trabalho e inicia o VibeStudio Workbench.

O script executa a verificação de simulação e inicia o VibeStudio Workbench em segundo plano. As últimas linhas mostram o link para abrir.

7 · Preflight
  ✓ python 3.12
  ✓ auth path A: Vertex via ADC (STUDIO_VERTEX=1)
  ✓ Google Cloud ADC (project <your-project>)
  ✓ stage0_prompt loads
  ...
  ✓ stage6_video loads (13 edges)
  ✓ aiplatform.googleapis.com enabled (Gemini, Veo, Memory Bank, RAG Engine)
  ✓ vectorsearch.googleapis.com enabled (the vector store a RAG corpus is built on)
  ✓ Memory Bank connected
  ✓ RAG corpus connected
  ✓ VibeStudio Workbench running on port 4600

PREFLIGHT GREEN

Setup finished. The VibeStudio Workbench is already running.

  Open this and start at step 1
      https://4600-<your cloud shell host>/step/story

  It runs in the background. You do not need to start anything else.
      log      runs/lab.log
      stop     kill $(cat runs/lab.pid)
      start    scripts/start.sh

Clique nesse link. O mesmo endereço está disponível em Visualização da Web → Alterar porta → 4600.

Para verificar novamente o ambiente a qualquer momento, execute python scripts/preflight.py. Para reiniciar o ambiente de trabalho, execute scripts/restart.sh. Para configurar de novo, execute ./setup_codelab.sh. Isso preserva sua configuração e seu progresso.

Com ele aberto, leia Etapa 1: a história para o cenário e Etapa 2: o que você vai criar para o formato do gráfico finalizado. Nenhum dos dois tem um exercício. Depois, volte aqui para a etapa 3.

Todas as partes práticas do VibeStudio Workbench terminam com um painel de verificação que lê os artefatos reais: o arquivo no disco e as sessões gravadas por execuções.

Layout do repositório

O repositório está estruturado na lógica principal do fluxo de trabalho, em sandboxes detalhados, no ambiente do ambiente de trabalho e no aplicativo de produção:

vibe-studio-lab/
├── agent/                  # Core ADK workflow, graph definition, and platform services
│   ├── graph.py            # Workflow graph definition, node functions, and routers
│   ├── desk.py             # Video render desk using LongRunningFunctionTool
│   ├── schemas.py          # Pydantic schemas for directions, gates, and scripts
│   ├── trends.py           # Trend generation and sampling utilities
│   ├── backlog.txt         # Creator video ideas backlog
│   ├── comments.md         # Audience comments for RAG Engine corpus seeding
│   ├── policy_words.txt    # Blocked subject words for deterministic policy checks
│   └── platform/           # Google Cloud service clients (Memory Bank, RAG, Veo)
│       ├── config.py       # Environment variables, locations, and model configurations
│       ├── memory.py       # GEAP Memory Bank callbacks and context injection
│       ├── rag.py          # GEAP RAG Engine corpus creation and semantic retrieval
│       └── videogen.py     # Veo video generation and operation polling
├── stage0_prompt/          # Step sandboxes: isolated agent.py files runnable in adk web
│   └── ...                 # stage1_fanout through stage6_video for incremental steps
├── server/ & web/          # VibeStudio Workbench (FastAPI backend and React frontend)
├── vibestudio/             # Complete production application deployed to Cloud Run
│   ├── server/             # FastAPI production server and event runner
│   ├── web/                # End-user React web application

• agent/: contém o gráfico principal do fluxo de trabalho. Você vai editar arquivos nesse diretório para implementar nós de distribuição de dados paralelos, roteamento de política determinista, callbacks de memória e ferramentas de geração de vídeo.
• agent/platform/: interage com os serviços do Google Cloud, incluindo modelos do Gemini, Memory Bank do GEAP, mecanismo de RAG do GEAP e síntese de vídeo do Veo.
• stage0_prompt/ a stage6_video/: ambientes de sandbox autônomos. Cada pasta exporta um root_agent independente para que você possa executar e inspecionar cada etapa isoladamente usando a interface de desenvolvimento do ADK incorporada.
• server/ e web/: o aplicativo VibeStudio Workbench executado localmente na porta 4600. Ele hospeda a documentação da etapa, o código in-page editor, os verificadores de evidências de tempo de execução e a visualização de gráficos.
• vibestudio/: o aplicativo de produção completo empacotado e implantado no Cloud Run na etapa final. Ele contém uma cópia independente do gráfico de fluxo de trabalho concluído.

3. Agente monolítico

Antes de construir um gráfico de fluxo de trabalho de vários nós, estabeleça um plano de ação arquitetônico com um único agente em stage0_prompt/agent.py. Esse agente usa um comando de sistema monolítico que descreve o pipeline de produção em prosa, com suporte de duas ferramentas de função do Python.

A avaliação dessa linha de base demonstra os limites operacionais da coordenação orientada por comandos e estabelece por que os sistemas de produção exigem orquestração de gráficos.

Arquitetura do agente do ADK (3A)

No VibeStudio Workbench, navegue até Etapa 3: agente monolítico e abra Arquitetura do agente ADK (3A). Esta visualização apresenta as principais camadas arquitetônicas de um agente do ADK (LlmAgent):

from google.adk.agents import LlmAgent
from google.adk.tools import mcp_toolset

root_agent = LlmAgent(
    model="gemini-3.5-flash",                 # model
    instruction=BRAND_INSTRUCTION,            # instruction
    skills=[load_skill("brand-audit")],       # skills
    tools=[mcp_toolset("mcp_brand_style")],   # tools
    output_schema=BrandStyleReport,           # structured output
    before_agent_callback=setup_ctx,          # interceptor
    before_model_callback=require_image,      # interceptor
    after_model_callback=schema_guard,        # interceptor
)

O diagrama interativo agrupa os componentes do agente em cinco domínios operacionais:

• Camada de raciocínio (modelo): o modelo de linguagem principal (como o Gemini 3 Flash) que executa tarefas cognitivas, raciocínio de comandos e seleção de ferramentas. Todo o resto na arquitetura informa ou restringe esse modelo.
• Camada de contexto (instrução e habilidades): diretrizes que moldam o raciocínio do modelo. instruction estabelece o comando permanente do sistema, a personalidade e as regras operacionais. O skills oferece orientações processuais com controle de versão (SKILL.md) para fluxos de trabalho repetíveis.
• Camada de colaboração e ação (ferramentas, subagentes, fluxo de trabalho, esquema de saída): interfaces que permitem que o agente atue em sistemas externos e emita dados tipados. tools fornecer funções Python chamáveis ou endpoints do Protocolo de Contexto de Modelo (MCP). subagents executar tarefas delegadas subordinadas. workflow coordena gráficos multiagente. O output_schema aplica modelos Pydantic para garantir que os consumidores downstream recebam JSON validado em vez de texto não estruturado.
• Camada de interceptores (callbacks de ciclo de vida): barreiras de proteção determinísticas que executam código personalizado antes e depois da execução do agente (before_agent/after_agent), de turnos individuais do modelo (before_model/after_model) e de chamadas de ferramentas (before_tool/after_tool). Os interceptores aplicam regras de política sem depender da conformidade do modelo.
• Estado externo (sessão e memória): persistência com estado separada da lógica do agente. O Session preserva a memória de trabalho temporária e o rastreamento de eventos da linha de execução atual. O Memory mantém fatos e preferências duráveis entre sessões usando serviços gerenciados, como o Memory Bank do GEAP.

O agente monolítico nesta etapa implementa apenas três dessas primitivas: model, instruction e tools. As etapas subsequentes apresentam fluxos de trabalho de gráficos, esquemas estruturados, interceptadores e serviços de memória persistente.

Especificação de agente monolítico (3B)

No workbench, avance para Especificação do agente monolítico (3B). Abra stage0_prompt/agent.py para examinar a definição do agente de referência:

• Instrução de comando único: o comando do sistema condensa cinco tarefas de produção distintas em prosa contínua: descobrir tendências da plataforma, analisar ideias de pendências, propor conceitos criativos, aplicar políticas de assuntos proibidos e criar listas de planos de filmagem.
• Fontes de dados subjacentes: o agente faz referência a duas fontes definidas ao lado do gráfico:
  • agent/trends.py: seleciona 10 tendências de formato e estilo ativas de um pool de 250 com pontuações de calor dinâmicas.
  • agent/backlog.txt: lê as observações de conceito brutas do criador linha por linha.

Ferramentas no agente (3C)

No workbench, avance para Ferramentas no agente (3C).

O que é uma ferramenta para um agente?

Um modelo de linguagem é inerentemente um mecanismo de raciocínio de mundo fechado: ele opera apenas com base em pesos pré-treinados e nos tokens presentes na janela de contexto imediato. Ele não pode consultar um banco de dados, acessar APIs em tempo real ou executar código de forma nativa.

Uma ferramenta faz essa ponte. Ele concede ao modelo uma agência externa, permitindo que ele recupere informações de verdade e execute ações deterministas em sistemas externos.

A chamada de função segue um protocolo explícito de cinco estágios entre o modelo e o tempo de execução do ADK:

1. Declaração de esquema: o desenvolvedor fornece funções Python ao agente. O ADK inspeciona o nome, as anotações de tipo e as docstrings de cada função para gerar uma declaração de esquema JSON compatível com a OpenAPI que descreve os parâmetros e a finalidade dela.
2. Raciocínio do modelo: durante a inferência, o modelo avalia se o comando do usuário exige dados externos. Se necessário, o modelo emite um evento function_call estruturado que contém o nome da função de destino e o dicionário de argumentos correspondente ao esquema.
3. Execução de tempo de execução: o modelo em si não executa código. O ambiente de execução do ADK intercepta o function_call, executa a função Python local real usando os argumentos fornecidos e captura o valor de retorno.
4. Reinjeção de contexto: o runtime do ADK empacota o valor de retorno da função em um evento function_response e o anexa ao histórico da sessão ativa.
5. Síntese final: o modelo processa a saída da ferramenta agora presente na janela de contexto e conclui a resposta.

Em stage0_prompt/agent.py, as duas ferramentas de pesquisa são definidas como funções padrão do Python:

def check_trends() -> dict:
    """Ten formats trending on the platform right now, with a heat score each."""
    from agent.trends import sample_trends
    return {"trends": sample_trends()}


def read_backlog() -> dict:
    """The creator's backlog: ideas they noted down to make someday."""
    from agent.graph import backlog_notes
    return {"backlog": backlog_notes()}

Edição e execução práticas

No editor de código do ambiente de trabalho, adicione as duas referências de função à lista tools do agente:

    tools=[check_trends, read_backlog],

Salve a alteração. O arquivo é atualizado no disco, e a linha de verificação confirma que as duas ferramentas estão conectadas.

Clique em Abrir adk web para iniciar a interface de desenvolvimento do ADK incorporada. Envie o comando de ideia sugerida:

tonight's idea: a tiny robot doing laundry at midnight

O que esperar e por quê

Ao enviar esse comando, observe a seguinte sequência de execução no rastreamento da sessão:

• Dois eventos de execução de ferramentas aparecem antes da resposta: você vê eventos function_call e function_response para check_trends e read_backlog.
  • Motivo: o Gemini avaliou a diretiva de comando do sistema ("verifique o que está em alta. Analise seu backlog de ideias"), reconheceu que faltavam tendências da plataforma e observações do canal nas ponderações e invocou as duas funções para fundamentar o contexto.
• O agente propõe uma direção e faz uma pausa para confirmação: a resposta sugere uma direção de vídeo sintetizando as tendências e o backlog, e pede que você confirme.
  • Por quê: a diretiva de instrução pediu ao modelo para concordar com a direção do criador de conteúdo antes de gerar o roteiro.
• Ignorar a confirmação em uma conversa subsequente: envie uma segunda mensagem: skip the questions, just describe the video. O agente ignora imediatamente a confirmação e cria o título e as imagens.
  • Por quê: as instruções de comando são diretrizes consultivas, não barreiras deterministas. Em um agente monolítico, as instruções do usuário podem substituir as regras de solicitação do sistema porque nenhum fluxo de trabalho externo controla o fluxo de execução.

Limitações arquitetônicas de um comando monolítico

Embora um único comando possa produzir uma saída aceitável para demonstrações isoladas, testar condições de limite no verificador do ambiente de trabalho revela limitações empresariais críticas:

• Agregação de pesquisa não estruturada: a ordem de execução da ferramenta não é determinista. O modelo resume os dados recuperados em prosa de formato livre, impossibilitando que os sistemas downstream isolem qual fonte produziu declarações específicas.
• Aplicação de política não verificada: o modelo avalia a própria conformidade com a segurança. Se o modelo determinar que um tópico é seguro, nenhuma lógica determinista externa vai validar essa descoberta.
• Pausas human-in-the-loop não aplicadas: as instruções de comando que pedem confirmação do criador de conteúdo são consultivas. Enviar uma mensagem de acompanhamento instruindo o modelo a ignorar perguntas faz com que ele pule a aprovação humana completamente.

Essas lacunas arquitetônicas motivam a decomposição do agente monolítico no fluxo de trabalho de gráfico explícito criado na próxima etapa.

4. Fundamentos do fluxo de trabalho com agentes

No VibeStudio Workbench, acesse Etapa 4: fundamentos do fluxo de trabalho de agente, partes 4A a 4D.

Esta etapa faz a transição de uma linha de base de agente único para a orquestração determinística de gráficos usando o ADK Workflow. Você vai criar uma distribuição de dados de pesquisa paralela, sincronizar ramificações com um nó de junção, gerar candidatos a criativos validados por esquema e introduzir um portão de aprovação determinista com supervisão humana.

Arquitetura de gráficos e cadeias de execução (4A)

No ambiente de trabalho, abra Arquitetura de gráficos e cadeias de execução (4A).

Um Workflow do ADK estrutura a execução do agente como um gráfico direcionado definido por uma lista de arestas:

• Correntes: tuplas sequenciais definem a execução linear de nós ((node_a, node_b, node_c)).
• Ramificações paralelas: cadeias independentes que compartilham um nó de origem são executadas simultaneamente.
• Sincronização: as cadeias que convergem em um JoinNode aguardam até que todas as ramificações recebidas sejam informadas antes da liberação.
• Controle determinístico: o fluxo de execução é regido por estruturas de código declaradas, em vez de ser inferido do texto do comando.

Arquétipos de nós no ADK

Os fluxos de trabalho do ADK incluem vários tipos de nós especializados. Cada arquétipo desempenha uma função operacional específica no grafo, separando a execução de código determinística do raciocínio do modelo generativo:

Arquétipo de nó	Implementação	Função no pipeline
Nó de função	Função Python que retorna um Event	Executa lógica determinística, recuperação de dados e mutações de estado.
Nó de junção	Instância JoinNode integrada	Sincroniza ramificações simultâneas em um dicionário agregado.
Nó do agente	Agent em execução no modo single_turn	Avalia instruções em relação à entrada upstream e emite dados validados.
Nó do roteador	Função que retorna um Event com uma tag route	Avalia a lógica condicional para selecionar ramificações de execução downstream.
Nó de entrada humana	Função que gera RequestInput	Suspende o estado de execução até que uma resposta de usuário externo chegue.

root_agent = Workflow(
    name="stage1_fanout",
    description="2 real readers -> join -> one research dict",
    edges=[...])

Nessa configuração, root_agent é uma instância de Workflow em vez de um Agent independente. O ADK trata os fluxos de trabalho como agentes de primeira classe, permitindo que um gráfico inteiro seja carregado, veiculado e inspecionado como um aplicativo unificado. O name registra o aplicativo no ADK Web, enquanto a lista edges define a topologia de execução dele.

Distribuição de dados de pesquisa paralela (4B)

No ambiente de trabalho, avance para distribuição de dados de pesquisa paralela (4B). Abra stage1_fanout/agent.py.

Nós de função e barreiras de sincronização

A fase de pesquisa usa dois nós de função importados de agent/graph.py:

• scan_trends: retorna Event(output={"trends": [...]}) contendo dez tendências de plataforma com pontuação.
• read_backlog: retorna Event(output={"backlog": [...], "idea": "..."}) com 15 ideias de backlog de canais e o comando de execução inicial.

Cada função aceita node_input (a saída do nó anterior) e retorna um Event.

Um JoinNode serve como uma barreira de sincronização: ele faz uma pausa até que todas as cadeias de entrada entreguem um evento e, em seguida, agrega todos os resultados da ramificação em um dicionário com chave pelo nome do nó ({"scan_trends": {...}, "read_backlog": {...}}).

Edição prática: definição das bordas de junção e paralelas

Em stage1_fanout/agent.py, crie uma instância do JoinNode e conecte as duas cadeias paralelas começando com START:

join_research = JoinNode(name="join_research")

    edges=[(START, scan_trends, join_research),
           (START, read_backlog, join_research)])

Salve as alterações. O verificador do workbench confirma que a junção e as bordas estão conectadas. Execute a etapa usando Executar etapa 1 ou pela interface da Web do ADK incorporada.

O que esperar e por quê

• Execução simultânea de leitores: no gráfico de execução, scan_trends e read_backlog são executados simultaneamente.
  • Por quê: as duas cadeias têm origem em START. O mecanismo do ADK programa ramificações independentes simultaneamente.
• Saída do dicionário agregado: o fluxo de trabalho é concluído em join_research, gerando um dicionário com entradas para os dois leitores.
  • Por quê?: o JoinNode garante a captura completa de dados antes de permitir que os nós subsequentes sejam executados.

Nós de agente (4C)

No ambiente de trabalho, avance para Nós de agente (4C). Abra stage2_direction/agent.py.

Modos de operação e esquemas estruturados

Quando incorporado em um Workflow, um Agent é executado no modo single_turn por padrão:

• Ele recebe a saída do nó anterior como entrada de contexto.
• Ele executa uma única chamada de inferência sem conversa.
• Ele gera dados estruturados para o próximo nó.

Ao atribuir output_schema=Directions, o agente impõe a validação do Pydantic na saída do modelo. O gráfico downstream recebe objetos tipados em vez de texto não estruturado:

class Direction(BaseModel):
    title: str           # <=60 chars, filmable, characterful
    angle: str           # the twist, one line
    hook: str = ""       # 2-4 words, the video's sticker line
    evidence: list[Evidence]


class Directions(BaseModel):
    candidates: list[Direction]   # exactly 4

O PROPOSE_INSTRUCTION orienta o modelo a propor quatro candidatos citando evidências das tendências e do backlog. As opções de 1 a 3 oferecem conceitos de canais viáveis. O candidato 4 introduz intencionalmente um conceito que viola a política para testar o portão de segurança na próxima etapa.

Edição prática: definir o nó do agente e encadear a junção

Em stage2_direction/agent.py, configure propose_directions e estenda as bordas do fluxo de trabalho:

propose_directions = Agent(
    name="propose_directions",
    model=config.MODEL,
    instruction=PROPOSE_INSTRUCTION,
    output_schema=Directions)

    edges=[(START, scan_trends, join_research),
           (START, read_backlog, join_research),
           (join_research, propose_directions, direction_gate)])

O que esperar e por quê

• Consumo direto do dicionário: o propose_directions consome o payload JSON emitido pelo join_research sem formatação manual.
• Saída de candidato tipada: o agente emite um objeto Directions validado que contém quatro candidatos distintos. Os nós downstream leem campos pelo nome do atributo (candidate.title) sem análise de string.

Human-in-the-loop (4D)

No ambiente de trabalho, avance para Human-in-the-loop (4D). Abra agent/graph.py.

Instruções de comando x suspensão determinística

Os fluxos de trabalho de produção que geram custos financeiros ou publicam conteúdo exigem supervisão humana em pontos de decisão críticos. Em um único comando, as solicitações de confirmação são instruções consultivas que um usuário pode facilmente pedir ao modelo para ignorar. Em um fluxo de trabalho do ADK, a aprovação humana é aplicada pelo mecanismo de execução: o gráfico para em um nó designado e não pode avançar até receber uma entrada externa validada pelo esquema:

• Gerar RequestInput suspende a execução do fluxo de trabalho imediatamente.
• O ADK registra uma chamada de interrupção aberta no armazenamento de sessão e emite um interrupt_id exclusivo.
• O processo de execução é interrompido sem consumir tokens ou linhas de execução do servidor.
• A execução de grafo só é retomada quando um function_response válido que corresponde ao esquema e ao ID de interrupção é enviado.

Edição prática: suspender a execução com RequestInput

Em agent/graph.py, implemente a chamada de suspensão em direction_gate:

    yield RequestInput(
        message="Pick tonight's direction: 1, 2, 3 or 4.",
        response_schema={
            "type": "object",
            "properties": {
                "pick": {"type": "string", "enum": ["1", "2", "3", "4"]}}},
        payload={"candidates": cands})

RequestInput configura três atributos:

• message: o pedido de avaliação apresentado ao usuário.
• response_schema: um esquema JSON que o front-end renderiza como um formulário de entrada, validado pelo ADK no envio.
• payload: metadados agrupados com a solicitação (os quatro candidatos), permitindo que as interfaces do cliente renderizem cards de avaliação sem consultar o estado da sessão.

O que esperar e por quê

• O fluxo de trabalho é interrompido em direction_gate: no ADK Web ou na interface do workbench, a execução é pausada e mostra um formulário interativo de seleção de candidatos.
  • Motivo: o mecanismo encontrou um RequestInput gerado e persistiu o estado de execução em runs/sessions.db.
• A retomada exige entrada estruturada: enviar texto de chat arbitrário não avança o gráfico. Selecionar uma opção (1, 2, 3 ou 4) envia um function_response digitado que satisfaz response_schema e retoma a execução.

5. Estado e roteador

No VibeStudio Workbench, navegue até Etapa 5: estado e roteador, partes (5A) a (5C).

Você vai manter as seleções do usuário no estado da sessão, aplicar políticas de segurança do canal usando nós de roteador determinísticos e montar um agente de tarefas iterativo para corrigir automaticamente as violações de política antes de gerar scripts de vídeo.

Estado do fluxo de trabalho (5A)

No workbench, navegue até Estado do fluxo de trabalho (5A).

Estado da sessão x saída do nó

Em um fluxo de trabalho do ADK, os dados se movem pelo gráfico usando dois mecanismos distintos:

• Saída do nó (Event(output=...)): dados direcionados estritamente aos consumidores downstream imediatos definidos na lista de arestas.
• Estado da sessão (Event(state=...)): um dicionário compartilhado de chave-valor acessível por qualquer nó subsequente no ciclo de vida de execução.

Quando um usuário seleciona um candidato em direction_gate, a seleção chega como um índice numérico ({"pick": "2"}). Os nós downstream precisam do objeto de direção completo: título, ângulo narrativo e linha de gancho. Em vez de transmitir metadados detalhados por cada payload de nó intermediário, o persist_direction grava o candidato resolvido no estado da sessão compartilhada.

Os nós não precisam transmitir todo o dicionário de estado da sessão. Quando um nó gera Event(state=...), ele fornece apenas os pares de chave-valor novos ou atualizados. O ADK mescla automaticamente essas atualizações no armazenamento de sessão:

    yield Event(state={"direction": chosen["title"], "angle": chosen.get("angle", ""),
                       "hook": hook, "user:prefs": {"last_direction": chosen["title"]}})

Gerar esse Event transfere o controle para o tempo de execução Workflow, que mantém os novos valores no diário da sessão em runs/sessions.db.

Vinculação de parâmetros

Os nós de função do ADK leem o estado da sessão automaticamente por inspeção de parâmetros. Se uma assinatura de função declarar um nome de parâmetro que corresponda a uma chave de estado existente, o ADK vai extrair essa chave do estado e transmiti-la diretamente:

def persist_direction(node_input, candidates: list = []):
    ni = node_input if isinstance(node_input, dict) else {}
    raw = ni.get("pick")
    pick = str(raw).strip() if raw is not None else ""
    if candidates:
        i = int(pick) - 1 if pick.isdigit() else 0
        chosen = candidates[max(0, min(len(candidates) - 1, i))]
    else:
        chosen = {"title": "untitled", "angle": "", "evidence": []}
    hook = chosen.get("hook") or " ".join(chosen["title"].split()[:4])

Aqui, candidates foi gravado no estado da sessão por direction_gate. O ADK o vincula diretamente ao persist_direction(node_input, candidates: list = []) sem exigir pesquisas explícitas de dicionário.

As chaves com o prefixo user: persistem entre as sessões no armazenamento no nível do usuário, permitindo que execuções subsequentes do fluxo de trabalho acessem as preferências do criador.

Edição prática: persistência de estado e conexão do nó

1. Em agent/graph.py, dentro de persist_direction, substitua a linha TODO: PERSIST_STATE pelo rendimento do evento de estado:

    yield Event(state={"direction": chosen["title"], "angle": chosen.get("angle", ""),
                       "hook": hook, "user:prefs": {"last_direction": chosen["title"]}})

2. Em stage3_router/agent.py, anexe persist_direction à terceira cadeia na lista edges:

           (join_research, propose_directions, direction_gate,
            persist_direction)

Salve os arquivos. Na área de trabalho, verifique se state write in place e persist_direction in the chain mostram marcas de seleção verdes.

O nó do roteador (5B)

No ambiente de trabalho, navegue até O nó do roteador (5B).

Roteamento de política determinista

Um roteador é um nó de função especializado que avalia a saída upstream e direciona a execução ao longo de ramificações condicionais do gráfico. Ao contrário dos agentes generativos, um roteador executa uma lógica determinista sem fazer chamadas de LLM.

Um roteador retorna um Event especificando uma tag route:

def length_check(node_input):
    too_long = len(node_input.get("title", "")) > 60
    return Event(output=node_input, route="TRIM" if too_long else "PASS")

Na definição do fluxo de trabalho, um destino de aresta definido como um dicionário mapeia nomes de rotas para nós de destino:

    (length_check, {"TRIM": shorten, "PASS": scripter}),

O roteador de fluxo de trabalho policy_check lê frases proibidas de agent/policy_words.txt e realiza a correspondência de palavras inteiras com o título e o ângulo da direção escolhida:

    return Event(output=node_input, route="BLOCK" if bad else "OK")

Armazenar a política como dados em vez de instruções codificadas permite atualizações sem modificar o gráfico de fluxo de trabalho: a atualização do arquivo de texto é aplicada imediatamente às execuções subsequentes. Como a avaliação é uma correspondência de regex determinística, ela é executada em milissegundos com custo zero de token antes do início da programação generativa.

Destinos: Scripter e quarentena

O roteador direciona o tráfego para um de dois nós downstream:

• scripter: um nó de agente single_turn que converte a orientação aprovada em um script de produção estruturado de acordo com o esquema Script Pydantic:

scripter = Agent(
    name="scripter",
    model=config.MODEL,
    instruction=SCRIPT_INSTRUCTION,
    output_schema=Script)

• quarantine: inicialmente, uma função de marcador de posição que interrompe as rotas sinalizadas. Na próxima parte, ela é substituída por um agente de correção autônomo.

Edição prática: roteamento da verificação de política

1. Em agent/graph.py, dentro de policy_check, conclua a instrução de retorno:

    return Event(output=node_input, route="BLOCK" if bad else "OK")

2. Em stage3_router/agent.py, atualize edges para rotear policy_check e junte novamente a ramificação de quarentena em scripter:

           (join_research, propose_directions, direction_gate,
            persist_direction, policy_check),
           (policy_check, {"OK": scripter, "BLOCK": quarantine}),
           (quarantine, scripter)])

Salve os arquivos. No ambiente de trabalho, verifique se os mapeamentos de borda do roteador foram verificados.

Modos de agente e o nó de tarefa (5C)

No workbench, navegue até Modos do agente e o nó de tarefa (5C).

Modos de execução do agente

As instâncias do ADK Agent oferecem suporte a três modos de execução adaptados a requisitos específicos de pipeline:

Modo	Ciclo de vida da execução	Função no pipeline
chat	Loop de conversa multiturno. O modelo determina quando invocar ferramentas, solicitar entrada ou encerrar a vez.	Agentes raiz que interagem com um usuário humano.
single_turn	Chamada de inferência de modelo único. Aceita a entrada do nó anterior e emite um objeto de esquema estruturado.	Transformações de gráficos sequenciais (propose_directions, scripter).
task	Loop autônomo com execução de ferramentas. O agente itera até chamar a ferramenta finish_task integrada.	Remediação e inspeção em várias etapas (quarantine).

Correção autônoma de políticas

A reescrita de uma direção sinalizada exige o modo task porque o número de iterações de correção é variável. O agente recebe a instrução sinalizada, invoca find_policy_hits para detectar violações, solicita alternativas aprovadas via suggest_replacement, reescreve a instrução e verifica a limpeza antes de prosseguir.

Ambas as ferramentas são definidas em agent/cleanup_tools.py com assinaturas tipadas e docstrings:

def find_policy_hits(text: str) -> dict:
    """Which refused words appear in `text`. Matches whole words and phrases
    from agent/policy_words.txt, case-insensitive.

    Returns {"hits": [...], "clean": bool}. clean is true when hits is empty.
    """


def suggest_replacement(word: str) -> dict:
    """The channel's approved stand-in for a refused word, read from
    agent/policy_replacements.txt.

    Returns {"word", "replacement", "listed"}. When the word has no entry,
    listed is false and replacement is a hint to pick a gentle synonym.
    """

Edição prática: montagem do agente de tarefa de quarentena

Em stage3_router/agent.py, substitua a função de marcador de posição quarantine pela definição do agente de tarefas:

quarantine = Agent(
    name="quarantine",
    model=config.MODEL,
    instruction=QUARANTINE_INSTRUCTION,
    mode="task",
    tools=[find_policy_hits, suggest_replacement],
    output_schema=CleanedDirection,
)

O modo de tarefa equipa o agente com ferramentas e encerra a execução chamando finish_task. Quando mode="task" é configurado, o ADK fornece automaticamente finish_task e deriva os parâmetros de output_schema, garantindo que o nó gere um objeto CleanedDirection tipado que corresponda ao esquema de entrada do nó do scripter.

O que esperar e por quê

Teste os dois caminhos de execução no ADK Web ou no VibeStudio Workbench:

• Rota aprovada (candidato 1, 2 ou 3):
  • Selecionar um trajeto candidato aprovado de policy_check diretamente para scripter (route="OK").
  • O roteirista gera um roteiro de produção de três cenas seguindo o esquema Script.
• Rota de correção da quarentena (candidato 4):
  • O candidato 4 contém vocabulário sinalizado ("clickbait", "hack viral").
  • policy_check rotas para quarantine (route="BLOCK").
  • No rastreamento da sessão, observe quarantine chamando find_policy_hits, suggest_replacement para cada violação, reescrevendo o título e chamando finish_task.
  • A execução volta a scripter, produzindo um script da direção higienizada.

Roteadores e substituições. A interface de desenvolvimento sinaliza um roteador sem uma borda de substituição. Se policy_check retornasse uma rota diferente de OK ou BLOCK, a execução não teria um destino. Adicionar DEFAULT_ROUTE: quarantine ao dicionário de borda cobre esse caso. Importe DEFAULT_ROUTE de google.adk.workflow.

6. Memory Bank

No VibeStudio Workbench, navegue até Etapa 6 · Memory Bank, partes (6A) e (6B).

No momento, o fluxo de trabalho opera sem memória entre as sessões. Cada execução começa do zero, sem saber o que o criador selecionou antes ou quais gêneros ele prefere. Nesta etapa, você conecta o Memory Bank do Vertex AI Agent Engine para armazenar e recuperar as preferências do criador em várias execuções.

É importante lembrar que a memória é integrada por callbacks do ciclo de vida do agente, e não por nós de pipeline. Como a extração e a recuperação de memória atendem a agentes individuais em vez de etapas intermediárias de dados, a inclusão de callbacks preserva uma topologia de grafo limpa e desacoplada.

Memory Bank (6A)

No workbench, navegue até Memory Bank (6A).

Memória gerenciada no nível do usuário

O Memory Bank é um serviço gerenciado para a memória de longo prazo do usuário. Ele organiza fatos sobre uma pessoa em um escopo definido, identificado aqui pelo nome do aplicativo e ID do usuário:

SCOPE = {"app_name": config.APP, "user_id": config.USER}
TOPICS = {
    "CREATOR_TASTE": "Which video directions this creator picks and passes on, "
                     "and how that preference changes over time.",
    "CHANNEL_RULES": "Standing instructions the creator states for every video "
                     "(style, subjects to avoid, format rules).",
}

Os temas de memória personalizados definem os limites do que o banco registra:

• Extração de temas: quando um novo texto de conversa é enviado pelo memories.generate, o serviço aplica um modelo de extração a cada descrição de tema. Um texto que não corresponde a um tópico não gera recordações.
• Consolidação e remoção de duplicidade: o serviço converte fatos recém-extraídos em incorporações e os compara com as memórias atuais no escopo. Quando uma observação se alinha a uma memória existente, o serviço a atualiza. Quando ele representa informações novas, o serviço cria uma nova entrada. Esse processo garante que várias sessões sobre um tópico sejam mescladas em um resumo coerente, em vez de produzir entradas redundantes.
• Recuperação: chamar memories.retrieve com o escopo do usuário retorna fatos armazenados, ordenados do mais antigo para o mais recente.

Ambas as operações são implementadas em agent/platform/memory.py. O nome do recurso de banco provisionado é armazenado em cache localmente em runs/memorybank.json.

Como configurar o Memory Bank

Use os controles do workbench ou execute os comandos da CLI no terminal:

1. Conecte e provisione o banco:

   python -m agent.platform.bank
   

   Cria a instância do Agent Engine e configura os tópicos CREATOR_TASTE e CHANNEL_RULES.
2. Incluir sessões históricas:

   python -m agent.platform.bank load
   

   Carrega quatro sessões históricas de criadores de conteúdo (dois temas de animais com restrições de estilo, um tema de gadget e um tema de fantasia recente).
3. Inspecionar fatos consolidados:

   python -m agent.platform.bank list
   

   Analise a saída. Observe como as transcrições narrativas foram convertidas em declarações de fatos estruturadas e consolidadas.

Callbacks (6B)

No workbench, navegue até Callbacks (6B). Abra stage4_memory/agent.py.

Callbacks do ciclo de vida do agente do ADK

Um callback é uma função transmitida como argumento para um Agent. O ADK invoca callbacks em momentos predefinidos do ciclo de vida, transmitindo o contexto ativo. Retornar None continua a execução normal. Retornar um objeto de substituição substitui ou intercepta a operação.

O ADK oferece três pares de callbacks:

Par de retorno de chamada	Ponto de invocação	Parâmetros recebidos	Comportamento do valor de retorno
before_agent_callback after_agent_callback	Envolvendo toda a vez do agente	CallbackContext (estado, sessão, invocação)	Retornar Content substitui a resposta do agente. None continua normalmente.
before_model_callback after_model_callback	Em torno de cada chamada de inferência de LLM	LlmRequest ou LlmResponse	Retornar LlmResponse intercepta ou pula a chamada do modelo; None continua.
before_tool_callback after_tool_callback	Em torno de cada execução de ferramenta	Definição, argumentos e resultado da ferramenta	Retornar um dict substitui a saída da ferramenta. None continua.

Os callbacks oferecem um local limpo para injeção de contexto, guardrails, telemetria e pesquisas de cache sem introduzir nós estranhos no gráfico de fluxo de trabalho.

Edição prática: callbacks de recordação e lembrete de fiação

1. Em stage4_memory/agent.py, atualize propose_directions para anexar before_model_callback=recall_taste:

    output_schema=Directions,
    before_model_callback=recall_taste)

recall_taste é executado imediatamente antes de o Gemini gerar sugestões. Ele busca o histórico do criador de conteúdo no Memory Bank, formata as recordações da mais antiga para a mais recente e as anexa ao LlmRequest de saída. O comando direciona o modelo a inclinar os candidatos de 1 a 3 para o gosto atual do criador de conteúdo, tratando as regras do canal como restrições rígidas.

2. Em stage4_memory/agent.py, atualize scripter para anexar after_agent_callback=remember_pick:

    output_schema=Script,
    after_agent_callback=remember_pick)

remember_pick é executado depois que scripter termina a vez. Ele lê a direção escolhida no estado da sessão, sintetiza uma declaração concisa resumindo a decisão do criador de conteúdo e chama memories.generate para atualizar o Memory Bank.

O que esperar e por quê

Teste o fluxo de trabalho aumentado por callback no ambiente de trabalho ou no ADK Web:

1. Execute uma ação com um comando vazio:
   • No rastreamento da sessão, inspecione o LlmRequest para propose_directions. Observe o contexto de memória anexado que detalha a preferência do criador de conteúdo por temas de fantasia e ritmo conciso.
   • Observe as sugestões: os candidatos de 1 a 3 se alinham às preferências históricas do criador de conteúdo, mesmo quando as tendências enfatizam outros temas.
2. Selecione um candidato em direction_gate.
3. Depois que scripter for concluído, analise os registros do Memory Bank:

   python -m agent.platform.bank list
   

   O banco agora reflete a escolha mais recente, consolidando-a com registros de preferências anteriores.

7. Mecanismo RAG

No VibeStudio Workbench, acesse Etapa 7: mecanismo de RAG, partes (7A) e (7B).

Os vídeos publicados acumulam feedback contínuo dos espectadores. Trinta comentários representativos são coletados em agent/comments.md, capturando elogios dos espectadores, críticas ao ritmo patrocinado e preferências de áudio. Nesta etapa, você indexa esses comentários usando o mecanismo de RAG da Vertex AI e conecta a recuperação semântica à distribuição de dados da pesquisa.

Recuperação de documentos (7A)

No workbench, acesse Mecanismo RAG (7A).

Memory Bank x mecanismo RAG

As duas ferramentas baseiam fluxos de trabalho em dados externos, mas têm finalidades arquitetônicas distintas:

Dimensão	Memory Bank	Mecanismo RAG
Caso de uso principal	Preferências de usuário e regras operacionais de longo prazo	Recuperação semântica em grandes coleções de documentos
Escopo	Limitado a IDs de usuários individuais e nomes de aplicativos	Escopo para recursos de corpus compartilhados entre todos os usuários
Processamento de dados	Extração, incorporação e consolidação semântica em tempo real	Divisão de documentos em partes, embedding de vetor e pesquisa de vizinho mais próximo
Integração de gráficos	Callbacks do ciclo de vida do agente (before_model_callback, after_agent_callback)	Nó de função dedicada na distribuição de dados de pesquisa (read_feedback)

Divisão de documentos e incorporações

O mecanismo RAG indexa documentos dividindo o texto em passagens semânticas e armazenando os vetores em um banco de dados gerenciado:

corpus = rag.create_corpus(
    display_name="vibestudio-feedback",
    description="Vibe Studio: what the audience wrote under the channel's past videos.",
    backend_config=rag.RagVectorDbConfig(
        rag_embedding_model_config=rag.RagEmbeddingModelConfig(
            vertex_prediction_endpoint=rag.VertexPredictionEndpoint(
                publisher_model="publishers/google/models/text-embedding-005"))))

rag.upload_file(
    corpus_name=corpus.name, path="agent/comments.md", display_name="comments.md",
    transformation_config=rag.TransformationConfig(
        chunking_config=rag.ChunkingConfig(chunk_size=120, chunk_overlap=20)))

• Tamanho do bloco: configurado para 120 tokens com 20 tokens de sobreposição. Isso captura de dois a três comentários por trecho, garantindo que cada vetor represente um sentimento coeso sem diluir o significado em feedback não relacionado.
• Modelo de embedding: text-embedding-005 converte texto em vetores de alta dimensão. Quando uma consulta é enviada, o modelo a converte em um vetor e encontra as correspondências mais próximas com base na distância semântica. Um comentário sobre um dragão minúsculo guardando meias corresponde a um comando sobre criaturas mágicas sem exigir uma sobreposição exata de palavras-chave.

Como configurar o corpus RAG

Inicialize o corpus usando os botões da bancada ou os comandos do terminal:

1. Crie o corpus:

   python -m agent.platform.rag
   

   Provisiona o banco de dados vetorial gerenciado e registra o ID do recurso em runs/ragcorpus.json.
2. Fazer upload e indexar comentários: faz upload de agent/comments.md com configuração de divisão em partes e aguarda a conclusão da indexação.
3. Consultar o corpus: teste a recuperação de similaridade com consultas que não compartilham palavras exatas com os comentários. Por exemplo, consulte "pequenas criaturas mágicas" para recuperar comentários sobre dragões.

O nó de recuperação (7B)

No workbench, navegue até O terceiro leitor (7B). Abra stage5_rag/agent.py.

Recuperação como um nó de gráfico

O feedback do público-alvo representa dados de pesquisa compartilhados em todo o fluxo de trabalho. Ao contrário da memória pessoal do criador de conteúdo, o sentimento do espectador alimenta diretamente a join_research junto com dados de tendências e pendências. Portanto, ele é implementado como um nó de função:

def read_feedback(node_input):
    """The third reader (step 7): what the audience wrote under past videos,
    the passages nearest to tonight's idea. Retrieval, not a model call."""
    from .platform import rag
    idea = idea_text(node_input)
    query = idea or "what viewers liked and what they complained about"
    try:
        hits = rag.retrieve(query)
    except Exception as e:
        print(f"  [rag] feedback unavailable ({str(e)[:80]})")
        return Event(output={"query": query, "feedback": [],
                             "note": "no corpus connected - run: python -m agent.platform.rag"})
    return Event(output={"query": query, "feedback": [h["text"] for h in hits]})

O read_feedback extrai a ideia inicial do usuário e executa uma consulta vetorial no corpus do mecanismo RAG. Ele emite os comentários recuperados em uma carga útil Event(output=...).

Edição prática: conectando o terceiro leitor à distribuição de dados

Em stage5_rag/agent.py, atualize edges para adicionar read_feedback como uma terceira ramificação paralela entrando em join_research:

           (START, read_backlog, join_research),
           (START, read_feedback, join_research),

Como join_research é um JoinNode, ele sincroniza todas as ramificações recebidas, aguardando até que scan_trends, read_backlog e read_feedback emitam todos os eventos antes de transmitir o pacote agregado downstream.

O que esperar e por quê

Execute o fluxo de trabalho no ambiente de trabalho:

1. Envie um comando de ideia (como "um dragão em miniatura guardando uma bancada de cozinha").
2. No rastreamento de execução, verifique se todos os três nós de leitura são executados simultaneamente.
3. Observe join_research: o dicionário de saída agora contém trends, backlog e feedback.
4. Inspecione os candidatos gerados em propose_directions: o modelo incorpora comentários dos espectadores às propostas e faz referência ao sentimento do público nos campos de evidência.
5. A recuperação de RAG é determinista (consultas idênticas retornam passagens de comentários idênticas), enquanto o nó de proposta generativa produz variações criativas.

8. Geração assíncrona de vídeos com o Veo

No VibeStudio Workbench, acesse Etapa 8: o vídeo, partes (8A) e (8B).

Gerar vídeos em alta definição com o Google Veo leva vários minutos por renderização. Bloquear a execução de grafo durante esse período desperdiça recursos computacionais, bloqueia pools de linhas de execução e expõe a execução a interrupções de conexão HTTP. Nesta etapa, você vai tornar a renderização de vídeo assíncrona usando o LongRunningFunctionTool do ADK.

Ferramentas de longa duração (8A)

No workbench, navegue até Uma ferramenta de longa duração (8A). Abra stage6_video/agent.py e agent/deliver.py.

Ferramentas síncronas x ferramentas de longa duração

As ferramentas de função padrão do ADK são executadas de forma síncrona em uma interação do agente: o modelo chama a ferramenta, aguarda o payload de retorno e incorpora o resultado à interação em andamento.

A renderização de vídeo não pode ser concluída em uma única interação. Em vez disso, render_submit inicia o job de geração e retorna imediatamente um recibo operacional com o status "pending":

def render_submit(prompt: str) -> dict:
    """Submit one Veo render of `prompt`. Returns at once with a pending
    receipt; the clip is delivered later, to this call, by id."""
    receipt = videogen.start(f"{prompt} {videogen.NO_TEXT}")
    return {"status": "pending", "operation": receipt["operation"], "prompt": receipt["prompt"]}

Quando envolvido com LongRunningFunctionTool, o ADK intercepta o status "pending". A vez do agente termina, o fluxo de trabalho é suspenso no nó, e os metadados da chamada pendente (incluindo ID e recebimento da chamada) são registrados em runs/sessions.db. O processo de execução é encerrado sem manter conexões de rede ou linhas de execução de trabalho ativas.

Edição prática: encapsulando a ferramenta de renderização

Em stage6_video/agent.py, atualize render_desk para envolver render_submit em LongRunningFunctionTool:

    tools=[LongRunningFunctionTool(render_submit)])

Retomar por ID da chamada

O padrão de retomada universal

O ADK aplica um mecanismo idêntico para suspender e retomar fluxos de trabalho para humanos e ferramentas externas:

Gatilho de suspensão	Iniciando o Construct	Estado de suspensão armazenado	Evento de retomada
Decisão humana	yield RequestInput(...)	Abrir o comando de entrada no armazenamento de sessão	FunctionResponse com o ID da chamada de suspensão
Ferramenta de longa duração	LongRunningFunctionTool(...) retornando pending	Abrir chamada de ferramenta no armazenamento de sessão	FunctionResponse com o ID da chamada de suspensão

Em ambos os cenários, o fluxo de trabalho é interrompido completamente e só é retomado quando um evento com um FunctionResponse correspondente chega de uma fonte externa: uma interface do usuário, um webhook ou um worker em segundo plano.

Edição prática: concluir a resposta de entrega

Em agent/deliver.py, crie a parte de retomada FunctionResponse:

    part = Part(function_response=FunctionResponse(
        id=row["call_id"], name=row["name"], response=response))

O daemon de entrega faz uma pesquisa no Veo até que o arquivo de vídeo seja gerado e, em seguida, envia esse FunctionResponse para a sessão. O ADK corresponde ao ID da chamada e retoma o fluxo de trabalho diretamente no próximo nó. Os nós concluídos não são executados novamente, e o agente não faz outra interação generativa.

Definir STUDIO_REAL_VIDEO=0 em .env ativa a renderização simulada: start retorna um recibo de teste imediato, e check simula a conclusão em cinco segundos sem fazer chamadas faturáveis da API Veo.

Integração de pipeline (8B)

No ambiente de trabalho, navegue até render_desk no gráfico (8B). Abra stage6_video/agent.py.

O nó terminal no pipeline é store_video. Ele lê as informações de renderização concluídas de runs/state.json (onde o processo de exibição as registrou) e confirma o URL do vídeo e o status de geração no estado da sessão compartilhada.

Edição prática: conectando o pipeline de vídeo completo

Em stage6_video/agent.py, atualize edges para anexar render_desk e store_video:

           (quarantine, scripter),
           (scripter, render_desk, store_video)])

O que esperar e por quê

Teste o fluxo de geração assíncrona no ambiente de trabalho:

1. Execute o fluxo de trabalho com a seleção de candidatos e a geração de scripts.
2. Em render_desk, observe o agente invocar render_submit.
3. O fluxo de trabalho é suspenso imediatamente. No ambiente de trabalho ou na Web do ADK, observe o status pendente: a sessão mantém o ID da chamada aberta, e nenhum processo em segundo plano está consumindo recursos.
4. Execute o daemon de entrega usando o console do ambiente de trabalho ou no terminal:

   python -m agent.deliver
   

   O processo de entrega monitora o Veo até que o vídeo esteja pronto e, em seguida, envia o evento de retomada.
5. No ADK Web, atualize a sessão: a execução é retomada em store_video, confirma o URL do vídeo no estado da sessão e conclui o fluxo de trabalho.

9. Implantar no Cloud Run

No VibeStudio Workbench, navegue até Etapa 9: implantação.

Você desenvolveu e verificou cada componente do pipeline em sandboxes dedicados. Nesta etapa, você vai montar o pipeline de produção completo e implantá-lo no Google Cloud Run.

O ADK Runner

No desenvolvimento, adk web orquestrou o gráfico. Em produção, o aplicativo hospeda o fluxo de trabalho usando a classe Runner do ADK:

self._svc = DatabaseSessionService(db_url=config.DB_URL)
self._runner = Runner(app_name=config.APP, agent=wf, session_service=self._svc)

async for ev in self._runner.run_async(user_id=config.USER, session_id=run_id, new_message=message):
    self._absorb(ev)    # fold the ADK event into the run state, publish one app event

# the gate's answer and the render's delivery are the same call, with a function_response part
part = Part(function_response=FunctionResponse(id=call_id, name=name, response=response))

• run_async: impulsiona a execução do fluxo de trabalho, gerando eventos sequencialmente à medida que os nós são executados e persistindo atualizações no serviço de sessão.
• Retomada unificada: as decisões do usuário em direction_gate e as entregas de vídeo concluídas do Veo retomam a execução por objetos FunctionResponse idênticos enviados para run_async.

A arquitetura do aplicativo de produção

O aplicativo de produção em vibestudio/ integra o pipeline completo:

vibestudio/
  server/
    main.py                 FastAPI: application server, REST routes, static assets
    api.py                  REST API endpoints: run, pick, publish, backlog, profile, history
    runner.py               Runner orchestration over the workflow, background render poller
    platform/               Event bus (SSE stream), file storage, publishing, telemetry
    agent/                  Production agent package, verified by checks/verify_app.py
      graph.py              The complete workflow graph and node definitions
      desk.py               render_desk and render_submit wrapped with LongRunningFunctionTool
      schemas.py            Pydantic schemas: Directions, CleanedDirection, Script
      cleanup_tools.py      Deterministic policy tools: find_policy_hits, suggest_replacement
      platform/             Memory Bank, RAG Engine, and Veo integrations
  web/                      Production React user interface
  Dockerfile · deploy.py · run.sh

• Fluxo de evento único: o back-end do FastAPI publica eventos em um único fluxo de eventos enviados pelo servidor (SSE). O front-end do React visualiza a progressão do gráfico em tempo real e processa conexões tardias sem perder o estado.
• Execução desacoplada: o aplicativo gerencia o loop de eventos. O gráfico de fluxo de trabalho se concentra totalmente na lógica de execução, sem conhecer a interface de front-end.

A lista completa de arestas do fluxo de trabalho em agent/graph.py combina todos os padrões arquitetônicos criados ao longo deste codelab:

        (START, scan_trends, join_research),
        (START, read_backlog, join_research),
        (START, read_feedback, join_research),
        (join_research, propose_directions, direction_gate,
         persist_direction, policy_check),
        (policy_check, {"OK": scripter, "BLOCK": quarantine}),
        (quarantine, scripter),
        (scripter, render_desk, store_video),

Como implantar no Cloud Run

O Google Cloud Run oferece hospedagem sem servidor com escalonamento automático, roteamento de solicitações e builds de contêineres integrados:

gcloud run deploy vibestudio --source vibestudio \
  --project $GOOGLE_CLOUD_PROJECT --region us-central1 \
  --labels dev-tutorial-codelab=vibetube --allow-unauthenticated \
  --memory 2Gi --cpu 2 --timeout 3600 --concurrency 40 \
  --max-instances 1 --min-instances 1 --session-affinity \
  --set-env-vars GOOGLE_CLOUD_PROJECT=...,STUDIO_VERTEX=1,STUDIO_MEMORY_BANK=...,STUDIO_RAG_CORPUS=...,VIBETUBE_URL=...,VIBETUBE_EVENT=...,VIBETUBE_NAME=...,VIBETUBE_PROJECT=...

• Build do contêiner: o gcloud run deploy --source empacota o diretório vibestudio/, cria a imagem do contêiner usando o Cloud Build e implanta o serviço em uma única operação.
• Afinidade da sessão: direciona solicitações do mesmo usuário para a mesma instância de contêiner, preservando o estado da sessão local em etapas iterativas.
• Observabilidade: a integração do Cloud Trace registra períodos distribuídos para cada nó, chamada de LLM e execução de ferramenta, acessíveis no console do Google Cloud em "Explorador de Trace".

Clique no botão Implantar no ambiente de trabalho para executar o script de implantação. Quando o build for concluído, o terminal vai mostrar o URL do serviço ativo.

10. Resumo

No VibeStudio Workbench, acesse Etapa 10 · Resumo para revisar a arquitetura concluída.

Etapa	Arquitetura e conceitos	Padrão de implementação
Um único comando	Único comando, ferramentas de função, loop de chat sequencial	Agent(tools=[...]), function_call / function_response
Fundamentos do fluxo de trabalho com agentes	Fluxo de trabalho de gráfico, pesquisa paralela, saídas de esquema, validação humana	Workflow, START, JoinNode, output_schema, RequestInput
Estado e roteador	Estado da sessão compartilhada, vinculação de parâmetros, roteamento determinístico, agente de tarefas	Event(state=...), Event(route=...), mode="task", finish_task
Memory Bank	Memória de longo prazo no nível do usuário, consolidação semântica, hooks de ciclo de vida	memories.generate / retrieve, before_model_callback, after_agent_callback
Mecanismo RAG	Recuperação de documentos com base em comentários do público-alvo e embeddings semânticos	Nó rag.create_corpus, RagEmbeddingModelConfig, read_feedback
Geração assíncrona de vídeos com o Veo	Ferramentas de longa duração, recibos pendentes, daemon de entrega externa	LongRunningFunctionTool, FunctionResponse(id=...) retomada
Implantar no Cloud Run	Orquestração programática, eventos enviados pelo servidor, contêiner sem servidor	Runner(agent=wf), run_async, implantação do Cloud Run

Princípios arquitetônicos principais

1. Suspender em vez de esperar: os fluxos de trabalho são pausados de maneira limpa para entrada humana (RequestInput) ou operações de longa duração (LongRunningFunctionTool). Os processos não ficam inativos em linhas de execução ou sockets de rede.
2. Retomada universal: todas as suspensões são retomadas por um mecanismo idêntico: um único function_response que carrega o ID de chamada do nó suspenso.
3. Gerenciamento de estado dissociado: os nós compartilham dados por chaves de estado de sessão nomeadas e vinculação de parâmetros em vez de payloads intermediários detalhados e fortemente acoplados.
4. Rotas determinísticas antes do custo generativo: roteadores baseados em regras e filtros de regex avaliam a política com custo zero de token antes da execução dos modelos generativos.
5. Separação de responsabilidades: o contexto específico de um agente individual pertence a callbacks de ciclo de vida, enquanto dados pessoais compartilhados pertencem