Przepływ pracy agenta z pakietem ADK

1. Wprowadzenie

Z tego ćwiczenia dowiesz się, jak tworzyć systemy oparte na agentach nowej generacji przy użyciu przepływów pracy i grafów w pakiecie Agent Development Kit (ADK). Wdrożysz typowe wzorce architektury, skoordynujesz interakcje z udziałem człowieka (HITL) i obsłużysz długotrwałe wykonywanie asynchroniczne. Zintegrujesz też firmowe bazy wiedzy i pamięć trwałą, aby dostosowywać i rozwijać zachowania agenta. Na koniec połącz te funkcje, aby utworzyć automatyczny potok generowania filmów.

Scenariusz

Prowadzisz kanał cyfrowy w VibeTube, który ma aktywnych odbiorców i stale rosnącą liczbę pomysłów na kreacje. Tworzenie każdego filmu wymaga ciągłego działania na wielu etapach: badania popularnych formatów, analizowania opinii widzów, opracowywania skryptów, sprawdzania zgodności z zasadami i generowania klipów wideo. Modele generatywne mogą tworzyć poszczególne komponenty, ale dostarczanie spójnych wersji wymaga skoordynowanej architektury agentów.

Aby zautomatyzować ten cykl życia, utworzysz VibeStudio. Ten proces oparty na agentach wykonuje rutynowe badania równolegle, przedstawia wyselekcjonowane opcje do zatwierdzenia przez człowieka, stosuje automatyczne bramki zasad przed wygenerowaniem filmu i zachowuje kontekst w kolejnych cyklach produkcyjnych.

Czego się dowiesz

• Podstawy inżynierii grafów: wieloetapowe architektury agentów wymagają jawnego przepływu sterowania i ustrukturyzowanych ścieżek wykonywania. Pakiet ADK Workflow tworzysz za pomocą krotek krawędziowych, punktu wejścia START, JoinNode do równoległego zwielokrotnienia wyjściowego i deterministycznych węzłów routera, aby sterować wykonaniem na podstawie stanu.
• Tryby agenta i wywołania zwrotne cyklu życia: specjalistyczne zadania wymagają odmiennych zachowań operacyjnych i określonych zabezpieczeń. Instancje ADK Agent konfigurujesz za pomocą trybów chat, single_turn i task z włączonymi narzędziami jako węzłów przepływu pracy, stosując przechwytujące za pomocą before_model_callback i after_agent_callback.
• Orkiestracja z udziałem człowieka: potoki produkcyjne są wstrzymywane w krytycznych punktach kontrolnych, aby umożliwić weryfikację przez człowieka. Implementujesz RequestInput, aby zawieszać wykonywanie przepływu pracy, wymuszać schematy strukturalnych odpowiedzi i wznawiać wykonywanie bez utrzymywania aktywnych procesów środowiska wykonawczego w stanie bezczynności.
• Hierarchiczna pamięć agenta: systemy produkcyjne oddzielają ulotny stan wykonania od trwałego kontekstu. Stan sesji krótkoterminowej zarządzasz za pomocą Event(state=...) i powiązania parametrów, a GEAP Bank zapamiętanych informacji służy do wyodrębniania, konsolidowania i utrwalania preferencji twórcy w różnych uruchomieniach.
• Powiązanie ze źródłem informacji przy użyciu baz wiedzy przedsiębiorstwa: autonomiczne agenty wymagają dynamicznego kontekstu domeny i opinii odbiorców. Łączysz korpus GEAP RAG Engine jako dedykowany węzeł pobierania w równoległym zwielokrotnieniu wyjściowym, aby semantycznie powiązać dane wyjściowe agenta ze źródłem informacji.
• Długotrwałe procesy i wdrażanie: renderowanie filmów multimodalnych odbywa się asynchronicznie i trwa dłużej. Implementujesz LongRunningFunctionTool z oczekującymi potwierdzeniami połączeń, aby zawieszać i wznawiać przepływ pracy według identyfikatora połączenia, a gotową ścieżkę wdrażasz za pomocą pakietu ADK Runner w Cloud Run.

Struktura tych ćwiczeń z programowania

To ćwiczenie stanowi punkt odniesienia w zakresie koncepcji i architektury. W każdej sekcji wyjaśniamy konstrukcje ADK zaimplementowane w odpowiednim kroku platformy, podajemy kod referencyjny i ustalamy podstawowe zasady projektowania. Przed wykonaniem odpowiedniego ćwiczenia w platformie sprawdź każdą sekcję.

Praktyczne ćwiczenia odbywają się w VibeStudio Workbench, czyli towarzyszącym interfejsie internetowym z interaktywnym edytorem kodu, weryfikatorami czasu działania i wbudowanym inspektorem pakietu ADK. Numeracja kroków w narzędziu jest zgodna z numeracją w tym ćwiczeniu, dzięki czemu możesz śledzić swoje postępy. Zmiany w grafie podstawowym są zachowywane na kolejnych etapach, a platforma automatycznie weryfikuje wymagania wstępne w miarę postępów.

Po wykonaniu ćwiczeń na platformie roboczej utworzysz kompleksową potokową platformę agentową i wdrożysz działającą aplikację VibeStudio w Cloud Run, aby generować treści wideo.

Środowisko składa się z 3 głównych komponentów: VibeStudio Workbench (lokalny interfejs internetowy do edytowania kodu i weryfikacji w czasie działania), backendu (pakiet ADK Workflow i piaskownice etapowe w agent/) oraz Google Cloud (modele Gemini, GEAP Bank zapamiętanych informacji, RAG Engine i generowanie filmów Veo).

2. Konfiguracja

Odbierz środki na warsztaty

Jeśli uczestniczysz w module prowadzonym przez instruktora, przydzieli on środki na Twój projekt Google Cloud. Postępuj zgodnie z instrukcjami prowadzącego, aby wykorzystać środki, i przed kontynuowaniem sprawdź, czy na Twoim koncie jest aktywna płatność.

Otwórz Cloud Shell

Cloud Shell to środowisko programistyczne w przeglądarce z preinstalowanymi narzędziami gcloud, Python i git.

Aby uruchomić Cloud Shell:

1. Otwórz konsolę Google Cloud.
2. W nagłówku u góry kliknij Aktywuj Cloud Shell (ikonę okna terminala).

Sesja terminala otworzy się u dołu okna przeglądarki.

Klonowanie i inicjowanie repozytorium

Aby skopiować projekt, uruchom w terminalu Cloud Shell te polecenia:

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

Monity konfiguracji

Podczas konfiguracji poprosimy Cię o podanie tych informacji:

• Identyfikator projektu Google Cloud: gdy pojawi się odpowiedni komunikat w setup_project.sh, naciśnij Enter, aby automatycznie utworzyć nowy projekt. Jeśli wolisz użyć istniejącego projektu (np. wstępnie przypisanego), wpisz identyfikator projektu i sprawdź, czy jest poprawny, a rozliczenia są aktywne.
• Kod wydarzenia: wpisz kod sali podany przez nauczyciela. Jeśli nie otrzymasz takiej wiadomości, skontaktuj się z asystentem nauczyciela lub sąsiadem. Jeśli wykonujesz to ćwiczenie w domu, naciśnij Enter, aby zaakceptować domyślne pomieszczenie sandbox.
• Wyświetlana nazwa kanału: wpisz swoje imię i nazwisko lub preferowany uchwyt kanału, gdy pojawi się odpowiedni komunikat w setup_codelab.sh, albo naciśnij Enter, aby zaakceptować domyślną nazwę wygenerowaną na podstawie Twojego konta Google.

Uruchom kolejno 2 skrypty konfiguracji:

./setup_project.sh
./setup_codelab.sh

