Agentischer Workflow mit dem ADK

1. Einführung

In diesem Codelab erfahren Sie, wie Sie mit Workflows und Diagrammen im Agent Development Kit (ADK) Agentensysteme der nächsten Generation erstellen. Sie implementieren gängige Architekturmuster, orchestrieren Human-in-the-Loop-Interaktionen (HITL) und verarbeiten die asynchrone Ausführung mit langer Ausführungszeit. Außerdem binden Sie Wissensdatenbanken des Unternehmens und persistenten Speicher ein, um das Verhalten von Agenten anzupassen und weiterzuentwickeln. Schließlich verbinden Sie diese Funktionen, um eine automatisierte Pipeline für die Videogenerierung zu erstellen.

Szenario

Du betreibst einen digitalen Kanal auf VibeTube mit einem aktiven Publikum und einem wachsenden Fundus an kreativen Ideen. Für die Produktion jedes Videos sind mehrere Phasen erforderlich: Recherche nach aktuellen Formaten, Berücksichtigung von Zuschauerfeedback, Entwicklung von Skripten, Prüfung der Richtlinienkonformität und Generierung von Videoclips. Generative Modelle können zwar einzelne Assets entwerfen, für konsistente Releases ist jedoch eine koordinierte KI-Agentenarchitektur erforderlich.

Um diesen Lebenszyklus zu automatisieren, erstellen Sie VibeStudio. Diese Pipeline führt Routine-Recherchen parallel aus, präsentiert kuratierte Optionen für die Genehmigung durch den Menschen, wendet automatisierte Richtlinien an, bevor Videos generiert werden, und behält den Kontext über Produktionsläufe hinweg bei.

Lerninhalte

• Grundlagen für die Entwicklung von Graphen: Multi-Step-Agentenarchitekturen erfordern einen expliziten Kontrollfluss und strukturierte Ausführungspfade. Sie erstellen ein ADK Workflow mit Edge-Tupeln, dem Einstiegspunkt START, JoinNode für die parallele Fan-Out-Aggregation und deterministischen Routerknoten, um die Ausführung basierend auf dem Status zu steuern.
• Agent-Modi und Lifecycle-Callbacks: Für spezielle Aufgaben sind unterschiedliche Betriebsverhalten und deterministische Schutzmaßnahmen erforderlich. Sie konfigurieren ADK-Agent-Instanzen mit den Modi chat, single_turn und task als Workflowknoten und wenden Interceptors mit before_model_callback und after_agent_callback an.
• Human-in-the-Loop-Orchestration: Produktionspipelines werden an wichtigen kreativen Prüfpunkten pausiert, damit ein Mensch eine Entscheidung treffen kann. Sie implementieren RequestInput, um die Workflow-Ausführung zu unterbrechen, strukturierte Antwortschemata zu erzwingen und die Ausführung fortzusetzen, ohne Leerlauf-Laufzeitprozesse aktiv zu halten.
• Hierarchischer Agentspeicher: In Produktionssystemen wird der kurzlebige Ausführungsstatus vom dauerhaften Kontext getrennt. Sie verwalten den kurzfristigen Sitzungsstatus mit Event(state=...) und der Parameterbindung und verbinden GEAP Memory Bank, um die Einstellungen des Creators über mehrere Läufe hinweg zu extrahieren, zusammenzufassen und beizubehalten.
• Fundierung mit unternehmenseigenen Wissensdatenbanken: Autonome Agents benötigen dynamischen Domänenkontext und Informationen zur Stimmung der Zielgruppe. Sie verbinden einen GEAP RAG Engine-Korpus als dedizierten Abrufknoten innerhalb des parallelen Fan-out, um Agentenausgaben semantisch zu fundieren.
• Lang andauernde Workflows und Bereitstellung: Das multimodale Videorendering erfolgt asynchron über einen längeren Zeitraum. Sie implementieren LongRunningFunctionTool mit ausstehenden Anrufbelegen, um den Workflow anhand der Anruf-ID zu pausieren und fortzusetzen, und stellen die fertige Pipeline mit dem ADK Runner in Cloud Run bereit.

Aufbau dieses Codelabs

Dieses Codelab dient als konzeptuelle und architektonische Referenz. In jedem Abschnitt werden die im entsprechenden Workbench-Schritt implementierten ADK-Konstrukte erläutert, Referenzcode bereitgestellt und grundlegende Designprinzipien festgelegt. Sehen Sie sich jeden Abschnitt an, bevor Sie die entsprechende Übung in der Workbench durchführen.

Die praktische Arbeit findet in der VibeStudio Workbench statt, einer begleitenden Weboberfläche mit einem interaktiven Code-Editor, Laufzeitprüfern und einem eingebetteten ADK-Inspector. Die Schrittnummerierung in der Workbench stimmt direkt mit diesem Codelab überein, damit Ihr Fortschritt synchronisiert bleibt. Grundlegende Änderungen am Diagramm bleiben über die einzelnen Schritte hinweg erhalten. Die Workbench prüft automatisch die Voraussetzungen, während Sie fortfahren.

Nach Abschluss der Workbench-Übungen haben Sie eine End-to-End-Pipeline mit Agenten zusammengestellt und eine funktionierende VibeStudio-Anwendung in Cloud Run bereitgestellt, um Videoinhalte zu generieren.

Die Umgebung besteht aus drei Hauptkomponenten: der VibeStudio Workbench (der lokalen Weboberfläche für die Codebearbeitung und Laufzeitüberprüfung), Ihrem Backend (dem ADK Workflow und den Stage-Sandboxes in agent/) und Google Cloud (Gemini-Modelle, GEAP Memory Bank, RAG Engine und Veo-Videogenerierung).

2. Einrichtung

Workshop-Guthaben einlösen

Wenn Sie an einem von einem Kursleiter geleiteten Lab teilnehmen, verteilt der Kursleiter das Guthaben für Ihr Google Cloud-Projekt. Folgen Sie der Anleitung des Kursleiters, um Ihre Guthabenpunkte einzulösen und dafür zu sorgen, dass die Abrechnung in Ihrem Konto aktiviert ist, bevor Sie fortfahren.

Cloud Shell öffnen

Cloud Shell ist eine browserbasierte Entwicklungsumgebung, in der gcloud, Python und Git vorinstalliert sind.

Gehen Sie wie folgt vor, um Cloud Shell zu starten:

1. Öffnen Sie die Google Cloud Console.
2. Klicken Sie in der Kopfzeile der oberen Navigationsleiste auf Cloud Shell aktivieren (das Terminalfenstersymbol).

Unten im Browserfenster wird eine Terminalsitzung geöffnet.

Repository klonen und initialisieren

Führen Sie im Cloud Shell-Terminal die folgenden Befehle aus, um das Projekt zu klonen:

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

Konfigurationsaufforderungen

Während der Einrichtung werden Sie zur Eingabe der folgenden Informationen aufgefordert:

• Google Cloud-Projekt-ID: Wenn Sie von setup_project.sh dazu aufgefordert werden, drücken Sie die Eingabetaste, um automatisch ein neues Projekt zu erstellen. Wenn Sie ein vorhandenes Projekt verwenden möchten (z. B. ein vorab zugewiesenes Projekt), geben Sie Ihre Projekt-ID ein und achten Sie darauf, dass die Schreibweise korrekt ist und die Abrechnung aktiviert ist.
• Eventcode: Geben Sie den von der Lehrkraft bereitgestellten Raumcode ein. Wenn Sie keine erhalten haben, fragen Sie einen Tutor oder einen Nachbarn. Wenn Sie dieses Lab zu Hause durchführen, drücken Sie die Eingabetaste, um den Standardraum sandbox zu übernehmen.
• Anzeigename des Kanals: Gib deinen Namen oder deinen bevorzugten Kanal-Alias ein, wenn du von setup_codelab.sh dazu aufgefordert wirst. Du kannst auch die Eingabetaste drücken, um den Standardnamen zu übernehmen, der aus deinem Google-Konto generiert wurde.

Führen Sie die beiden Einrichtungsskripts in der folgenden Reihenfolge aus:

./setup_project.sh
./setup_codelab.sh

