Flujo de trabajo de agentes con el ADK

1. Introducción

En este codelab, se te guía para crear sistemas basados en agentes de próxima generación con flujos de trabajo y gráficos en el Kit de desarrollo de agentes (ADK). Implementarás patrones de arquitectura comunes, organizarás interacciones con interacción humana (HITL) y controlarás la ejecución asíncrona de larga duración. También integrarás bases de conocimiento empresarial y memoria persistente para personalizar y desarrollar el comportamiento del agente. Por último, conectarás estas capacidades para impulsar una canalización automatizada de generación de videos.

Situación hipotética

Tienes un canal digital en VibeTube con un público activo y una lista de ideas creativas en expansión. La producción de cada video requiere una ejecución continua en varias etapas: investigación de formatos populares, síntesis de comentarios de los usuarios, desarrollo de guiones, verificación del cumplimiento de las políticas y generación de clips de video. Los modelos generativos pueden crear borradores de recursos individuales, pero para lanzar versiones coherentes se requiere una arquitectura de agentes coordinada.

Para automatizar este ciclo de vida, compilarás VibeStudio. Esta canalización basada en agentes ejecuta investigaciones de rutina en paralelo, presenta opciones seleccionadas para la aprobación con interacción humana, aplica puertas de políticas automatizadas antes de generar el video y conserva el contexto en todas las ejecuciones de producción.

Qué aprenderá

• Fundamentos de la ingeniería de grafos: Las arquitecturas de agentes de varios pasos requieren un flujo de control explícito y rutas de ejecución estructuradas. Compilas un Workflow del ADK con tuplas de borde, el punto de entrada START, JoinNode para la agregación de fan-out paralela y nodos de router determinísticos para dirigir la ejecución según el estado.
• Modos y devoluciones de llamada de ciclo de vida del agente: Las tareas especializadas requieren comportamientos operativos distintos y rieles determinísticos. Configuras instancias de Agent del ADK con los modos chat, single_turn y task habilitado para herramientas como nodos de flujo de trabajo, y aplicas interceptores con before_model_callback y after_agent_callback.
• Orquestación con interacción humana: Las canalizaciones de producción se pausan para que se realice una evaluación humana en los puntos de control creativos críticos. Implementas RequestInput para suspender la ejecución del flujo de trabajo, aplicar esquemas de respuesta estructurados y reanudar la ejecución sin mantener activos los procesos de tiempo de ejecución inactivos.
• Memoria jerárquica del agente: Los sistemas de producción separan el estado de ejecución efímero del contexto duradero. Administras el estado de la sesión a corto plazo con Event(state=...) y la vinculación de parámetros, y conectas GEAP Memory Bank para extraer, consolidar y conservar las preferencias del creador en todas las ejecuciones.
• Fundamentación con bases de conocimiento empresarial: Los agentes autónomos requieren contexto dinámico del dominio y opiniones del público. Conectas un corpus del motor de RAG de GEAP como un nodo de recuperación dedicado dentro del fan-out paralelo para fundamentar semánticamente los resultados del agente.
• Flujos de trabajo y la implementación de larga duración: La renderización de video multimodal funciona de forma asíncrona durante períodos prolongados. Implementas LongRunningFunctionTool con recibos de llamadas pendientes para suspender y reanudar el flujo de trabajo por ID de llamada, y, luego, implementas la canalización finalizada con el ADK Runner en Cloud Run.

Cómo se organiza este codelab

Este codelab sirve como referencia conceptual y arquitectónica. En cada sección, se explican las construcciones del ADK implementadas en el paso correspondiente del banco de trabajo, se proporciona código de referencia y se establecen principios de diseño básicos. Revisa cada sección antes de completar el ejercicio correspondiente en el banco de trabajo.

El trabajo práctico se realiza en VibeStudio Workbench, una interfaz web complementaria que incluye un editor de código interactivo, verificadores de tiempo de ejecución y un inspector de ADK integrado. La numeración de los pasos en el banco de trabajo se alinea directamente con este codelab para mantener sincronizado tu progreso. Las ediciones de gráficos fundamentales persisten en todos los pasos, y el banco de trabajo verifica automáticamente los requisitos previos a medida que avanzas.

Cuando completes los ejercicios del banco de trabajo, ensamblarás una canalización de agente integral y, luego, implementarás una aplicación de VibeStudio en ejecución en Cloud Run para generar contenido de video.

El entorno consta de tres componentes principales: VibeStudio Workbench (la interfaz web local para la edición de código y la verificación del tiempo de ejecución), tu backend (el ADK Workflow y las zonas de pruebas de etapa en agent/) y Google Cloud (modelos de Gemini, GEAP Memory Bank, RAG Engine y generación de video de Veo).

2. Configuración

Reclama los créditos de tu taller

Si asistes a un lab dirigido por un instructor, este distribuirá créditos para tu proyecto de Google Cloud. Sigue las instrucciones del instructor para canjear tus créditos y asegúrate de que la facturación esté activa en tu cuenta antes de continuar.

Abre Cloud Shell

Cloud Shell es un entorno de desarrollo basado en navegador con gcloud, Python y git preinstalados.

Para iniciar Cloud Shell, haz lo siguiente:

1. Navega a la consola de Google Cloud.
2. En el encabezado de navegación superior, haz clic en Activar Cloud Shell (el ícono de ventana de terminal).

Se abrirá una sesión de terminal en la parte inferior de la ventana del navegador.

Clona e inicializa el repositorio

Ejecuta los siguientes comandos en la terminal de Cloud Shell para clonar el proyecto:

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

Mensajes de configuración

Durante la configuración, se te solicitarán los siguientes detalles:

• ID del proyecto de Google Cloud: Cuando setup_project.sh te lo solicite, presiona Intro para crear un proyecto nuevo automáticamente. Si prefieres usar un proyecto existente (como un proyecto preasignado), ingresa tu ID del proyecto y asegúrate de que la ortografía sea correcta y de que la facturación esté activa.
• Código del evento: Ingresa el código de la sala que te proporcionó el instructor. Si no recibiste uno, consulta con un asistente de enseñanza o un vecino. Si completas este lab en casa, presiona Intro para aceptar la habitación sandbox predeterminada.
• Nombre visible del canal: Ingresa tu nombre o el identificador de canal que prefieras cuando setup_codelab.sh te lo solicite, o presiona Intro para aceptar el valor predeterminado generado a partir de tu Cuenta de Google.

Ejecuta las dos secuencias de comandos de configuración en orden:

./setup_project.sh
./setup_codelab.sh