• setup_project.sh: tworzy lub ponownie wykorzystuje projekt Google Cloud z aktywnym rozliczeniem, zapisuje identyfikator projektu w zmiennej ~/project_id.txt i konfiguruje aktywny kontekst gcloud.
• setup_codelab.sh: instaluje uv i zależności Pythona w .venv, włącza wymagane interfejsy Google Cloud API, konfiguruje ustawienia kanału w .env, weryfikuje dostęp do modelu za pomocą Gemini, udostępnia zasoby Banku zapamiętanych informacji i RAG, tworzy interfejs platformy roboczej i uruchamia VibeStudio Workbench.

Skrypt przeprowadza kontrolę wstępną i uruchamia VibeStudio Workbench w tle. W ostatnich wierszach wyświetla się link do otwarcia.

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

Kliknij ten link. Ten sam adres jest dostępny w sekcji Podgląd w przeglądarce → Zmień port → 4600.

Aby w dowolnym momencie ponownie sprawdzić środowisko, uruchom polecenie python scripts/preflight.py. Aby ponownie uruchomić platformę, wpisz scripts/restart.sh. Aby ponownie skonfigurować urządzenie, uruchom ./setup_codelab.sh. Zachowa to Twoją konfigurację i postępy.

Po otwarciu przeczytaj krok 1. Historia, aby poznać scenariusz, oraz krok 2. Co utworzysz, aby dowiedzieć się, jak będzie wyglądać gotowy wykres. Żaden z nich nie zawiera ćwiczenia. Następnie wróć tutaj, aby przejść do kroku 3.

Każda praktyczna część VibeStudio Workbench kończy się panelem weryfikacyjnym, który odczytuje rzeczywiste artefakty: plik na dysku i sesje zapisane przez uruchomienia.

Układ repozytorium

Repozytorium jest podzielone na podstawową logikę przepływu pracy, piaskownice krok po kroku, środowisko robocze i aplikację produkcyjną:

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/: zawiera podstawowy wykres przepływu pracy. W tym katalogu będziesz edytować pliki, aby wdrożyć równoległe węzły zwielokrotnienia wyjściowego, deterministyczne routing zasad, wywołania zwrotne pamięci i narzędzia do generowania filmów.
• agent/platform/: współpracuje z usługami Google Cloud, w tym z modelami Gemini, GEAP Bank zapamiętanych informacji, GEAP RAG Engine i Veo do syntezy wideo.
• stage0_prompt/–stage6_video/: autonomiczne środowiska piaskownicy. Każdy folder eksportuje samodzielny plik root_agent, dzięki czemu możesz uruchamiać i sprawdzać każdy krok osobno za pomocą wbudowanego interfejsu programistycznego ADK.
• server/ i web/: aplikacja VibeStudio Workbench działająca lokalnie na porcie 4600. Zawiera dokumentację kroków, edytor kodu na stronie, weryfikatory dowodów w czasie działania i wizualizację grafu.
• vibestudio/: kompletna aplikacja produkcyjna spakowana i wdrożona w Cloud Run w ostatnim kroku. Zawiera własną, samodzielną kopię ukończonego wykresu przepływu pracy.

3. Agent monolityczny

Przed utworzeniem wykresu przepływu pracy z wieloma węzłami ustalasz podstawową architekturę z jednym agentem w stage0_prompt/agent.py. Ten agent korzysta z monolitycznego promptu systemowego opisującego potok produkcyjny w formie prozy, który jest obsługiwany przez 2 narzędzia funkcji Pythona.

Ocena tej wartości bazowej pokazuje granice operacyjne koordynacji opartej na promptach i wyjaśnia, dlaczego systemy produkcyjne wymagają orkiestracji grafów.

Architektura agenta ADK (3A)

W VibeStudio Workbench przejdź do Kroku 3. Agent monolityczny i otwórz Architekturę agenta ADK (3A). Ten widok przedstawia podstawowe warstwy architektury agenta 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
)

Interaktywny diagram dzieli komponenty agenta na 5 obszarów operacyjnych:

• Warstwa rozumowania (model): podstawowy model językowy (np. Gemini 3 Flash) wykonujący zadania kognitywne, rozumowanie na podstawie promptów i wybór narzędzi. Wszystkie pozostałe elementy architektury dostarczają informacji o tym modelu lub go ograniczają.
• Warstwa kontekstu (instrukcje i umiejętności): wytyczne kształtujące rozumowanie modelu. instruction określa stały prompt systemowy, personę i reguły operacyjne. skills udostępniać wersjonowane, proceduralne wskazówki (SKILL.md) dotyczące powtarzalnych przepływów pracy;
• Warstwa współpracy i działań (narzędzia, subagenci, przepływ pracy, schemat wyjściowy): interfejsy umożliwiające agentowi działanie w systemach zewnętrznych i emitowanie danych o określonym typie. tools udostępniać wywoływalne funkcje Pythona lub punkty końcowe protokołu Model Context Protocol (MCP); subagents wykonywać podrzędne delegowane zadania; workflow koordynuje wykresy wielu agentów. output_schema stosuje modele Pydantic, aby zagwarantować, że odbiorcy otrzymają zweryfikowany plik JSON zamiast nieustrukturyzowanego tekstu.
• Warstwa przechwytująca (wywołania zwrotne cyklu życia): deterministyczne zabezpieczenia wykonujące niestandardowy kod przed i po wykonaniu przez agenta (before_agent/after_agent), poszczególne tury modelu (before_model/after_model) i wywołania narzędzi (before_tool/after_tool). Przechwytujące egzekwują reguły zasad bez polegania na zgodności modelu.
• Stan zewnętrzny (sesja i pamięć): trwałość stanu oddzielona od logiki agenta. Session zachowuje tymczasową pamięć roboczą i ślad zdarzeń dla bieżącego wątku wykonania. Memory utrzymuje trwałe fakty i ustawienia dotyczące różnych sesji, korzystając z usług zarządzanych, takich jak GEAP Bank zapamiętanych informacji.

Agent monolityczny w tym kroku implementuje tylko 3 z tych elementów: model, instruction i tools. W kolejnych krokach przedstawimy przepływy pracy z wykorzystaniem wykresów, schematy strukturalne, przechwytywanie i usługi pamięci trwałej.

Specyfikacja agenta monolitycznego (3B)

W platformie Workbench przejdź do sekcji Monolithic agent specification (3B) (Specyfikacja agenta monolitycznego (3B)). Otwórz stage0_prompt/agent.py, aby sprawdzić definicję agenta podstawowego:

• Instrukcja w jednym prompcie: prompt systemowy łączy 5 różnych zadań produkcyjnych w ciągły tekst: odkrywanie trendów na platformie, przeglądanie pomysłów z zaległości, proponowanie koncepcji kreatywnych, egzekwowanie zasad dotyczących zabronionych tematów i tworzenie list ujęć.
• Źródła danych: agent odwołuje się do 2 źródeł zdefiniowanych obok wykresu:
  • agent/trends.py: próbkuje 10 aktywnych trendów formatu i stylu z puli 250 trendów z dynamicznymi wynikami popularności.
  • agent/backlog.txt: odczytuje surowe notatki koncepcyjne twórcy wiersz po wierszu.

Narzędzia w agencie (3C)

W platformie przejdź do sekcji Narzędzia w agencie (3C).

Czym jest narzędzie dla agenta?

Model językowy jest z natury silnikiem wnioskowania w zamkniętym świecie: działa wyłącznie na podstawie wstępnie wytrenowanych wag i tokenów znajdujących się w jego bezpośrednim oknie kontekstu. Nie może natywnie wysyłać zapytań do bazy danych, uzyskiwać dostępu do interfejsów API w czasie rzeczywistym ani wykonywać kodu.

Narzędzie przekracza tę granicę. Zapewnia to modelowi zewnętrzną sprawczość, umożliwiając mu pobieranie prawdziwych informacji i wykonywanie deterministycznych działań w systemach zewnętrznych.

