Workflow agentico con ADK

1. Introduzione

Questo codelab ti guida nella creazione di sistemi basati su agenti di nuova generazione utilizzando workflow e grafici in Agent Development Kit (ADK). Implementerai pattern architetturali comuni, orchestrerai interazioni human-in-the-loop (HITL) e gestirai l'esecuzione asincrona a esecuzione prolungata. Integrerai anche le knowledge base aziendali e la memoria persistente per personalizzare e far evolvere il comportamento dell'agente. Infine, collegherai queste funzionalità per creare una pipeline automatizzata di generazione di video.

Lo scenario

Gestisci un canale digitale su VibeTube con un pubblico attivo e un backlog in espansione di idee creative. La produzione di ogni video richiede un'esecuzione continua in più fasi: ricerca dei formati di tendenza, sintesi del feedback degli spettatori, sviluppo dei copioni, verifica della conformità alle norme e generazione di video clip. I modelli generativi possono creare bozze di singoli asset, ma per rilasci coerenti è necessaria un'architettura di agenti orchestrata.

Per automatizzare questo ciclo di vita, creerai VibeStudio. Questa pipeline agentica esegue ricerche di routine in parallelo, presenta opzioni curate per l'approvazione human-in-the-loop, applica gate di policy automatizzati prima di generare il video e conserva il contesto durante le esecuzioni di produzione.

Cosa imparerai

• Fondamenti di ingegneria dei grafi: le architetture degli agenti multi-step richiedono un flusso di controllo esplicito e percorsi di esecuzione strutturati. Crei un ADK Workflow utilizzando tuple edge, il punto di ingresso START, JoinNode per l'aggregazione parallela di fan-out e nodi router deterministici per indirizzare l'esecuzione in base allo stato.
• Modalità e callback del ciclo di vita degli agenti: le attività specializzate richiedono comportamenti operativi distinti e guardrail deterministici. Configura le istanze ADK Agent utilizzando le modalità chat, single_turn e task abilitate per gli strumenti come nodi del flusso di lavoro, applicando gli intercettori con before_model_callback e after_agent_callback.
• Orchestrazione human-in-the-loop: le pipeline di produzione si interrompono per il giudizio umano in corrispondenza di punti di controllo creativi critici. Implementa RequestInput per sospendere l'esecuzione del flusso di lavoro, applicare schemi di risposta strutturati e riprendere l'esecuzione senza mantenere attivi i processi di runtime inattivi.
• Memoria gerarchica dell'agente: i sistemi di produzione separano lo stato di esecuzione effimero dal contesto durevole. Gestisci lo stato della sessione a breve termine utilizzando Event(state=...) e l'associazione dei parametri e connetti GEAP Memory Bank per estrarre, consolidare e rendere persistenti le preferenze dei creator tra le esecuzioni.
• Fondamento con le knowledge base aziendali: gli agenti autonomi richiedono un contesto di dominio dinamico e il sentiment del pubblico. Collega un corpus di RAG Engine GEAP come nodo di recupero dedicato all'interno del fan-out parallelo per basare semanticamente gli output dell'agente.
• Workflow a lunga esecuzione e deployment: il rendering video multimodale viene eseguito in modo asincrono per periodi di tempo prolungati. Implementa LongRunningFunctionTool con le ricevute delle chiamate in attesa per sospendere e riprendere il flusso di lavoro in base all'ID chiamata e esegui il deployment della pipeline completata utilizzando l'ADK Runner su Cloud Run.

Come è organizzato questo codelab

Questo codelab funge da riferimento concettuale e architetturale. Ogni sezione spiega i costrutti dell'ADK implementati nel passaggio del workbench corrispondente, fornisce il codice di riferimento e stabilisce i principi di progettazione di base. Esamina ogni sezione prima di completare l'esercizio corrispondente nel workbench.

Il lavoro pratico si svolge in VibeStudio Workbench, un'interfaccia web complementare con un editor di codice interattivo, verificatori di runtime e un ispettore ADK incorporato. La numerazione dei passaggi nel workbench è allineata direttamente a questo codelab per mantenere sincronizzati i tuoi progressi. Le modifiche al grafico di base vengono mantenute nei vari passaggi e il workbench verifica automaticamente i prerequisiti man mano che avanzi.

Al termine degli esercizi del workbench, assemblerai una pipeline agentica end-to-end ed eseguirai il deployment di un'applicazione VibeStudio in esecuzione su Cloud Run per generare contenuti video.

L'ambiente è costituito da tre componenti principali: VibeStudio Workbench (l'interfaccia web locale per la modifica del codice e la verifica in fase di runtime), il backend (l'ADK Workflow e le sandbox di staging in agent/) e Google Cloud (modelli Gemini, GEAP Memory Bank, motore RAG e generazione di video Veo).

2. Configurazione

Richiedere i crediti per il workshop

Se partecipi a un lab guidato da un istruttore, quest'ultimo distribuirà i crediti per il tuo progetto Google Cloud. Segui le istruzioni dell'insegnante per utilizzare i tuoi crediti e assicurati che la fatturazione sia attiva sul tuo account prima di continuare.

Apri Cloud Shell

Cloud Shell è un ambiente di sviluppo basato su browser con gcloud, Python e git preinstallati.

Per avviare Cloud Shell:

1. Vai alla console Google Cloud.
2. Nell'intestazione di navigazione in alto, fai clic su Attiva Cloud Shell (l'icona della finestra del terminale).

Si apre una sessione del terminale nella parte inferiore della finestra del browser.

Clona e inizializza il repository

Esegui questi comandi nel terminale Cloud Shell per clonare il progetto:

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

Richieste di configurazione

Durante la configurazione, ti verranno richiesti i seguenti dettagli:

• ID progetto Google Cloud: quando richiesto da setup_project.sh, premi Invio per creare automaticamente un nuovo progetto. Se preferisci utilizzare un progetto esistente (ad esempio un progetto preassegnato), inserisci l'ID progetto e assicurati che l'ortografia sia corretta e che la fatturazione sia attiva.
• Codice evento: inserisci il codice della stanza fornito dall'insegnante. Se non ne hai ricevuto uno, chiedi a un assistente all'insegnamento o a un vicino. Se stai completando questo lab a casa, premi Invio per accettare la stanza sandbox predefinita.
• Nome visualizzato del canale: inserisci il tuo nome o l'handle del canale che preferisci quando ti viene richiesto da setup_codelab.sh oppure premi Invio per accettare il nome predefinito generato dal tuo Account Google.

Esegui i due script di configurazione in ordine:

./setup_project.sh
./setup_codelab.sh