• setup_project.sh: Crea o reutiliza un proyecto de Google Cloud con facturación activa, guarda el ID del proyecto en ~/project_id.txt y configura el contexto activo de gcloud.
• setup_codelab.sh: Instala uv y las dependencias de Python en .venv, habilita las APIs de Google Cloud requeridas, configura los parámetros del canal en .env, verifica el acceso al modelo con Gemini, aprovisiona los recursos de Memory Bank y RAG, compila la interfaz del banco de trabajo y, luego, inicia VibeStudio Workbench.

La secuencia de comandos ejecuta la verificación previa y, luego, inicia VibeStudio Workbench en segundo plano. En sus últimas líneas, se muestra el vínculo para abrirlo.

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

Haz clic en ese vínculo. La misma dirección está disponible en Vista previa en la Web → Cambiar puerto → 4600.

Para volver a verificar el entorno en cualquier momento, ejecuta python scripts/preflight.py. Para reiniciar el banco de trabajo, ejecuta scripts/restart.sh. Para volver a configurarlo, ejecuta ./setup_codelab.sh, que conserva tu configuración y progreso.

Con el archivo abierto, lee el paso 1, La historia, para conocer la situación, y el paso 2, Lo que compilarás, para conocer la forma del gráfico terminado. Ninguno de los dos tiene un ejercicio. Luego, regresa aquí para el paso 3.

Cada parte práctica de VibeStudio Workbench termina con un panel de verificación que lee los artefactos reales: el archivo en el disco y las sesiones escritas por las ejecuciones.

Diseño del repositorio

El repositorio se estructura en la lógica del flujo de trabajo principal, los entornos de pruebas paso a paso, el entorno de banco de trabajo y la aplicación de producción:

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/: Contiene el gráfico principal del flujo de trabajo. Editarás archivos en este directorio para implementar nodos de fan-out paralelos, enrutamiento de políticas determinístico, devoluciones de llamada de memoria y herramientas de generación de video.
• agent/platform/: Se comunica con los servicios de Google Cloud, incluidos los modelos de Gemini, el banco de memoria de GEAP, el motor de RAG de GEAP y la síntesis de video de Veo.
• stage0_prompt/ a stage6_video/: Entornos de zona de pruebas autónomos. Cada carpeta exporta un root_agent independiente para que puedas ejecutar e inspeccionar cada paso de forma aislada a través de la interfaz de desarrollo del ADK integrada.
• server/ y web/: La aplicación VibeStudio Workbench que se ejecuta de forma local en el puerto 4600. Contiene la documentación de los pasos, el editor de código in-page, los verificadores de evidencia del tiempo de ejecución y la visualización de gráficos.
• vibestudio/: Es la aplicación de producción completa empaquetada y, luego, implementada en Cloud Run en el paso final. Contiene su propia copia independiente del grafo de flujo de trabajo completado.

3. Agente monolítico

Antes de construir un gráfico de flujo de trabajo de varios nodos, establece una referencia arquitectónica con un solo agente en stage0_prompt/agent.py. Este agente se basa en una instrucción del sistema monolítica que describe la canalización de producción en prosa, con el respaldo de dos herramientas de funciones de Python.

La evaluación de este modelo de referencia demuestra los límites operativos de la coordinación basada en instrucciones y establece por qué los sistemas de producción requieren la organización de gráficos.

Arquitectura de agentes del ADK (3A)

En VibeStudio Workbench, navega a Paso 3: Agente monolítico y abre Arquitectura del agente de ADK (3A). En esta vista, se presentan las capas arquitectónicas principales de un agente del 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
)

El diagrama interactivo agrupa los componentes del agente en cinco dominios operativos:

• Capa de razonamiento (modelo): Es el modelo de lenguaje principal (como Gemini 3 Flash) que ejecuta tareas cognitivas, razonamiento de instrucciones y selección de herramientas. Todo lo demás en la arquitectura informa o limita este modelo.
• Capa de contexto (instrucciones y habilidades): Son directivas que dan forma al razonamiento del modelo. instruction establece las instrucciones del sistema, el arquetipo y las reglas operativas permanentes. skills proporciona orientación procedimental con versiones (SKILL.md) para flujos de trabajo repetibles.
• Capa de colaboración y acción (herramientas, subagentes, flujo de trabajo, esquema de salida): Interfaces que permiten que el agente actúe en sistemas externos y emita datos escritos. tools proporcionar funciones de Python invocables o extremos del Protocolo de contexto del modelo (MCP) subagents ejecutar tareas subordinadas delegadas workflow coordina gráficos multiagente. output_schema aplica modelos de Pydantic para garantizar que los consumidores posteriores reciban JSON validado en lugar de texto no estructurado.
• Capa de interceptores (devoluciones de llamada de ciclo de vida): Son rieles determinísticos que ejecutan código personalizado antes y después de la ejecución del agente (before_agent/after_agent), los turnos individuales del modelo (before_model/after_model) y las llamadas a herramientas (before_tool/after_tool). Los interceptores aplican las reglas de política sin depender del cumplimiento del modelo.
• Estado externo (sesión y memoria): Persistencia con estado separada de la lógica del agente. Session conserva la memoria de trabajo transitoria y el registro de eventos del subproceso de ejecución actual. Memory mantiene hechos y preferencias duraderos entre sesiones con servicios administrados, como GEAP Memory Bank.

El agente monolítico de este paso solo implementa tres de estas primitivas: model, instruction y tools. En los pasos posteriores, se presentan los flujos de trabajo de gráficos, los esquemas estructurados, los interceptores y los servicios de memoria persistente.

Especificación del agente monolítico (3B)

En el banco de trabajo, avanza a Especificación del agente monolítico (3B). Abre stage0_prompt/agent.py para examinar la definición del agente de referencia:

• Instrucción de una sola instrucción: La instrucción del sistema condensa cinco tareas de producción distintas en prosa continua: descubrir tendencias de la plataforma, revisar ideas pendientes, proponer conceptos creativos, aplicar políticas sobre temas prohibidos y redactar listas de tomas.
• Fuentes de datos subyacentes: El agente hace referencia a dos fuentes definidas junto al gráfico:
  • agent/trends.py: Muestra diez tendencias activas de formato y estilo de un grupo de 250 con puntuaciones de calor dinámicas.
  • agent/backlog.txt: Lee las notas conceptuales sin procesar del creador línea por línea.

Herramientas en el agente (3C)

En el banco de trabajo, avanza a Herramientas en el agente (3C).

¿Qué es una herramienta para un agente?

Un modelo de lenguaje es, por naturaleza, un motor de razonamiento de mundo cerrado: opera únicamente con los pesos entrenados previamente y los tokens presentes en su ventana de contexto inmediata. No puede consultar una base de datos, acceder a APIs en tiempo real ni ejecutar código de forma nativa.

Un objeto Tool une este límite. Le otorga al modelo una agencia externa, lo que le permite recuperar información de referencia y ejecutar acciones determinísticas en sistemas externos.