• setup_project.sh: Erstellt oder verwendet ein Google Cloud-Projekt mit aktiver Abrechnung wieder, speichert die Projekt-ID in ~/project_id.txt und konfiguriert den aktiven gcloud-Kontext.
• setup_codelab.sh: Installiert uv und Python-Abhängigkeiten in .venv, aktiviert die erforderlichen Google Cloud-APIs, konfiguriert die Kanaleinstellungen in .env, prüft den Modellzugriff mit Gemini, stellt Memory Bank- und RAG-Ressourcen bereit, erstellt die Workbench-Oberfläche und startet die VibeStudio Workbench.

Das Skript führt die Preflight-Prüfung aus und startet VibeStudio Workbench im Hintergrund. In den letzten Zeilen wird der Link zum Öffnen angezeigt.

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

Klicken Sie auf diesen Link. Dieselbe Adresse ist unter Webvorschau → Port ändern → 4600 verfügbar.

Wenn Sie die Umgebung noch einmal prüfen möchten, führen Sie python scripts/preflight.py aus. Führen Sie scripts/restart.sh aus, um die Workbench neu zu starten. Wenn Sie die Einrichtung noch einmal durchführen möchten, führen Sie ./setup_codelab.sh aus. Ihre Konfiguration und Ihr Fortschritt bleiben dabei erhalten.

Lesen Sie sich das Szenario in Schritt 1: Die Geschichte und die Form des fertigen Diagramms in Schritt 2: Was Sie erstellen durch. Keines der beiden hat eine Übung. Kehren Sie dann hierher zurück, um mit Schritt 3 fortzufahren.

Jeder praktische Teil der VibeStudio Workbench endet mit einem Bestätigungsfeld, in dem die tatsächlichen Artefakte gelesen werden: die Datei auf der Festplatte und die von den Ausführungen geschriebenen Sitzungen.

Repository-Layout

Das Repository ist in die Kern-Workflow-Logik, schrittweise Sandboxes, die Workbench-Umgebung und die Produktionsanwendung unterteilt:

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/: Enthält das Kernworkflowdiagramm. Sie bearbeiten Dateien in diesem Verzeichnis, um parallele Fan-Out-Knoten, deterministisches Richtlinienrouting, Memory-Callbacks und Videogenerierungstools zu implementieren.
• agent/platform/: Schnittstellen zu Google Cloud-Diensten, einschließlich Gemini-Modellen, GEAP Memory Bank, GEAP RAG Engine und Veo-Videosynthese.
• stage0_prompt/ bis stage6_video/: Eigenständige Sandbox-Umgebungen. Für jeden Ordner wird ein eigenständiges root_agent exportiert. So können Sie jeden Schritt über die eingebettete ADK-Entwicklungsoberfläche isoliert ausführen und prüfen.
• server/ und web/: Die VibeStudio Workbench-Anwendung, die lokal auf Port 4600 ausgeführt wird. Hier finden Sie die Schrittdokumentation, den In-Page-Code-Editor, die Laufzeit-Beweisprüfer und die Diagrammvisualisierung.
• vibestudio/: Die vollständige Produktionsanwendung, die im letzten Schritt verpackt und in Cloud Run bereitgestellt wird. Sie enthält eine eigene eigenständige Kopie des abgeschlossenen Workflow-Diagramms.

3. Monolithischer Agent

Bevor Sie ein Workflowdiagramm mit mehreren Knoten erstellen, legen Sie mit einem einzelnen Agent in stage0_prompt/agent.py eine architektonische Baseline fest. Dieser Agent basiert auf einem monolithischen System-Prompt, der die Produktionspipeline in Prosa beschreibt und von zwei Python-Funktionstools unterstützt wird.

Die Auswertung dieser Baseline zeigt die betrieblichen Grenzen der promptgesteuerten Koordination und verdeutlicht, warum für Produktionssysteme eine Graph-Orchestrierung erforderlich ist.

ADK-Agentenarchitektur (3A)

Gehen Sie in der VibeStudio Workbench zu Schritt 3: Monolithischer Agent und öffnen Sie ADK-Agentenarchitektur (3A). In dieser Ansicht werden die wichtigsten Architekturschichten eines ADK-Agents (LlmAgent) dargestellt:

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
)

Im interaktiven Diagramm werden die Agent-Komponenten in fünf operative Bereiche unterteilt:

• Reasoning-Ebene (Modell): Das Kern-Language Model (z. B. Gemini 3 Flash), das kognitive Aufgaben, Prompt-Reasoning und die Toolauswahl ausführt. Alles andere in der Architektur informiert oder schränkt dieses Modell ein.
• Kontextschicht (Anweisung und Fähigkeiten): Richtlinien, die die Argumentation des Modells beeinflussen. instruction legt den permanenten Systemprompt, die Rolle und die Betriebsregeln fest. skills bietet versionsbezogene, prozedurale Anleitungen (SKILL.md) für wiederholbare Workflows.
• Zusammenarbeits- und Aktionsebene (Tools, Sub-Agents, Workflow, Ausgabeschema): Schnittstellen, die es dem Agent ermöglichen, auf externe Systeme zuzugreifen und typisierte Daten auszugeben. tools bereitstellen, die aufrufbare Python-Funktionen oder MCP-Endpunkte (Model Context Protocol) sind. subagents untergeordnete delegierte Aufgaben ausführen. workflow koordiniert Multi-Agenten-Diagramme. Mit output_schema werden Pydantic-Modelle angewendet, um sicherzustellen, dass nachgelagerte Nutzer validiertes JSON anstelle von unstrukturiertem Text erhalten.
• Interceptor-Ebene (Lifecycle-Callbacks): Deterministische Schutzmaßnahmen, die benutzerdefinierten Code vor und nach der Ausführung des Agenten (before_agent/after_agent), einzelnen Modell-Turns (before_model/after_model) und Tool-Aufrufen (before_tool/after_tool) ausführen. Interceptors erzwingen Richtlinienregeln, ohne auf die Modell-Compliance angewiesen zu sein.
• Externer Status (Sitzung und Speicher): Statusbehaftete Persistenz, die von der Agent-Logik getrennt ist. Session behält den temporären Arbeitsspeicher und den Ereignistrace für den aktuellen Ausführungsthread bei. Memory speichert dauerhafte sitzungsübergreifende Fakten und Einstellungen mithilfe von verwalteten Diensten wie GEAP Memory Bank.

Der monolithische Agent in diesem Schritt implementiert nur drei dieser Primitiven: model, instruction und tools. In den folgenden Schritten werden Diagramm-Workflows, strukturierte Schemas, Interceptors und nichtflüchtige Speicherdienste vorgestellt.

Monolithische Agentenspezifikation (3B)

Gehen Sie in der Workbench zu Monolithic agent specification (3B) (Spezifikation für monolithischen Agenten (3B)). Öffnen Sie stage0_prompt/agent.py, um die Baseline-Agent-Definition zu prüfen:

• Anweisung mit einem einzelnen Prompt: Der Systemprompt fasst fünf verschiedene Produktionsaufgaben in einem fortlaufenden Text zusammen: Plattformtrends ermitteln, Ideen im Backlog prüfen, kreative Konzepte vorschlagen, Richtlinien zu verbotenen Themen durchsetzen und Shotlists entwerfen.
• Zugrunde liegende Datenquellen: Der Agent verweist auf zwei Quellen, die neben dem Diagramm definiert sind:
  • agent/trends.py: Es werden zehn aktive Format- und Stiltrends aus einem Pool von 250 Trends mit dynamischen Hitzepunkten ausgewählt.
  • agent/backlog.txt: Liest die Rohkonzeptnotizen des Creators zeilenweise vor.

Tools in KI-Agenten (3C)

Gehen Sie in der Workbench zu Tools in Agent (3C).

Was ist ein Tool für einen Agenten?

Ein Language Model ist von Natur aus eine Closed-World-Engine: Es arbeitet ausschließlich mit vortrainierten Gewichten und den Tokens, die sich im unmittelbaren Kontextfenster befinden. Sie kann nicht nativ Datenbanken abfragen, auf Echtzeit-APIs zugreifen oder Code ausführen.