Wywoływanie narzędzi odbywa się zgodnie z wyraźnym 5-etapowym protokołem między modelem a środowiskiem wykonawczym ADK:

1. Deklaracja schematu: deweloper udostępnia agentowi funkcje Pythona. ADK sprawdza nazwę, adnotacje typu i ciągi dokumentacyjne każdej funkcji, aby wygenerować deklarację schematu JSON zgodną z OpenAPI, która opisuje jej parametry i przeznaczenie.
2. Uzasadnienie modelu: podczas wnioskowania model ocenia, czy prompt użytkownika wymaga danych zewnętrznych. W razie potrzeby model emituje zdarzenie strukturalne function_call zawierające nazwę funkcji docelowej i słownik argumentów zgodny ze schematem.
3. Wykonanie w czasie działania: model nie wykonuje kodu. Środowisko wykonawcze ADK przechwytuje function_call, wykonuje rzeczywistą lokalną funkcję Pythona przy użyciu podanych argumentów i przechwytuje wartość zwrotną.
4. Ponowne wstrzykiwanie kontekstu: środowisko wykonawcze ADK pakuje wartość zwrotną funkcji w zdarzenie function_response i dołącza je do historii aktywnej sesji.
5. Ostateczna synteza: model przetwarza dane wyjściowe narzędzia, które znajdują się teraz w oknie kontekstu, i kończy odpowiedź.

W stage0_prompt/agent.py oba narzędzia badawcze są zdefiniowane jako standardowe funkcje Pythona:

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

Praktyczna edycja i wykonywanie

W edytorze kodu platformy dodaj 2 odwołania do funkcji na listę tools agenta:

    tools=[check_trends, read_backlog],

Zapisz zmiany. Plik zostanie zaktualizowany na dysku, a w wierszu weryfikacji pojawi się potwierdzenie, że oba narzędzia są połączone.

Kliknij Open adk web (Otwórz ADK w internecie), aby uruchomić osadzony interfejs programowania ADK. Wyślij prompta z sugerowanym pomysłem:

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

Czego możesz się spodziewać i dlaczego

Gdy wyślesz ten prompt, w śladzie sesji zobaczysz tę sekwencję wykonania:

• Przed odpowiedzią pojawiają się 2 zdarzenia wykonania narzędzia: widzisz zdarzenia function_call i function_response dla check_trends i read_backlog.
  • Uzasadnienie: Gemini ocenił dyrektywę systemową („sprawdź, co jest popularne, przejrzyj zaległe pomysły”), stwierdził, że w jego wagach brakuje trendów na platformie i notatek o kanale, i wywołał obie funkcje, aby ugruntować kontekst.
• Agent proponuje kierunek i zatrzymuje się, aby uzyskać potwierdzenie: odpowiedź sugeruje kierunek filmu, łącząc trendy i zaległości, i prosi o potwierdzenie.
  • Uzasadnienie: instrukcja nakazywała modelowi uzgodnienie kierunku z twórcą przed wygenerowaniem skryptu.
• Pomijanie potwierdzenia w kolejnej turze: wyślij drugą wiadomość: skip the questions, just describe the video. Agent natychmiast pomija potwierdzenie i przygotowuje tytuł oraz ujęcia.
  • Uzasadnienie: instrukcje w prompcie są wytycznymi, a nie deterministycznymi barierami. W przypadku agenta monolitycznego instrukcje użytkownika mogą zastępować obowiązujące reguły promptu systemowego, ponieważ przepływ wykonania nie jest kontrolowany przez żaden zewnętrzny proces.

Ograniczenia architektoniczne monolitycznego prompta

Pojedynczy prompt może generować akceptowalne wyniki w przypadku odosobnionych wersji demonstracyjnych, ale testowanie warunków brzegowych w weryfikatorze platformy roboczej ujawnia istotne ograniczenia dla przedsiębiorstw:

• Niestrukturalne zbieranie informacji: kolejność wykonywania narzędzi jest niedeterministyczna. Model podsumowuje pobrane dane w formie swobodnego tekstu, co uniemożliwia systemom niższego rzędu wyodrębnienie źródła konkretnych twierdzeń.
• Egzekwowanie zasad bez weryfikacji: model ocenia własną zgodność z zasadami bezpieczeństwa. Jeśli model uzna, że temat jest bezpieczny, żadna zewnętrzna logika deterministyczna nie zweryfikuje tego wyniku.
• Nieegzekwowane wstrzymania z udziałem człowieka: instrukcje w prompcie, które wymagają potwierdzenia od twórcy, mają charakter doradczy. Wysłanie wiadomości z prośbą o pominięcie pytań powoduje, że model całkowicie pomija zatwierdzenie przez człowieka.

Te luki w architekturze uzasadniają rozbicie monolitycznego agenta na jawny przepływ pracy z wykresem, który utworzymy w następnym kroku.

4. Podstawy przepływu pracy z agentem

W VibeStudio Workbench przejdź do Step 4 · Agentic workflow fundamentals, części 4A–4D.

W tym kroku przechodzimy od pojedynczego agenta do deterministycznej orkiestracji wykresów za pomocą ADK Workflow. Utworzysz równoległe zwielokrotnienie wyjściowe badań, zsynchronizujesz gałęzie za pomocą węzła łączenia, wygenerujesz kandydatów na kreacje zweryfikowane pod kątem schematu i wprowadzisz deterministyczną bramkę zatwierdzania z udziałem człowieka.

Architektura wykresu i łańcuchy wykonania (4A)

W obszarze roboczym otwórz Architekturę grafu i łańcuchy wykonywania (4A).

Pakiet ADK Workflow strukturyzuje wykonywanie agenta jako graf skierowany zdefiniowany przez listę krawędzi:

• Łańcuchy: sekwencyjne krotki definiują liniowe wykonywanie węzłów ((node_a, node_b, node_c)).
• Równoległe gałęzie: niezależne łańcuchy, które mają wspólny węzeł początkowy, są wykonywane równocześnie.
• Synchronizacja: łańcuchy zbiegające się w JoinNode oczekują na zgłoszenie wszystkich przychodzących gałęzi przed zwolnieniem.
• Kontrola deterministyczna: przepływ wykonywania podlega zadeklarowanym strukturom kodu, a nie jest wywnioskowany z tekstu prompta.

Archetypy węzłów w pakiecie ADK

Przepływy pracy ADK składają się z kilku wyspecjalizowanych typów węzłów. Każdy archetyp pełni w grafie określoną rolę operacyjną, oddzielając deterministyczne wykonywanie kodu od wnioskowania modelu generatywnego:

Archetyp węzła	Implementacja	Rola w potoku
Węzeł funkcji	Funkcja Pythona zwracająca Event	Wykonuje deterministyczną logikę, pobieranie danych i zmiany stanu.
Węzeł łączenia	Wbudowana instancja JoinNode	Synchronizuje równoległe gałęzie w zagregowanym słowniku.
Węzeł agenta	Agent – działa w trybie single_turn	Sprawdza instrukcje na podstawie danych wejściowych z wyższego poziomu i generuje zweryfikowane dane.
Węzeł routera	Funkcja zwracająca Event z tagiem route	Ocenia logikę warunkową, aby wybrać gałęzie wykonania podrzędnego.
Węzeł danych wejściowych od użytkownika	Funkcja zwracająca RequestInput	Zawiesza stan wykonania do czasu otrzymania odpowiedzi od użytkownika zewnętrznego.

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

W tej konfiguracji root_agent jest instancją Workflow, a nie samodzielnym Agent. ADK traktuje przepływy pracy jako agenty pierwszej klasy, co umożliwia wczytywanie, obsługiwanie i sprawdzanie całego wykresu jako ujednoliconej aplikacji. name rejestruje aplikację w ADK Web, a lista edges określa jej topologię wykonania.

Równoległe zwielokrotnienie wyjściowe wyszukiwania (4B)