El llamado a herramientas sigue un protocolo explícito de cinco etapas entre el modelo y el tiempo de ejecución del ADK:

1. Declaración de esquema: El desarrollador proporciona funciones de Python al agente. El ADK inspecciona el nombre, las anotaciones de tipo y las cadenas de documentación de cada función para generar una declaración de esquema JSON compatible con OpenAPI que describa sus parámetros y su propósito.
2. Razonamiento del modelo: Durante la inferencia, el modelo evalúa si la instrucción del usuario requiere datos externos. Si es necesario, el modelo emite un evento function_call estructurado que contiene el nombre de la función objetivo y el diccionario de argumentos que coinciden con el esquema.
3. Ejecución en el tiempo de ejecución: El modelo en sí no ejecuta código. El tiempo de ejecución del ADK intercepta el function_call, ejecuta la función de Python local real con los argumentos proporcionados y captura el valor de devolución.
4. Reinyección de contexto: El tiempo de ejecución del ADK empaqueta el valor de devolución de la función en un evento function_response y lo agrega al historial de la sesión activa.
5. Síntesis final: El modelo procesa el resultado de la herramienta que ahora está presente en su ventana de contexto y completa su respuesta.

En stage0_prompt/agent.py, las dos herramientas de investigación se definen como funciones estándar de 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()}

Edición y ejecución prácticas

En el editor de código del banco de trabajo, agrega las dos referencias de funciones a la lista tools del agente:

    tools=[check_trends, read_backlog],

Guarde el cambio. El archivo se actualiza en el disco y la fila de verificación confirma que ambas herramientas están conectadas.

Haz clic en Open adk web para iniciar la interfaz de desarrollo del ADK integrada. Envía la instrucción de idea sugerida:

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

Qué esperar y por qué

Cuando envíes esta instrucción, observa la siguiente secuencia de ejecución en el registro de la sesión:

• Aparecen dos eventos de ejecución de herramientas antes de la respuesta: Verás los eventos function_call y function_response para check_trends y read_backlog.
  • Por qué: Gemini evaluó la directiva de la instrucción del sistema ("verifica qué es tendencia y consulta tu lista de ideas pendientes"), reconoció que carecía de tendencias de la plataforma y notas del canal en sus ponderaciones, y, luego, invocó ambas funciones para fundamentar su contexto.
• El agente propone una dirección y hace una pausa para que la confirmes: La respuesta sugiere una dirección de video que sintetiza las tendencias y el backlog, y te pide que la confirmes.
  • Por qué: La directiva de instrucción le pidió al modelo que acordara la dirección con el creador antes de generar el guion.
• Omitir la confirmación en un turno de seguimiento: Envía un segundo mensaje: skip the questions, just describe the video. El agente omite inmediatamente la confirmación y redacta el título y las tomas.
  • Por qué: Las instrucciones de la instrucción son lineamientos orientativos en lugar de barreras determinísticas. En un agente monolítico, las instrucciones del usuario pueden anular las reglas permanentes de la instrucción del sistema porque ningún flujo de trabajo externo controla el flujo de ejecución.

Limitaciones arquitectónicas de una instrucción monolítica

Si bien una sola instrucción puede producir resultados aceptables para demostraciones aisladas, probar las condiciones límite en el verificador del banco de trabajo revela limitaciones empresariales críticas:

• Agregación de investigación no estructurada: El orden de ejecución de la herramienta no es determinístico. El modelo resume los datos recuperados en prosa de formato libre, lo que imposibilita que los sistemas posteriores aíslen qué fuente produjo afirmaciones específicas.
• Aplicación de políticas no verificadas: El modelo evalúa su propio cumplimiento de la seguridad. Si el modelo determina que un tema es seguro, no se valida ese hallazgo con lógica determinística externa.
• Pausas con interacción humana no obligatorias: Las instrucciones de la instrucción que solicitan la confirmación del creador son orientativas. Si envías un mensaje de seguimiento en el que se le indica al modelo que omita las preguntas, este omitirá por completo la aprobación humana.

Estas brechas arquitectónicas motivan la descomposición del agente monolítico en el flujo de trabajo de grafo explícito que se compila en el siguiente paso.

4. Fundamentos del flujo de trabajo de agentes

En VibeStudio Workbench, navega a Paso 4: Aspectos básicos del flujo de trabajo agentic, partes 4A a 4D.

En este paso, se pasa de un modelo de referencia de un solo agente a una organización determinística de gráficos con el ADK Workflow. Crearás un fan-out de investigación paralelo, sincronizarás las ramas con un nodo de unión, generarás candidatos de creatividades validados por el esquema y presentarás una puerta de aprobación determinística con interacción humana.

Arquitectura de gráficos y cadenas de ejecución (4A)

En el banco de trabajo, abre Graph architecture and execution chains (4A).

Un ADK Workflow estructura la ejecución del agente como un grafo dirigido definido por una lista de bordes:

• Cadenas: Las tuplas secuenciales definen la ejecución lineal de nodos ((node_a, node_b, node_c)).
• Ramas paralelas: Las cadenas independientes que comparten un nodo de origen se ejecutan de forma simultánea.
• Sincronización: Las cadenas que convergen en un JoinNode esperan hasta que todas las ramas entrantes envían su informe antes de liberarse.
• Control determinístico: El flujo de ejecución se rige por estructuras de código declaradas en lugar de inferirse a partir del texto de la instrucción.

Arquetipos de nodos en el ADK

Los flujos de trabajo del ADK componen varios tipos de nodos especializados. Cada arquetipo desempeña un rol operativo específico en el grafo, lo que separa la ejecución de código determinístico del razonamiento del modelo generativo:

Arquetipo de nodo	Implementación	Rol en la canalización
Nodo de función	Función de Python que devuelve un Event	Ejecuta lógica determinística, recuperación de datos y mutaciones de estado.
Nodo de unión	Instancia de JoinNode integrada	Sincroniza las ramas simultáneas en un diccionario agregado.
Nodo de agente	Agent se ejecuta en modo single_turn	Evalúa las instrucciones en función de la entrada upstream y emite datos validados.
Nodo de router	Función que devuelve un Event con una etiqueta route	Evalúa la lógica condicional para seleccionar las ramas de ejecución posteriores.
Nodo de entrada humana	Función de rendimiento RequestInput	Suspende el estado de ejecución hasta que llega una respuesta externa del usuario.

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

En esta configuración, root_agent es una instancia de Workflow en lugar de un Agent independiente. El ADK trata los flujos de trabajo como agentes de primera clase, lo que permite cargar, publicar y revisar un gráfico completo como una aplicación unificada. El name registra la aplicación en ADK Web, mientras que la lista edges define su topología de ejecución.

Fan-out de investigación paralela (4B)