Ein Tool schließt diese Lücke. Es gewährt dem Modell externe Autonomie, sodass es Informationen abrufen und deterministische Aktionen in externen Systemen ausführen kann.

Toolaufrufe folgen einem expliziten fünfstufigen Protokoll zwischen dem Modell und der ADK-Laufzeit:

1. Schemadeklaration: Der Entwickler stellt dem Agent Python-Funktionen zur Verfügung. Das ADK prüft den Namen, die Typanmerkungen und die Docstrings jeder Funktion, um eine OpenAPI-kompatible JSON-Schemadeklaration zu generieren, die ihre Parameter und ihren Zweck beschreibt.
2. Modell-Reasoning: Während der Inferenz bewertet das Modell, ob für den Prompt des Nutzers externe Daten erforderlich sind. Bei Bedarf gibt das Modell ein strukturiertes function_call-Ereignis aus, das den Namen der Zielfunktion und das Argument-Dictionary enthält, das dem Schema entspricht.
3. Laufzeitausführung: Das Modell selbst führt keinen Code aus. Die ADK-Laufzeit fängt function_call ab, führt die tatsächliche lokale Python-Funktion mit den bereitgestellten Argumenten aus und erfasst den Rückgabewert.
4. Kontext-Re-Injection: Die ADK-Laufzeit packt den Rückgabewert der Funktion in ein function_response-Ereignis und hängt ihn an den aktiven Sitzungsverlauf an.
5. Abschließende Synthese: Das Modell verarbeitet die Tool-Ausgabe, die sich jetzt in seinem Kontextfenster befindet, und schließt seine Antwort ab.

In stage0_prompt/agent.py werden die beiden Recherchetools als Standard-Python-Funktionen definiert:

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

Praktische Bearbeitung und Ausführung

Fügen Sie im Code-Editor der Workbench die beiden Funktionsreferenzen in die tools-Liste des Agenten ein:

    tools=[check_trends, read_backlog],

Speichern Sie die Änderung. Die Datei wird auf der Festplatte aktualisiert und in der Bestätigungszeile wird bestätigt, dass beide Tools verbunden sind.

Klicken Sie auf Open adk web (ADK-Weboberfläche öffnen), um die eingebettete ADK-Entwicklungsoberfläche zu starten. Senden Sie den vorgeschlagenen Prompt für die Idee:

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

Was Sie erwartet und warum

Wenn Sie diesen Prompt senden, sehen Sie im Sitzungstrace die folgende Ausführungsreihenfolge:

• Zwei Ereignisse zur Toolausführung werden vor der Antwort angezeigt: Sie sehen die Ereignisse function_call und function_response für check_trends und read_backlog.
  • Grund: Gemini hat die Systemprompt-Anweisung („check what is trending. look at your backlog of ideas“) ausgewertet, erkannt, dass in den Gewichten Plattformtrends und Kanalnotizen fehlen, und beide Funktionen aufgerufen, um den Kontext zu fundieren.
• Der KI-Agent schlägt eine Richtung vor und wartet auf Bestätigung: In der Antwort wird eine Videorichtung vorgeschlagen, die Trends und Backlog berücksichtigt. Sie werden aufgefordert, diese zu bestätigen.
  • Warum: In der Anweisung wurde das Modell aufgefordert, sich mit dem Creator auf die Richtung zu einigen, bevor das Skript generiert wird.
• Bestätigung in einem Folgeturn umgehen: Senden Sie eine zweite Nachricht: skip the questions, just describe the video. Der KI-Agent umgeht die Bestätigung sofort und entwirft Titel und Bilder.
  • Grund: Prompt-Anweisungen sind Empfehlungen und keine deterministischen Barrieren. Bei einem monolithischen Agenten können Nutzeranweisungen die bestehenden Systemprompt-Regeln überschreiben, da kein externer Workflow den Ausführungsablauf steuert.

Architektonische Einschränkungen eines monolithischen Prompts

Ein einzelner Prompt kann für isolierte Demos akzeptable Ergebnisse liefern. Wenn Sie jedoch Grenzbedingungen im Workbench-Prüftool testen, werden kritische Einschränkungen für Unternehmen deutlich:

• Aggregation unstrukturierter Daten: Die Reihenfolge der Tool-Ausführung ist nicht deterministisch. Das Modell fasst abgerufene Daten in Freiformprosa zusammen, sodass nachgelagerte Systeme nicht isolieren können, aus welcher Quelle bestimmte Behauptungen stammen.
• Nicht bestätigte Richtliniendurchsetzung: Das Modell bewertet seine eigene Sicherheitskonformität. Wenn das Modell feststellt, dass ein Thema sicher ist, wird diese Feststellung nicht durch eine externe deterministische Logik validiert.
• Nicht erzwungene Human-in-the-Loop-Pausen: Die Aufforderung zur Bestätigung durch den Creator ist nur ein Hinweis. Wenn Sie eine Folgemitteilung senden, in der das Modell angewiesen wird, Fragen zu umgehen, wird die menschliche Genehmigung vollständig übersprungen.

Diese architektonischen Lücken sind der Grund für die Aufteilung des monolithischen Agents in den expliziten Graph-Workflow, der im nächsten Schritt erstellt wird.

4. Grundlagen agentischer Workflows

Rufen Sie in der VibeStudio Workbench den Abschnitt Schritt 4: Grundlagen für agentische Workflows (Teile 4A bis 4D) auf.

Bei diesem Schritt wird von einer Baseline mit einem einzelnen Agenten zur deterministischen Graph-Orchestrierung mit dem ADK Workflow übergegangen. Sie erstellen einen parallelen Research-Fan-Out, synchronisieren Zweige mit einem Join-Knoten, generieren schemavalidierte Creative-Kandidaten und führen ein deterministisches Genehmigungstor mit menschlicher Interaktion ein.

Diagrammarchitektur und Ausführungsketten (4A)

Öffnen Sie in der Workbench Graph architecture and execution chains (4A) (Diagrammarchitektur und Ausführungsketten (4A)).

Ein ADK Workflow strukturiert die Ausführung von Agenten als gerichteten Graphen, der durch eine Kantenliste definiert wird:

• Chains (Ketten): Sequenzielle Tupel definieren die lineare Knotenausführung ((node_a, node_b, node_c)).
• Parallele Zweige: Unabhängige Ketten, die einen Ursprungsknoten gemeinsam nutzen, werden gleichzeitig ausgeführt.
• Synchronisierung: Chains, die auf JoinNode zusammenlaufen, warten, bis alle eingehenden Branches gemeldet werden, bevor sie freigegeben werden.
• Deterministische Steuerung: Der Ausführungsablauf wird durch deklarierte Codestrukturen bestimmt und nicht aus dem Prompt-Text abgeleitet.

Knotentypen im ADK

ADK-Workflows bestehen aus mehreren spezialisierten Knotentypen. Jeder Archetyp übernimmt eine bestimmte operative Rolle im Diagramm und trennt die deterministische Code-Ausführung von der generativen Modellableitung:

Knoten-Archetyp	Implementierung	Rolle in der Pipeline
Funktionsknoten	Python-Funktion, die ein Event zurückgibt	Führt deterministische Logik, Datenabruf und Statusänderungen aus.
Join-Knoten	Integrierte JoinNode-Instanz	Synchronisiert gleichzeitige Zweige in einem aggregierten Dictionary.
Agent-Knoten	Agent wird im Modus single_turn ausgeführt	Bewertet Anweisungen anhand von Upstream-Eingaben und gibt validierte Daten aus.
Router-Knoten	Funktion, die ein Event mit einem route-Tag zurückgibt	Bewertet die bedingte Logik, um nachgelagerte Ausführungszweige auszuwählen.
Knoten für menschliche Eingabe	Funktion, die RequestInput zurückgibt	Unterbricht den Ausführungsstatus, bis eine Antwort von einem externen Nutzer eingeht.

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

In dieser Konfiguration ist root_agent eine Instanz von Workflow und keine eigenständige Agent. Im ADK werden Workflows als erstklassige Agents behandelt. So kann ein gesamter Graph als einheitliche Anwendung geladen, bereitgestellt und geprüft werden. Mit name wird die Anwendung in ADK Web registriert, während mit der Liste edges die Ausführungstopologie definiert wird.