• setup_project.sh: crea o riutilizza un progetto Google Cloud con fatturazione attiva, salva l'ID progetto in ~/project_id.txt e configura il contesto gcloud attivo.
• setup_codelab.sh: installa uv e le dipendenze Python in .venv, attiva le API Google Cloud richieste, configura le impostazioni del canale in .env, verifica l'accesso al modello con Gemini, esegue il provisioning delle risorse Memory Bank e RAG, crea l'interfaccia del workbench e avvia VibeStudio Workbench.

Lo script esegue il controllo preflight e avvia VibeStudio Workbench in background. Le ultime righe mostrano il link da aprire.

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

Fai clic su questo link. Lo stesso indirizzo è disponibile in Anteprima web → Cambia porta → 4600.

Per ricontrollare l'ambiente in qualsiasi momento, esegui python scripts/preflight.py. Per riavviare il workbench, esegui scripts/restart.sh. Per configurarlo di nuovo, esegui ./setup_codelab.sh, che conserva la configurazione e i progressi.

Una volta aperto, leggi lo step 1, The story, per lo scenario e lo step 2, What you build, per la forma del grafico finito. Nessuno dei due ha un esercizio. Poi torna qui per il passaggio 3.

Ogni parte pratica di VibeStudio Workbench termina con un pannello di verifica che legge gli artefatti reali: il file sul disco e le sessioni scritte dalle esecuzioni.

Layout del repository

Il repository è strutturato in logica del workflow principale, sandbox passo passo, ambiente workbench e applicazione di produzione:

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 il grafico del workflow principale. Modificherai i file in questa directory per implementare nodi di fan-out paralleli, routing deterministico delle policy, callback di memoria e strumenti di generazione video.
• agent/platform/: interfacce con i servizi Google Cloud, inclusi i modelli Gemini, GEAP Memory Bank, GEAP RAG Engine e la sintesi video di Veo.
• stage0_prompt/ fino a stage6_video/: ambienti sandbox autonomi. Ogni cartella esporta un root_agent autonomo, in modo da poter eseguire e ispezionare ogni passaggio in isolamento tramite l'interfaccia di sviluppo ADK incorporata.
• server/ e web/: l'applicazione VibeStudio Workbench in esecuzione localmente sulla porta 4600. Ospita la documentazione dei passaggi, l'editor di codice in-page, i verificatori delle prove di runtime e la visualizzazione dei grafici.
• vibestudio/: l'applicazione di produzione completa pacchettizzata e sottoposta a deployment in Cloud Run nell'ultimo passaggio. Contiene una propria copia autonoma del grafico del flusso di lavoro completato.

3. Agente monolitico

Prima di creare un grafico del workflow multi-nodo, stabilisci una base di riferimento dell'architettura con un singolo agente in stage0_prompt/agent.py. Questo agente si basa su un prompt di sistema monolitico che descrive la pipeline di produzione in prosa, supportato da due strumenti di funzioni Python.

La valutazione di questa baseline dimostra i limiti operativi del coordinamento basato sui prompt e stabilisce perché i sistemi di produzione richiedono l'orchestrazione dei grafici.

Architettura dell'agente ADK (3A)

In VibeStudio Workbench, vai al Passaggio 3 - Agente monolitico e apri Architettura dell'agente ADK (3A). Questa visualizzazione mostra i livelli architetturali principali di un agente 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
)

Il diagramma interattivo raggruppa i componenti dell'agente in cinque domini operativi:

• Livello di ragionamento (modello): il modello linguistico di base (ad esempio Gemini 3 Flash) che esegue attività cognitive, ragionamento dei prompt e selezione degli strumenti. Tutto il resto dell'architettura informa o vincola questo modello.
• Livello di contesto (istruzioni e skill): direttive che modellano il ragionamento del modello. instruction stabilisce il prompt di sistema permanente, la persona e le regole operative. skills forniscono indicazioni procedurali con controllo delle versioni (SKILL.md) per workflow ripetibili.
• Livello di collaborazione e azione (strumenti, subagenti, flusso di lavoro, schema di output): interfacce che consentono all'agente di agire su sistemi esterni ed emettere dati digitati. tools forniscono funzioni Python chiamabili o endpoint Model Context Protocol (MCP). subagents esegue le attività delegate subordinate. workflow coordina i grafici multi-agente. output_schema applica i modelli Pydantic per garantire che i consumatori downstream ricevano JSON convalidati anziché testo non strutturato.
• Livello di intercettazione (callback del ciclo di vita): guardrail deterministici che eseguono codice personalizzato prima e dopo l'esecuzione dell'agente (before_agent/after_agent), i turni del modello individuale (before_model/after_model) e le chiamate di strumenti (before_tool/after_tool). Gli intercettori applicano le regole dei criteri senza fare affidamento sulla conformità del modello.
• Stato esterno (sessione e memoria): persistenza stateful separata dalla logica dell'agente. Session conserva la memoria di lavoro temporanea e la traccia degli eventi per il thread di esecuzione corrente. Memory mantiene fatti e preferenze duraturi tra le sessioni utilizzando servizi gestiti come GEAP Memory Bank.

L'agente monolitico in questo passaggio implementa solo tre di queste primitive: model, instruction e tools. I passaggi successivi introducono flussi di lavoro grafici, schemi strutturati, intercettori e servizi di memoria persistente.

Specifica dell'agente monolitico (3B)

In Workbench, vai a Monolithic agent specification (3B). Apri stage0_prompt/agent.py per esaminare la definizione dell'agente di base:

• Istruzione con un solo prompt: il prompt di sistema condensa cinque attività di produzione distinte in un testo continuo: scoprire le tendenze della piattaforma, esaminare le idee in attesa, proporre concetti creativi, applicare le norme relative agli argomenti vietati e redigere le liste delle inquadrature.
• Origini dati sottostanti: l'agente fa riferimento a due origini definite accanto al grafico:
  • agent/trends.py: campioni di 10 tendenze attive di formattazione e stile da un pool di 250 con punteggi di calore dinamici.
  • agent/backlog.txt: Legge le note del concept grezze del creator riga per riga.

Strumenti in Agente (3C)

Nel workbench, vai a Strumenti nell'agente (3C).

Che cos'è uno strumento per un agente?

Un modello linguistico è intrinsecamente un motore di ragionamento a mondo chiuso: opera esclusivamente su pesi preaddestrati e sui token presenti nella sua finestra contestuale immediata. Non può eseguire query su un database, accedere alle API in tempo reale o eseguire codice in modo nativo.

Uno strumento colma questo confine. Concede al modello un'agenzia esterna, consentendogli di recuperare informazioni verificate ed eseguire azioni deterministiche in sistemi esterni.

L'invocazione di strumenti segue un protocollo esplicito in cinque fasi tra il modello e il runtime dell'ADK:

