Google ADK 멀티 에이전트: Workflow Agent · 구조화 출력 · Callback으로 만드는 YouTube Shorts 메이커

graph TD
    A["ShortsProducerAgent<br>(LlmAgent · 오케스트레이터)"] -->|AgentTool| B["ContentPlannerAgent<br>(LlmAgent · output_schema)"]
    A -->|AgentTool| C["AssetGeneratorAgent<br>(ParallelAgent)"]
    A -->|AgentTool| D["VideoAssemblerAgent<br>(LlmAgent + assemble_video)"]
    C --> E["ImageGeneratorAgent<br>(SequentialAgent)"]
    C --> F["VoiceGeneratorAgent<br>(LlmAgent + generate_narrations)"]
    E --> G["PromptBuilderAgent<br>(LlmAgent · output_schema)"]
    E --> H["ImageBuilder<br>(LlmAgent + generate_images)"]

# asset_generator/agent.py — 이미지·음성을 병렬 실행
from google.adk.agents import ParallelAgent

asset_generator_agent = ParallelAgent(
    name="AssetGeneratorAgent",
    description=ASSET_GENERATOR_DESCRIPTION,
    sub_agents=[image_generator_agent, voice_generator_agent],
)

# image_generator/agent.py — 프롬프트 빌더 → 이미지 빌더 순차 실행
from google.adk.agents import SequentialAgent

image_generator_agent = SequentialAgent(
    name="ImageGeneratorAgent",
    sub_agents=[prompt_builder_agent, image_builder_agent],
)

# content_planner/agent.py
from pydantic import BaseModel, Field

class SceneOutput(BaseModel):
    id: int = Field(description="Scene ID number")
    narration: str = Field(description="Narration text for the scene")
    visual_description: str = Field(description="Detailed description for image generation")
    embedded_text: str = Field(description="Text overlay for the image")
    embedded_text_location: str = Field(description="Where to position the text (e.g., 'top center')")
    duration: int = Field(description="Duration in seconds for this scene")

class ContentPlanOutput(BaseModel):
    topic: str
    total_duration: int = Field(description="Total video duration in seconds (max 20)")
    scenes: list[SceneOutput]

content_planner_agent = Agent(
    name="ContentPlannerAgent",
    description=CONTENT_PLANNER_DESCRIPTION,
    instruction=CONTENT_PLANNER_PROMPT,
    model=MODEL,
    output_schema=ContentPlanOutput,       # 출력 구조 강제
    output_key="content_planner_output",   # 결과를 state에 저장
)

# prompt_builder/prompt.py (발췌)
PROMPT_BUILDER_PROMPT = """
You are the PromptBuilderAgent ...
## Your Task:
Take the structured content plan: {content_planner_output} and create
optimized vertical image generation prompts for each scene ...
"""

# image_builder/tools.py (발췌)
async def generate_images(tool_context: ToolContext):
    # state에서 최적화된 프롬프트를 가져온다
    optimized_prompts = tool_context.state.get("prompt_builder_output").get("optimized_prompts")
    # 이미 만든 이미지는 재생성하지 않도록 기존 아티팩트 확인 (멱등성)
    existing_artifacts = await tool_context.list_artifacts()

    for prompt in optimized_prompts:
        filename = f"scene_{prompt['scene_id']}_image.jpeg"
        if filename in existing_artifacts:
            continue
        image = client.images.generate(
            model="gpt-image-1.5", prompt=prompt["enhanced_prompt"],
            n=1, quality="low", output_format="jpeg", size="1024x1536",
        )
        image_bytes = base64.b64decode(image.data[0].b64_json)  # base64 디코딩
        artifact = types.Part(inline_data=types.Blob(mime_type="image/jpeg", data=image_bytes))
        await tool_context.save_artifact(filename=filename, artifact=artifact)
    ...

# voice_generator/tools.py (발췌)
async def generate_narrations(tool_context, voice, voice_instructions):
    existing_artifacts = await tool_context.list_artifacts()
    for instruction in voice_instructions:
        filename = f"scene_{instruction['scene_id']}_narration.mp3"
        if filename in existing_artifacts:
            continue
        with client.audio.speech.with_streaming_response.create(
            model="gpt-4o-mini-tts", voice=voice,
            input=instruction["input"], instructions=instruction["instructions"],
        ) as response:
            audio_data = response.read()
        artifact = types.Part(inline_data=types.Blob(mime_type="audio/mpeg", data=audio_data))
        await tool_context.save_artifact(filename=filename, artifact=artifact)
    ...

# video_assembler/tools.py (발췌)
async def assemble_video(tool_context: ToolContext) -> str:
    # state에서 장면 타이밍을 읽고
    scenes = tool_context.state.get("content_planner_output", {}).get("scenes", [])
    # 저장된 모든 아티팩트(scene_N_image.jpeg / scene_N_narration.mp3)를 불러온다
    existing_artifacts = await tool_context.list_artifacts()
    ...
    image_artifact = await tool_context.load_artifact(filename=image_name)
    ...
    # FFmpeg로 이미지+오디오를 장면 길이에 맞춰 합성 → 1080x1920 MP4
    subprocess.run(ffmpeg_cmd, capture_output=True, text=True, check=True)
    # 결과 영상을 최종 아티팩트로 저장
    await tool_context.save_artifact(filename="youtube_short_final.mp4", artifact=artifact)

# agent.py
def before_model_callback(callback_context: CallbackContext, llm_request: LlmRequest):
    last_message = llm_request.contents[-1]
    if last_message.role == "user":
        text = last_message.parts[0].text
        if "hummus" in text:   # 차단 키워드 예시
            return LlmResponse(content=types.Content(
                parts=[types.Part(text="Sorry I can't help with that.")], role="model",
            ))
    return None  # None이면 정상 동작(모델 호출 진행)

shorts_producer_agent = Agent(
    name="ShortsProducerAgent",
    ...
    before_model_callback=before_model_callback,
)