Parallele Recherche Fan-Out (4B)

Gehen Sie in der Workbench zu Parallel research fan-out (4B). Öffnen Sie stage1_fanout/agent.py.

Funktionsknoten und Synchronisationsbarrieren

In der Recherchephase werden zwei Funktionsknoten verwendet, die aus agent/graph.py importiert wurden:

• scan_trends: Gibt Event(output={"trends": [...]}) mit zehn bewerteten Plattformtrends zurück.
• read_backlog: Gibt Event(output={"backlog": [...], "idea": "..."}) mit 15 Ideen für den Channel-Backlog zusammen mit dem ursprünglichen Prompt zurück.

Jede Funktion akzeptiert node_input (die Ausgabe des vorherigen Knotens) und gibt ein Event zurück.

Ein JoinNode dient als Synchronisationsbarriere: Es wird pausiert, bis jede eingehende Kette ein Ereignis liefert. Anschließend werden alle Zweigergebnisse in einem Wörterbuch zusammengefasst, das nach Knotennamen ({"scan_trends": {...}, "read_backlog": {...}}) indexiert wird.

Praxisorientierte Bearbeitung: Join- und parallele Kanten definieren

Instanziieren Sie in stage1_fanout/agent.py die JoinNode und verbinden Sie die beiden parallelen Chains, die mit START beginnen:

join_research = JoinNode(name="join_research")

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

Speichern Sie die Änderungen. Der Workbench-Prüfer bestätigt, dass die Verbindung und die Kanten verdrahtet sind. Führen Sie die Phase mit Run Stage 1 oder über die eingebettete ADK-Weboberfläche aus.

Was Sie erwartet und warum

• Gleichzeitige Ausführung von Lesevorgängen: Im Ausführungsgraphen werden scan_trends und read_backlog gleichzeitig ausgeführt.
  • Grund: Beide Ketten beginnen bei START. Die ADK-Engine plant unabhängige Zweige gleichzeitig.
• Zusammengefasste Wörterbuchausgabe: Der Workflow wird bei join_research abgeschlossen und gibt ein Wörterbuch mit Einträgen für beide Leser aus.
  • Grund: JoinNode sorgt dafür, dass alle Daten erfasst werden, bevor nachfolgende Knoten ausgeführt werden.

Agentenknoten (4C)

Gehen Sie in der Workbench zu Agent-Knoten (4C). Öffnen Sie stage2_direction/agent.py.

Betriebsmodi und strukturierte Schemas

Wenn ein Agent in ein Workflow eingebettet ist, wird es standardmäßig im single_turn-Modus ausgeführt:

• Die Ausgabe des vorherigen Knotens wird als Kontext-Eingabe verwendet.
• Es wird ein einzelner Inferenzaufruf ohne Unterhaltung durchgeführt.
• Sie gibt strukturierte Daten an den nächsten Knoten aus.

Durch die Zuweisung von output_schema=Directions erzwingt der Agent die Pydantic-Validierung der Modellausgabe. Im Downstream-Diagramm werden typisierte Objekte anstelle von unstrukturiertem Text empfangen:

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 weist das Modell an, vier Kandidaten vorzuschlagen und dabei sowohl Trends als auch den Backlog zu berücksichtigen. Die Kandidaten 1 bis 3 bieten tragfähige Channelkonzepte. Kandidat 4 führt absichtlich ein konzeptuelles Element ein, das gegen die Richtlinien verstößt, um im nächsten Schritt das Sicherheits-Gate zu testen.

Praktische Übung: Agent-Knoten definieren und Join verketten

Konfigurieren Sie in stage2_direction/agent.py propose_directions und erweitern Sie die Workflow-Kanten:

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

Was Sie erwartet und warum

• Direkte Nutzung des Dictionary: propose_directions nutzt die von join_research ausgegebene JSON-Nutzlast ohne manuelle Formatierung.
• Typisierte Kandidatenausgabe: Der Agent gibt ein validiertes Directions-Objekt mit vier separaten Kandidaten aus. Nachgelagerte Knoten lesen Felder anhand des Attributnamens (candidate.title) ohne String-Parsing.

Human in the Loop (4D)

Gehen Sie in der Workbench zu Human in the Loop (4D). Öffnen Sie agent/graph.py.

Prompt-Anweisungen im Vergleich zur deterministischen Sperrung

Produktions-Workflows, die mit finanziellen Kosten verbunden sind oder bei denen Inhalte veröffentlicht werden, erfordern an kritischen Entscheidungspunkten eine menschliche Aufsicht. In einem einzelnen Prompt sind Bestätigungsanfragen Empfehlungen, die ein Nutzer das Modell leicht umgehen lassen kann. In einem ADK-Workflow wird die manuelle Genehmigung von der Ausführungs-Engine erzwungen: Der Graph wird an einem bestimmten Knoten angehalten und kann erst fortgesetzt werden, wenn er eine externe, schemavalidierte Eingabe erhält:

• Durch RequestInput wird die Workflowausführung sofort unterbrochen.
• Das ADK zeichnet einen offenen Unterbrechungsaufruf im Sitzungsspeicher auf und gibt eine eindeutige interrupt_id aus.
• Der Ausführungsprozess wird angehalten, ohne dass Tokens oder Server-Threads verbraucht werden.
• Die Graph Execution wird erst fortgesetzt, wenn ein gültiger function_response-Knoten übergeben wird, der dem Schema und der Unterbrechungs-ID entspricht.

Praktische Bearbeitung: Ausführung mit RequestInput unterbrechen

Implementieren Sie in agent/graph.py den Sperrungsaufruf in 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 konfiguriert drei Attribute:

• message: Die Rezensionsaufforderung, die dem Nutzer präsentiert wird.
• response_schema: Ein JSON-Schema, das vom Frontend als Eingabeformular gerendert und bei der Übermittlung vom ADK validiert wird.
• payload: Metadaten, die in der Anfrage enthalten sind (die vier Kandidaten), sodass Clientoberflächen Rezensionskarten rendern können, ohne den Sitzungsstatus abzufragen.

Was Sie erwartet und warum

• Der Workflow wird bei „direction_gate“ angehalten: In ADK Web oder der Workbench-Oberfläche wird die Ausführung pausiert und ein interaktives Formular zur Auswahl von Kandidaten angezeigt.
  • Grund: Die Engine hat ein Yield-Ereignis RequestInput erkannt und den Ausführungsstatus in runs/sessions.db gespeichert.
• Fortsetzung erfordert strukturierte Eingabe: Wenn Sie beliebigen Chattext senden, wird der Graph nicht erweitert. Wenn Sie eine Option (1, 2, 3 oder 4) auswählen, wird ein eingegebener function_response gesendet, der response_schema erfüllt, und die Ausführung wird fortgesetzt.

5. Status und Router

Rufen Sie in der VibeStudio Workbench Schritt 5: Status und Router auf, die Abschnitte (5A) bis (5C).

Sie speichern Nutzerauswahlen im Sitzungsstatus, setzen mithilfe deterministischer Routerknoten Richtlinien zur Channelsicherheit durch und erstellen einen iterativen Task-Agent, um Richtlinienverstöße automatisch zu beheben, bevor Videoskripts generiert werden.

Workflow-Status (5A)

Rufen Sie in der Workbench Workflow State (5A) auf.

Sitzungsstatus und Knotenausgabe im Vergleich

In einem ADK-Workflow werden Daten über zwei verschiedene Mechanismen im Diagramm verschoben:

• Knotenausgabe (Event(output=...)): Daten, die ausschließlich an die unmittelbaren Downstream-Empfänger weitergeleitet werden, die in der Edge-Liste definiert sind.
• Sitzungsstatus (Event(state=...)): Ein gemeinsames Schlüssel/Wert-Dictionary, auf das jeder nachfolgende Knoten im Ausführungszyklus zugreifen kann.