En el banco de trabajo, avanza a Parallel research fan-out (4B). Abre stage1_fanout/agent.py.

Nodos de función y barreras de sincronización

La fase de investigación usa dos nodos de función importados de agent/graph.py:

• scan_trends: Devuelve Event(output={"trends": [...]}) que contiene diez tendencias de la plataforma con puntuación.
• read_backlog: Devuelve Event(output={"backlog": [...], "idea": "..."}) que contiene quince ideas de tareas pendientes del canal junto con la instrucción de ejecución inicial.

Cada función acepta node_input (la salida del nodo anterior) y devuelve un Event.

Un JoinNode sirve como barrera de sincronización: se detiene hasta que cada cadena entrante entrega un evento y, luego, agrega todos los resultados de las ramas en un diccionario con el nombre del nodo como clave ({"scan_trends": {...}, "read_backlog": {...}}).

Edición práctica: cómo definir las uniones y los bordes paralelos

En stage1_fanout/agent.py, crea una instancia de JoinNode y conecta las dos cadenas paralelas que comienzan en START:

join_research = JoinNode(name="join_research")

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

Guarda los cambios. El verificador de la mesa de trabajo confirma que la unión y los bordes están conectados. Ejecuta la etapa con Run Stage 1 o a través de la interfaz web integrada del ADK.

Qué esperar y por qué

• Ejecución simultánea de lectores: En el gráfico de ejecución, scan_trends y read_backlog se ejecutan de forma simultánea.
  • Motivo: Ambas cadenas se originan en START. El motor del ADK programa ramas independientes de forma simultánea.
• Resultado del diccionario agregado: El flujo de trabajo se completa en join_research y genera un diccionario con entradas para ambos lectores.
  • Motivo: JoinNode garantiza la captura completa de datos antes de permitir que se ejecuten los nodos posteriores.

Nodos de agentes (4C)

En el banco de trabajo, avanza a Nodos de agentes (4C). Abre stage2_direction/agent.py.

Modos de operación y esquemas estructurados

Cuando se incorpora dentro de un Workflow, un Agent se ejecuta en modo single_turn de forma predeterminada:

• Recibe el resultado del nodo anterior como entrada de contexto.
• Ejecuta una sola llamada de inferencia sin una conversación fluida.
• Genera datos estructurados para el siguiente nodo.

Al asignar output_schema=Directions, el agente aplica la validación de Pydantic en el resultado del modelo. El gráfico de nivel inferior recibe objetos escritos en lugar de prosa no estructurada:

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

PROPOSE_INSTRUCTION le indica al modelo que proponga cuatro candidatos citando evidencia de las tendencias y el backlog. Los candidatos del 1 al 3 ofrecen conceptos de canales viables. El candidato 4 introduce intencionalmente un concepto que incumple la política para probar la puerta de seguridad en el siguiente paso.

Edición práctica: Definición del nodo del agente y encadenamiento de la unión

En stage2_direction/agent.py, configura propose_directions y extiende los bordes del flujo de trabajo:

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)])

Qué esperar y por qué

• Consumo directo del diccionario: propose_directions consume la carga útil de JSON que emite join_research sin necesidad de un formato manual.
• Salida de candidato escrita: El agente emite un objeto Directions validado que contiene cuatro candidatos discretos. Los nodos posteriores leen los campos por nombre de atributo (candidate.title) sin analizar cadenas.

Con interacción humana (4D)

En el banco de trabajo, avanza a Human-in-the-loop (4D). Abre agent/graph.py.

Instrucciones de la instrucción frente a la suspensión determinística

Los flujos de trabajo de producción que generan costos financieros o publican contenido requieren supervisión humana en los puntos de decisión críticos. En una sola instrucción, las solicitudes de confirmación son instrucciones de asesoramiento que un usuario puede solicitar fácilmente al modelo que omita. En un flujo de trabajo del ADK, el motor de ejecución aplica la aprobación humana: el gráfico se detiene en un nodo designado y no puede avanzar hasta que recibe una entrada externa validada por el esquema:

• La instrucción RequestInput suspende la ejecución del flujo de trabajo de inmediato.
• El ADK registra una llamada de interrupción abierta en el almacén de sesiones y emite un interrupt_id único.
• El proceso de ejecución se detiene sin consumir tokens ni subprocesos del servidor.
• La ejecución por grafos se reanuda solo cuando se envía un function_response válido que coincide con el esquema y el ID de interrupción.

Edición práctica: Suspender la ejecución con RequestInput

En agent/graph.py, implementa la llamada de suspensión dentro de 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 tres atributos:

• message: Es el mensaje de revisión que se le presenta al usuario.
• response_schema: Es un esquema JSON que el frontend renderiza como un formulario de entrada y que el ADK valida cuando se envía.
• payload: Son metadatos incluidos en la solicitud (los cuatro candidatos) que permiten que las interfaces del cliente rendericen tarjetas de opiniones sin consultar el estado de la sesión.

Qué esperar y por qué

• El flujo de trabajo se detiene en direction_gate: En la Web del ADK o en la interfaz del banco de trabajo, la ejecución se pausa y se muestra un formulario interactivo de selección de candidatos.
  • Por qué: El motor encontró un RequestInput diferido y persistió el estado de ejecución en runs/sessions.db.
• La reanudación requiere una entrada estructurada: Enviar texto de chat arbitrario no avanza el gráfico. Si seleccionas una opción (1, 2, 3 o 4), se envía un function_response escrito que satisface response_schema y se reanuda la ejecución.

5. Estado y router

En VibeStudio Workbench, navega a Paso 5: Estado y Router, partes (5A) a (5C).

Conservarás las selecciones del usuario en el estado de la sesión, aplicarás las políticas de seguridad del canal con nodos de enrutador determinísticos y ensamblarás un agente de tareas iterativo para corregir automáticamente los incumplimientos de políticas antes de generar guiones de video.

Estado del flujo de trabajo (5A)

En Workbench, navega a Estado del flujo de trabajo (5A).

Estado de la sesión vs. salida del nodo

En un flujo de trabajo del ADK, los datos se mueven por el grafo a través de dos mecanismos distintos:

• Salida del nodo (Event(output=...)): Son los datos dirigidos estrictamente a los consumidores inmediatos posteriores definidos en la lista de aristas.
• Estado de la sesión (Event(state=...)): Es un diccionario compartido de par clave-valor al que puede acceder cualquier nodo posterior en el ciclo de vida de la ejecución.

Cuando un usuario selecciona un candidato en direction_gate, la selección llega como un índice numérico ({"pick": "2"}). Los nodos posteriores necesitan el objeto de dirección completo: título, ángulo narrativo y gancho. En lugar de pasar metadatos detallados a través de cada carga útil de nodo intermedio, persist_direction escribe el candidato resuelto en el estado de sesión compartido.