W narzędziu przejdź do sekcji Parallel research zwielokrotnienie wyjściowe (4B). Otwórz stage1_fanout/agent.py.

Węzły funkcji i bariery synchronizacji

W fazie badań używane są 2 węzły funkcji zaimportowane z agent/graph.py:

• scan_trends: zwraca Event(output={"trends": [...]}) zawierający 10 ocenionych trendów platformy.
• read_backlog: Zwraca Event(output={"backlog": [...], "idea": "..."}) 15 pomysłów na zaległe treści na kanale wraz z początkowym promptem.

Każda funkcja przyjmuje argument node_input (dane wyjściowe poprzedniego węzła) i zwraca wartość Event.

Węzeł JoinNode pełni funkcję bariery synchronizacji: wstrzymuje działanie, dopóki każdy łańcuch przychodzący nie dostarczy zdarzenia, a następnie agreguje wszystkie wyniki gałęzi w słowniku z kluczami w postaci nazw węzłów ({"scan_trends": {...}, "read_backlog": {...}}).

Edycja ręczna: definiowanie złączenia i równoległych krawędzi

W stage1_fanout/agent.py utwórz instancję JoinNode i połącz 2 równoległe łańcuchy, zaczynając od START:

join_research = JoinNode(name="join_research")

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

Zapisz zmiany. Weryfikator stołu warsztatowego potwierdza, że połączenie i krawędzie są połączone przewodami. Uruchom etap, klikając Uruchom etap 1 lub korzystając z osadzonego interfejsu internetowego ADK.

Czego możesz się spodziewać i dlaczego

• Jednoczesne wykonywanie czytnika: na wykresie wykonania scan_trends i read_backlog są wykonywane jednocześnie.
  • Uzasadnienie: oba łańcuchy pochodzą z lokalizacji START. Silnik ADK planuje niezależne gałęzie jednocześnie.
• Zbiorcze dane wyjściowe słownika: przepływ pracy kończy się na etapie join_research, a wynikiem jest słownik z wpisami dla obu czytników.
  • Dlaczego: JoinNode zapewnia pełne przechwytywanie danych przed zezwoleniem na wykonanie kolejnych węzłów.

Węzły agenta (4C)

W środowisku Workbench przejdź do sekcji Węzły agenta (4C). Otwórz stage2_direction/agent.py.

Tryby działania i schematy strukturalne

Gdy element Agent jest umieszczony w Workflow, domyślnie działa w trybie single_turn:

• Jako dane wejściowe kontekstu otrzymuje dane wyjściowe poprzedniego węzła.
• Wykonuje pojedyncze wywołanie wnioskowania bez konwersacji.
• Przekazuje uporządkowane dane do następnego węzła.

Przypisując output_schema=Directions, agent wymusza weryfikację danych wyjściowych modelu za pomocą biblioteki Pydantic. Wykres podrzędny otrzymuje obiekty z określonym typem zamiast nieustrukturyzowanego tekstu:

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 nakazuje modelowi zaproponowanie 4 kandydatów z odwołaniem do dowodów pochodzących zarówno z trendów, jak i z zaległości. Kandydaci 1–3 oferują realne koncepcje kanałów. Kandydat 4 celowo wprowadza koncepcję naruszającą zasady, aby w następnym kroku przetestować zabezpieczenie.

Edycja praktyczna: definiowanie węzła agenta i łączenie złączenia

W sekcji stage2_direction/agent.py skonfiguruj propose_directions i rozszerz krawędzie przepływu pracy:

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

Czego możesz się spodziewać i dlaczego

• Bezpośrednie wykorzystanie słownika: propose_directions wykorzystuje ładunek JSON wyemitowany przez join_research bez ręcznego formatowania.
• Wpisane dane wyjściowe kandydata: agent emituje zweryfikowany obiekt Directions zawierający 4 oddzielne kandydatury. Węzły podrzędne odczytują pola według nazwy atrybutu (candidate.title) bez analizowania ciągów znaków.

Z udziałem człowieka (4D)

W platformie przejdź do sekcji Human-in-the-loop (4D). Otwórz agent/graph.py.

Instrukcje promptów a zawieszenie deterministyczne

Procesy produkcyjne, które generują koszty finansowe lub publikują treści, wymagają nadzoru człowieka w kluczowych momentach podejmowania decyzji. W przypadku pojedynczego prompta prośby o potwierdzenie to instrukcje pomocnicze, które użytkownik może łatwo pominąć. W przepływie pracy ADK zatwierdzenie przez człowieka jest wymuszane przez silnik wykonawczy: wykres zatrzymuje się w wyznaczonym węźle i nie może przejść dalej, dopóki nie otrzyma zewnętrznych danych wejściowych zweryfikowanych pod kątem schematu:

• Wywołanie funkcji RequestInput natychmiast wstrzymuje wykonanie przepływu pracy.
• ADK rejestruje w pamięci sesji wywołanie przerwania otwartego i wydaje unikalny identyfikator interrupt_id.
• Proces wykonywania zostaje wstrzymany bez zużywania tokenów ani wątków serwera.
• Wykonanie wykresu zostanie wznowione dopiero po przesłaniu prawidłowego function_response zgodnego ze schematem i identyfikatorem przerwania.

Edycja praktyczna: wstrzymywanie wykonywania za pomocą RequestInput

W agent/graph.py zaimplementuj wywołanie zawieszenia w 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 konfiguruje 3 atrybuty:

• message: prompt z prośbą o opinię wyświetlany użytkownikowi.
• response_schema: schemat JSON, który interfejs renderuje jako formularz wejściowy, a ADK weryfikuje po przesłaniu.
• payload: metadane dołączone do żądania (4 kandydaci), które umożliwiają interfejsom klienta renderowanie kart opinii bez wysyłania zapytań o stan sesji.

Czego możesz się spodziewać i dlaczego

• Przepływ pracy zatrzymuje się na etapie direction_gate: w ADK Web lub w interfejsie platformy roboczej działanie zostaje wstrzymane i wyświetla się interaktywny formularz wyboru kandydatów.
  • Uzasadnienie: silnik napotkał wywołanie RequestInput i zapisał stan wykonania w runs/sessions.db.
• Wznowienie wymaga strukturalnych danych wejściowych: wysłanie dowolnego tekstu na czacie nie powoduje przejścia do kolejnego węzła. Wybranie opcji (1, 2, 3 lub 4) powoduje przesłanie wpisanego znaku function_response, który spełnia warunek response_schema, i wznowienie wykonywania.

5. Stan i router

W VibeStudio Workbench przejdź do Step 5 · State and Router, części (5A)–(5C).

Będziesz utrwalać wybory użytkowników w stanie sesji, egzekwować zasady bezpieczeństwa kanału za pomocą deterministycznych węzłów routera i tworzyć iteracyjnego agenta zadań, który będzie automatycznie usuwać naruszenia zasad przed wygenerowaniem skryptów wideo.

Stan przepływu pracy (5A)

W Workbenchu otwórz Workflow State (5A) (Stan przepływu pracy (5A)).

Stan sesji a dane wyjściowe węzła

W przepływie pracy ADK dane przemieszczają się po wykresie za pomocą 2 różnych mechanizmów:

• Dane wyjściowe węzła (Event(output=...)): dane kierowane ściśle do bezpośrednich odbiorców zdefiniowanych na liście krawędzi.
• Stan sesji (Event(state=...)): współdzielony słownik klucz-wartość dostępny dla każdego kolejnego węzła w cyklu życia wykonania.

Gdy użytkownik wybierze kandydata w punkcie direction_gate, wybór zostanie przekazany jako indeks liczbowy ({"pick": "2"}). Węzły podrzędne potrzebują pełnego obiektu kierunku: tytułu, kąta narracji i wstępu. Zamiast przekazywać szczegółowe metadane przez każdy ładunek węzła pośredniego, persist_direction zapisuje rozwiązanie w stanie sesji współdzielonej.