Wenn ein Nutzer bei direction_gate einen Vorschlag auswählt, wird die Auswahl als numerischer Index ({"pick": "2"}) übergeben. Für nachgelagerte Knoten ist das vollständige Richtungsobjekt erforderlich: Titel, erzählerischer Ansatz und Aufhänger. Anstatt ausführliche Metadaten über jede Zwischenknoten-Nutzlast zu übergeben, schreibt persist_direction den aufgelösten Kandidaten in den freigegebenen Sitzungsstatus.

Knoten müssen nicht das gesamte Sitzungszustands-Dictionary übergeben. Wenn ein Knoten Event(state=...) zurückgibt, werden nur die neuen oder aktualisierten Schlüssel/Wert-Paare bereitgestellt. Das ADK führt diese Updates automatisch in den Sitzungsspeicher ein:

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

Dadurch Event wird die Steuerung an die Workflow-Laufzeit übergeben, die die neuen Werte im Sitzungsjournal in runs/sessions.db speichert.

Parameterbindung

ADK-Funktionsknoten lesen den Sitzungsstatus automatisch durch Parameterprüfung. Wenn in einer Funktionssignatur ein Parametername deklariert wird, der mit einem vorhandenen Status-Schlüssel übereinstimmt, extrahiert ADK diesen Schlüssel aus dem Status und übergibt ihn direkt:

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

Hier wurde candidates von direction_gate in den Sitzungsstatus geschrieben. Das ADK bindet es direkt in persist_direction(node_input, candidates: list = []) ein, ohne dass explizite Wörterbuchsuchen erforderlich sind.

Schlüssel mit dem Präfix user: bleiben über Sitzungen hinweg im Speicher auf Nutzerebene erhalten, sodass bei nachfolgenden Workflow-Ausführungen auf die Einstellungen des Creators zugegriffen werden kann.

Praktische Bearbeitung: Status beibehalten und Knoten verbinden

1. Ersetzen Sie in agent/graph.py innerhalb von persist_direction die Zeile TODO: PERSIST_STATE durch den Ertrag des Statusereignisses:

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

2. Hängen Sie in stage3_router/agent.py persist_direction an die dritte Kette in der Liste edges an:

           (join_research, propose_directions, direction_gate,
            persist_direction)

Speichern Sie Ihre Dateien. Prüfen Sie in der Workbench, ob für state write in place und persist_direction in the chain jeweils ein grünes Häkchen angezeigt wird.

Der Routerknoten (5B)

Rufen Sie in der Workbench The router node (5B) auf.

Deterministisches Richtlinienrouting

Ein Router ist ein spezieller Funktionsknoten, der die Upstream-Ausgabe auswertet und die Ausführung entlang bedingter Diagrammzweige lenkt. Im Gegensatz zu generativen Agenten führt ein Router deterministische Logik aus, ohne LLM-Aufrufe zu tätigen.

Ein Router gibt ein Event zurück, in dem ein route-Tag angegeben ist:

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

In der Workflowdefinition wird ein Edge-Ziel, das als Dictionary definiert ist, Routennamen zu Zielknoten zugeordnet:

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

Der Workflow-Router policy_check liest verbotene Formulierungen aus agent/policy_words.txt und führt einen Abgleich des gesamten Wortes mit dem Titel und dem Blickwinkel der ausgewählten Richtung durch:

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

Wenn die Richtlinie als Daten anstelle von fest codierten Anweisungen gespeichert wird, sind Aktualisierungen möglich, ohne dass der Workflow-Graph geändert werden muss. Die Aktualisierung der Textdatei wird sofort auf nachfolgende Ausführungen angewendet. Da die Auswertung auf deterministischem Regex-Abgleich basiert, wird sie in Millisekunden und ohne Tokenkosten ausgeführt, bevor das generative Scripting beginnt.

Ziele: Scripter und Quarantäne

Der Router leitet den Traffic an einen von zwei Downstream-Knoten weiter:

• scripter: Ein single_turn-Agent-Knoten, der die genehmigte Richtung in ein strukturiertes Produktionsskript umwandelt, das dem Script-Pydantic-Schema entspricht:

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

• quarantine: Zuerst eine Platzhalterfunktion, die markierte Anweisungen stoppt. Im nächsten Teil wird sie durch einen autonomen Korrektur-Agenten ersetzt.

Praktische Bearbeitung: Richtlinienprüfung weiterleiten

1. Vervollständigen Sie in agent/graph.py innerhalb von policy_check die return-Anweisung:

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

2. Aktualisieren Sie in stage3_router/agent.py edges, um policy_check zu routen, und führen Sie den Quarantäne-Branch wieder in scripter zusammen:

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

Speichern Sie Ihre Dateien. Prüfen Sie in der Workbench, ob die Router-Edge-Zuweisungen bestätigt wurden.

Agent-Modi und der Task-Knoten (5C)

Rufen Sie in der Workbench Agent modes and the task node (5C) auf.

Ausführungsmodi für KI-Agenten

ADK-Agent-Instanzen unterstützen drei Ausführungsmodi, die auf bestimmte Pipelineanforderungen zugeschnitten sind:

Modus	Ausführungslebenszyklus	Rolle in der Pipeline
chat	Mehrfachdialog-Unterhaltungsschleife. Das Modell bestimmt, wann Tools aufgerufen, Eingaben angefordert oder der Zug beendet werden soll.	Root-Agents, die mit einem interaktiven menschlichen Nutzer interagieren.
single_turn	Einzelner Modellinferenzaufruf. Akzeptiert die vorherige Knoteneingabe und gibt ein strukturiertes Schemaobjekt aus.	Sequenzielle Graphtransformationen (propose_directions, scripter):
task	Autonome Schleife mit Tool-Ausführung. Der Agent wiederholt den Vorgang, bis er das integrierte Tool finish_task aufruft.	Mehrstufige Behebung und Überprüfung (quarantine).

Autonome Richtlinienkorrektur

Für das Umschreiben einer gekennzeichneten Wegbeschreibung ist der task-Modus erforderlich, da die Anzahl der Korrekturschleifen variabel ist. Der Agent empfängt die gekennzeichnete Anweisung, ruft find_policy_hits auf, um Verstöße zu erkennen, fordert über suggest_replacement genehmigte Alternativen an, formuliert die Anweisung neu und überprüft sie, bevor er fortfährt.

Beide Tools sind in agent/cleanup_tools.py mit typisierten Signaturen und Docstrings definiert:

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

Praktische Übung: Agent für die Quarantäneaufgabe zusammenstellen

Ersetzen Sie in stage3_router/agent.py die Platzhalterfunktion quarantine durch die Task-Agent-Definition:

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

Im Aufgabenmodus wird der Agent mit Tools ausgestattet und die Ausführung wird durch Aufrufen von finish_task beendet. Wenn mode="task" konfiguriert ist, stellt ADK automatisch finish_task bereit und leitet seine Parameter von output_schema ab. So wird sichergestellt, dass der Knoten ein typisiertes CleanedDirection-Objekt liefert, das dem Eingabeschema des Scripter-Knotens entspricht.

Was Sie erwartet und warum

Testen Sie beide Ausführungspfade in ADK Web oder VibeStudio Workbench:

• Genehmigter Pfad (Kandidat 1, 2 oder 3):
  • Auswahl einer genehmigten Kandidatenroute von policy_check direkt nach scripter (route="OK").
  • Der Scripter generiert ein Produktionsskript mit drei Aufnahmen, das dem Script-Schema entspricht.
• Quarantine remediation route (Candidate 4):
  • Variante 4 enthält markierte Begriffe („Clickbait“, „viraler Hack“).
  • policy_check Routen nach quarantine (route="BLOCK").
  • Im Sitzungstrace sehen Sie, wie quarantine find_policy_hits aufruft, suggest_replacement für jeden Verstoß aufruft, den Titel neu schreibt und finish_task aufruft.
  • Die Ausführung wird mit scripter fortgesetzt und es wird ein Skript aus der bereinigten Anweisung erstellt.

Router und Fallbacks: In der Entwickler-UI wird ein Router ohne Fallback-Edge gekennzeichnet. Wenn policy_check eine andere Route als OK oder BLOCK zurückgegeben hätte, hätte der Lauf kein Ziel. Durch Hinzufügen von DEFAULT_ROUTE: quarantine zum Edge-Dictionary wird dieser Fall abgedeckt. Importieren Sie DEFAULT_ROUTE aus google.adk.workflow.