1. Dichiarazione dello schema: lo sviluppatore fornisce le funzioni Python all'agente. ADK esamina il nome, le annotazioni di tipo e le docstring di ogni funzione per generare una dichiarazione dello schema JSON compatibile con OpenAPI che descriva i parametri e lo scopo.
2. Ragionamento del modello: durante l'inferenza, il modello valuta se il prompt dell'utente richiede dati esterni. Se necessario, il modello genera un evento function_call strutturato contenente il nome della funzione di destinazione e il dizionario degli argomenti corrispondente allo schema.
3. Esecuzione del runtime: il modello stesso non esegue codice. L'ambiente di runtime dell'ADK intercetta function_call, esegue la funzione Python locale effettiva utilizzando gli argomenti forniti e acquisisce il valore restituito.
4. Re-iniezione del contesto: l'ambiente di runtime dell'ADK inserisce il valore restituito dalla funzione in un evento function_response e lo aggiunge alla cronologia della sessione attiva.
5. Sintesi finale: il modello elabora l'output dello strumento ora presente nella finestra contestuale e completa la risposta.

In stage0_prompt/agent.py, i due strumenti di ricerca sono definiti come funzioni Python standard:

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

Modifica ed esecuzione pratiche

Nell'editor di codice del workbench, aggiungi i due riferimenti alle funzioni all'elenco tools dell'agente:

    tools=[check_trends, read_backlog],

Salva la modifica. Il file viene aggiornato sul disco e la riga di verifica conferma che entrambi gli strumenti sono collegati.

Fai clic su Apri ADK web per avviare l'interfaccia di sviluppo ADK incorporata. Invia il prompt per l'idea suggerita:

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

Cosa aspettarsi e perché

Quando invii questo prompt, osserva la seguente sequenza di esecuzione nella traccia della sessione:

• Prima della risposta vengono visualizzati due eventi di esecuzione dello strumento: vengono visualizzati gli eventi function_call e function_response per check_trends e read_backlog.
  • Perché: Gemini ha valutato la direttiva del prompt di sistema ("controlla le tendenze. esamina l'elenco delle idee"), ha riconosciuto che mancavano le tendenze della piattaforma e le note del canale nei suoi pesi e ha richiamato entrambe le funzioni per basare il suo contesto.
• L'agente propone una direzione e si interrompe per la conferma: la risposta suggerisce una direzione per i video sintetizzando le tendenze e il backlog e ti chiede di confermare.
  • Perché: la direttiva ha chiesto al modello di concordare la direzione con il creator prima di generare lo script.
• Ignorare la conferma in un turno successivo: invia un secondo messaggio: skip the questions, just describe the video. L'agente ignora immediatamente la conferma e crea una bozza del titolo e degli scatti.
  • Motivo: le istruzioni del prompt sono linee guida consultive anziché barriere deterministiche. In un agente monolitico, le istruzioni dell'utente possono ignorare le regole del prompt di sistema predefinito perché nessun flusso di lavoro esterno controlla il flusso di esecuzione.

Limitazioni architetturali di un prompt monolitico

Sebbene un singolo prompt possa produrre un output accettabile per demo isolate, il test delle condizioni limite nel verificatore del workbench rivela limitazioni aziendali critiche:

• Aggregazione della ricerca non strutturata: l'ordine di esecuzione degli strumenti non è deterministico. Il modello riassume i dati recuperati in un testo in formato libero, rendendo impossibile per i sistemi downstream isolare l'origine che ha prodotto affermazioni specifiche.
• Applicazione di policy non verificate: il modello valuta la propria conformità alla sicurezza. Se il modello determina che un argomento è sicuro, nessuna logica deterministica esterna convalida il risultato.
• Pause Human-in-the-Loop non applicate: le istruzioni del prompt che richiedono la conferma del creator sono di natura consultiva. L'invio di un messaggio di follow-up che istruisce il modello a ignorare le domande fa sì che salti completamente l'approvazione umana.

Queste lacune architetturali motivano la scomposizione dell'agente monolitico nel flusso di lavoro del grafico esplicito creato nel passaggio successivo.

4. Nozioni di base sui workflow agentici

In VibeStudio Workbench, vai a Passaggio 4 - Nozioni di base del workflow agentico, parti 4A-4D.

Questo passaggio esegue la transizione da una baseline a singolo agente all'orchestrazione deterministica dei grafici utilizzando ADK Workflow. Creerai un fan-out di ricerca parallela, sincronizzerai i rami con un nodo di unione, genererai candidati creativi con schema convalidato e introdurrai un gate di approvazione deterministico human-in-the-loop.

Architettura del grafico e catene di esecuzione (4A)

Nel workbench, apri Architettura del grafico e catene di esecuzione (4A).

Un Workflow ADK struttura l'esecuzione dell'agente come un grafico diretto definito da un elenco di archi:

• Catene: le tuple sequenziali definiscono l'esecuzione lineare dei nodi ((node_a, node_b, node_c)).
• Rami paralleli: le catene indipendenti che condividono un nodo di origine vengono eseguite contemporaneamente.
• Sincronizzazione: le catene che convergono su un JoinNode attendono che tutti i rami in entrata vengano segnalati prima del rilascio.
• Controllo deterministico: il flusso di esecuzione è regolato da strutture di codice dichiarate anziché essere dedotto dal testo del prompt.

Archetipi di nodi in ADK

I flussi di lavoro ADK sono composti da diversi tipi di nodi specializzati. Ogni archetipo svolge un ruolo operativo specifico nel grafico, separando l'esecuzione del codice deterministico dal ragionamento del modello generativo:

Archetipo del nodo	Implementazione	Ruolo nella pipeline
Nodo funzione	Funzione Python che restituisce un Event	Esegue la logica deterministica, il recupero dei dati e le mutazioni di stato.
Nodo di unione	Istanza JoinNode integrata	Sincronizza i rami simultanei in un dizionario aggregato.
Nodo agente	Agent in esecuzione in modalità single_turn	Valuta le istruzioni rispetto all'input upstream ed emette dati convalidati.
Router node	Funzione che restituisce un Event con un tag route	Valuta la logica condizionale per selezionare i rami di esecuzione downstream.
Nodo di input umano	Funzione di generazione RequestInput	Sospende lo stato di esecuzione finché non arriva una risposta da un utente esterno.

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

In questa configurazione, root_agent è un'istanza di Workflow anziché un Agent autonomo. L'ADK considera i flussi di lavoro come agenti di prima classe, consentendo di caricare, pubblicare e ispezionare un intero grafico come un'applicazione unificata. name registra l'applicazione in ADK Web, mentre l'elenco edges ne definisce la topologia di esecuzione.

Fan-out della ricerca parallela (4B)

Nel workbench, vai a Parallel research fan-out (4B). Apri stage1_fanout/agent.py.