Węzły nie muszą przekazywać całego słownika stanów sesji. Gdy węzeł zwraca Event(state=...), dostarcza tylko nowe lub zaktualizowane pary klucz-wartość. ADK automatycznie scala te aktualizacje w pamięci sesji:

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

Przekazanie tej Event kontroli do środowiska wykonawczego Workflow powoduje zapisanie nowych wartości w dzienniku sesji w runs/sessions.db.

Powiązanie parametrów

Węzły funkcji ADK automatycznie odczytują stan sesji poprzez sprawdzanie parametrów. Jeśli sygnatura funkcji deklaruje nazwę parametru pasującą do istniejącego klucza stanu, ADK wyodrębnia ten klucz ze stanu i przekazuje go bezpośrednio:

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

W tym przypadku wartość candidates została zapisana w stanie sesji przez direction_gate. ADK wiąże go bezpośrednio z persist_direction(node_input, candidates: list = []) bez konieczności jawnego wyszukiwania w słowniku.

Klucze z prefiksem user: są przechowywane w pamięci na poziomie użytkownika i są dostępne w kolejnych sesjach, co umożliwia późniejszym uruchomieniom procesu uzyskiwanie dostępu do preferencji twórcy.

Ręczna edycja: utrwalanie stanu i podłączanie węzła

1. W pliku agent/graph.py w sekcji persist_direction zastąp wiersz TODO: PERSIST_STATE wartością zwracaną zdarzenia stanu:

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

2. W zbiorze stage3_router/agent.py dodaj persist_direction do trzeciego łańcucha na liście edges:

           (join_research, propose_directions, direction_gate,
            persist_direction)

Zapisz pliki. W środowisku roboczym sprawdź, czy przy ikonach state write in place i persist_direction in the chain wyświetlają się zielone znaczniki wyboru.

Węzeł routera (5B)

W środowisku Workbench otwórz Węzeł routera (5B).

Deterministyczny routing oparty na zasadach

Router to wyspecjalizowany węzeł funkcji, który ocenia dane wyjściowe z węzłów nadrzędnych i kieruje wykonaniem wzdłuż gałęzi wykresu warunkowego. W przeciwieństwie do agentów generatywnych router wykonuje logikę deterministyczną bez wywoływania LLM.

Router zwraca Event określający 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")

W definicji przepływu pracy krawędź docelowa zdefiniowana jako słownik mapuje nazwy tras na węzły docelowe:

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

Router przepływu pracy policy_check odczytuje zabronione frazy z agent/policy_words.txt i przeprowadza dopasowanie całych słów do tytułu i kąta wybranego kierunku:

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

Przechowywanie zasad w formie danych zamiast zakodowanych na stałe instrukcji umożliwia aktualizowanie ich bez modyfikowania wykresu przepływu pracy: aktualizacja pliku tekstowego jest natychmiast stosowana w przypadku kolejnych uruchomień. Ocena jest deterministycznym dopasowywaniem wyrażeń regularnych, więc trwa milisekundy i nie generuje kosztów w postaci tokenów.

Miejsca docelowe: Scripter i kwarantanna

Router kieruje ruch do jednego z 2 węzłów podrzędnych:

• scripter: węzeł agenta single_turn, który przekształca zatwierdzoną wskazówkę w strukturalny skrypt produkcyjny zgodny ze Script schematem Pydantic:

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

• quarantine: początkowo funkcja zastępcza, która wstrzymuje oznaczone instrukcje. W następnej części zastąpiona przez autonomicznego agenta naprawczego.

Praktyczna edycja: kierowanie sprawdzaniem zasad

1. W sekcji agent/graph.py w obszarze policy_check uzupełnij instrukcję powrotu:

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

2. W stage3_router/agent.py zaktualizuj edges, aby kierować do policy_check, i połącz ponownie gałąź kwarantanny z scripter:

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

Zapisz pliki. W platformie sprawdź, czy mapowania serwerów granicznych routera zostały zweryfikowane.

Tryby agenta i węzeł zadania (5C)

W obszarze roboczym przejdź do sekcji Tryby agenta i węzeł zadania (5C).

Tryby wykonywania agenta

Instancje ADKAgent obsługują 3 tryby wykonywania dostosowane do konkretnych wymagań potoku:

Tryb	Cykl życia wykonania	Rola w potoku
chat	Wieloetapowa pętla konwersacyjna. Model określa, kiedy wywołać narzędzia, poprosić o dane wejściowe lub zakończyć turę.	Agenci główni, którzy wchodzą w interakcje z użytkownikami.
single_turn	Wywołanie wnioskowania z użyciem jednego modelu. Akceptuje dane wejściowe z poprzedniego węzła i emituje obiekt schematu strukturalnego.	Sekwencyjne przekształcenia wykresu (propose_directions, scripter).
task	Autonomiczna pętla z wykonywaniem narzędzi. Agent wykonuje iteracje, dopóki nie wywoła wbudowanego narzędzia finish_task.	Wieloetapowe działania zaradcze i inspekcja (quarantine).

Autonomiczne działania naprawcze w przypadku naruszenia zasad

Przeredagowanie oznaczonego kierunku wymaga trybu task, ponieważ liczba iteracji korekty jest zmienna. Agent otrzymuje oznaczoną instrukcję, wywołuje funkcję find_policy_hits, aby wykryć naruszenia, prosi o zatwierdzone alternatywy za pomocą funkcji suggest_replacement, przepisuje instrukcję i sprawdza, czy nie zawiera ona nieodpowiednich treści, zanim przejdzie dalej.

Oba narzędzia są zdefiniowane w agent/cleanup_tools.py z podpisanymi sygnaturami i ciągami dokumentacyjnymi:

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

Praktyczna edycja: tworzenie agenta zadania kwarantanny

W stage3_router/agent.py zastąp funkcję zastępczą quarantine definicją agenta zadań:

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

Tryb zadania wyposaża agenta w narzędzia i kończy wykonywanie, wywołując metodę finish_task. Gdy skonfigurujesz mode="task", ADK automatycznie udostępni finish_task i wygeneruje jego parametry z output_schema, dzięki czemu węzeł będzie zwracać obiekt CleanedDirection o określonym typie, który będzie zgodny ze schematem wejściowym węzła skryptu.

Czego możesz się spodziewać i dlaczego

Przetestuj obie ścieżki wykonania w ADK Web lub VibeStudio Workbench:

• Zatwierdzona ścieżka (kandydat 1, 2 lub 3):
  • Wybór zatwierdzonego kandydata powoduje przekierowanie z policy_check bezpośrednio do scripter (route="OK").
  • Scenarzysta tworzy 3-ujęciowy scenariusz produkcji zgodny ze schematem Script.
• Ścieżka kwarantanny (kandydat 4):
  • Kandydat 4 zawiera słowa oznaczone jako nieodpowiednie („clickbait”, „viral hack”).
  • policy_check tras do: quarantine (route="BLOCK").
  • W śladzie sesji obserwuj wywołanie funkcji quarantine, która wywołuje funkcję find_policy_hits, która wywołuje funkcję suggest_replacement w przypadku każdego naruszenia, przepisuje tytuł i wywołuje funkcję finish_task.
  • Wykonanie dołącza do scripter, tworząc skrypt na podstawie oczyszczonego kierunku.

Routery i wersje zapasowe Interfejs deweloperski oznacza router bez krawędzi rezerwowej. Jeśli policy_check zwróciło trasę inną niż OK lub BLOCK, bieg nie miałby miejsca docelowego. Dodanie DEFAULT_ROUTE: quarantine do słownika krawędzi obejmuje ten przypadek; zaimportuj DEFAULT_ROUTE z google.adk.workflow.

6. Bank zapamiętanych informacji