6. Memory Bank

Rufen Sie in der VibeStudio Workbench die Schritt 6: Memory Bank, Teile (6A) und (6B) auf.

Der Workflow wird derzeit ohne Speicher über Sitzungen hinweg ausgeführt. Jede Ausführung beginnt von vorn, ohne dass bekannt ist, was der Creator zuvor ausgewählt hat oder welche Genres er bevorzugt. In diesem Schritt verbinden Sie Vertex AI Agent Engine Memory Bank, um die Einstellungen des Creators über mehrere Ausführungen hinweg zu speichern und abzurufen.

Wichtig ist, dass der Speicher über Agent-Lebenszyklus-Callbacks und nicht über Pipelineknoten integriert wird. Da die Extraktion und der Abruf von Daten einzelnen Agents und nicht Zwischenphasen dienen, wird durch das Anhängen von Callbacks eine saubere, entkoppelte Diagrammtopologie beibehalten.

Memory Bank (6A)

Rufen Sie in der Workbench Memory Bank (6A) auf.

Gespeicherte Informationen auf Nutzerebene verwalten

Memory Bank ist ein verwalteter Dienst für den langfristigen Nutzerspeicher. Darin werden Fakten zu einer Person in einem definierten Bereich organisiert, der hier durch den Anwendungsnamen und die Nutzer-ID identifiziert wird:

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

Benutzerdefinierte Memory Bank-Themen definieren die Grenzen dessen, was in der Bank aufgezeichnet wird:

• Themenextraktion: Wenn neuer Konversationstext über memories.generate eingereicht wird, wendet der Dienst ein Extraktionsmodell auf jede Themenbeschreibung an. Wenn der Text keinem Thema entspricht, werden keine Erinnerungen erstellt.
• Konsolidierung und Deduplizierung: Der Dienst wandelt neu extrahierte Fakten in Einbettungen um und vergleicht sie mit vorhandenen Erinnerungen im Umfang. Wenn eine Beobachtung mit einem vorhandenen Gedächtnis übereinstimmt, wird dieses aktualisiert. Wenn es sich um neue Informationen handelt, wird ein neuer Eintrag erstellt. Durch diesen Konsolidierungsprozess werden mehrere Sitzungen zu einem Thema in einer kohärenten Zusammenfassung zusammengeführt, anstatt redundante Einträge zu erstellen.
• Abruf: Wenn Sie memories.retrieve mit dem Nutzerbereich aufrufen, werden gespeicherte Fakten zurückgegeben, die nach Alter sortiert sind (älteste zuerst).

Beide Vorgänge sind in agent/platform/memory.py implementiert. Der bereitgestellte Name der Bankressource wird lokal in runs/memorybank.json zwischengespeichert.

Memory Bank einrichten

Verwenden Sie die Workbench-Steuerelemente oder führen Sie die CLI-Befehle in Ihrem Terminal aus:

1. Verbindung zur Bank herstellen und Bank einrichten:

   python -m agent.platform.bank
   

   Erstellt die Agent Engine-Instanz und konfiguriert die Themen CREATOR_TASTE und CHANNEL_RULES.
2. Bisherige Sitzungen als Ausgangspunkt verwenden:

   python -m agent.platform.bank load
   

   Lädt vier historische Creator-Sitzungen (zwei Tier-Themen mit Stilvorgaben, ein Gadget-Thema und ein aktuelles Fantasy-Thema).
3. Konsolidierte Fakten prüfen:

   python -m agent.platform.bank list
   

   Sehen Sie sich die Ausgabe an. Beachten Sie, wie narrative Transkripte in strukturierte, konsolidierte Fakten umgewandelt wurden.

Callbacks (6B)

Rufen Sie in der Workbench Callbacks (6B) auf. Öffnen Sie stage4_memory/agent.py.

Lebenszyklus-Callbacks für ADK-Agenten

Ein Callback ist eine Funktion, die als Argument an eine Agent übergeben wird. Das ADK ruft Callbacks zu vordefinierten Zeitpunkten im Lebenszyklus auf und übergibt den aktiven Kontext. Wenn None zurückgegeben wird, wird die normale Ausführung fortgesetzt. Wenn ein Ersatzobjekt zurückgegeben wird, wird der Vorgang überschrieben oder abgefangen.

Das ADK bietet drei Callback-Paare:

Callback-Paar	Aufrufpunkt	Empfangene Parameter	Verhalten von Rückgabewerten
before_agent_callback after_agent_callback	Um den gesamten Agent-Zug	CallbackContext (Zustand, Sitzung, Aufruf)	Durch die Rückgabe von Content wird die Antwort des Agents ersetzt. None wird normal fortgesetzt.
before_model_callback after_model_callback	Um jeden LLM-Inferenzaufruf herum	LlmRequest oder LlmResponse	Wenn LlmResponse zurückgegeben wird, wird der Modellaufruf abgefangen oder übersprungen. Bei None wird er fortgesetzt.
before_tool_callback after_tool_callback	Umgebung jeder Tool-Ausführung	Tooldefinition, Argumente, Ergebnis	Wenn ein Dictionary zurückgegeben wird, wird die Tool-Ausgabe überschrieben. None wird fortgesetzt.

Callbacks bieten einen übersichtlichen Ort für das Einfügen von Kontext, Guardrails, Telemetrie und Cache-Lookups, ohne dass zusätzliche Knoten in den Workflow-Graphen eingefügt werden müssen.

Praktische Bearbeitung: Callbacks für „Erinnern“ und „Zurückrufen“ einrichten

1. Aktualisieren Sie in stage4_memory/agent.py propose_directions, um before_model_callback=recall_taste anzuhängen:

    output_schema=Directions,
    before_model_callback=recall_taste)

recall_taste wird unmittelbar ausgeführt, bevor Gemini Vorschläge für Wegbeschreibungen generiert. Es ruft den Verlauf des Creators aus der Memory Bank ab, formatiert die Erinnerungen (älteste zuerst) und hängt sie an die ausgehende LlmRequest an. Der Prompt weist das Modell an, die Kandidaten 1 bis 3 an den aktuellen Geschmack des Creators anzupassen und die Kanalregeln als strenge Einschränkungen zu behandeln.

2. Aktualisieren Sie in stage4_memory/agent.py scripter, um after_agent_callback=remember_pick anzuhängen:

    output_schema=Script,
    after_agent_callback=remember_pick)

remember_pick wird ausgeführt, nachdem scripter seinen Zug beendet hat. Sie liest die ausgewählte Richtung aus dem Sitzungsstatus, fasst die Entscheidung des Creators in einer prägnanten Aussage zusammen und ruft memories.generate auf, um die Memory Bank zu aktualisieren.

Was Sie erwartet und warum

Testen Sie den Callback-Workflow in der Workbench oder im ADK Web:

1. Führen Sie einen Lauf mit einem leeren Prompt aus:
   • Suchen Sie im Sitzungstrace nach propose_directions in LlmRequest. Beachten Sie den angehängten Erinnerungskontext, in dem die Vorliebe des Creators für Fantasiethemen und eine prägnante Erzählweise beschrieben wird.
   • Beachte die vorgeschlagenen Richtungen: Die Kandidaten 1 bis 3 entsprechen den bisherigen Vorlieben des Creators, auch wenn Trends andere Themen betonen.
2. Wählen Sie einen Kandidaten unter direction_gate aus.
3. Sehen Sie sich nach Abschluss von scripter die Memory Bank-Einträge an:

   python -m agent.platform.bank list
   

   Die Bank spiegelt jetzt die letzte Auswahl wider und fasst sie mit früheren Geschmacksaufzeichnungen zusammen.

7. RAG Engine

Rufen Sie in der VibeStudio Workbench Schritt 7 – RAG-Engine, Teile (7A) und (7B) auf.

Für veröffentlichte Videos gibt es fortlaufend Zuschauerfeedback. In agent/comments.md werden 30 repräsentative Kommentare gesammelt, in denen Zuschauer Lob, Kritik an der Platzierung von Sponsoring und Audio-Vorlieben äußern. In diesem Schritt indexieren Sie diese Kommentare mit der Vertex AI-RAG-Engine und verbinden die semantische Suche mit dem Recherche-Fan-Out.