Nodi funzione e barriere di sincronizzazione

La fase di ricerca utilizza due nodi di funzione importati da agent/graph.py:

• scan_trends: restituisce Event(output={"trends": [...]}) contenente dieci tendenze della piattaforma con punteggio.
• read_backlog: Restituisce Event(output={"backlog": [...], "idea": "..."}) contenente quindici idee per i contenuti del canale insieme al prompt di esecuzione iniziale.

Ogni funzione accetta node_input (l'output del nodo precedente) e restituisce un Event.

Un JoinNode funge da barriera di sincronizzazione: si mette in pausa finché ogni catena in entrata non genera un evento, poi aggrega tutti i risultati dei rami in un dizionario con chiave in base al nome del nodo ({"scan_trends": {...}, "read_backlog": {...}}).

Modifica pratica: definizione dei bordi uniti e paralleli

In stage1_fanout/agent.py, crea un'istanza di JoinNode e collega le due catene parallele a partire da START:

join_research = JoinNode(name="join_research")

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

Salva le modifiche. Il verificatore del banco di lavoro conferma che la giunzione e i bordi sono collegati. Esegui la fase utilizzando Esegui fase 1 o tramite l'interfaccia web ADK incorporata.

Cosa aspettarsi e perché

• Esecuzione simultanea del lettore: nel grafico di esecuzione, scan_trends e read_backlog vengono eseguiti contemporaneamente.
  • Motivo: entrambe le catene hanno origine in START. Il motore ADK pianifica i rami indipendenti contemporaneamente.
• Output del dizionario aggregato: il flusso di lavoro viene completato alle ore join_research, generando un dizionario con voci per entrambi i lettori.
  • Perché: JoinNode garantisce l'acquisizione completa dei dati prima di consentire l'esecuzione dei nodi successivi.

Nodi agenti (4C)

Nel workbench, vai a Agent nodes (4C). Apri stage2_direction/agent.py.

Modalità operative e schemi strutturati

Se incorporato in un Workflow, un Agent viene eseguito in modalità single_turn per impostazione predefinita:

• Riceve l'output del nodo precedente come input di contesto.
• Esegue una singola chiamata di inferenza senza conversazioni avanti e indietro.
• Restituisce i dati strutturati al nodo successivo.

Assegnando output_schema=Directions, l'agente applica la convalida Pydantic all'output del modello. Il grafico downstream riceve oggetti digitati anziché prosa non strutturata:

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 indica al modello di proporre quattro candidati citando prove sia delle tendenze che del backlog. I candidati da 1 a 3 offrono concetti di canale validi. Il candidato 4 introduce intenzionalmente un concetto che viola le norme per testare il meccanismo di sicurezza nel passaggio successivo.

Modifica pratica: definizione del nodo dell'agente e concatenazione del join

In stage2_direction/agent.py, configura propose_directions ed estendi i bordi del flusso di lavoro:

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

Cosa aspettarsi e perché

• Utilizzo diretto del dizionario: propose_directions utilizza il payload JSON emesso da join_research senza formattazione manuale.
• Output del candidato digitato: l'agente emette un oggetto Directions convalidato contenente quattro candidati discreti. I nodi downstream leggono i campi in base al nome dell'attributo (candidate.title) senza l'analisi della stringa.

Human-in-the-loop (4D)

Nel workbench, vai a Human-in-the-loop (4D). Apri agent/graph.py.

Istruzioni del prompt e sospensione deterministica

I flussi di lavoro di produzione che comportano costi finanziari o pubblicano contenuti richiedono la supervisione umana nei punti decisionali critici. In un singolo prompt, le richieste di conferma sono istruzioni consultive che un utente può facilmente richiedere al modello di ignorare. In un flusso di lavoro ADK, l'approvazione umana viene applicata dal motore di esecuzione: il grafico si arresta in un nodo designato e non può avanzare finché non riceve un input esterno con schema convalidato:

• La generazione di RequestInput sospende immediatamente l'esecuzione del flusso di lavoro.
• ADK registra una chiamata di interruzione aperta nell'archivio delle sessioni ed emette un interrupt_id univoco.
• Il processo di esecuzione si arresta senza consumare token o thread del server.
• L'esecuzione del grafico riprende solo quando viene inviato un function_response valido corrispondente allo schema e all'ID interruzione.

Modifica pratica: sospensione dell'esecuzione con RequestInput

In agent/graph.py, implementa la chiamata di sospensione all'interno di 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 tre attributi:

• message: il prompt di recensione presentato all'utente.
• response_schema: uno schema JSON che il frontend esegue il rendering come modulo di input, convalidato da ADK al momento dell'invio.
• payload: Metadati inclusi nella richiesta (i quattro candidati), che consentono alle interfacce client di eseguire il rendering delle schede di recensione senza eseguire query sullo stato della sessione.

Cosa aspettarsi e perché

• Il flusso di lavoro si interrompe in direction_gate: in ADK Web o nell'interfaccia del workbench, l'esecuzione si interrompe e viene visualizzato un modulo interattivo di selezione dei candidati.
  • Perché: il motore ha rilevato un RequestInput e ha reso persistente lo stato di esecuzione in runs/sessions.db.
• La ripresa richiede un input strutturato: l'invio di testo della chat arbitrario non fa avanzare il grafico. Selezionando un'opzione (1, 2, 3 o 4) viene inviato un function_response digitato che soddisfa response_schema e l'esecuzione riprende.

5. Stato e router

In VibeStudio Workbench, vai a Passaggio 5: stato e router, parti da (5A) a (5C).

Conserverai le selezioni degli utenti nello stato della sessione, applicherai le norme di sicurezza del canale utilizzando nodi router deterministici e assemblerai un agente di attività iterativo per correggere automaticamente le violazioni delle norme prima di generare i copioni dei video.

Stato del workflow (5A)

In Workbench, vai a Stato del flusso di lavoro (5A).

Stato della sessione e output del nodo

In un flusso di lavoro ADK, i dati si spostano nel grafico attraverso due meccanismi distinti:

• Output del nodo (Event(output=...)): dati indirizzati rigorosamente ai consumatori downstream immediati definiti nell'elenco dei nodi periferici.
• Stato sessione (Event(state=...)): un dizionario coppia chiave-valore condiviso accessibile da qualsiasi nodo successivo nel ciclo di vita dell'esecuzione.

Quando un utente seleziona un candidato in direction_gate, la selezione arriva come indice numerico ({"pick": "2"}). I nodi downstream hanno bisogno dell'oggetto di regia completo: titolo, angolazione della narrazione e frase agganciante. Anziché trasmettere metadati dettagliati tramite ogni payload del nodo intermedio, persist_direction scrive il candidato risolto nello stato della sessione condivisa.

I nodi non devono superare l'intero dizionario degli stati della sessione. Quando un nodo genera Event(state=...), fornisce solo le coppie chiave-valore nuove o aggiornate. L'ADK unisce automaticamente questi aggiornamenti nello store delle sessioni:

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

In questo modo, Event passa il controllo al runtime Workflow, che salva i nuovi valori nel journal della sessione in runs/sessions.db.

Associazione dei parametri

I nodi funzione ADK leggono automaticamente lo stato della sessione tramite l'ispezione dei parametri. Se la firma di una funzione dichiara un nome di parametro corrispondente a una chiave di stato esistente, ADK estrae la chiave dallo stato e la trasmette direttamente:

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

Qui, candidates è stato scritto nello stato della sessione da direction_gate. ADK lo associa direttamente a persist_direction(node_input, candidates: list = []) senza richiedere ricerche esplicite nel dizionario.

Le chiavi con il prefisso user: vengono mantenute tra le sessioni nello spazio di archiviazione a livello utente, consentendo alle esecuzioni successive del flusso di lavoro di accedere alle preferenze del creator.

Modifica manuale: mantenimento dello stato e collegamento del nodo

1. In agent/graph.py, all'interno di persist_direction, sostituisci la riga TODO: PERSIST_STATE con il rendimento dell'evento di stato:

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

2. In stage3_router/agent.py, aggiungi persist_direction alla terza catena nell'elenco edges:

           (join_research, propose_directions, direction_gate,
            persist_direction)

Salva i file. Nel workbench, verifica che state write in place e persist_direction in the chain mostrino entrambi segni di spunta verdi.

Il nodo router (5B)

In Workbench, vai a The router node (5B).

Policy di routing deterministico

Un router è un nodo di funzione specializzato che valuta l'output upstream e indirizza l'esecuzione lungo i rami condizionali del grafico. A differenza degli agenti generativi, un router esegue una logica deterministica senza effettuare chiamate LLM.

Un router restituisce un Event che specifica un 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")

Nella definizione del flusso di lavoro, una destinazione edge definita come dizionario mappa i nomi delle route ai nodi di destinazione:

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

Il router del flusso di lavoro policy_check legge le frasi vietate da agent/policy_words.txt ed esegue la corrispondenza esatta con il titolo e l'angolazione della direzione scelta:

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

L'archiviazione delle norme come dati anziché come istruzioni hardcoded consente gli aggiornamenti senza modificare il grafico del flusso di lavoro: l'aggiornamento del file di testo viene applicato immediatamente alle esecuzioni successive. Poiché la valutazione è una corrispondenza di espressioni regolari deterministica, viene eseguita in millisecondi a costo zero dei token prima dell'inizio dello scripting generativo.

Destinazioni: Scripter e quarantena

Il router indirizza il traffico a uno dei due nodi downstream:

• scripter: un nodo agente single_turn che converte la direzione approvata in un copione di produzione strutturato conforme allo schema Pydantic Script:

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

• quarantine: inizialmente una funzione segnaposto che interrompe le indicazioni segnalate, sostituita nella parte successiva da un agente di correzione autonomo.

Modifica pratica: routing del controllo delle policy

1. In agent/graph.py, all'interno di policy_check, completa la dichiarazione return:

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

2. In stage3_router/agent.py, aggiorna edges per indirizzare policy_check e unisci di nuovo il ramo della quarantena in scripter:

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

Salva i file. Nel workbench, verifica che le mappature edge del router siano verificate.

Modalità dell'agente e nodo dell'attività (5C)

Nel workbench, vai a Modalità agente e nodo di attività (5C).

Modalità di esecuzione dell'agente

Le istanze ADK Agent supportano tre modalità di esecuzione personalizzate in base a requisiti specifici della pipeline:

Modalità	Ciclo di vita dell'esecuzione	Ruolo nella pipeline
chat	Ciclo conversazionale multi-turno. Il modello determina quando richiamare gli strumenti, richiedere input o terminare il turno.	Agenti root che interagiscono con un utente umano interattivo.
single_turn	Singola chiamata di inferenza del modello. Accetta l'input del nodo precedente ed emette un oggetto schema strutturato.	Trasformazioni del grafico sequenziale (propose_directions, scripter).
task	Loop autonomo con esecuzione dello strumento. L'agente esegue iterazioni fino a quando non chiama lo strumento finish_task integrato.	Ispezione e correzione in più passaggi (quarantine).

Correzione automatica delle policy

La riscrittura di una direzione segnalata richiede la modalità task perché il numero di iterazioni di correzione è variabile. L'agente riceve l'indicazione segnalata, richiama find_policy_hits per rilevare le violazioni, richiede alternative approvate tramite suggest_replacement, riscrive l'indicazione e verifica che sia pulita prima di procedere.

Entrambi gli strumenti sono definiti in agent/cleanup_tools.py con firme e docstring digitati:

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

Modifica pratica: assemblaggio dell'agente di attività di quarantena

In stage3_router/agent.py, sostituisci la funzione segnaposto quarantine con la definizione dell'agente di attività:

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

La modalità Attività fornisce all'agente gli strumenti e termina l'esecuzione chiamando finish_task. Quando mode="task" è configurato, ADK fornisce automaticamente finish_task e deriva i relativi parametri da output_schema, garantendo che il nodo produca un oggetto CleanedDirection digitato corrispondente allo schema di input del nodo scripter.

Cosa aspettarsi e perché

Testa entrambi i percorsi di esecuzione in ADK Web o VibeStudio Workbench:

• Percorso approvato (candidato 1, 2 o 3):
  • Selezione di un percorso candidato approvato da policy_check direttamente a scripter (route="OK").
  • L'autore dello script genera uno script di produzione a tre inquadrature in conformità allo schema Script.
• Percorso di correzione della quarantena (candidato 4):
  • Il candidato 4 contiene un vocabolario segnalato ("clickbait", "trucco virale").
  • policy_check percorsi per quarantine (route="BLOCK").
  • Nella traccia della sessione, osserva quarantine che chiama find_policy_hits, che chiama suggest_replacement per ogni violazione, che riscrive il titolo e che chiama finish_task.
  • L'esecuzione si unisce a scripter, producendo un copione dalla direzione sanificata.

Router e fallback. L'UI di sviluppo segnala un router senza un edge di fallback. Se policy_check avesse restituito un percorso diverso da OK o BLOCK, la corsa non avrebbe avuto una destinazione. L'aggiunta di DEFAULT_ROUTE: quarantine al dizionario edge copre questo caso; importa DEFAULT_ROUTE da google.adk.workflow.

6. Memory Bank

In VibeStudio Workbench, vai al Passaggio 6 · Memory Bank, parti (6A) e (6B).

Al momento il flusso di lavoro opera senza memoria tra le sessioni. Ogni esecuzione inizia da zero, senza sapere cosa ha selezionato in precedenza il creator o quali generi preferisce. In questo passaggio, colleghi Vertex AI Agent Engine Memory Bank per archiviare e recuperare le preferenze del creator durante le esecuzioni.

Fondamentalmente, la memoria è integrata tramite i callback del ciclo di vita dell'agente anziché i nodi della pipeline. Poiché l'estrazione e il recupero della memoria servono singoli agenti anziché fasi intermedie dei dati, l'aggiunta di callback preserva una topologia del grafico pulita e disaccoppiata.

Memory Bank (6A)

In Workbench, vai a Memory Bank (6A).

Memoria a livello utente gestita

Memory Bank è un servizio gestito per la memoria a lungo termine degli utenti. Organizza i fatti relativi a una persona in un ambito definito, identificato qui dal nome dell'applicazione e dall'ID utente:

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

Gli argomenti della memoria personalizzata definiscono i limiti di ciò che la banca registra:

• Estrazione degli argomenti: quando viene inviato un nuovo testo della conversazione tramite memories.generate, il servizio applica un modello di estrazione a ogni descrizione dell'argomento. Il testo che non corrisponde a un argomento non produce ricordi.
• Consolidamento ed eliminazione dei duplicati: il servizio converte i fatti appena estratti in incorporamenti e li confronta con i ricordi esistenti nell'ambito. Quando un'osservazione corrisponde a un ricordo esistente, il servizio lo aggiorna. Quando rappresenta nuove informazioni, il servizio crea una nuova voce. Questo processo di consolidamento garantisce che più sessioni su un argomento vengano unite in un riepilogo coerente anziché produrre voci ridondanti.
• Recupero: la chiamata a memories.retrieve con l'ambito utente restituisce i fatti memorizzati, ordinati dal meno recente al più recente.

Entrambe le operazioni sono implementate in agent/platform/memory.py. Il nome della risorsa banca di cui è stato eseguito il provisioning viene memorizzato nella cache locale in runs/memorybank.json.

Configurazione di Memory Bank

Utilizza i controlli del workbench o esegui i comandi della CLI nel terminale:

1. Connetti e configura la banca:

   python -m agent.platform.bank
   

   Crea l'istanza Agent Engine e configura gli argomenti CREATOR_TASTE e CHANNEL_RULES.
2. Inizializza le sessioni storiche:

   python -m agent.platform.bank load
   

   Carica quattro sessioni storiche per i creator (due temi di animali con vincoli di stile, un tema di gadget e un tema fantasy recente).
3. Controllare i fatti consolidati:

   python -m agent.platform.bank list
   

   Esamina l'output. Nota come le trascrizioni narrative sono state convertite in affermazioni strutturate e consolidate dei fatti.

Callback (6B)

In Workbench, vai a Callback (6B). Apri stage4_memory/agent.py.

Callback del ciclo di vita dell'agente ADK

Un callback è una funzione passata come argomento a un Agent. L'ADK richiama i callback in momenti del ciclo di vita predefiniti, passando il contesto attivo. La restituzione di None continua l'esecuzione normale; la restituzione di un oggetto sostitutivo esegue l'override o intercetta l'operazione.

ADK fornisce tre coppie di callback:

Coppia di callback	Punto di chiamata	Parametri ricevuti	Comportamento del valore restituito
before_agent_callback after_agent_callback	Intorno all'intera svolta dell'agente	CallbackContext (stato, sessione, chiamata)	Content sostituisce la risposta dell'agente; None procede normalmente.
before_model_callback after_model_callback	Intorno a ogni chiamata di inferenza LLM	LlmRequest o LlmResponse	Il ritorno di LlmResponse intercetta o salta la chiamata del modello; None procede.
before_tool_callback after_tool_callback	Intorno a ogni esecuzione dello strumento	Definizione, argomenti e risultato dello strumento	La restituzione di un dizionario sostituisce l'output dello strumento; None procede.

I callback forniscono una posizione pulita per l'inserimento del contesto, le misure di protezione, la telemetria e le ricerche nella cache senza introdurre nodi estranei nel grafico del flusso di lavoro.

Modifica pratica: richiamo dei cavi e callback dei promemoria

1. In stage4_memory/agent.py, aggiorna propose_directions per allegare before_model_callback=recall_taste:

    output_schema=Directions,
    before_model_callback=recall_taste)