W VibeStudio Workbench przejdź do kroku 6 · Bank zapamiętanych informacji, części (6A) i (6B).

Obecnie przepływ pracy działa bez pamięci między sesjami. Każde wykonanie zaczyna się od zera, bez wiedzy o tym, co wcześniej wybrał twórca lub jakie gatunki preferuje. W tym kroku połączysz Bank zapamiętanych informacji Vertex AI Agent Engine, aby przechowywać i pobierać preferencje twórcy w różnych uruchomieniach.

Co ważne, pamięć jest zintegrowana za pomocą wywołań zwrotnych cyklu życia agenta, a nie węzłów potoku. Ponieważ wyodrębnianie i pobieranie pamięci służy poszczególnym agentom, a nie pośrednim etapom danych, dołączanie wywołań zwrotnych zachowuje czystą, odseparowaną topologię wykresu.

Bank zapamiętanych informacji (6A)

W Workbench otwórz Bank zapamiętanych informacji (6A).

Zarządzana pamięć na poziomie użytkownika

Bank zapamiętanych informacji to usługa zarządzana służąca do przechowywania długoterminowej pamięci użytkownika. Porządkuje fakty dotyczące osoby w określonym zakresie, który jest tu identyfikowany przez nazwę aplikacji i identyfikator użytkownika:

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

Tematy pamięci niestandardowej określają granice tego, co bank rejestruje:

• Wyodrębnianie tematu: gdy nowy tekst rozmowy zostanie przesłany za pomocą interfejsu memories.generate, usługa zastosuje model wyodrębniania do każdego opisu tematu. Tekst, który nie pasuje do tematu, nie generuje wspomnień.
• Konsolidacja i usuwanie duplikatów: usługa przekształca nowo wyodrębnione fakty w wektory i porównuje je z istniejącymi wspomnieniami w zakresie. Gdy obserwacja jest zgodna z istniejącym wspomnieniem, usługa aktualizuje to wspomnienie. Jeśli zawiera nowe informacje, usługa tworzy nowy wpis. Ten proces konsolidacji sprawia, że wiele sesji dotyczących danego tematu jest łączonych w spójne podsumowanie, zamiast tworzyć nadmiarowe wpisy.
• Pobieranie: wywołanie funkcji memories.retrieve w zakresie użytkownika zwraca zapisane fakty w kolejności od najstarszych do najnowszych.

Obie operacje są wdrażane w agent/platform/memory.py. Nazwa udostępnionego zasobu banku jest buforowana lokalnie w runs/memorybank.json.

Konfigurowanie banku zapamiętanych informacji

Użyj elementów sterujących stacji roboczej lub uruchom polecenia interfejsu wiersza poleceń w terminalu:

1. Połącz i udostępnij bank:

   python -m agent.platform.bank
   

   Tworzy instancję Agent Engine i konfiguruje tematy CREATOR_TASTE i CHANNEL_RULES.
2. Wypełnij historyczne sesje:

   python -m agent.platform.bank load
   

   Wczytuje 4 sesje historyczne twórców (2 motywy zwierzęce z ograniczeniami dotyczącymi stylu, 1 motyw gadżetów i 1 ostatni motyw fantasy).
3. Sprawdzanie skonsolidowanych informacji:

   python -m agent.platform.bank list
   

   Sprawdź dane wyjściowe. Zwróć uwagę, jak transkrypcje narracyjne zostały przekształcone w uporządkowane, skonsolidowane stwierdzenia faktów.

Wywołania zwrotne (6B)

W środowisku Workbench otwórz Callbacks (6B). Otwórz stage4_memory/agent.py.

Wywołania zwrotne cyklu życia agenta ADK

Wywołanie zwrotne to funkcja przekazywana jako argument do Agent. ADK wywołuje wywołania zwrotne w predefiniowanych momentach cyklu życia, przekazując aktywny kontekst. Zwrócenie wartości None powoduje kontynuowanie normalnego wykonywania, a zwrócenie obiektu zastępczego zastępuje lub przechwytuje operację.

ADK udostępnia 3 pary wywołań zwrotnych:

Para wywołań zwrotnych	Punkt wywołania	Otrzymane parametry	Zachowanie wartości zwracanej
before_agent_callback after_agent_callback	Obejmuje całą turę agenta	CallbackContext (stan, sesja, wywołanie)	Powrót Content zastępuje odpowiedź agenta, a None działa normalnie.
before_model_callback after_model_callback	Wokół każdego wywołania wnioskowania LLM	LlmRequest lub LlmResponse	Zwrócenie LlmResponse przerywa lub pomija wywołanie modelu, a zwrócenie None kontynuuje wywołanie.
before_tool_callback after_tool_callback	Wokół każdego wykonania narzędzia	Definicja narzędzia, argumenty, wynik	Zwrócenie słownika zastępuje dane wyjściowe narzędzia; None jest kontynuowane.

Wywołania zwrotne zapewniają czyste miejsce do wstrzykiwania kontekstu, zabezpieczeń, telemetrii i wyszukiwania w pamięci podręcznej bez wprowadzania do wykresu przepływu pracy zbędnych węzłów.

Ręczna edycja: wycofanie okablowania i wywołania zwrotne zapamiętywania

1. W stage4_memory/agent.py zaktualizuj propose_directions, aby dołączyć before_model_callback=recall_taste:

    output_schema=Directions,
    before_model_callback=recall_taste)

recall_taste jest wykonywana bezpośrednio przed wygenerowaniem przez Gemini proponowanych wskazówek. Pobiera historię twórcy z Banku zapamiętanych informacji, formatuje wspomnienia, zaczynając od najstarszych, i dołącza je do wychodzącego LlmRequest. Prompt nakazuje modelowi, aby kandydaci 1–3 byli bardziej zgodni z aktualnymi preferencjami twórcy, a reguły kanału traktował jako ścisłe ograniczenia.

2. W stage4_memory/agent.py zaktualizuj scripter, aby dołączyć after_agent_callback=remember_pick:

    output_schema=Script,
    after_agent_callback=remember_pick)

remember_pick jest uruchamiana po zakończeniu tury przez scripter. Odczytuje wybrany kierunek ze stanu sesji, tworzy zwięzłe podsumowanie decyzji twórcy i wywołuje funkcję memories.generate, aby zaktualizować Bank zapamiętanych informacji.

Czego możesz się spodziewać i dlaczego

Przetestuj przepływ pracy z wywołaniem zwrotnym w środowisku roboczym lub w interfejsie internetowym ADK:

1. Wykonaj działanie z pustym promptem:
   • W śladzie sesji sprawdź LlmRequest dla propose_directions. Zwróć uwagę na dodany kontekst pamięci, który zawiera informacje o preferencjach twórcy dotyczące motywów fantasy i zwięzłego tempa.
   • Obserwuj proponowane kierunki: kandydaci 1–3 są zgodni z dotychczasowymi preferencjami twórcy, nawet jeśli trendy podkreślają inne tematy.
2. Wybierz kandydata w direction_gate.
3. Po zakończeniu scripter sprawdź rekordy w Banku zapamiętanych informacji:

   python -m agent.platform.bank list
   

   Bank odzwierciedla teraz najnowszy wybór, łącząc go z poprzednimi zapisami dotyczącymi gustu.

7. RAG Engine

W VibeStudio Workbench przejdź do Kroku 7. Silnik RAG, części (7A) i (7B).

Opublikowane filmy gromadzą bieżące opinie widzów. W agent/comments.md zebrano 30 reprezentatywnych komentarzy, w których widzowie chwalą film, krytykują tempo reklam i wyrażają preferencje dotyczące dźwięku. W tym kroku indeksujesz te komentarze za pomocą Vertex AI RAG Engine i łączysz wyszukiwanie semantyczne ze zwielokrotnieniem wyjściowym wyszukiwania.

Wyszukiwanie w dokumentach (7A)