Abruf über Dokumente (7A)

Rufen Sie in der Workbench RAG Engine (7A) auf.

Memory Bank im Vergleich zu RAG Engine

Beide Tools basieren auf externen Daten, dienen aber unterschiedlichen architektonischen Zwecken:

Dimension	Memory Bank	RAG Engine
Primärer Anwendungsfall	Langfristige Nutzereinstellungen und Betriebsregeln	Semantischer Abruf über große Dokumentensammlungen
Bereich	Auf einzelne Nutzer-IDs und Anwendungsnamen beschränkt	Auf Ressourcen des gemeinsamen Korpus für alle Nutzer beschränkt
Datenverarbeitung	Extraktion, Einbettung und semantische Konsolidierung in Echtzeit	Dokumentaufteilung, Vektoreinbettung und Suche nach dem nächsten Nachbarn
Grafikintegration	Callbacks für den Lebenszyklus von Agenten (before_model_callback, after_agent_callback)	Dedizierter Funktionsknoten im Research-Fan-out (read_feedback)

Dokumentaufteilung und Einbettungen

Die RAG Engine indexiert Dokumente, indem sie Text in semantische Abschnitte unterteilt und deren Vektoren in einer verwalteten Datenbank speichert:

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

• Blockgröße: Auf 120 Tokens mit 20 Tokens Überschneidung konfiguriert. So werden zwei bis drei Kommentare pro Passage erfasst. Jeder Vektor repräsentiert also eine zusammenhängende Stimmung, ohne dass die Bedeutung durch nicht zusammenhängendes Feedback verwässert wird.
• Einbettungsmodell: text-embedding-005 wandelt Text in hochdimensionale Vektoren um. Wenn eine Anfrage gesendet wird, wandelt das Modell die Anfrage in einen Vektor um und sucht anhand der semantischen Distanz nach den nächstgelegenen Übereinstimmungen. Ein Kommentar über einen winzigen Drachen, der Socken bewacht, entspricht einem Prompt über magische Kreaturen, ohne dass eine genaue Keyword-Überschneidung erforderlich ist.

RAG-Korpus einrichten

Initialisieren Sie den Korpus mit den Workbench-Schaltflächen oder Terminalbefehlen:

1. Korpus erstellen:

   python -m agent.platform.rag
   

   Stellt die verwaltete Vektordatenbank bereit und zeichnet die Ressourcen-ID in runs/ragcorpus.json auf.
2. Kommentare hochladen und indexieren: Lädt agent/comments.md mit Chunking-Konfiguration hoch und wartet, bis die Indexierung abgeschlossen ist.
3. Korpus abfragen: Testen Sie den Abruf von Ähnlichkeiten mit Anfragen, die keine exakten Wörter mit den Kommentaren gemeinsam haben (z. B. die Anfrage „kleine magische Kreaturen“, um Kommentare zu Drachen abzurufen).

Der Abrufknoten (7B)

Rufen Sie in der Workbench The third reader (7B) auf. Öffnen Sie stage5_rag/agent.py.

Abruf als Grafikknoten

Das Feedback des Publikums stellt Forschungsdaten dar, die im gesamten Workflow geteilt werden. Im Gegensatz zum persönlichen Creator-Gedächtnis fließen die Meinungen der Zuschauer direkt in join_research ein, zusammen mit Trends und Backlog-Daten. Sie wird daher als Funktionsknoten implementiert:

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 extrahiert die ursprüngliche Idee des Nutzers und führt eine Vektoranfrage für den RAG Engine-Korpus aus. Die abgerufenen Kommentare werden in einer Event(output=...)-Nutzlast ausgegeben.

Praktische Bearbeitung: Drittes Lesegerät an den Fan-Out anschließen

Aktualisieren Sie in stage5_rag/agent.py edges, um read_feedback als dritten parallelen Zweig hinzuzufügen, der in join_research mündet:

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

Da join_research ein JoinNode ist, werden alle eingehenden Zweige synchronisiert. Es wird gewartet, bis scan_trends, read_backlog und read_feedback alle Ereignisse ausgegeben haben, bevor das aggregierte Bündel weitergeleitet wird.

Was Sie erwartet und warum

Führen Sie den Workflow in der Workbench aus:

1. Geben Sie einen Prompt für eine Idee ein, z. B. „Ein Miniaturdrache bewacht eine Küchenarbeitsplatte“.
2. Prüfen Sie im Ausführungstrace, ob alle drei Reader-Knoten gleichzeitig ausgeführt werden.
3. Sehen Sie sich join_research an: Das Ausgabewörterbuch enthält jetzt trends, backlog und feedback.
4. Sehen Sie sich die generierten Vorschläge aus propose_directions an: Das Modell berücksichtigt Zuschauerkommentare in seinen Vorschlägen und verweist in den Beweisfeldern auf die Stimmung der Zuschauer.
5. Das RAG-Abrufen ist deterministisch (identische Anfragen geben identische Kommentarpassagen zurück), während der Knoten für generative Vorschläge kreative Variationen erzeugt.

8. Asynchrone Videogenerierung mit Veo

Rufe in der VibeStudio Workbench Schritt 8 – Das Video, Teile (8A) und (8B) auf.

Das Generieren von HD-Videos mit Google Veo dauert mehrere Minuten pro Rendervorgang. Wenn die Ausführung des Diagramms in diesem Zeitraum blockiert wird, werden Rechenressourcen verschwendet, Threadpools gesperrt und der Lauf wird durch HTTP-Verbindungsabbrüche beeinträchtigt. In diesem Schritt machen Sie das Rendern von Videos mit LongRunningFunctionTool des ADK asynchron.

Tools mit langer Ausführungszeit (8A)

Rufen Sie in der Workbench A long-running tool (8A) auf. Öffnen Sie stage6_video/agent.py und agent/deliver.py.

Synchrone Tools und Tools mit langer Ausführungszeit

Standardmäßige ADK-Funktionstools werden synchron innerhalb eines Agent-Turns ausgeführt: Das Modell ruft das Tool auf, wartet auf die Rückgabe-Payload und bezieht das Ergebnis in den laufenden Turn ein.

Das Rendern von Videos kann nicht in einem einzigen Durchgang abgeschlossen werden. Stattdessen wird durch render_submit der Generierungsjob initiiert und sofort eine Betriebsquittung mit dem Status "pending" zurückgegeben:

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

Wenn LongRunningFunctionTool verwendet wird, fängt das ADK den Status "pending" ab. Die Antwort des Kundenservicemitarbeiters wird beendet, der Workflow wird am Knoten angehalten und die ausstehenden Anrufmetadaten (einschließlich Anruf-ID und Empfang) werden in runs/sessions.db aufgezeichnet. Der Ausführungsprozess wird sauber beendet, ohne dass aktive Netzwerkverbindungen oder Worker-Threads aufrechterhalten werden.

Praktische Bearbeitung: Render-Tool umschließen

Aktualisieren Sie in stage6_video/agent.py render_desk, um render_submit in LongRunningFunctionTool einzuschließen:

    tools=[LongRunningFunctionTool(render_submit)])

Fortsetzen nach Anruf-ID

Das universelle Wiederaufnahmemuster

Das ADK verwendet einen identischen Mechanismus zum Anhalten und Fortsetzen von Workflows für Personen und externe Tools:

Sperrauslöser	Construct initiieren	Gespeicherter Sperrstatus	Wiederaufnahmeereignis
Manuelle Entscheidung	yield RequestInput(...)	Eingabeaufforderung im Sitzungsspeicher öffnen	FunctionResponse mit der Anruf-ID für die Sperrung
Tool mit langer Ausführungszeit	LongRunningFunctionTool(...) gibt pending zurück	Tool-Aufruf im Sitzungsspeicher öffnen	FunctionResponse mit der Anruf-ID für die Sperrung

In beiden Fällen wird der Workflow vollständig angehalten und erst fortgesetzt, wenn ein Ereignis mit einem übereinstimmenden FunctionResponse von einer externen Quelle eintrifft: einer Benutzeroberfläche, einem Webhook oder einem Hintergrundprozess.