recall_taste viene eseguito immediatamente prima che Gemini generi le indicazioni candidate. Recupera la cronologia del creator da Memory Bank, formatta i ricordi in ordine cronologico e li aggiunge al LlmRequest in uscita. Il prompt indica al modello di orientare i candidati da 1 a 3 verso i gusti attuali del creator, trattando le regole del canale come vincoli rigorosi.

2. In stage4_memory/agent.py, aggiorna scripter per allegare after_agent_callback=remember_pick:

    output_schema=Script,
    after_agent_callback=remember_pick)

remember_pick viene eseguito dopo che scripter ha completato il suo turno. Legge la direzione scelta dallo stato della sessione, sintetizza una dichiarazione concisa che riassume la decisione del creator e chiama memories.generate per aggiornare la Memory Bank.

Cosa aspettarsi e perché

Testa il flusso di lavoro aumentato con i callback nel workbench o nell'ADK Web:

1. Esegui una corsa con un prompt vuoto:
   • Nella traccia della sessione, esamina LlmRequest per propose_directions. Nota il contesto del ricordo aggiunto che descrive in dettaglio la preferenza del creator per i temi fantasy e il ritmo conciso.
   • Osserva le indicazioni proposte: i candidati da 1 a 3 sono in linea con le preferenze storiche del creator anche quando le tendenze enfatizzano altri argomenti.
