1. 소개
이 Codelab에서는 에이전트 개발 키트 (ADK)의 워크플로와 그래프를 사용하여 차세대 에이전트형 시스템을 빌드하는 방법을 안내합니다. 일반적인 아키텍처 패턴을 구현하고, 인간 참여형 (HITL) 상호작용을 조정하고, 장기 실행 비동기 실행을 처리합니다. 또한 엔터프라이즈 기술 자료와 영구 메모리를 통합하여 에이전트 동작을 맞춤설정하고 발전시킵니다. 마지막으로 이러한 기능을 연결하여 자동 동영상 생성 파이프라인을 구축합니다.
시나리오
활성 시청자가 있고 창의적인 아이디어가 계속 쌓이는 VibeTube의 디지털 채널을 운영하고 있습니다. 각 동영상을 제작하려면 트렌드 형식 조사, 시청자 의견 종합, 스크립트 개발, 정책 준수 확인, 동영상 클립 생성 등 여러 단계에 걸쳐 지속적으로 실행해야 합니다. 생성형 모델은 개별 애셋을 초안으로 작성할 수 있지만 일관된 출시를 제공하려면 오케스트레이션된 에이전트 아키텍처가 필요합니다.
이 수명 주기를 자동화하기 위해 VibeStudio를 빌드합니다. 이 에이전트 파이프라인은 일상적인 조사를 병렬로 실행하고, 인간 참여형 승인을 위해 선별된 옵션을 제시하고, 동영상을 생성하기 전에 자동화된 정책 게이트를 적용하고, 프로덕션 실행 전반에서 컨텍스트를 유지합니다.
학습 내용
- 그래프 엔지니어링 기반: 다단계 에이전트 아키텍처에는 명시적인 제어 흐름과 구조화된 실행 경로가 필요합니다. 상태에 따라 실행을 안내하는 결정적 라우터 노드와 함께 에지 튜플,
START진입점, 병렬 팬아웃 집계를 위한JoinNode를 사용하여 ADKWorkflow를 빌드합니다. - 에이전트 모드 및 수명 주기 콜백: 전문화된 작업에는 고유한 운영 동작과 결정론적 가이드라인이 필요합니다.
chat,single_turn, 도구 지원task모드를 워크플로 노드로 사용하여 ADKAgent인스턴스를 구성하고before_model_callback및after_agent_callback로 인터셉터를 적용합니다. - 인간 참여형 오케스트레이션: 중요한 크리에이티브 체크포인트에서 인간의 판단을 위해 프로덕션 파이프라인이 일시중지됩니다.
RequestInput를 구현하여 워크플로 실행을 일시중지하고, 구조화된 응답 스키마를 적용하고, 유휴 런타임 프로세스를 활성 상태로 유지하지 않고 실행을 재개합니다. - 계층적 에이전트 메모리: 프로덕션 시스템은 일시적인 실행 상태를 지속 가능한 컨텍스트와 분리합니다.
Event(state=...)및 매개변수 바인딩을 사용하여 단기 세션 상태를 관리하고 GEAP 메모리 뱅크를 연결하여 여러 실행에서 크리에이터 환경설정을 추출, 통합, 유지합니다. - 엔터프라이즈 기술 자료를 사용한 그라운딩: 자율 에이전트에는 동적 도메인 컨텍스트와 잠재고객 감정이 필요합니다. GEAP RAG 엔진 코퍼스를 병렬 팬아웃 내의 전용 검색 노드로 연결하여 에이전트 출력을 의미적으로 그라운딩합니다.
- 장기 실행 워크플로 및 배포: 멀티모달 동영상 렌더링은 장기간에 걸쳐 비동기식으로 작동합니다. 대기 중인 호출 영수증으로
LongRunningFunctionTool를 구현하여 호출 ID로 워크플로를 일시중지 및 재개하고, Cloud Run에서 ADKRunner를 사용하여 완료된 파이프라인을 배포합니다.
이 Codelab의 구성
이 Codelab은 개념 및 아키텍처 참조 역할을 합니다. 각 섹션에서는 해당 워크벤치 단계에서 구현된 ADK 구조를 설명하고, 참조 코드를 제공하며, 핵심 설계 원칙을 설정합니다. 워크벤치에서 해당 연습을 완료하기 전에 각 섹션을 검토하세요.
실습은 대화형 코드 편집기, 런타임 검증 도구, 내장 ADK 검사기가 포함된 동반 웹 인터페이스인 VibeStudio Workbench에서 진행됩니다. 워크벤치의 단계 번호는 이 Codelab과 직접 일치하므로 진행 상황을 동기화할 수 있습니다. 기본 그래프 수정사항은 단계 전반에 걸쳐 유지되며, 진행하면서 워크벤치에서 필수 요건을 자동으로 확인합니다.
워크벤치 연습을 완료하면 엔드 투 엔드 에이전트형 파이프라인을 조립하고 실행 중인 VibeStudio 애플리케이션을 Cloud Run에 배포하여 동영상 콘텐츠를 생성합니다.
환경은 VibeStudio Workbench (코드 편집 및 런타임 확인을 위한 로컬 웹 인터페이스), 백엔드 (ADK Workflow 및 agent/의 스테이지 샌드박스), Google Cloud (Gemini 모델, GEAP 메모리 뱅크, RAG 엔진, Veo 동영상 생성)의 세 가지 기본 구성요소로 구성됩니다.
2. 설정
워크숍 크레딧 신청하기
강사 주도 실습에 참여하는 경우 강사가 Google Cloud 프로젝트의 크레딧을 배포합니다. 강사의 안내에 따라 크레딧을 사용하고 계정에서 결제가 활성화되어 있는지 확인한 후 계속 진행하세요.
Cloud Shell 열기
Cloud Shell은 gcloud, Python, git이 사전 설치된 브라우저 기반 개발 환경입니다.
Cloud Shell을 실행하려면 다음 단계를 따르세요.
- Google Cloud 콘솔로 이동합니다.
- 상단 탐색 헤더에서 Cloud Shell 활성화 (터미널 창 아이콘)를 클릭합니다.
터미널 세션이 브라우저 창 하단에 열립니다.
저장소 클론 및 초기화
Cloud Shell 터미널에서 다음 명령어를 실행하여 프로젝트를 클론합니다.
git clone https://github.com/gca-americas/vibetube-studio cd ~/vibetube-studio
구성 프롬프트
설정 중에 다음 세부정보를 입력하라는 메시지가 표시됩니다.
- Google Cloud 프로젝트 ID:
setup_project.sh에서 메시지가 표시되면 Enter 키를 눌러 새 프로젝트를 자동으로 만듭니다. 기존 프로젝트 (예: 사전 할당된 프로젝트)를 사용하려면 프로젝트 ID를 입력하고 철자가 올바르며 결제가 활성 상태인지 확인하세요. - 이벤트 코드: 강사가 제공한 회의실 코드를 입력합니다. 받지 못한 경우 조교 또는 이웃에게 문의하세요. 집에서 이 실습을 완료하는 경우 Enter를 눌러 기본
sandbox방을 수락합니다. - 채널 표시 이름:
setup_codelab.sh에서 메시지가 표시되면 이름 또는 원하는 채널 핸들을 입력하거나 Enter 키를 눌러 Google 계정에서 생성된 기본값을 수락합니다.
다음 두 설정 스크립트를 순서대로 실행합니다.
./setup_project.sh ./setup_codelab.sh
setup_project.sh: 활성 결제가 있는 Google Cloud 프로젝트를 생성하거나 재사용하고, 프로젝트 ID를~/project_id.txt에 저장하고, 활성gcloud컨텍스트를 구성합니다.setup_codelab.sh:uv및 Python 종속 항목을.venv에 설치하고, 필요한 Google Cloud API를 사용 설정하고,.env에서 채널 설정을 구성하고, Gemini로 모델 액세스를 확인하고, 메모리 뱅크 및 RAG 리소스를 프로비저닝하고, 워크벤치 인터페이스를 빌드하고, VibeStudio 워크벤치를 시작합니다.
스크립트는 실행 전 검사를 실행하고 백그라운드에서 VibeStudio Workbench를 시작합니다. 마지막 줄에는 열 링크가 표시됩니다.
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
링크를 클릭합니다. 웹 미리보기 → 포트 변경 → 4600에서도 동일한 주소를 확인할 수 있습니다.
언제든지 환경을 다시 확인하려면 python scripts/preflight.py를 실행합니다. 워크벤치를 다시 시작하려면 scripts/restart.sh를 실행합니다. 다시 설정하려면 ./setup_codelab.sh를 실행하세요. 그러면 구성과 진행 상황이 유지됩니다.
이 파일을 열고 시나리오의 1단계, 스토리와 완성된 그래프의 모양을 나타내는 2단계, 빌드할 항목을 읽습니다. 둘 다 운동이 없습니다. 그런 다음 여기로 돌아와 3단계를 진행합니다.
VibeStudio Workbench의 모든 실습 부분은 실제 아티팩트(디스크의 파일과 실행으로 작성된 세션)를 읽는 확인 패널로 끝납니다.
저장소 레이아웃
저장소는 핵심 워크플로 로직, 단계별 샌드박스, 워크벤치 환경, 프로덕션 애플리케이션으로 구성됩니다.
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/: 핵심 워크플로 그래프를 포함합니다. 이 디렉터리의 파일을 수정하여 병렬 팬아웃 노드, 결정적 정책 라우팅, 메모리 콜백, 동영상 생성 도구를 구현합니다.agent/platform/: Gemini 모델, GEAP 메모리 뱅크, GEAP RAG 엔진, Veo 동영상 합성 등 Google Cloud 서비스와 인터페이스합니다.stage0_prompt/~stage6_video/: 자체 포함 샌드박스 환경입니다. 각 폴더는 독립형root_agent를 내보내므로 내장된 ADK 개발 인터페이스를 통해 각 단계를 격리된 상태로 실행하고 검사할 수 있습니다.server/및web/: 포트 4600에서 로컬로 실행되는 VibeStudio Workbench 애플리케이션입니다. 단계 문서, 인페이지 코드 편집기, 런타임 증거 검증 도구, 그래프 시각화를 호스팅합니다.vibestudio/: 최종 단계에서 패키징되어 Cloud Run에 배포된 완전한 프로덕션 애플리케이션입니다. 완료된 워크플로 그래프의 자체 독립형 복사본이 포함되어 있습니다.
3. 모놀리식 에이전트
다중 노드 워크플로 그래프를 구성하기 전에 stage0_prompt/agent.py에서 단일 에이전트로 아키텍처 기준을 설정합니다. 이 에이전트는 두 개의 Python 함수 도구로 지원되는 산문으로 프로덕션 파이프라인을 설명하는 모놀리식 시스템 프롬프트를 사용합니다.
이 기준을 평가하면 프롬프트 기반 조정의 운영 경계가 명확해지고 프로덕션 시스템에 그래프 오케스트레이션이 필요한 이유가 설명됩니다.
ADK 에이전트 아키텍처 (3A)
VibeStudio Workbench에서 Step 3 · Monolithic agent로 이동하여 ADK agent architecture (3A)를 엽니다. 이 뷰에는 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
)
대화형 다이어그램은 에이전트 구성요소를 5가지 운영 도메인으로 그룹화합니다.
- 추론 레이어 (모델): 인지 작업, 프롬프트 추론, 도구 선택을 실행하는 핵심 언어 모델 (예: Gemini 3 Flash) 아키텍처의 다른 모든 요소는 이 모델에 정보를 제공하거나 제약합니다.
- 컨텍스트 레이어 (요청 사항 및 스킬): 모델 추론을 형성하는 지시어입니다.
instruction는 영구 시스템 프롬프트, 페르소나, 운영 규칙을 설정합니다.skills반복 가능한 워크플로를 위한 버전이 지정된 절차 안내 (SKILL.md)를 제공합니다. - 협업 및 작업 레이어 (도구, 하위 에이전트, 워크플로, 출력 스키마): 에이전트가 외부 시스템에서 작동하고 유형이 지정된 데이터를 내보낼 수 있도록 지원하는 인터페이스
tools호출 가능한 Python 함수 또는 모델 컨텍스트 프로토콜 (MCP) 엔드포인트를 제공합니다.subagents하위 위임 작업을 실행합니다.workflow는 멀티 에이전트 그래프를 조정합니다.output_schema는 Pydantic 모델을 적용하여 다운스트림 소비자가 구조화되지 않은 텍스트 대신 검증된 JSON을 수신하도록 보장합니다. - 인터셉터 레이어 (수명 주기 콜백): 에이전트 실행 (
before_agent/after_agent), 개별 모델 턴 (before_model/after_model), 도구 호출 (before_tool/after_tool) 전후에 맞춤 코드를 실행하는 결정적 가드레일입니다. 인터셉터는 모델 규정 준수에 의존하지 않고 정책 규칙을 적용합니다. - 외부 상태 (세션 및 메모리): 에이전트 로직과 분리된 스테이트풀 지속성
Session는 현재 실행 스레드의 임시 작업 메모리와 이벤트 추적을 보존합니다.Memory는 GEAP 메모리 뱅크와 같은 관리 서비스를 사용하여 내구성 있는 교차 세션 사실과 환경설정을 유지합니다.
이 단계의 모놀리식 에이전트는 이러한 기본 요소 중 model, instruction, tools의 세 가지만 구현합니다. 다음 단계에서는 그래프 워크플로, 구조화된 스키마, 인터셉터, 영구 메모리 서비스를 소개합니다.
모놀리식 에이전트 사양 (3B)
워크벤치에서 모놀리식 에이전트 사양 (3B)으로 이동합니다. stage0_prompt/agent.py를 열어 기준 에이전트 정의를 검사합니다.
- 단일 프롬프트 명령: 시스템 프롬프트는 플랫폼 트렌드 파악, 백로그 아이디어 검토, 광고 소재 컨셉 제안, 금지된 주제 정책 시행, 장면 목록 초안 작성이라는 다섯 가지 개별 제작 작업을 연속적인 산문으로 압축합니다.
- 기본 데이터 소스: 에이전트는 그래프 옆에 정의된 두 소스를 참조합니다.
agent/trends.py: 동적 열 점수가 있는 250개 풀에서 활성 형식 및 스타일 트렌드 10개를 샘플링합니다.agent/backlog.txt: 크리에이터의 원시 개념 메모를 줄 단위로 읽습니다.
에이전트의 도구 (3C)
워크벤치에서 에이전트의 도구 (3C)로 이동합니다.
에이전트에게 도구란 무엇인가요?
언어 모델은 본질적으로 폐쇄형 추론 엔진입니다. 사전 학습된 가중치와 즉각적인 컨텍스트 윈도우에 있는 토큰만으로 작동합니다. 데이터베이스를 기본적으로 쿼리하거나, 실시간 API에 액세스하거나, 코드를 실행할 수 없습니다.
도구는 이 경계를 연결합니다. 모델에 외부 대리인 권한을 부여하여 모델이 그라운드 트루스 정보를 검색하고 외부 시스템에서 결정론적 작업을 실행할 수 있습니다.
도구 호출은 모델과 ADK 런타임 간의 명시적인 5단계 프로토콜을 따릅니다.
- 스키마 선언: 개발자가 에이전트에 Python 함수를 제공합니다. ADK는 각 함수의 이름, 유형 주석, docstring을 검사하여 매개변수와 목적을 설명하는 OpenAPI 호환 JSON 스키마 선언을 생성합니다.
- 모델 추론: 추론 중에 모델은 사용자의 프롬프트에 외부 데이터가 필요한지 평가합니다. 필요한 경우 모델은 스키마와 일치하는 타겟 함수 이름과 인수 사전이 포함된 구조화된
function_call이벤트를 내보냅니다. - 런타임 실행: 모델 자체는 코드를 실행하지 않습니다. ADK 런타임은
function_call를 가로채고, 제공된 인수를 사용하여 실제 로컬 Python 함수를 실행하고, 반환 값을 캡처합니다. - 컨텍스트 재주입: ADK 런타임은 함수 반환 값을
function_response이벤트로 패키징하고 활성 세션 기록에 추가합니다. - 최종 종합: 모델이 이제 컨텍스트 창에 있는 도구 출력을 처리하고 대답을 완료합니다.
stage0_prompt/agent.py에서 두 연구 도구는 표준 Python 함수로 정의됩니다.
def check_trends() -> dict:
"""Ten formats trending on the platform right now, with a heat score each."""
from agent.trends import sample_trends
return {"trends": sample_trends()}
def read_backlog() -> dict:
"""The creator's backlog: ideas they noted down to make someday."""
from agent.graph import backlog_notes
return {"backlog": backlog_notes()}
직접 수정 및 실행
워크벤치 코드 편집기에서 두 함수 참조를 에이전트의 tools 목록에 추가합니다.
tools=[check_trends, read_backlog],
변경사항을 저장합니다. 파일이 디스크에서 업데이트되고 확인 행에서 두 도구가 모두 연결되어 있음을 확인합니다.
Open adk web을 클릭하여 내장된 ADK 개발 인터페이스를 실행합니다. 추천 아이디어 프롬프트를 전송합니다.
tonight's idea: a tiny robot doing laundry at midnight
예상되는 상황 및 이유
이 프롬프트를 보내면 세션 추적에서 다음 실행 시퀀스가 표시됩니다.
- 대답 전에 두 가지 도구 실행 이벤트가 표시됨:
check_trends및read_backlog에 대한function_call및function_response이벤트가 표시됩니다.- 이유: Gemini는 시스템 프롬프트 지시문 ('인기 급상승을 확인하고 아이디어 백로그를 살펴봐')을 평가하고, 가중치에 플랫폼 트렌드와 채널 메모가 부족하다는 것을 인식하고, 컨텍스트를 그라운딩하기 위해 두 함수를 모두 호출했습니다.
- 상담사가 방향을 제안하고 확인을 위해 잠시 멈춤: 대답에서 트렌드와 백로그를 종합한 동영상 방향을 제안하고 확인을 요청합니다.
- 이유: 스크립트를 생성하기 전에 크리에이터와 방향을 합의하라고 모델에 요청하는 명령 지시어가 있습니다.
- 후속 턴에서 확인 건너뛰기: 두 번째 메시지를 보냅니다(
skip the questions, just describe the video). 에이전트가 확인을 즉시 우회하고 제목과 장면을 작성합니다.- 이유: 프롬프트 안내는 결정론적 장벽이 아닌 권장 가이드라인입니다. 모놀리식 에이전트에서는 외부 워크플로가 실행 흐름을 제어하지 않으므로 사용자 안내가 기존 시스템 프롬프트 규칙을 재정의할 수 있습니다.
모놀리식 프롬프트의 아키텍처 제한사항
단일 프롬프트로 격리된 데모에서 허용 가능한 출력을 생성할 수 있지만 워크벤치 검증 도구에서 경계 조건을 테스트하면 다음과 같은 중요한 엔터프라이즈 제한사항이 드러납니다.
- 비구조화된 연구 집계: 도구 실행 순서가 비결정적입니다. 모델은 검색된 데이터를 자유 형식의 산문으로 요약하므로 다운스트림 시스템에서 특정 주장을 생성한 소스를 격리할 수 없습니다.
- 미인증 정책 시행: 모델이 자체 안전 규정 준수를 평가합니다. 모델에서 주제가 안전하다고 판단하면 외부 결정론적 로직에서 이 발견 사항을 검증하지 않습니다.
- 강제되지 않는 인간 참여형 일시중지: 크리에이터 확인을 요청하는 프롬프트 안내는 권장사항입니다. 모델이 질문을 우회하도록 지시하는 후속 메시지를 보내면 모델이 사람의 승인을 완전히 건너뜁니다.
이러한 아키텍처 격차로 인해 다음 단계에서 빌드되는 명시적 그래프 워크플로로 모놀리식 에이전트를 분해해야 합니다.
4. 에이전트형 워크플로 기본사항
VibeStudio Workbench에서 Step 4 · Agentic workflow fundamentals, parts 4A through 4D로 이동합니다.
이 단계에서는 단일 에이전트 기준에서 ADK Workflow를 사용한 결정적 그래프 오케스트레이션으로 전환합니다. 병렬 연구 팬아웃을 빌드하고, 조인 노드로 브랜치를 동기화하고, 스키마 검증된 광고 소재 후보를 생성하고, 결정론적 human-in-the-loop 승인 게이트를 도입합니다.
그래프 아키텍처 및 실행 체인 (4A)
작업대에서 그래프 아키텍처 및 실행 체인 (4A)을 엽니다.
ADK Workflow는 에이전트 실행을 다음과 같은 에지 목록으로 정의된 방향 그래프로 구조화합니다.
- 체인: 순차적 튜플은 선형 노드 실행 (
(node_a, node_b, node_c))을 정의합니다. - 병렬 브랜치: 원본 노드를 공유하는 독립 체인이 동시에 실행됩니다.
- 동기화:
JoinNode로 수렴되는 체인은 모든 수신 브랜치가 보고될 때까지 기다린 후 해제됩니다. - 결정적 제어: 실행 흐름은 프롬프트 텍스트에서 추론되는 대신 선언된 코드 구조에 의거하여 관리됩니다.
ADK의 노드 원형
ADK 워크플로는 여러 전문 노드 유형으로 구성됩니다. 각 원형은 그래프에서 특정 운영 역할을 수행하여 결정적 코드 실행과 생성 모델 추론을 분리합니다.
노드 원형 | 구현 | 파이프라인의 역할 |
함수 노드 |
| 결정적 로직, 데이터 가져오기, 상태 변형을 실행합니다. |
노드 결합 | 내장 | 동시 브랜치를 집계된 사전으로 동기화합니다. |
에이전트 노드 |
| 업스트림 입력에 대해 명령어를 평가하고 검증된 데이터를 내보냅니다. |
라우터 노드 |
| 조건부 로직을 평가하여 다운스트림 실행 브랜치를 선택합니다. |
사람 입력 노드 |
| 외부 사용자 응답이 도착할 때까지 실행 상태를 일시중지합니다. |
root_agent = Workflow(
name="stage1_fanout",
description="2 real readers -> join -> one research dict",
edges=[...])
이 구성에서 root_agent는 독립형 Agent이 아닌 Workflow의 인스턴스입니다. ADK는 워크플로를 일급 에이전트로 취급하여 전체 그래프를 통합 애플리케이션으로 로드, 제공, 검사할 수 있습니다. name는 ADK 웹에 애플리케이션을 등록하고 edges 목록은 실행 토폴로지를 정의합니다.
병렬 연구 팬아웃 (4B)
워크벤치에서 병렬 연구 팬아웃 (4B)으로 이동합니다. stage1_fanout/agent.py를 엽니다.
함수 노드 및 동기화 장벽
연구 단계에서는 agent/graph.py에서 가져온 두 개의 함수 노드를 사용합니다.
scan_trends: 점수가 매겨진 플랫폼 트렌드 10개가 포함된Event(output={"trends": [...]})을 반환합니다.read_backlog: 초기 실행 프롬프트와 함께 15개의 채널 백로그 아이디어가 포함된Event(output={"backlog": [...], "idea": "..."})을 반환합니다.
각 함수는 node_input (이전 노드의 출력)를 허용하고 Event를 반환합니다.
JoinNode는 동기화 장벽 역할을 합니다. 모든 인바운드 체인이 이벤트를 전달할 때까지 일시중지한 다음 모든 브랜치 결과를 노드 이름 ({"scan_trends": {...}, "read_backlog": {...}})으로 키가 지정된 사전으로 집계합니다.
실습 수정: 조인 및 평행 모서리 정의
stage1_fanout/agent.py에서 JoinNode를 인스턴스화하고 START에서 시작하는 두 개의 병렬 체인을 연결합니다.
join_research = JoinNode(name="join_research")
edges=[(START, scan_trends, join_research),
(START, read_backlog, join_research)])
변경사항을 저장합니다. 워크벤치 검증 도구는 조인과 가장자리가 연결되어 있는지 확인합니다. 1단계 실행을 사용하거나 내장된 ADK 웹 인터페이스를 통해 단계를 실행합니다.
예상되는 상황 및 이유
- 동시 리더 실행: 실행 그래프에서
scan_trends와read_backlog가 동시에 실행됩니다.- 이유: 두 체인이
START에서 시작됩니다. ADK 엔진은 독립적인 브랜치를 동시에 예약합니다.
- 이유: 두 체인이
- 집계된 사전 출력: 워크플로가
join_research에서 완료되어 두 리더의 항목이 포함된 사전을 출력합니다.- 이유:
JoinNode는 후속 노드가 실행되도록 허용하기 전에 완전한 데이터 캡처를 보장합니다.
- 이유:
상담사 노드 (4C)
워크벤치에서 에이전트 노드 (4C)로 이동합니다. stage2_direction/agent.py를 엽니다.
작동 모드 및 구조화된 스키마
Workflow 내에 삽입된 경우 Agent은 기본적으로 single_turn 모드로 실행됩니다.
- 이 노드는 이전 노드의 출력을 컨텍스트 입력으로 받습니다.
- 대화형 왕복 없이 단일 추론 호출을 실행합니다.
- 구조화된 데이터를 다음 노드로 출력합니다.
output_schema=Directions를 할당하면 에이전트가 모델 출력에 Pydantic 검증을 적용합니다. 다운스트림 그래프는 비구조화된 산문 대신 유형이 지정된 객체를 수신합니다.
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은 모델에 트렌드와 백로그 모두의 증거를 인용하여 4명의 후보자를 제안하도록 지시합니다. 후보 1~3은 실행 가능한 채널 개념을 제공합니다. 후보 4는 다음 단계에서 안전 게이트를 테스트하기 위해 정책을 위반하는 개념을 의도적으로 도입합니다.
실습 수정: 에이전트 노드 정의 및 조인 연결
stage2_direction/agent.py에서 propose_directions를 구성하고 워크플로 가장자리를 확장합니다.
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)])
예상되는 상황 및 이유
- 직접 사전 소비:
propose_directions는 수동 형식 지정 없이join_research에서 내보낸 JSON 페이로드를 소비합니다. - 유형이 지정된 후보 출력: 에이전트가 4개의 개별 후보가 포함된 검증된
Directions객체를 내보냅니다. 다운스트림 노드는 문자열 파싱 없이 속성 이름 (candidate.title)으로 필드를 읽습니다.
인간 참여형(Human-In-The-Loop)(4D)
워크벤치에서 인간 참여형 (4D)으로 이동합니다. agent/graph.py를 엽니다.
프롬프트 요청 사항과 결정론적 정지 비교
재정적 비용이 발생하거나 콘텐츠를 게시하는 프로덕션 워크플로에는 중요한 의사 결정 지점에서 사람의 감독이 필요합니다. 단일 프롬프트에서 확인 요청은 사용자가 모델에 쉽게 프롬프트를 입력하여 우회할 수 있는 권장사항입니다. ADK 워크플로에서 사람의 승인은 실행 엔진에 의해 적용됩니다. 그래프는 지정된 노드에서 중지되며 외부의 스키마 검증된 입력을 받을 때까지 진행할 수 없습니다.
RequestInput를 생성하면 워크플로 실행이 즉시 일시 중지됩니다.- ADK는 세션 저장소에 열린 인터럽트 호출을 기록하고 고유한
interrupt_id를 발행합니다. - 실행 프로세스가 토큰이나 서버 스레드를 사용하지 않고 중지됩니다.
- 스키마 및 인터럽트 ID와 일치하는 유효한
function_response가 제출된 경우에만 지연 실행이 재개됩니다.
실습 수정: RequestInput으로 실행 일시중지
agent/graph.py에서 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는 다음 세 가지 속성을 구성합니다.
message: 사용자에게 표시되는 리뷰 프롬프트입니다.response_schema: 프런트엔드에서 입력 양식으로 렌더링하고 제출 시 ADK에서 검증하는 JSON 스키마입니다.payload: 요청과 함께 번들로 제공되는 메타데이터 (4개의 후보)로, 클라이언트 인터페이스가 세션 상태를 쿼리하지 않고도 리뷰 카드를 렌더링할 수 있습니다.
예상되는 상황 및 이유
- direction_gate에서 워크플로가 중지됨: ADK Web 또는 워크벤치 인터페이스에서 실행이 일시중지되고 대화형 후보 선택 양식이 표시됩니다.
- 이유: 엔진에서 생성된
RequestInput가 발생했고 실행 상태가runs/sessions.db에 유지되었습니다.
- 이유: 엔진에서 생성된
- 재개에는 구조화된 입력이 필요함: 임의의 채팅 텍스트를 전송해도 그래프가 진행되지 않습니다. 옵션 (1, 2, 3 또는 4)을 선택하면
response_schema을 충족하는 입력된function_response이 제출되고 실행이 재개됩니다.
5. 상태 및 라우터
VibeStudio Workbench에서 5단계 · 상태 및 라우터의 (5A)~(5C) 부분으로 이동합니다.
사용자 선택사항을 세션 상태에 유지하고, 결정적 라우터 노드를 사용하여 채널 안전 정책을 적용하고, 동영상 스크립트를 생성하기 전에 정책 위반을 자동으로 해결하기 위해 반복적인 작업 에이전트를 어셈블합니다.
워크플로 상태 (5A)
워크벤치에서 Workflow State (5A)로 이동합니다.
세션 상태와 노드 출력 비교
ADK 워크플로에서 데이터는 다음 두 가지 메커니즘을 통해 그래프를 이동합니다.
- 노드 출력 (
Event(output=...)): 에지 목록에 정의된 직계 다운스트림 소비자로만 전송되는 데이터입니다. - 세션 상태 (
Event(state=...)): 실행 수명 주기의 후속 노드에서 액세스할 수 있는 공유 키-값 딕셔너리입니다.
사용자가 direction_gate에서 후보를 선택하면 선택이 숫자 색인 ({"pick": "2"})으로 도착합니다. 다운스트림 노드에는 제목, 내러티브 각도, 후크 라인 등 완전한 방향 객체가 필요합니다. 모든 중간 노드 페이로드를 통해 자세한 메타데이터를 전달하는 대신 persist_direction은 해결된 후보를 공유 세션 상태에 씁니다.
노드는 전체 세션 상태 딕셔너리를 전달할 필요가 없습니다. 노드가 Event(state=...)를 생성하면 새로 추가되거나 업데이트된 키-값 쌍만 제공합니다. ADK는 이러한 업데이트를 세션 저장소에 자동으로 병합합니다.
yield Event(state={"direction": chosen["title"], "angle": chosen.get("angle", ""),
"hook": hook, "user:prefs": {"last_direction": chosen["title"]}})
이 Event를 생성하면 Workflow 런타임에 제어권이 넘어가고, 런타임은 새 값을 runs/sessions.db의 세션 저널에 유지합니다.
매개변수 바인딩
ADK 함수 노드는 파라미터 검사를 통해 세션 상태를 자동으로 읽습니다. 함수 서명에서 기존 상태 키와 일치하는 매개변수 이름을 선언하면 ADK가 상태에서 해당 키를 추출하여 직접 전달합니다.
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])
여기서 candidates은 direction_gate에 의해 세션 상태에 작성되었습니다. ADK는 명시적 사전 조회가 필요 없이 persist_direction(node_input, candidates: list = [])에 직접 바인딩합니다.
user:로 시작하는 키는 사용자 수준 스토리지의 세션 간에 유지되므로 후속 워크플로 실행에서 크리에이터 환경설정에 액세스할 수 있습니다.
실습 수정: 상태 유지 및 노드 연결
agent/graph.py에서persist_direction내의TODO: PERSIST_STATE줄을 상태 이벤트 yield로 바꿉니다.
yield Event(state={"direction": chosen["title"], "angle": chosen.get("angle", ""),
"hook": hook, "user:prefs": {"last_direction": chosen["title"]}})
stage3_router/agent.py에서edges목록의 세 번째 체인에persist_direction을 추가합니다.
(join_research, propose_directions, direction_gate,
persist_direction)
파일을 저장합니다. 작업대에서 state write in place 및 persist_direction in the chain 모두 녹색 체크표시가 표시되는지 확인합니다.
라우터 노드 (5B)
워크벤치에서 라우터 노드 (5B)로 이동합니다.
확정적 정책 라우팅
라우터는 업스트림 출력을 평가하고 조건부 그래프 브랜치를 따라 실행을 지시하는 특수 함수 노드입니다. 생성형 에이전트와 달리 라우터는 LLM 호출 없이 결정론적 로직을 실행합니다.
라우터는 route 태그를 지정하는 Event를 반환합니다.
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")
워크플로 정의에서 사전으로 정의된 에지 타겟은 경로 이름을 대상 노드에 매핑합니다.
(length_check, {"TRIM": shorten, "PASS": scripter}),
워크플로 라우터는 policy_check agent/policy_words.txt에서 금지된 문구를 읽고 선택한 방향의 제목 및 각도에 대해 전체 단어 일치를 실행합니다.
return Event(output=node_input, route="BLOCK" if bad else "OK")
하드코딩된 명령어 대신 정책을 데이터로 저장하면 워크플로 그래프를 수정하지 않고 업데이트할 수 있습니다. 텍스트 파일을 업데이트하면 후속 실행에 즉시 적용됩니다. 평가는 결정적 정규식 일치이므로 생성 스크립트가 시작되기 전에 토큰 비용 없이 밀리초 단위로 실행됩니다.
대상: 스크립터 및 격리
라우터는 트래픽을 다음 두 다운스트림 노드 중 하나로 전달합니다.
scripter: 승인된 방향을ScriptPydantic 스키마를 준수하는 구조화된 제작 스크립트로 변환하는single_turn에이전트 노드입니다.
scripter = Agent(
name="scripter",
model=config.MODEL,
instruction=SCRIPT_INSTRUCTION,
output_schema=Script)
quarantine: 처음에는 플래그가 지정된 방향을 중지하는 자리표시자 함수로, 다음 부분에서 자율 시정 에이전트로 대체됩니다.
직접 수정: 정책 확인 라우팅
agent/graph.py의policy_check내에서 return 문을 완료합니다.
return Event(output=node_input, route="BLOCK" if bad else "OK")
stage3_router/agent.py에서policy_check를 라우팅하도록edges를 업데이트하고 스팸 격리 저장소 브랜치를scripter에 다시 조인합니다.
(join_research, propose_directions, direction_gate,
persist_direction, policy_check),
(policy_check, {"OK": scripter, "BLOCK": quarantine}),
(quarantine, scripter)])
파일을 저장합니다. 워크벤치에서 라우터 에지 매핑이 확인되었는지 확인합니다.
에이전트 모드 및 작업 노드 (5C)
워크벤치에서 Agent modes and the task node (5C)로 이동합니다.
에이전트 실행 모드
ADK Agent 인스턴스는 특정 파이프라인 요구사항에 맞게 조정된 세 가지 실행 모드를 지원합니다.
모드 | 실행 수명 주기 | 파이프라인의 역할 |
| 멀티턴 대화 루프 모델은 도구를 호출하거나, 입력을 요청하거나, 차례를 종료할 시기를 결정합니다. | 대화형 인간 사용자를 응대하는 루트 에이전트 |
| 단일 모델 추론 호출입니다. 이전 노드 입력을 수락하고 구조화된 스키마 객체를 내보냅니다. | 순차적 그래프 변환 ( |
| 도구 실행이 포함된 자율 루프입니다. 에이전트는 내장된 | 다단계 수정 및 검사 ( |
자율 정책 수정
플래그가 지정된 방향을 다시 작성하려면 수정 반복 횟수가 가변적이므로 task 모드가 필요합니다. 상담사는 신고된 요청을 수신하고, find_policy_hits를 호출하여 위반사항을 감지하고, suggest_replacement를 통해 승인된 대안을 요청하고, 요청을 다시 작성하고, 깨끗한지 확인한 후 계속 진행합니다.
두 도구 모두 유형이 지정된 서명과 독스트링을 사용하여 agent/cleanup_tools.py에 정의되어 있습니다.
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.
"""
실습 수정: 스팸 격리 저장소 작업 에이전트 조립
stage3_router/agent.py에서 자리표시자 quarantine 함수를 작업 에이전트 정의로 바꿉니다.
quarantine = Agent(
name="quarantine",
model=config.MODEL,
instruction=QUARANTINE_INSTRUCTION,
mode="task",
tools=[find_policy_hits, suggest_replacement],
output_schema=CleanedDirection,
)
작업 모드는 에이전트에게 도구를 제공하고 finish_task를 호출하여 실행을 종료합니다. mode="task"가 구성되면 ADK는 자동으로 finish_task를 제공하고 output_schema에서 매개변수를 파생하여 노드가 스크립터 노드의 입력 스키마와 일치하는 유형이 지정된 CleanedDirection 객체를 생성하도록 합니다.
예상되는 상황 및 이유
ADK 웹 또는 VibeStudio Workbench에서 두 실행 경로를 모두 테스트합니다.
- 승인된 경로 (후보 1, 2 또는 3):
policy_check에서scripter(route="OK")로 바로 연결되는 승인된 후보 경로를 선택합니다.- 스크립터는
Script스키마를 준수하는 3컷 프로덕션 스크립트를 생성합니다.
- 격리 해결 경로 (후보 4):
- 후보 4에 신고된 어휘 ('클릭베이트', '바이럴 해킹')가 포함되어 있습니다.
policy_check개의 경로가quarantine(route="BLOCK")로 이동합니다.- 세션 추적에서
quarantine가find_policy_hits를 호출하고, 각 위반에 대해suggest_replacement를 호출하고, 제목을 다시 작성하고,finish_task를 호출하는 것을 확인합니다. - 실행이
scripter에 다시 참여하여 정리된 방향에서 스크립트를 생성합니다.
6. 메모리 뱅크
VibeStudio Workbench에서 Step 6 · 메모리 뱅크, (6A) 및 (6B) 파트로 이동합니다.
현재 워크플로는 세션 간에 메모리 없이 작동합니다. 각 실행은 처음부터 시작되며, 크리에이터가 이전에 선택한 항목이나 선호하는 장르를 알지 못합니다. 이 단계에서는 여러 실행에서 크리에이터 환경설정을 저장하고 검색할 수 있도록 Vertex AI Agent Engine 메모리 뱅크를 연결합니다.
중요한 점은 메모리가 파이프라인 노드 대신 에이전트 수명 주기 콜백을 통해 통합된다는 것입니다. 메모리 추출 및 검색은 중간 데이터 단계가 아닌 개별 에이전트를 제공하므로 콜백을 연결하면 깨끗하고 분리된 그래프 토폴로지가 유지됩니다.
메모리 뱅크 (6A)
워크벤치에서 메모리 뱅크 (6A)로 이동합니다.
관리 사용자 수준 메모리
메모리 뱅크는 장기 사용자 메모리를 위한 관리형 서비스입니다. 정의된 범위(여기서는 애플리케이션 이름과 사용자 ID로 식별됨)에 따라 사용자에 관한 사실을 정리합니다.
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).",
}
맞춤 메모리 주제는 뱅크에서 기록하는 항목의 경계를 정의합니다.
- 주제 추출:
memories.generate를 통해 새로운 대화 텍스트가 제출되면 서비스는 각 주제 설명에 추출 모델을 적용합니다. 주제와 일치하지 않는 텍스트는 메모리를 생성하지 않습니다. - 통합 및 중복 제거: 서비스는 새로 추출된 사실을 임베딩으로 변환하고 범위 내의 기존 메모리와 비교합니다. 관찰 내용이 기존 메모리와 일치하면 서비스에서 해당 메모리를 업데이트합니다. 새로운 정보를 나타내는 경우 서비스는 새 항목을 만듭니다. 이 통합 프로세스를 통해 주제에 관한 여러 세션이 중복 항목을 생성하는 대신 일관된 요약으로 병합됩니다.
- 검색: 사용자 범위로
memories.retrieve를 호출하면 저장된 사실이 가장 오래된 순서대로 반환됩니다.
두 작업 모두 agent/platform/memory.py에 구현되어 있습니다. 프로비저닝된 은행 리소스 이름은 runs/memorybank.json에 로컬로 캐시됩니다.
메모리 뱅크 설정
워크벤치 컨트롤을 사용하거나 터미널에서 CLI 명령어를 실행합니다.
- 은행 연결 및 프로비저닝:
python -m agent.platform.bankCREATOR_TASTE및CHANNEL_RULES주제를 구성합니다. - 이전 세션 시드:
python -m agent.platform.bank load - 통합된 사실 검사:
python -m agent.platform.bank list
콜백 (6B)
워크벤치에서 콜백 (6B)으로 이동합니다. stage4_memory/agent.py를 엽니다.
ADK 에이전트 수명 주기 콜백
콜백은 Agent에 인수로 전달되는 함수입니다. ADK는 사전 정의된 수명 주기 시점에 콜백을 호출하여 활성 컨텍스트를 전달합니다. None를 반환하면 정상적인 실행이 계속되고, 대체 객체를 반환하면 작업이 재정의되거나 차단됩니다.
ADK는 다음 세 쌍의 콜백을 제공합니다.
콜백 쌍 | 호출 지점 | 수신된 매개변수 | 반환 값 동작 |
| 전체 에이전트 턴을 둘러싸는 영역 |
|
|
| 각 LLM 추론 호출을 둘러싸는 |
|
|
| 각 도구 실행을 둘러싸는 | 도구 정의, 인수, 결과 | dict를 반환하면 도구 출력이 재정의되고 |
콜백은 워크플로 그래프에 불필요한 노드를 도입하지 않고 컨텍스트 삽입, 가드레일, 원격 분석, 캐시 조회를 위한 깔끔한 위치를 제공합니다.
실습 수정: 회수 및 기억 콜백 연결
stage4_memory/agent.py에서before_model_callback=recall_taste를 연결하도록propose_directions를 업데이트합니다.
output_schema=Directions,
before_model_callback=recall_taste)
recall_taste는 Gemini가 후보 방향을 생성하기 직전에 실행됩니다. 메모리 뱅크에서 크리에이터의 기록을 가져오고, 가장 오래된 추억부터 순서대로 형식을 지정하고, 이를 전송되는 LlmRequest에 추가합니다. 프롬프트는 모델이 채널 규칙을 엄격한 제약 조건으로 취급하면서 후보 1~3을 크리에이터의 현재 취향에 맞추도록 지시합니다.
stage4_memory/agent.py에서after_agent_callback=remember_pick를 연결하도록scripter를 업데이트합니다.
output_schema=Script,
after_agent_callback=remember_pick)
remember_pick는 scripter의 차례가 완료된 후에 실행됩니다. 세션 상태에서 선택한 방향을 읽고, 크리에이터의 결정을 요약하는 간결한 문장을 합성하고, memories.generate를 호출하여 메모리 뱅크를 업데이트합니다.
예상되는 상황 및 이유
작업대 또는 ADK 웹에서 콜백으로 보강된 워크플로를 테스트합니다.
- 빈 프롬프트로 실행을 실행합니다.
- 세션 추적에서
LlmRequest의propose_directions를 검사합니다. 판타지 테마와 간결한 속도에 대한 크리에이터의 선호도를 자세히 설명하는 추억 컨텍스트가 추가되었습니다. - 제안된 방향을 살펴보세요. 트렌드에서 다른 주제를 강조하더라도 후보자 1~3은 크리에이터의 과거 선호도와 일치합니다.
- 세션 추적에서
direction_gate에서 후보자를 선택합니다.scripter가 완료되면 메모리 뱅크 레코드를 검토합니다.python -m agent.platform.bank list
7. RAG Engine
VibeStudio Workbench에서 Step 7 · RAG Engine, (7A) 및 (7B) 파트로 이동합니다.
게시된 동영상에는 지속적인 시청자 의견이 누적됩니다. agent/comments.md에는 30개의 대표적인 댓글이 수집되어 시청자의 칭찬, 스폰서 광고의 페이싱에 대한 비판, 오디오 환경설정을 파악할 수 있습니다. 이 단계에서는 Vertex AI RAG Engine을 사용하여 이러한 댓글의 색인을 생성하고 시맨틱 검색을 리서치 팬아웃에 연결합니다.
문서 기반 검색 (7A)
워크벤치에서 RAG Engine (7A)으로 이동합니다.
메모리 뱅크와 RAG Engine 비교
두 도구 모두 외부 데이터를 기반으로 워크플로를 실행하지만 아키텍처상의 목적은 서로 다릅니다.
측정기준 | 메모리 뱅크 | RAG Engine |
기본 사용 사례 | 장기 사용자 환경설정 및 운영 규칙 | 대규모 문서 컬렉션에 대한 시맨틱 검색 |
범위 | 개별 사용자 ID 및 애플리케이션 이름으로 범위가 지정됩니다. | 모든 사용자의 공유 말뭉치 리소스에 범위가 지정됩니다. |
데이터 처리 | 실시간 추출, 삽입, 시맨틱 통합 | 문서 청킹, 벡터 임베딩, 최근접 이웃 검색 |
그래프 통합 | 에이전트 수명 주기 콜백 ( | 연구 팬아웃의 전용 함수 노드 ( |
문서 청크 처리 및 임베딩
RAG Engine은 텍스트를 시맨틱 구절로 나누고 관리형 데이터베이스에 벡터를 저장하여 문서를 색인화합니다.
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)))
- 청크 크기: 겹치는 토큰 20개와 함께 토큰 120개로 구성됩니다. 이렇게 하면 단락당 2~3개의 댓글이 캡처되어 각 벡터가 관련 없는 의견 전반에 걸쳐 의미를 희석하지 않고 일관된 감정을 나타냅니다.
- 임베딩 모델:
text-embedding-005는 텍스트를 고차원 벡터로 변환합니다. 쿼리가 제출되면 모델은 쿼리를 벡터로 변환하고 의미론적 거리를 기반으로 가장 가까운 일치 항목을 찾습니다. 양말을 지키는 작은 용에 관한 댓글은 정확한 키워드 중복 없이 마법의 생물에 관한 프롬프트와 일치합니다.
RAG 코퍼스 설정
워크벤치 버튼 또는 터미널 명령어를 사용하여 코퍼스를 초기화합니다.
- 코퍼스 만들기:
python -m agent.platform.ragruns/ragcorpus.json에 리소스 ID를 기록합니다. - 댓글 업로드 및 색인 생성: 청크 구성으로
agent/comments.md를 업로드하고 색인 생성이 완료될 때까지 기다립니다. - 코퍼스 쿼리: 댓글과 정확한 단어를 공유하지 않는 쿼리로 유사성 검색을 테스트합니다 (예: '작은 마법 생물'을 쿼리하여 용에 관한 댓글을 검색).
검색 노드 (7B)
워크벤치에서 세 번째 리더 (7B)로 이동합니다. stage5_rag/agent.py를 엽니다.
그래프 노드로 가져오기
시청자 의견은 워크플로 전반에서 공유되는 연구 데이터를 나타냅니다. 개인 크리에이터 메모리와 달리 시청자 감정은 트렌드 및 백로그 데이터와 함께 join_research에 직접 제공됩니다. 따라서 함수 노드로 구현됩니다.
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는 사용자의 초기 아이디어를 추출하고 RAG Engine 코퍼스에 대해 벡터 쿼리를 실행합니다. 검색된 댓글을 Event(output=...) 페이로드로 내보냅니다.
직접 수정: 세 번째 리더를 팬아웃에 연결
stage5_rag/agent.py에서 join_research에 들어가는 세 번째 병렬 브랜치로 read_feedback을 추가하도록 edges를 업데이트합니다.
(START, read_backlog, join_research),
(START, read_feedback, join_research),
join_research는 JoinNode이므로 모든 수신 브랜치를 동기화하고 scan_trends, read_backlog, read_feedback이 모든 이벤트를 내보낼 때까지 기다린 후 집계된 번들을 다운스트림으로 전달합니다.
예상되는 상황 및 이유
워크벤치에서 워크플로를 실행합니다.
- 아이디어 프롬프트 (예: '주방 카운터를 지키는 소형 용')를 제출합니다.
- 실행 추적에서 세 개의 리더 노드가 모두 동시에 실행되는지 확인합니다.
join_research를 관찰합니다. 이제 출력 사전에는trends,backlog,feedback가 포함됩니다.propose_directions에서 생성된 후보를 검사합니다. 모델은 시청자 댓글을 제안에 통합하고 증거 필드에서 시청자 감정을 참조합니다.- RAG 검색은 결정적 (동일한 쿼리는 동일한 댓글 구절을 반환)인 반면 생성 제안 노드는 창의적인 변형을 생성합니다.
8. Veo를 사용한 비동기 동영상 생성
VibeStudio Workbench에서 Step 8 · The video, parts (8A) and (8B)로 이동합니다.
Google Veo로 고화질 동영상을 생성하려면 렌더링당 몇 분이 걸립니다. 이 기간 동안 지연 실행을 차단하면 컴퓨팅 리소스가 낭비되고, 스레드 풀이 잠기며, 실행이 HTTP 연결 드롭아웃에 노출됩니다. 이 단계에서는 ADK의 LongRunningFunctionTool를 사용하여 동영상 렌더링을 비동기식으로 만듭니다.
장기 실행 도구 (8A)
워크벤치에서 A long-running tool (8A)로 이동합니다. stage6_video/agent.py과 agent/deliver.py을 엽니다.
동기 도구와 장기 실행 도구
표준 ADK 함수 도구는 에이전트 턴 내에서 동기식으로 실행됩니다. 모델이 도구를 호출하고, 반환 페이로드를 기다리고, 결과를 진행 중인 턴에 통합합니다.
동영상 렌더링이 단일 턴 내에 완료될 수 없습니다. 대신 render_submit은 생성 작업을 시작하고 상태가 "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"]}
LongRunningFunctionTool로 래핑하면 ADK가 "pending" 상태를 가로챕니다. 상담사의 턴이 종료되고 워크플로가 노드에서 일시중지되며 대기 중인 통화 메타데이터 (통화 ID 및 영수증 포함)가 runs/sessions.db에 기록됩니다. 실행 프로세스가 활성 네트워크 연결이나 작업자 스레드를 유지하지 않고 깔끔하게 종료됩니다.
실습 수정: 렌더링 도구 래핑
stage6_video/agent.py에서 render_submit를 LongRunningFunctionTool로 래핑하도록 render_desk를 업데이트합니다.
tools=[LongRunningFunctionTool(render_submit)])
통화 ID로 재개
범용 재개 패턴
ADK는 사람과 외부 도구 모두에 워크플로를 일시중지하고 재개하는 동일한 메커니즘을 적용합니다.
정지 트리거 | Construct 시작 | 저장된 정지 상태 | 재개 이벤트 |
사람의 결정 |
| 세션 저장소에서 입력 프롬프트 열기 | 정지 호출 ID를 전달하는 |
장기 실행 도구 |
| 세션 저장소에서 도구 호출 열기 | 정지 호출 ID를 전달하는 |
두 시나리오 모두에서 워크플로가 완전히 중지되고 일치하는 FunctionResponse가 외부 소스(사용자 인터페이스, 웹훅 또는 백그라운드 작업자)에서 도착할 때만 재개됩니다.
직접 수정: 배송 응답 완료
agent/deliver.py에서 재개 FunctionResponse 부분을 구성합니다.
part = Part(function_response=FunctionResponse(
id=row["call_id"], name=row["name"], response=response))
전송 데몬은 동영상 파일이 생성될 때까지 Veo를 폴링한 다음 이 FunctionResponse을 세션에 디스패치합니다. ADK는 호출 ID를 일치시키고 다음 노드에서 워크플로를 직접 재개합니다. 완료된 노드는 다시 실행되지 않으며 에이전트는 다른 생성 턴을 수행하지 않습니다.
.env에서 STUDIO_REAL_VIDEO=0를 설정하면 모의 렌더링이 사용 설정됩니다. start는 즉시 테스트 영수증을 반환하고 check는 청구 가능한 Veo API 호출 없이 5초 내에 완료를 시뮬레이션합니다.
파이프라인 통합 (8B)
워크벤치에서 render_desk in the graph (8B)로 이동합니다. stage6_video/agent.py를 엽니다.
파이프라인의 터미널 노드는 store_video입니다. runs/state.json (전송 프로세스에서 기록한 위치)에서 완료된 렌더링 정보를 읽고 동영상 URL과 생성 상태를 공유 세션 상태에 커밋합니다.
실습 편집: 전체 동영상 파이프라인 연결
stage6_video/agent.py에서 render_desk와 store_video를 추가하도록 edges을 업데이트합니다.
(quarantine, scripter),
(scripter, render_desk, store_video)])
예상되는 상황 및 이유
워크벤치에서 비동기 생성 흐름을 테스트합니다.
- 후보 선택 및 스크립트 생성을 통해 워크플로를 실행합니다.
render_desk에서 에이전트가render_submit를 호출하는 것을 관찰합니다.- 워크플로가 즉시 일시중지됩니다. 워크벤치 또는 ADK 웹에서 대기 상태를 관찰합니다. 세션이 열린 호출 ID를 보유하고 백그라운드 프로세스가 리소스를 사용하지 않습니다.
- 워크벤치 콘솔 또는 터미널에서 전송 데몬을 실행합니다.
python -m agent.deliver - ADK Web에서 세션을 새로고침합니다.
store_video에서 실행이 재개되고, 동영상 URL이 세션 상태에 커밋되고, 워크플로가 완료됩니다.
9. Cloud Run에 배포
VibeStudio Workbench에서 Step 9 · Deploy로 이동합니다.
전용 샌드박스에서 파이프라인의 각 구성요소를 개발하고 확인했습니다. 이 단계에서는 전체 프로덕션 파이프라인을 어셈블하고 Google Cloud Run에 배포합니다.
ADK 러너
개발에서는 adk web가 그래프를 오케스트레이션했습니다. 프로덕션에서 애플리케이션은 ADK의 Runner 클래스를 사용하여 워크플로를 호스팅합니다.
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: 워크플로 실행을 유도하여 노드가 실행될 때 순차적으로 이벤트를 생성하고 세션 서비스에 업데이트를 유지합니다.- 통합 재개:
direction_gate에서의 사용자 결정과 Veo의 완료된 동영상 전송 모두run_async에 제출된 동일한FunctionResponse객체를 통해 실행을 재개합니다.
프로덕션 애플리케이션 아키텍처
vibestudio/의 프로덕션 애플리케이션은 전체 파이프라인을 통합합니다.
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
- 단일 이벤트 스트림: FastAPI 백엔드는 단일 서버 전송 이벤트 (SSE) 스트림에 이벤트를 게시합니다. React 프런트엔드는 그래프 진행 상황을 실시간으로 시각화하고 상태 손실 없이 지연된 연결을 처리합니다.
- 분리된 실행: 애플리케이션이 이벤트 루프를 관리합니다. 워크플로 그래프는 실행 로직에만 초점을 맞추고 프런트엔드 인터페이스는 인식하지 못합니다.
agent/graph.py의 전체 워크플로 에지 목록은 이 Codelab 전반에 걸쳐 빌드된 모든 아키텍처 패턴을 결합합니다.
(START, scan_trends, join_research),
(START, read_backlog, join_research),
(START, read_feedback, join_research),
(join_research, propose_directions, direction_gate,
persist_direction, policy_check),
(policy_check, {"OK": scripter, "BLOCK": quarantine}),
(quarantine, scripter),
(scripter, render_desk, store_video),
Cloud Run에 배포
Google Cloud Run은 자동 확장, 요청 라우팅, 통합 컨테이너 빌드를 지원하는 서버리스 호스팅을 제공합니다.
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=...
- 컨테이너 빌드:
gcloud run deploy --source는vibestudio/디렉터리를 패키징하고, Cloud Build를 사용하여 컨테이너 이미지를 빌드하고, 단일 작업으로 서비스를 배포합니다. - 세션 어피니티: 동일한 사용자의 요청을 동일한 컨테이너 인스턴스로 전달하여 반복 단계에서 로컬 세션 상태를 유지합니다.
- 관측 가능성: Cloud Trace 통합은 모든 노드, LLM 호출, 도구 실행에 대해 분산 스팬을 기록하며, Google Cloud 콘솔의 Trace 탐색기에서 액세스할 수 있습니다.
워크벤치에서 배포 버튼을 클릭하여 배포 스크립트를 실행합니다. 빌드가 완료되면 터미널에 라이브 서비스 URL이 표시됩니다.
10. 요약
VibeStudio Workbench에서 Step 10 · Summary로 이동하여 완료된 아키텍처를 검토합니다.
단계 | 아키텍처 및 개념 | 구현 패턴 |
단일 프롬프트 | 단일 프롬프트, 함수 도구, 순차적 채팅 루프 |
|
에이전트형 워크플로 기본사항 | 그래프 워크플로, 병렬 연구, 스키마 출력, 사람 게이트 |
|
상태 및 라우터 | 공유 세션 상태, 매개변수 바인딩, 결정적 라우팅, 태스크 에이전트 |
|
메모리 뱅크 | 사용자 수준 장기 메모리, 시맨틱 통합, 수명 주기 후크 |
|
RAG Engine | 잠재고객 댓글, 시맨틱 임베딩을 통한 문서 검색 |
|
Veo를 사용한 비동기 동영상 생성 | 장기 실행 도구, 대기 중인 영수증, 외부 전송 데몬 |
|
Cloud Run에 배포 | 프로그래매틱 오케스트레이션, 서버 전송 이벤트, 서버리스 컨테이너 |
|
핵심 아키텍처 원칙
- 대신 일시중지: 워크플로는 사람의 입력 (
RequestInput) 또는 장기 실행 작업 (LongRunningFunctionTool)을 위해 깔끔하게 일시중지됩니다. 프로세스는 스레드 또는 네트워크 소켓에서 유휴 상태로 대기하지 않습니다. - 범용 재개: 모든 정지는 정지된 노드의 호출 ID를 전달하는 단일
function_response를 통해 동일한 메커니즘으로 재개됩니다. - 분리된 상태 관리: 노드는 긴밀하게 결합된 중간 페이로드 대신 명명된 세션 상태 키와 매개변수 바인딩을 통해 데이터를 공유합니다.
- 생성 비용 전 결정적 라우팅: 규칙 기반 라우터와 정규식 필터는 생성 모델이 실행되기 전에 토큰 비용이 0인 정책을 평가합니다.
- 관심사 분리: 개별 에이전트와 관련된 컨텍스트는 수명 주기 콜백에 속하고 공유 데이터 종속 항목은