Los nodos no necesitan pasar todo el diccionario de estado de la sesión. Cuando un nodo genera Event(state=...), solo proporciona los pares clave-valor nuevos o actualizados. El ADK combina automáticamente estas actualizaciones en el almacén de sesiones:

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

Cuando se genera este Event, se entrega el control al tiempo de ejecución de Workflow, que persiste los valores nuevos en el registro de la sesión en runs/sessions.db.

Vinculación de parámetros

Los nodos de función del ADK leen el estado de la sesión automáticamente a través de la inspección de parámetros. Si la firma de una función declara un nombre de parámetro que coincide con una clave de estado existente, el ADK extrae esa clave del estado y la pasa directamente:

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])

Aquí, candidates se escribió en el estado de la sesión con direction_gate. El ADK lo vincula directamente a persist_direction(node_input, candidates: list = []) sin necesidad de búsquedas explícitas en el diccionario.

Las claves con el prefijo user: persisten en todas las sesiones en el almacenamiento a nivel del usuario, lo que permite que las ejecuciones posteriores del flujo de trabajo accedan a las preferencias del creador.

Edición práctica: cómo conservar el estado y conectar el nodo

1. En agent/graph.py, dentro de persist_direction, reemplaza la línea TODO: PERSIST_STATE por el rendimiento del evento de estado:

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

2. En stage3_router/agent.py, agrega persist_direction a la tercera cadena de la lista edges:

           (join_research, propose_directions, direction_gate,
            persist_direction)

Guarda tus archivos. En el banco de trabajo, verifica que state write in place y persist_direction in the chain muestren marcas de verificación verdes.

El nodo del router (5B)

En el banco de trabajo, navega a El nodo del router (5B).

Enrutamiento determinístico basado en políticas

Un router es un nodo de función especializado que evalúa el resultado de la transmisión ascendente y dirige la ejecución a lo largo de las ramas condicionales del gráfico. A diferencia de los agentes generativos, un router ejecuta lógica determinística sin realizar llamadas a LLM.

Un router devuelve un Event que especifica una etiqueta 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")

En la definición del flujo de trabajo, un destino de borde definido como un diccionario asigna nombres de rutas a nodos de destino:

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

El enrutador de flujo de trabajo policy_check lee las frases prohibidas de agent/policy_words.txt y realiza la correlación de palabras completas con el título y el ángulo de la dirección elegida:

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

Almacenar la política como datos en lugar de instrucciones codificadas permite realizar actualizaciones sin modificar el gráfico de flujo de trabajo: la actualización del archivo de texto se aplica de inmediato a las ejecuciones posteriores. Dado que la evaluación es una coincidencia de regex determinística, se ejecuta en milisegundos con un costo de cero tokens antes de que comience la generación de código.

Destinos: Scripter y cuarentena

El router dirige el tráfico a uno de los dos nodos de transmisión:

• scripter: Es un nodo de agente single_turn que convierte la dirección aprobada en un guion de producción estructurado que se ajusta al esquema Script de Pydantic:

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

• quarantine: Inicialmente, es una función de marcador de posición que detiene las instrucciones marcadas, reemplazada en la siguiente parte por un agente de corrección autónomo.

Edición práctica: Enruta la verificación de políticas

1. En agent/graph.py, dentro de policy_check, completa la sentencia return:

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

2. En stage3_router/agent.py, actualiza edges para que dirija a policy_check y vuelve a unir la rama de cuarentena en scripter:

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

Guarda tus archivos. En el banco de trabajo, verifica que se hayan verificado las asignaciones de borde del router.

Modos del agente y el nodo de tareas (5C)

En Workbench, navega a Agent modes and the task node (5C).

Modos de ejecución del agente

Las instancias de ADK Agent admiten tres modos de ejecución adaptados a los requisitos específicos de la canalización:

Modo	Ciclo de vida de la ejecución	Rol en la canalización
chat	Es un bucle conversacional de varios turnos. El modelo determina cuándo invocar herramientas, solicitar entrada o finalizar el turno.	Agentes raíz que interactúan con un usuario humano.
single_turn	Es una sola llamada de inferencia del modelo. Acepta la entrada del nodo anterior y emite un objeto de esquema estructurado.	Transformaciones de gráficos secuenciales (propose_directions, scripter).
task	Es un bucle autónomo con ejecución de herramientas. El agente itera hasta que llama a la herramienta integrada finish_task.	Inspección y corrección de varios pasos (quarantine).

Corrección autónoma de políticas

Para reescribir una dirección marcada, se requiere el modo task, ya que la cantidad de iteraciones de corrección es variable. El agente recibe la instrucción marcada, invoca find_policy_hits para detectar incumplimientos, solicita alternativas aprobadas a través de suggest_replacement, vuelve a escribir la instrucción y verifica que no haya incumplimientos antes de continuar.