2. Seleziona un candidato all'indirizzo direction_gate.
3. Al termine di scripter, esamina i record di Memory Bank:

   python -m agent.platform.bank list
   

   La banca ora riflette l'ultima scelta, consolidandola con i record dei gusti precedenti.

7. RAG Engine

In VibeStudio Workbench, vai a Passaggio 7 - RAG Engine, parti (7A) e (7B).

I video pubblicati accumulano feedback continui degli spettatori. In agent/comments.md vengono raccolti 30 commenti rappresentativi, che catturano i complimenti degli spettatori, le critiche al ritmo della sponsorizzazione e le preferenze audio. In questo passaggio, indicizza questi commenti utilizzando Vertex AI RAG Engine e collega il recupero semantico al fan-out della ricerca.

Recupero dai documenti (7A)

In Workbench, vai a RAG Engine (7A).

Memory Bank e RAG Engine

Entrambi gli strumenti basano i workflow su dati esterni, ma hanno scopi architetturali distinti:

Dimensione	Memory Bank	RAG Engine
Caso d'uso principale	Preferenze utente a lungo termine e regole operative	Recupero semantico su grandi raccolte di documenti
Ambito	Limitati a singoli ID utente e nomi delle applicazioni	Limitato alle risorse del corpus condivise tra tutti gli utenti
Elaborazione dei dati	Estrazione, incorporamento e consolidamento semantico in tempo reale	Suddivisione dei documenti in blocchi, vector embedding e ricerca del vicino più prossimo
Integrazione del grafico	Callback del ciclo di vita dell'agente (before_model_callback, after_agent_callback)	Nodo di funzione dedicato nel fan-out della ricerca (read_feedback)