W obszarze roboczym otwórz RAG Engine (7A).

Bank zapamiętanych informacji a RAG Engine

Oba narzędzia opierają przepływy pracy na danych zewnętrznych, ale służą do różnych celów architektonicznych:

Wymiar	Bank zapamiętanych informacji	RAG Engine
Główny przypadek użycia	Długoterminowe preferencje użytkowników i reguły operacyjne	Wyszukiwanie semantyczne w dużych zbiorach dokumentów
Zakres	Ograniczone do poszczególnych identyfikatorów użytkowników i nazw aplikacji	Ograniczone do zasobów korpusu udostępnionych wszystkim użytkownikom
Przetwarzanie danych	Wyodrębnianie, osadzanie i konsolidacja semantyczna w czasie rzeczywistym	Dzielenie dokumentu na fragmenty, wektory dystrybucyjne i wyszukiwanie najbliższych sąsiadów
Integracja z wykresem	Wywołania zwrotne cyklu życia agenta (before_model_callback, after_agent_callback)	Dedykowany węzeł funkcji w zwielokrotnieniu wyjściowym wyszukiwania (read_feedback)

Dzielenie dokumentu na fragmenty i wektory dystrybucyjne

RAG Engine indeksuje dokumenty, dzieląc tekst na fragmenty semantyczne i przechowując ich wektory w zarządzanej bazie danych:

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

• Rozmiar fragmentu: skonfigurowany na 120 tokenów z 20 tokenami nakładania się. W ten sposób zbierane są 2–3 komentarze do każdego fragmentu, dzięki czemu każdy wektor reprezentuje spójne nastawienie bez rozwadniania znaczenia w przypadku niezwiązanych ze sobą opinii.
• Model wektora dystrybucyjnego: text-embedding-005 przekształca tekst w wektory o dużej liczbie wymiarów. Gdy przesyłane jest zapytanie, model konwertuje je na wektor i znajduje najbliższe dopasowania na podstawie odległości semantycznej. Komentarz o małym smoku pilnującym skarpetek pasuje do promptu o magicznych stworzeniach bez konieczności dokładnego pokrywania się słów kluczowych.

Konfigurowanie korpusu RAG

Zainicjuj korpus za pomocą przycisków platformy lub poleceń terminala:

1. Utwórz korpus:

   python -m agent.platform.rag
   

   Zapewnia obsługę administracyjną zarządzanej bazy danych wektorów i zapisuje identyfikator zasobu w runs/ragcorpus.json.
2. Przesyłanie i indeksowanie komentarzy: przesyła agent/comments.md z konfiguracją dzielenia na części i czeka na zakończenie indeksowania.
3. Wysyłaj zapytania do korpusu: testuj wyszukiwanie podobieństw za pomocą zapytań, które nie zawierają dokładnie tych samych słów co komentarze (np. zapytanie „małe magiczne stworzenia” może zwrócić komentarze o smokach).

Węzeł pobierania (7B)

W workbenchu otwórz The third reader (7B) (Trzeci czytnik (7B)). Otwórz stage5_rag/agent.py.

Pobieranie jako węzła wykresu

Opinie odbiorców to dane z badań udostępniane w ramach procesu. W przeciwieństwie do pamięci twórcy osobistego opinie widzów są przesyłane bezpośrednio do join_research wraz z danymi o trendach i zaległościach. Dlatego jest ona zaimplementowana jako węzeł funkcji:

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 wyodrębnia początkowy pomysł użytkownika i wykonuje zapytanie wektorowe w korpusie RAG Engine. Zwraca pobrane komentarze w Event(output=...) ładunku.

Praktyczna edycja: podłączanie trzeciego czytnika do zwielokrotnienia wyjściowego

W stage5_rag/agent.py zaktualizuj edges, aby dodać read_feedback jako trzecią gałąź równoległą wchodzącą do join_research:

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

Ponieważ join_research to JoinNode, synchronizuje wszystkie przychodzące gałęzie, czekając, aż scan_trends, read_backlog i read_feedback wyemitują zdarzenia, zanim przekaże zagregowany pakiet dalej.

Czego możesz się spodziewać i dlaczego

Uruchom przepływ pracy w środowisku roboczym:

1. Prześlij prompt z pomysłem (np. „miniaturowy smok strzegący blatu kuchennego”).
2. W śladzie wykonania sprawdź, czy wszystkie 3 węzły czytnika są wykonywane jednocześnie.
3. Obserwuj join_research: słownik wyjściowy zawiera teraz trends, backlog i feedback.
4. Sprawdź wygenerowane propozycje w propose_directions: model uwzględnia w swoich propozycjach komentarze widzów i odnosi się do opinii odbiorców w polach dowodów.
5. Zwróć uwagę, że pobieranie RAG jest deterministyczne (identyczne zapytania zwracają identyczne fragmenty komentarzy), a węzeł propozycji generatywnej tworzy kreatywne warianty.

8. Asynchroniczne generowanie filmów za pomocą Veo

W VibeStudio Workbench przejdź do Kroku 8. Film, części (8A) i (8B).

Generowanie filmu w wysokiej rozdzielczości za pomocą Google Veo wymaga kilku minut na renderowanie. Blokowanie wykonania wykresu w tym okresie powoduje marnowanie zasobów obliczeniowych, blokowanie pul wątków i naraża działanie na przerwy w połączeniu HTTP. W tym kroku renderowanie wideo staje się asynchroniczne dzięki użyciu funkcji LongRunningFunctionTool ADK.

Narzędzia długotrwałe (8A)

W workbenchu otwórz Narzędzie działające długo (8A). Otwórz stage6_video/agent.py i agent/deliver.py.

Narzędzia synchroniczne a narzędzia działające przez dłuższy czas

Standardowe narzędzia funkcji pakietu ADK są wykonywane synchronicznie w ramach tury agenta: model wywołuje narzędzie, czeka na zwrócenie ładunku i włącza wynik do bieżącej tury.

Renderowanie filmu nie może zostać ukończone w jednej turze. Zamiast tego render_submit inicjuje zadanie generowania i natychmiast zwraca potwierdzenie operacji ze stanem "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"]}

Gdy jest otoczony tagiem LongRunningFunctionTool, ADK przechwytuje stan "pending". Tura agenta dobiega końca, przepływ pracy zostaje wstrzymany w węźle, a dane oczekującego połączenia (w tym identyfikator połączenia i potwierdzenie) są rejestrowane w runs/sessions.db. Proces wykonywania kończy się bez utrzymywania aktywnych połączeń sieciowych ani wątków roboczych.

Praktyczna edycja: zawijanie narzędzia do renderowania

W pliku stage6_video/agent.py zaktualizuj element render_desk, aby umieścić element render_submit w elemencie LongRunningFunctionTool:

    tools=[LongRunningFunctionTool(render_submit)])

Wznawianie według identyfikatora połączenia

Uniwersalny wzorzec wznowienia

ADK stosuje identyczny mechanizm wstrzymywania i wznawiania przepływów pracy zarówno w przypadku ludzi, jak i narzędzi zewnętrznych:

Wyzwalacz zawieszenia	Inicjowanie budowania	Stan zawieszenia przechowywany	Zdarzenie wznowienia
Decyzja człowieka	yield RequestInput(...)	Otwieranie promptu wejściowego w pamięci sesji	FunctionResponse zawierające identyfikator połączenia w sprawie zawieszenia,
Narzędzie długotrwałe	LongRunningFunctionTool(...) powracający pending	Otwieranie wywołania narzędzia w pamięci sesji	FunctionResponse zawierające identyfikator połączenia w sprawie zawieszenia,

W obu przypadkach przepływ pracy zostaje całkowicie wstrzymany i wznawia się dopiero wtedy, gdy z zewnętrznego źródła (interfejsu użytkownika, webhooka lub procesu działającego w tle) nadejdzie zdarzenie z pasującym FunctionResponse.