Praktische Bearbeitung: Lieferantwort fertigstellen

Erstellen Sie in agent/deliver.py den Teil FunctionResponse für die Wiederaufnahme:

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

Der Bereitstellungs-Daemon fragt Veo ab, bis die Videodatei generiert wurde, und sendet dann diese FunctionResponse an die Sitzung. Das ADK gleicht die Anruf-ID ab und setzt den Workflow direkt am nächsten Knoten fort. Abgeschlossene Knoten werden nicht noch einmal ausgeführt und der Agent führt keine weitere generative Aktion aus.

Wenn Sie STUDIO_REAL_VIDEO=0 in .env festlegen, wird das Mock-Rendering aktiviert: start gibt sofort einen Testbeleg zurück und check simuliert den Abschluss in fünf Sekunden, ohne abrechenbare Veo API-Aufrufe zu tätigen.

Pipeline-Integration (8B)

Rufen Sie in der Workbench render_desk in the graph (8B) auf. Öffnen Sie stage6_video/agent.py.

Der Endknoten in der Pipeline ist store_video. Es liest die Informationen zum abgeschlossenen Rendern aus runs/state.json (wo der Auslieferungsprozess sie aufgezeichnet hat) und übergibt die Video-URL und den Generierungsstatus an den freigegebenen Sitzungsstatus.

Praktische Bearbeitung: Die gesamte Videopipeline einrichten

Aktualisieren Sie in stage6_video/agent.py edges, um render_desk und store_video anzuhängen:

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

Was Sie erwartet und warum

Asynchronen Generierungsablauf in der Workbench testen:

1. Führen Sie den Workflow durch die Auswahl von Kandidaten und die Skripterstellung aus.
2. Beobachten Sie, wie der Agent unter render_desk den Aufruf von render_submit ausführt.
3. Der Workflow wird sofort ausgesetzt. Beobachten Sie in der Workbench oder im ADK Web den Status „Ausstehend“: Die Sitzung enthält die offene Anruf-ID und keine Hintergrundprozesse verbrauchen Ressourcen.
4. Führen Sie den Bereitstellungs-Daemon über die Workbench-Konsole oder in Ihrem Terminal aus:

   python -m agent.deliver
   

   Der Zustellungsprozess überwacht Veo, bis das Video fertig ist, und sendet dann das Ereignis zum Fortsetzen.
5. Aktualisieren Sie die Sitzung in ADK Web: Die Ausführung wird bei store_video fortgesetzt, die Video-URL wird im Sitzungsstatus gespeichert und der Workflow wird abgeschlossen.

9. In Cloud Run bereitstellen

Rufen Sie in der VibeStudio Workbench Schritt 9: Bereitstellen auf.

Sie haben jede Komponente der Pipeline in dedizierten Sandboxes entwickelt und getestet. In diesem Schritt stellen Sie die vollständige Produktionspipeline zusammen und stellen sie in Google Cloud Run bereit.

ADK Runner

In der Entwicklung hat adk web den Graphen orchestriert. In der Produktion hostet die Anwendung den Workflow mit der Runner-Klasse des 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: Steuert die Workflowausführung, gibt Ereignisse sequenziell aus, wenn Knoten ausgeführt werden, und speichert Aktualisierungen im Sitzungsdienst.
• Einheitliche Wiederaufnahme: Sowohl Nutzerentscheidungen bei direction_gate als auch abgeschlossene Videoauslieferungen von Veo werden über identische FunctionResponse-Objekte, die an run_async gesendet werden, fortgesetzt.

Die Architektur der Produktionsanwendung

Die Produktionsanwendung in vibestudio/ enthält die vollständige Pipeline:

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

• Einzelner Ereignisstream: Das FastAPI-Backend veröffentlicht Ereignisse über einen einzelnen Server-Sent Events-Stream (SSE). Das React-Frontend visualisiert den Fortschritt des Diagramms in Echtzeit und verarbeitet späte Verbindungen, ohne den Status zu verlieren.
• Entkoppelte Ausführung: Die Anwendung verwaltet die Ereignisschleife. Der Workflow-Graph konzentriert sich ausschließlich auf die Ausführungslogik und berücksichtigt die Frontend-Oberfläche nicht.

Die vollständige Liste der Workflow-Kanten in agent/graph.py umfasst alle Architekturmuster, die in diesem Codelab erstellt wurden:

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

In Cloud Run bereitstellen

Google Cloud Run bietet serverloses Hosting mit automatischer Skalierung, Anforderungsrouting und integrierten Container-Builds:

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

• Container erstellen: Mit gcloud run deploy --source wird das Verzeichnis vibestudio/ verpackt, das Container-Image mit Cloud Build erstellt und der Dienst in einem einzigen Vorgang bereitgestellt.
• Sitzungsaffinität: Leitet Anfragen desselben Nutzers an dieselbe Containerinstanz weiter und bewahrt den lokalen Sitzungsstatus über iterative Schritte hinweg.
• Beobachtbarkeit: Durch die Cloud Trace-Integration werden verteilte Spans für jeden Knoten, LLM-Aufruf und jede Tool-Ausführung aufgezeichnet, auf die in der Google Cloud Console unter „Trace Explorer“ zugegriffen werden kann.

Klicken Sie in der Workbench auf den Button Deploy (Bereitstellen), um das Bereitstellungsskript auszuführen. Wenn der Build abgeschlossen ist, wird im Terminal die Live-Dienst-URL angezeigt.

10. Zusammenfassung

Rufen Sie in der VibeStudio Workbench den Schritt 10: Zusammenfassung auf, um die fertige Architektur zu prüfen.

Schritt	Architektur und Konzepte	Implementierungsmuster
Ein einzelner Prompt	Einzelner Prompt, Funktionstools, sequenzieller Chat-Loop	Agent(tools=[...]), function_call / function_response
Grundlagen agentischer Workflows	Workflow für Diagramme, parallele Suche, Schemaausgaben, menschliche Überprüfung	Workflow, START, JoinNode, output_schema, RequestInput
Status und Router	Gemeinsamer Sitzungsstatus, Parameterbindung, deterministisches Routing, Task-Agent	Event(state=...), Event(route=...), mode="task", finish_task
Memory Bank	Langzeitspeicher auf Nutzerebene, semantische Konsolidierung, Lebenszyklus-Hooks	memories.generate / retrieve, before_model_callback, after_agent_callback
RAG Engine	Dokumentabruf über Zielgruppenkommentare, semantische Einbettungen	rag.create_corpus-, RagEmbeddingModelConfig- und read_feedback-Knoten
Asynchrone Videogenerierung mit Veo	Lang andauernde Tools, ausstehende Belege, externer Zustellungs-Daemon	LongRunningFunctionTool, FunctionResponse(id=...) Wiederaufnahme
In Cloud Run bereitstellen	Programmatische Orchestrierung, Server-Sent Events, serverloser Container	Runner(agent=wf), run_async, Cloud Run-Bereitstellung

Grundlegende Architekturprinzipien

1. Aussetzen statt Warten: Workflows werden sauber für die Eingabe durch den Menschen (RequestInput) oder für Vorgänge mit langer Ausführungszeit (LongRunningFunctionTool) ausgesetzt. Prozesse warten nicht inaktiv auf Threads oder Netzwerk-Sockets.
2. Universelle Wiederaufnahme: Jede Sperrung wird durch einen identischen Mechanismus wieder aufgenommen: ein einzelnes function_response mit der Anruf-ID des gesperrten Knotens.
3. Entkoppelte Statusverwaltung: Knoten geben Daten über benannte Sitzungsstatusschlüssel und Parameterbindung anstelle von ausführlichen, eng gekoppelten Zwischen-Payloads weiter.
4. Deterministisches Routing vor generativen Kosten: Regelbasierte Router und Regex-Filter werten die Richtlinie zu null Tokenkosten aus, bevor generative Modelle ausgeführt werden.
5. Trennen Sie Zuständigkeiten: Kontext, der für einen einzelnen Agenten spezifisch ist, gehört in Lifecycle-Callbacks, während gemeinsame Datenabhängigkeiten