Chunking ed embedding dei documenti

RAG Engine indicizza i documenti dividendo il testo in passaggi semantici e memorizzando i relativi vettori in un database gestito:

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

• Dimensione chunk: configurata su 120 token con 20 token di sovrapposizione. In questo modo vengono acquisiti da due a tre commenti per passaggio, garantendo che ogni vettore rappresenti un sentimento coeso senza diluire il significato in feedback non correlati.
• Modello di embedding: text-embedding-005 converte il testo in vettori ad alta dimensione. Quando viene inviata una query, il modello la converte in un vettore e trova le corrispondenze più vicine in base alla distanza semantica. Un commento su un minuscolo drago che fa la guardia ai calzini corrisponde a un prompt su creature magiche senza richiedere una sovrapposizione esatta delle parole chiave.

Configurazione del corpus RAG

Inizializza il corpus utilizzando i pulsanti del workbench o i comandi del terminale:

1. Crea il corpus:

   python -m agent.platform.rag
   

   Esegue il provisioning del database vettoriale gestito e registra l'ID risorsa in runs/ragcorpus.json.
2. Carica e indicizza i commenti: carica agent/comments.md con la configurazione di suddivisione in blocchi e attende il completamento dell'indicizzazione.
3. Esegui query sul corpus: testa il recupero della somiglianza con query che non condividono parole esatte con i commenti (ad esempio, esegui la query "piccole creature magiche" per recuperare i commenti sui draghi).

Il nodo di recupero (7B)

In Workbench, vai a The third reader (7B). Apri stage5_rag/agent.py.

Recupero come nodo del grafico

Il feedback sul pubblico rappresenta i dati di ricerca condivisi nel flusso di lavoro. A differenza della memoria personale del creator, il sentiment degli spettatori viene inserito direttamente in join_research insieme ai dati su tendenze e backlog. Pertanto, viene implementato come nodo di funzione:

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 estrae l'idea iniziale dell'utente ed esegue una query vettoriale sul corpus di RAG Engine. Emette i commenti recuperati in un payload Event(output=...).

Modifica pratica: collegamento del terzo lettore al fan-out

In stage5_rag/agent.py, aggiorna edges per aggiungere read_feedback come terzo ramo parallelo che entra in join_research:

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

Poiché join_research è un JoinNode, sincronizza tutti i rami in entrata, attendendo che scan_trends, read_backlog e read_feedback abbiano emesso tutti gli eventi prima di passare il bundle aggregato a valle.

Cosa aspettarsi e perché

Esegui il workflow in Workbench:

1. Invia un prompt per un'idea (ad esempio "un drago in miniatura che sorveglia il bancone della cucina").
2. Nella traccia di esecuzione, verifica che tutti e tre i nodi del lettore vengano eseguiti contemporaneamente.
3. Osserva join_research: il suo dizionario di output ora contiene trends, backlog e feedback.
4. Esamina i candidati generati da propose_directions: il modello incorpora i commenti degli spettatori nelle sue proposte e fa riferimento al sentiment del pubblico nei campi delle prove.
5. Tieni presente che il recupero RAG è deterministico (query identiche restituiscono passaggi di commenti identici), mentre il nodo di proposta generativa produce variazioni creative.

8. Generazione asincrona di video con Veo

In VibeStudio Workbench, vai al passaggio 8 - Il video, parti (8A) e (8B).

La generazione di video in alta definizione con Google Veo richiede diversi minuti per rendering. Il blocco dell'esecuzione del grafico durante questo periodo spreca risorse di calcolo, blocca i pool di thread ed espone l'esecuzione a interruzioni della connessione HTTP. In questo passaggio, il rendering video viene reso asincrono utilizzando LongRunningFunctionTool di ADK.

Strumenti a esecuzione prolungata (8A)

Nel workbench, vai a Uno strumento a esecuzione prolungata (8A). Apri stage6_video/agent.py e agent/deliver.py.

Strumenti sincroni e strumenti a esecuzione prolungata

Gli strumenti di funzione ADK standard vengono eseguiti in modo sincrono all'interno di un turno dell'agente: il modello chiama lo strumento, attende il payload di ritorno e incorpora il risultato nel turno in corso.

Il rendering video non può essere completato in un solo turno. render_submit avvia il job di generazione e restituisce immediatamente una ricevuta operativa con stato "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"]}

Se racchiuso tra LongRunningFunctionTool, ADK intercetta lo stato "pending". Il turno dell'agente termina, il workflow viene sospeso nel nodo e i metadati della chiamata in attesa (inclusi ID chiamata e ricevuta) vengono registrati in runs/sessions.db. Il processo di esecuzione termina in modo pulito senza mantenere connessioni di rete attive o thread di lavoro.

Modifica pratica: wrapping dello strumento di rendering

In stage6_video/agent.py, aggiorna render_desk per aggregare render_submit in LongRunningFunctionTool:

    tools=[LongRunningFunctionTool(render_submit)])

Ripresa per ID chiamata

Il pattern di ripresa universale

L'ADK applica un meccanismo identico per sospendere e riprendere i flussi di lavoro sia per gli utenti che per gli strumenti esterni:

Trigger sospensione	Avvio di Costruisci	Stato di sospensione memorizzato	Evento di ripristino
Decisione umana	yield RequestInput(...)	Apri la richiesta di input nello store delle sessioni	FunctionResponse che riporta l'ID chiamata di sospensione
Strumento a lunga esecuzione	LongRunningFunctionTool(...) in restituzione il giorno pending	Apri la chiamata allo strumento nell'archivio delle sessioni	FunctionResponse che riporta l'ID chiamata di sospensione

In entrambi gli scenari, il flusso di lavoro si interrompe completamente e riprende solo quando arriva da una fonte esterna un evento con un FunctionResponse corrispondente: un'interfaccia utente, un webhook o un worker in background.

Modifica pratica: completamento della risposta di consegna

In agent/deliver.py, crea la parte FunctionResponse di ripresa:

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