Ręczna edycja: uzupełnianie odpowiedzi na dostawę

W agent/deliver.py utwórz część FunctionResponse dotyczącą wznowienia:

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

Demon dostarczania wysyła zapytania do Veo, dopóki nie zostanie wygenerowany plik wideo, a następnie wysyła ten plik FunctionResponse do sesji. ADK dopasowuje identyfikator połączenia i wznawia przepływ pracy bezpośrednio w następnym węźle. Ukończone węzły nie są wykonywane ponownie, a agent nie wykonuje kolejnego ruchu generatywnego.

Ustawienie STUDIO_REAL_VIDEO=0 w .env włącza renderowanie próbne: start zwraca natychmiastowy rachunek testowy, a check symuluje zakończenie w ciągu 5 sekund bez wykonywania płatnych wywołań interfejsu Veo API.

Integracja potoku (8B)

W środowisku roboczym otwórz render_desk na wykresie (8B). Otwórz stage6_video/agent.py.

Węzłem końcowym w potoku jest store_video. Odczytuje informacje o ukończonym renderowaniu z runs/state.json (gdzie proces wyświetlania je zarejestrował) i zapisuje adres URL filmu oraz stan generowania w udostępnionym stanie sesji.

Praktyczna edycja: podłączanie całego potoku wideo

W stage6_video/agent.py zaktualizuj edges, aby dodać render_desk i store_video:

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

Czego możesz się spodziewać i dlaczego

Przetestuj asynchroniczny przepływ generowania w środowisku:

1. Przeprowadź proces od wyboru kandydata po wygenerowanie skryptu.
2. W render_desk obserwuj wywołanie przez agenta funkcji render_submit.
3. Przepływ pracy zostanie natychmiast zawieszony. W środowisku roboczym lub ADK Web sprawdź stan oczekiwania: sesja zawiera otwarty identyfikator połączenia, a żadne procesy w tle nie zużywają zasobów.
4. Uruchom demona dostarczania za pomocą konsoli platformy lub w terminalu:

   python -m agent.deliver
   

   Proces dostarczania monitoruje Veo, dopóki film nie będzie gotowy, a następnie wysyła zdarzenie wznowienia.
5. W ADK Web odśwież sesję: wykonanie zostanie wznowione w punkcie store_video, adres URL filmu zostanie zapisany w stanie sesji, a proces zostanie zakończony.

9. Wdrożenie w Cloud Run

W VibeStudio Workbench przejdź do Kroku 9. Wdróż.

Każdy komponent potoku został opracowany i zweryfikowany w odpowiednich piaskownicach. W tym kroku składasz kompletny potok produkcyjny i wdrażasz go w Google Cloud Run.

The ADK Runner

W trakcie opracowywania, adk web zorganizował graf. W środowisku produkcyjnym aplikacja hostuje przepływ pracy za pomocą klasy Runner pakietu 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: uruchamia wykonywanie przepływu pracy, generując zdarzenia sekwencyjnie w miarę wykonywania węzłów i utrwalając aktualizacje w usłudze sesji.
• Ujednolicone wznawianie: zarówno decyzje użytkownika w punkcie direction_gate, jak i dostarczone przez Veo ukończone filmy wznawiają działanie za pomocą identycznych obiektów FunctionResponse przesyłanych do run_async.

Architektura aplikacji produkcyjnej

Aplikacja produkcyjna w vibestudio/ integruje pełną ścieżkę:

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

• Strumień pojedynczych zdarzeń: backend FastAPI publikuje zdarzenia w jednym strumieniu zdarzeń wysyłanych przez serwer (SSE). Interfejs React wizualizuje postęp wykresu w czasie rzeczywistym i obsługuje późne połączenia bez utraty stanu.
• Oddzielne wykonywanie: aplikacja zarządza pętlą zdarzeń. Wykres przepływu pracy koncentruje się wyłącznie na logice wykonywania, nie uwzględniając interfejsu.

Pełna lista krawędzi przepływu pracy w agent/graph.py łączy wszystkie wzorce architektoniczne utworzone w ramach tego laboratorium:

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

Wdrażanie w Cloud Run

Google Cloud Run zapewnia hosting bezserwerowy z automatycznym skalowaniem, routingiem żądań i zintegrowanymi kompilacjami kontenerów:

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

• Kompilacja kontenera: gcloud run deploy --source pakuje katalog vibestudio/, tworzy obraz kontenera za pomocą Cloud Build i wdraża usługę w ramach jednej operacji.
• Koligacja sesji: kieruje żądania od tego samego użytkownika do tej samej instancji kontenera, zachowując lokalny stan sesji w kolejnych krokach.
• Obserwowanie: integracja z Cloud Trace rejestruje rozproszone zakresy dla każdego węzła, wywołania LLM i wykonania narzędzia, które są dostępne w konsoli Google Cloud w Eksploratorze logów czasu.

Aby uruchomić skrypt wdrażania, kliknij przycisk Wdróż w środowisku roboczym. Po zakończeniu kompilacji w terminalu wyświetli się adres URL usługi na żywo.

10. Podsumowanie

W VibeStudio Workbench przejdź do Kroku 10. Podsumowanie, aby sprawdzić ukończoną architekturę.

Krok	Architektura i koncepcje	Wzorzec implementacji
pojedynczy prompt,	Pojedynczy prompt, narzędzia funkcji, sekwencyjna pętla czatu	Agent(tools=[...]), function_call / function_response
Podstawy przepływu pracy z agentem	Przepływ pracy w formie grafu, równoległe badania, dane wyjściowe schematu, weryfikacja przez człowieka	Workflow, START, JoinNode, output_schema, RequestInput
Stan i router	Stan sesji współdzielonej, powiązanie parametrów, deterministyczne przekierowywanie, agent zadań	Event(state=...), Event(route=...), mode="task", finish_task
Bank zapamiętanych informacji	Pamięć długotrwała na poziomie użytkownika, konsolidacja semantyczna, punkty zaczepienia cyklu życia	memories.generate / retrieve, before_model_callback, after_agent_callback
RAG Engine	Wyszukiwanie dokumentów na podstawie komentarzy odbiorców i osadzania semantycznego	Węzeł rag.create_corpus, RagEmbeddingModelConfig, read_feedback
Asynchroniczne generowanie filmów za pomocą Veo	Długotrwałe narzędzia, oczekujące potwierdzenia, zewnętrzny demon dostarczania	LongRunningFunctionTool, FunctionResponse(id=...) wznowienie
Wdrożenie w Cloud Run	Orkiestracja programowa, zdarzenia wysyłane przez serwer, kontener bezserwerowy	Runner(agent=wf), run_async, wdrożenie Cloud Run

Podstawowe zasady architektury

1. Zawieszanie zamiast czekania: procesy są zawieszane w sposób uporządkowany, aby umożliwić wprowadzenie danych przez użytkownika (RequestInput) lub wykonanie długotrwałych operacji (LongRunningFunctionTool). Procesy nie czekają bezczynnie na wątkach ani gniazdach sieciowych.
2. Uniwersalne wznawianie: każde zawieszenie jest wznawiane za pomocą identycznego mechanizmu: pojedynczego sygnału function_response zawierającego identyfikator wywołania zawieszonego węzła.
3. Oddzielne zarządzanie stanem: węzły udostępniają dane za pomocą nazwanych kluczy stanu sesji i powiązań parametrów zamiast szczegółowych, ściśle powiązanych ładunków pośrednich.
4. Deterministyczne kierowanie przed generatywnym kosztem: routery oparte na regułach i filtry wyrażeń regularnych oceniają zasady przy zerowym koszcie tokena przed uruchomieniem modeli generatywnych.
5. Rozdzielenie odpowiedzialności: kontekst specyficzny dla poszczególnych agentów należy do wywołań zwrotnych cyklu życia, a wspólne zależności danych