Ambas herramientas se definen en agent/cleanup_tools.py con firmas escritas y cadenas de documentación:

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.
    """

Edición práctica: Ensamblaje del agente de tareas de cuarentena

En stage3_router/agent.py, reemplaza la función de marcador de posición quarantine por la definición del agente de tareas:

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

El modo de tareas equipa al agente con herramientas y finaliza la ejecución llamando a finish_task. Cuando se configura mode="task", el ADK proporciona automáticamente finish_task y deriva sus parámetros de output_schema, lo que garantiza que el nodo genere un objeto CleanedDirection escrito que coincida con el esquema de entrada del nodo del guionista.

Qué esperar y por qué

Prueba ambas rutas de ejecución en la Web del ADK o en VibeStudio Workbench:

• Ruta aprobada (candidato 1, 2 o 3):
  • Selecciona una ruta candidata aprobada desde policy_check directamente hasta scripter (route="OK").
  • El guionista genera un guion de producción de 3 tomas que se ajusta al esquema Script.
• Ruta de corrección de cuarentena (candidato 4):
  • La opción 4 contiene vocabulario marcado ("cebo de clics", "truco viral").
  • policy_check rutas a quarantine (route="BLOCK").
  • En el registro de seguimiento de la sesión, observa que quarantine llama a find_policy_hits, llama a suggest_replacement para cada incumplimiento, reescribe el título y llama a finish_task.
  • La ejecución se une a scripter y produce un guion a partir de la dirección filtrada.

Routers y rutas de respaldo. La IU de desarrollo marca un router sin un borde de resguardo. Si policy_check hubiera devuelto una ruta diferente de OK o BLOCK, el recorrido no tendría destino. Agregar DEFAULT_ROUTE: quarantine al diccionario de borde abarca ese caso; importa DEFAULT_ROUTE desde google.adk.workflow.

6. Memory Bank

En VibeStudio Workbench, navega a Paso 6 · Memory Bank, partes (6A) y (6B).

Actualmente, el flujo de trabajo funciona sin memoria entre sesiones. Cada ejecución comienza desde cero, sin saber qué seleccionó el creador anteriormente ni qué géneros prefiere. En este paso, conectarás Vertex AI Agent Engine Memory Bank para almacenar y recuperar las preferencias del creador en diferentes ejecuciones.

Es fundamental que la memoria se integre a través de devoluciones de llamada del ciclo de vida del agente en lugar de nodos de canalización. Dado que la extracción y la recuperación de la memoria sirven a agentes individuales en lugar de etapas de datos intermedias, adjuntar devoluciones de llamada preserva una topología de gráfico limpia y desacoplada.

Memory Bank (6A)

En Workbench, navega a Memory Bank (6A).

Memoria administrada a nivel del usuario

Memory Bank es un servicio administrado para la memoria del usuario a largo plazo. Organiza los datos sobre una persona en un alcance definido, que aquí se identifica por el nombre de la aplicación y el ID de usuario:

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).",
}

Los temas de memoria personalizados definen los límites de lo que registra el banco:

• Extracción de temas: Cuando se envía texto de conversación nuevo a través de memories.generate, el servicio aplica un modelo de extracción a cada descripción del tema. El texto que no coincide con un tema no genera recuerdos.
• Consolidación y eliminación de duplicados: El servicio convierte los hechos recién extraídos en incorporaciones y los compara con los recuerdos existentes en el alcance. Cuando una observación coincide con un recuerdo existente, el servicio actualiza ese recuerdo. Cuando representa información nueva, el servicio crea una entrada nueva. Este proceso de consolidación garantiza que varias sesiones sobre un tema se fusionen en un resumen coherente en lugar de producir entradas redundantes.
• Recuperación: Llamar a memories.retrieve con el alcance del usuario devuelve hechos almacenados, ordenados del más antiguo al más reciente.

Ambas operaciones se implementan en agent/platform/memory.py. El nombre del recurso bancario aprovisionado se almacena en caché de forma local en runs/memorybank.json.

Cómo configurar Memory Bank

Usa los controles del banco de trabajo o ejecuta los comandos de la CLI en tu terminal:

1. Conecta y aprovisiona el banco:

   python -m agent.platform.bank
   

   Crea la instancia de Agent Engine y configura los temas CREATOR_TASTE y CHANNEL_RULES.
2. Propaga sesiones históricas:

   python -m agent.platform.bank load
   

   Carga cuatro sesiones históricas de creadores (dos temas de animales con restricciones de estilo, un tema de dispositivos y un tema de fantasía reciente).
3. Inspecciona los hechos consolidados:

   python -m agent.platform.bank list
   

   Examina el resultado. Observa cómo las transcripciones narrativas se convirtieron en declaraciones de hechos estructuradas y consolidadas.

Devoluciones de llamada (6B)

En el banco de trabajo, navega a Devoluciones de llamada (6B). Abre stage4_memory/agent.py.

Devoluciones de llamada del ciclo de vida del agente del ADK

Una devolución de llamada es una función que se pasa como argumento a un Agent. El ADK invoca devoluciones de llamada en momentos predefinidos del ciclo de vida y pasa el contexto activo. Devolver None continúa la ejecución normal; devolver un objeto de reemplazo anula o intercepta la operación.

El ADK proporciona tres pares de devoluciones de llamada:

Par de devolución de llamada	Punto de invocación	Parámetros recibidos	Comportamiento del valor de retorno
before_agent_callback after_agent_callback	Rodea todo el turno del agente.	CallbackContext (estado, sesión, invocación)	La devolución de Content reemplaza la respuesta del agente; None continúa con normalidad.
before_model_callback after_model_callback	Rodea cada llamada de inferencia del LLM	LlmRequest o LlmResponse	Si se devuelve LlmResponse, se intercepta o se omite la llamada al modelo; si se devuelve None, se continúa.
before_tool_callback after_tool_callback	Alrededor de cada ejecución de la herramienta	Definición, argumentos y resultado de la herramienta	Si se devuelve un dict, se anula el resultado de la herramienta y se continúa con None.

Las devoluciones de llamada proporcionan una ubicación limpia para la inyección de contexto, los rieles de protección, la telemetría y las búsquedas en caché sin introducir nodos externos en el gráfico de flujo de trabajo.

Edición práctica: Conexión de devoluciones de llamada de recuperación y recuerdo

1. En stage4_memory/agent.py, actualiza propose_directions para adjuntar before_model_callback=recall_taste:

    output_schema=Directions,
    before_model_callback=recall_taste)

recall_taste se ejecuta inmediatamente antes de que Gemini genere direcciones candidatas. Recupera el historial del creador de Memory Bank, da formato a los recuerdos de más antiguo a más reciente y los agrega al objeto LlmRequest saliente. La instrucción dirige al modelo para que los candidatos de inclinación 1 a 3 se acerquen al gusto actual del creador y, al mismo tiempo, traten las reglas del canal como restricciones estrictas.

2. En stage4_memory/agent.py, actualiza scripter para adjuntar after_agent_callback=remember_pick:

    output_schema=Script,
    after_agent_callback=remember_pick)

remember_pick se ejecuta después de que scripter completa su turno. Lee la dirección elegida del estado de la sesión, sintetiza una declaración concisa que resume la decisión del creador y llama a memories.generate para actualizar Memory Bank.

Qué esperar y por qué

Prueba el flujo de trabajo aumentado con devolución de llamada en el banco de trabajo o en la Web del ADK:

1. Ejecuta una ejecución con una instrucción vacía:
   • En el registro de seguimiento de la sesión, inspecciona LlmRequest para ver propose_directions. Observa el contexto de memoria agregado que detalla la preferencia del creador por los temas de fantasía y el ritmo conciso.
   • Observa las direcciones propuestas: Los candidatos del 1 al 3 se alinean con las preferencias históricas del creador, incluso cuando las tendencias enfatizan otros temas.
2. Selecciona un candidato en direction_gate.
3. Después de que se complete scripter, revisa los registros de Memory Bank:

   python -m agent.platform.bank list
   

   Ahora, el banco refleja la elección más reciente y la consolida con los registros de preferencias anteriores.

7. RAG Engine

En VibeStudio Workbench, navega a Paso 7 · RAG Engine, partes (7A) y (7B).

Los videos publicados acumulan comentarios continuos de los usuarios. En agent/comments.md, se recopilan treinta comentarios representativos que capturan los elogios de los usuarios, las críticas sobre el ritmo de los videos patrocinados y las preferencias de audio. En este paso, indexarás estos comentarios con Vertex AI RAG Engine y conectarás la recuperación semántica en el fan-out de la investigación.

Recuperación sobre documentos (7A)

En Workbench, navega a RAG Engine (7A).

Comparación entre Memory Bank y RAG Engine

Ambas herramientas basan los flujos de trabajo en datos externos, pero tienen propósitos arquitectónicos distintos:

Dimensión	Memory Bank	RAG Engine
Caso de uso principal	Preferencias del usuario a largo plazo y reglas operativas	Recuperación semántica en grandes colecciones de documentos
Alcance	Se limita a los IDs de usuario y nombres de aplicaciones individuales	Se limita a los recursos del corpus compartido entre todos los usuarios.
Procesamiento de datos	Extracción, incorporación y consolidación semántica en tiempo real	Fragmentación de documentos, incorporación de vectores y búsqueda de vecinos más cercanos
Integración de grafos	Devoluciones de llamada del ciclo de vida del agente (before_model_callback, after_agent_callback)	Nodo de función dedicado en fan-out de investigación (read_feedback)

Fragmentación y embeddings de documentos

RAG Engine indexa documentos dividiendo el texto en pasajes semánticos y almacenando sus vectores en una base de datos administrada:

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)))

• Tamaño del fragmento: Se configuró en 120 tokens con 20 tokens de superposición. Esto captura de dos a tres comentarios por pasaje, lo que garantiza que cada vector represente un sentimiento cohesivo sin diluir el significado en comentarios no relacionados.
• Modelo de embedding: text-embedding-005 convierte texto en vectores de alta dimensión. Cuando se envía una búsqueda, el modelo la convierte en un vector y encuentra las coincidencias más cercanas según la distancia semántica. Un comentario sobre un dragón pequeño que cuida calcetines coincide con una instrucción sobre criaturas mágicas sin requerir una coincidencia exacta de palabras clave.

Configura el corpus de RAG

Inicializa el corpus con los botones del banco de trabajo o los comandos de la terminal:

1. Crea el corpus:

   python -m agent.platform.rag
   

   Aprovisiona la base de datos vectorial administrada y registra el ID del recurso en runs/ragcorpus.json.
2. Upload and index comments: Sube agent/comments.md con la configuración de fragmentación y espera a que se complete la indexación.
3. Consulta el corpus: Prueba la recuperación de similitud con búsquedas que no compartan palabras exactas con los comentarios (por ejemplo, busca "criaturas mágicas pequeñas" para recuperar comentarios sobre dragones).

El nodo de recuperación (7B)

En Workbench, navega a The third reader (7B). Abre stage5_rag/agent.py.

Recuperación como nodo de gráfico

Los comentarios del público representan los datos de investigación que se comparten en todo el flujo de trabajo. A diferencia de la memoria personal del creador, el sentimiento de los usuarios se incorpora directamente a join_research junto con los datos de tendencias y de la lista de tareas pendientes. Por lo tanto, se implementa como un nodo de función:

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]})

read_feedback extrae la idea inicial del usuario y ejecuta una consulta vectorial en el corpus de RAG Engine. Emite los comentarios recuperados en una carga útil de Event(output=...).

Edición práctica: Cableado del tercer lector en el fan-out

En stage5_rag/agent.py, actualiza edges para agregar read_feedback como una tercera rama paralela que ingresa a join_research:

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

Como join_research es un JoinNode, sincroniza todas las ramas entrantes y espera hasta que scan_trends, read_backlog y read_feedback hayan emitido todos los eventos antes de pasar el paquete agregado a la siguiente etapa.

Qué esperar y por qué

Ejecuta el flujo de trabajo en el banco de trabajo:

1. Envía una instrucción de idea (como "un dragón en miniatura que cuida una encimera de cocina").
2. En el registro de ejecución, verifica que los tres nodos de lectura se ejecuten de forma simultánea.
3. Observa join_research: Su diccionario de salida ahora contiene trends, backlog y feedback.
4. Inspecciona los candidatos generados en propose_directions: El modelo incorpora los comentarios de los usuarios en sus propuestas y hace referencia al sentimiento del público en los campos de evidencia.
5. Ten en cuenta que la recuperación de RAG es determinística (las consultas idénticas devuelven pasajes de comentarios idénticos), mientras que el nodo de propuesta generativa produce variaciones creativas.

8. Generación asíncrona de video con Veo

En VibeStudio Workbench, navega a Paso 8: El video, partes (8A) y (8B).

Generar videos en alta definición con Google Veo requiere varios minutos por renderización. Bloquear la ejecución por grafos durante este período desperdicia recursos de cómputo, bloquea grupos de subprocesos y expone la ejecución a interrupciones de la conexión HTTP. En este paso, harás que la renderización de video sea asíncrona con el elemento LongRunningFunctionTool del ADK.

Herramientas de larga duración (8A)

En Workbench, navega a A long-running tool (8A). Abre stage6_video/agent.py y agent/deliver.py.

Herramientas síncronas vs. herramientas de ejecución prolongada

Las herramientas de funciones del ADK estándar se ejecutan de forma síncrona dentro de un turno del agente: el modelo llama a la herramienta, espera la carga útil de devolución y, luego, incorpora el resultado al turno en curso.

La renderización de video no se puede completar en un solo turno. En su lugar, render_submit inicia el trabajo de generación y, de inmediato, devuelve un recibo operativo con el estado "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"]}

Cuando se incluye en LongRunningFunctionTool, el ADK intercepta el estado "pending". Finaliza el turno del agente, el flujo de trabajo se suspende en el nodo y los metadatos de la llamada pendiente (incluidos el ID y el recibo de la llamada) se registran en runs/sessions.db. El proceso de ejecución sale de forma limpia sin mantener conexiones de red activas ni subprocesos de trabajo.

Edición práctica: cómo unir la herramienta de renderización

En stage6_video/agent.py, actualiza render_desk para unir render_submit en LongRunningFunctionTool:

    tools=[LongRunningFunctionTool(render_submit)])

Reanudación por ID de llamada

El patrón de reanudación universal

El ADK aplica un mecanismo idéntico para suspender y reanudar flujos de trabajo tanto para humanos como para herramientas externas:

Activador de suspensión	Cómo iniciar Construct	Estado de suspensión almacenado	Evento de reanudación
Decisión humana	yield RequestInput(...)	Abrir la instrucción de entrada en el almacén de sesiones	FunctionResponse que contiene el ID de la llamada de suspensión
Herramienta de larga duración	LongRunningFunctionTool(...) muestra pending	Abrir la llamada a herramienta en el almacén de sesiones	FunctionResponse que contiene el ID de la llamada de suspensión

En ambos casos, el flujo de trabajo se detiene por completo y se reanuda solo cuando llega un evento con un FunctionResponse coincidente desde una fuente externa: una interfaz de usuario, un webhook o un trabajador en segundo plano.

Edición práctica: Cómo completar la respuesta de entrega

En agent/deliver.py, construye la parte de reanudación FunctionResponse:

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

El daemon de entrega sondea Veo hasta que se genera el archivo de video y, luego, envía este FunctionResponse a la sesión. El ADK coincide con el ID de llamada y reanuda el flujo de trabajo directamente en el siguiente nodo. Los nodos completados no se vuelven a ejecutar, y el agente no toma otro turno generativo.

Configurar STUDIO_REAL_VIDEO=0 en .env habilita la renderización simulada: start devuelve un recibo de prueba inmediato y check simula la finalización en cinco segundos sin realizar llamadas facturables a la API de Veo.

Integración de canalizaciones (8B)

En el banco de trabajo, navega a render_desk en el gráfico (8B). Abre stage6_video/agent.py.

El nodo terminal de la canalización es store_video. Lee la información de renderización completada de runs/state.json (donde se registró el proceso de entrega) y confirma la URL del video y el estado de generación en el estado de sesión compartido.

Edición práctica: Cableado de la canalización de video completa

En stage6_video/agent.py, actualiza edges para agregar render_desk y store_video:

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

Qué esperar y por qué

Prueba el flujo de generación asíncrono en el banco de trabajo:

1. Ejecuta el flujo de trabajo a través de la selección de candidatos y la generación de secuencias de comandos.
2. En render_desk, observa cómo el agente invoca render_submit.
3. El flujo de trabajo se suspende de inmediato. En el banco de trabajo o en ADK Web, observa el estado pendiente: la sesión contiene el ID de llamada abierta y ningún proceso en segundo plano consume recursos.
4. Ejecuta el daemon de entrega con la consola de la estación de trabajo o en tu terminal:

   python -m agent.deliver
   

   El proceso de entrega supervisa a Veo hasta que el video está listo y, luego, envía el evento de reanudación.
5. En ADK Web, actualiza la sesión: la ejecución se reanuda en store_video, confirma la URL del video en el estado de la sesión y completa el flujo de trabajo.

9. Implementa en Cloud Run

En VibeStudio Workbench, navega a Paso 9: Implementar.

Desarrollaste y verificaste cada componente de la canalización en zonas de pruebas dedicadas. En este paso, ensamblarás la canalización de producción completa y la implementarás en Google Cloud Run.

El ejecutor de ADK

Durante el desarrollo, adk web orquestó el gráfico. En producción, la aplicación aloja el flujo de trabajo con la clase Runner del 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: Controla la ejecución del flujo de trabajo, genera eventos de forma secuencial a medida que se ejecutan los nodos y persiste las actualizaciones en el servicio de sesión.
• Reanudación unificada: Las decisiones del usuario en direction_gate y las entregas de videos completadas de Veo reanudan la ejecución a través de objetos FunctionResponse idénticos enviados a run_async.

La arquitectura de la aplicación de producción

La aplicación de producción en vibestudio/ integra la canalización completa:

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

• Flujo de eventos único: El backend de FastAPI publica eventos en un solo flujo de eventos enviados por el servidor (SSE). El frontend de React visualiza el progreso del gráfico en tiempo real y controla las conexiones tardías sin perder el estado.
• Ejecución desacoplada: La aplicación administra el bucle de eventos. El gráfico de flujo de trabajo se enfoca por completo en la lógica de ejecución, sin tener en cuenta la interfaz de frontend.

La lista de bordes del flujo de trabajo completo en agent/graph.py combina todos los patrones arquitectónicos creados a lo largo de este 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),

Implementa en Cloud Run

Google Cloud Run proporciona alojamiento sin servidores con escalamiento automático, enrutamiento de solicitudes y compilaciones de contenedores integradas:

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=...

• Compilación de contenedor: gcloud run deploy --source empaqueta el directorio vibestudio/, compila la imagen de contenedor con Cloud Build y, luego, implementa el servicio en una sola operación.
• Afinidad de sesión: Dirige las solicitudes del mismo usuario a la misma instancia del contenedor, lo que preserva el estado de la sesión local en los pasos iterativos.
• Observabilidad: La integración de Cloud Trace registra intervalos distribuidos para cada nodo, llamada a LLM y ejecución de herramientas, a los que se puede acceder en la consola de Google Cloud en el Explorador de Trace.

Haz clic en el botón Implementar en el banco de trabajo para ejecutar la secuencia de comandos de implementación. Cuando se complete la compilación, el terminal mostrará la URL del servicio activo.

10. Resumen

En VibeStudio Workbench, ve a Paso 10: Resumen para revisar la arquitectura completa.

Paso	Arquitectura y conceptos	Patrón de implementación
Una sola instrucción	Instrucción única, herramientas de funciones y bucle de chat secuencial	Agent(tools=[...]), function_call / function_response
Fundamentos del flujo de trabajo de agentes	Flujo de trabajo de gráficos, investigación paralela, resultados de esquemas, puerta humana	Workflow, START, JoinNode, output_schema y RequestInput
Estado y router	Estado de sesión compartido, vinculación de parámetros, enrutamiento determinístico, agente de tareas	Event(state=...), Event(route=...), mode="task", finish_task
Memory Bank	Memoria a largo plazo a nivel del usuario, consolidación semántica y hooks de ciclo de vida	memories.generate / retrieve, before_model_callback, after_agent_callback
RAG Engine	Recuperación de documentos a partir de comentarios del público y de incorporaciones semánticas	Nodo rag.create_corpus, RagEmbeddingModelConfig, read_feedback
Generación asíncrona de video con Veo	Herramientas de ejecución prolongada, recibos pendientes y daemon de entrega externo	LongRunningFunctionTool, reanudación de FunctionResponse(id=...)
Implementa en Cloud Run	Orquestación programática, eventos enviados por el servidor y contenedor sin servidores	Runner(agent=wf), run_async, implementación de Cloud Run

Principios básicos de arquitectura

1. Suspender en lugar de esperar: Los flujos de trabajo se pausan de forma limpia para la entrada humana (RequestInput) o las operaciones de larga duración (LongRunningFunctionTool). Los procesos no esperan inactivos en los subprocesos ni en los sockets de red.
2. Reanudación universal: Cada suspensión se reanuda a través de un mecanismo idéntico: un solo function_response que lleva el ID de llamada del nodo suspendido.
3. Administración de estados desacoplada: Los nodos comparten datos a través de claves de estado de sesión con nombre y vinculación de parámetros en lugar de cargas útiles intermedias detalladas y estrechamente acopladas.
4. Enrutamiento determinístico antes del costo generativo: Los enrutadores basados en reglas y los filtros de regex evalúan la política con un costo de cero tokens antes de que se ejecuten los modelos generativos.
5. Separación de responsabilidades: El contexto específico de un agente individual pertenece a las devoluciones de llamada del ciclo de vida, mientras que las dependencias de datos compartidos pertenecen