Il daemon di pubblicazione esegue il polling di Veo finché il file video non viene generato, quindi invia questo FunctionResponse alla sessione. L'ADK corrisponde all'ID chiamata e riprende il flusso di lavoro direttamente al nodo successivo. I nodi completati non vengono rieseguiti e l'agente non esegue un altro turno generativo.

L'impostazione di STUDIO_REAL_VIDEO=0 in .env consente il rendering simulato: start restituisce una ricevuta di test immediata e check simula il completamento in cinque secondi senza effettuare chiamate all'API Veo fatturabili.

Integrazione della pipeline (8B)

Nel workbench, vai a render_desk nel grafico (8B). Apri stage6_video/agent.py.

Il nodo terminale della pipeline è store_video. Legge le informazioni di rendering completate da runs/state.json (dove è stato registrato il processo di pubblicazione) e registra l'URL del video e lo stato di generazione nello stato della sessione condivisa.

Modifica pratica: cablaggio della pipeline video completa

In stage6_video/agent.py, aggiorna edges per aggiungere render_desk e store_video:

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

Cosa aspettarsi e perché

Testa il flusso di generazione asincrono nel workbench:

1. Esegui il flusso di lavoro tramite la selezione dei candidati e la generazione dello script.
2. In render_desk, osserva l'agente richiamare render_submit.
3. Il flusso di lavoro viene sospeso immediatamente. Nel workbench o in ADK Web, osserva lo stato in attesa: la sessione contiene l'ID chiamata aperto e nessun processo in background consuma risorse.
4. Esegui il daemon di pubblicazione utilizzando la console workbench o nel terminale:

   python -m agent.deliver
   

   La procedura di pubblicazione monitora Veo finché il video non è pronto, quindi invia l'evento di ripresa.
5. In ADK Web, aggiorna la sessione: l'esecuzione riprende da store_video, esegue il commit dell'URL del video nello stato della sessione e completa il flusso di lavoro.

9. Esegui il deployment in Cloud Run

In VibeStudio Workbench, vai al Passaggio 9: esegui il deployment.

Hai sviluppato e verificato ogni componente della pipeline in sandbox dedicate. In questo passaggio, assembla la pipeline di produzione completa ed esegui il deployment su Google Cloud Run.

ADK Runner

Durante lo sviluppo, adk web ha orchestrato il grafico. In produzione, l'applicazione ospita il flusso di lavoro utilizzando la classe Runner dell'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: gestisce l'esecuzione del flusso di lavoro, generando eventi in sequenza man mano che i nodi vengono eseguiti e mantenendo gli aggiornamenti al servizio di sessione.
• Ripresa unificata: sia le decisioni dell'utente in corrispondenza di direction_gate sia le consegne dei video completate da Veo riprendono l'esecuzione tramite oggetti FunctionResponse identici inviati a run_async.

Architettura dell'applicazione di produzione

L'applicazione di produzione in vibestudio/ integra la pipeline 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

• Stream di eventi singolo: il backend FastAPI pubblica gli eventi in un unico stream Server-Sent Events (SSE). Il frontend React visualizza l'avanzamento del grafico in tempo reale e gestisce le connessioni tardive senza perdere lo stato.
• Esecuzione disaccoppiata: l'applicazione gestisce il ciclo degli eventi. Il grafico del flusso di lavoro si concentra interamente sulla logica di esecuzione, senza tenere conto dell'interfaccia frontend.

L'elenco completo dei bordi del flusso di lavoro in agent/graph.py combina ogni pattern architetturale creato in questo 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),

Deployment in Cloud Run

Google Cloud Run fornisce hosting serverless con scalabilità automatica, routing delle richieste e build di container integrati:

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 container: gcloud run deploy --source pacchettizza la directory vibestudio/, crea l'immagine container utilizzando Cloud Build ed esegue il deployment del servizio in un'unica operazione.
• Affinità di sessione: indirizza le richieste dello stesso utente alla stessa istanza container, preservando lo stato della sessione locale nei passaggi iterativi.
• Osservabilità: l'integrazione di Cloud Trace registra gli span distribuiti per ogni nodo, chiamata LLM ed esecuzione di strumenti, accessibili nella console Google Cloud in Esplora tracce.

Fai clic sul pulsante Esegui il deployment nel workbench per eseguire lo script di deployment. Al termine della build, il terminale visualizza l'URL del servizio live.

10. Riepilogo

In VibeStudio Workbench, vai a Passaggio 10: Riepilogo per rivedere l'architettura completata.

Passaggio	Architettura e concetti	Pattern di implementazione
Un singolo prompt	Prompt singolo, strumenti di funzione, ciclo di chat sequenziale	Agent(tools=[...]), function_call / function_response
Nozioni di base sui workflow agentici	Flusso di lavoro del grafico, ricerca parallela, output dello schema, gate umano	Workflow, START, JoinNode, output_schema, RequestInput
Stato e router	Stato della sessione condivisa, binding dei parametri, routing deterministico, agente attività	Event(state=...), Event(route=...), mode="task", finish_task
Memory Bank	Memoria a lungo termine a livello utente, consolidamento semantico, hook del ciclo di vita	memories.generate / retrieve, before_model_callback, after_agent_callback
RAG Engine	Recupero di documenti basato su commenti del pubblico, incorporamenti semantici	Nodo rag.create_corpus, RagEmbeddingModelConfig, read_feedback
Generazione asincrona di video con Veo	Strumenti a esecuzione prolungata, ricevute in attesa, daemon di consegna esterno	LongRunningFunctionTool, FunctionResponse(id=...) ripresa
Esegui il deployment in Cloud Run	Orchestrazione programmatica, eventi inviati dal server, container serverless	Runner(agent=wf), run_async, deployment di Cloud Run

Principi architettonici fondamentali

1. Sospensione anziché attesa: i flussi di lavoro vengono messi in pausa in modo pulito per l'input umano (RequestInput) o per le operazioni a esecuzione prolungata (LongRunningFunctionTool). I processi non rimangono inattivi su thread o socket di rete.
2. Ripresa universale: ogni sospensione viene ripristinata tramite un meccanismo identico: un singolo function_response che trasporta l'ID chiamata del nodo sospeso.
3. Gestione dello stato disaccoppiata: i nodi condividono i dati tramite chiavi di stato della sessione denominate e binding dei parametri anziché payload intermedi dettagliati e strettamente accoppiati.
4. Routing deterministico prima del costo generativo: i router basati su regole e i filtri regex valutano le norme a costo zero dei token prima dell'esecuzione dei modelli generativi.
5. Separazione delle problematiche: il contesto specifico di un singolo agente appartiene ai callback del ciclo di vita, mentre le dipendenze dei dati condivisi appartengono