大家好！歡迎來到「Build on Google AI」工程挑戰的第 29 天。

經過了將近一個月的奮戰，我們從基礎的 API 呼叫，一路打怪升級：我們解決了 Token 爆炸、馴服了 LLM 輸出強型別 JSON、設計了防止 Agent 吵架的無窮迴圈機制、加入了人類授權 (HITL) 煞車，甚至優化了 Cloud Run 的冷啟動效能。

在企業級軟體工程中，單一技術點的突破只是起點，「系統整合 (System Integration)」才是決勝的關鍵。

今天，是我們程式碼實戰的「大結局」。我們不講新套件，我們要將這 28 天的心血結晶，用 Google ADK 的 Workflow 織成一張完美的網。我們要展示：如何只給系統「一句模糊的話」，它就能在底層啟動一支跨部門 AI 團隊，自動跑完所有流程，最後端出一份精美、零邏輯錯誤的影音分鏡企劃書！

終極架構：三步走的完美閉環
我們的最終系統架構包含三個核心節點：

需求擴展 (PM Node)： 接收一句話，擴寫成包含背景、受眾、調性的具體 Spec。

創意生成與驗證 (Director Node)： 根據 Spec 產出強型別 JSON，並透過 Pydantic 驗證秒數邏輯，若出錯則自動重試 (Self-Correction)。

展示渲染 (Render Node)： 將冰冷的 JSON 物件，轉換為帶有 Emoji 與表格排版的 Markdown，直接推送到 Web 戰情室。

終極實戰程式碼 (main.py)
這份程式碼是你這一個月來的「畢業作品」。它結構清晰、職責分離，完全展現了現代 Agentic System 的架構美學。

import logging
from typing import List, Optional
from pydantic import BaseModel, Field, model_validator, ValidationError
from google.adk import Agent, Workflow

logging.basicConfig(level=logging.INFO, format='%(message)s')
logger = logging.getLogger(__name__)

# ==========================================
# 1. 領域模型與全局狀態 (Domain Models & State)
# ==========================================
class StoryboardScene(BaseModel):
    scene_number: int = Field(description="分鏡編號")
    duration_seconds: int = Field(description="該分鏡秒數")
    visual_description: str = Field(description="畫面視覺描述")
    camera_movement: str = Field(description="運鏡指示")
    voiceover: str = Field(description="旁白台詞")

class VideoStoryboard(BaseModel):
    video_title: str
    total_duration_seconds: int
    scenes: List[StoryboardScene]

    @model_validator(mode='after')
    def check_duration_sum(self) -> 'VideoStoryboard':
        calculated_sum = sum(scene.duration_seconds for scene in self.scenes)
        if calculated_sum != self.total_duration_seconds:
            raise ValueError(f"邏輯錯誤：分鏡秒數加總 ({calculated_sum}s) 必須等於總時長 ({self.total_duration_seconds}s)。")
        return self

class AppState(BaseModel):
    user_input: str = ""
    pm_spec: str = ""
    final_storyboard: Optional[VideoStoryboard] = None
    ui_markdown: str = ""

# ==========================================
# 2. 核心智能體定義 (Agents)
# ==========================================
pm_agent = Agent(
    name="pm_agent",
    model="gemini-2.5-flash",
    instruction="你是一個資深產品經理。請將使用者的簡短需求，擴寫成一份包含『目標受眾、核心訊息、視覺調性』的影片規格書。",
)

director_agent = Agent(
    name="director_agent",
    model="gemini-2.5-pro",
    instruction="你是專業導演。請根據 PM 規格書，產出精準的分鏡腳本。嚴格遵守時間限制。",
    response_schema=VideoStoryboard
)

# ==========================================
# 3. 工作流節點實作 (Node Operations)
# ==========================================
def pm_node(state: AppState) -> AppState:
    logger.info("🧠 [PM Node] 正在解析並擴寫使用者需求...")
    response = pm_agent.invoke(state.user_input)
    state.pm_spec = response.text
    return state

def director_node(state: AppState) -> AppState:
    logger.info("🎬 [Director Node] 正在進行創意分鏡與邏輯驗證...")
    current_prompt = f"請根據以下規格書製作分鏡：\n{state.pm_spec}"
    
    # 內建自我修正迴圈 (Self-Correction)
    for attempt in range(3):
        try:
            storyboard = director_agent.invoke(current_prompt)
            state.final_storyboard = storyboard
            logger.info("✅ [Director Node] 分鏡表生成且邏輯驗證通過！")
            return state
        except ValidationError as e:
            logger.warning(f"⚠️ [Director Node] 邏輯錯誤，進行第 {attempt+1} 次自我修正...")
            current_prompt += f"\n\n你上一次的產出有誤：{str(e)}。請修正秒數配置。"
            
    raise RuntimeError("Director Agent 無法產出符合邏輯的分鏡表。")

def render_node(state: AppState) -> AppState:
    logger.info("🎨 [Render Node] 正在渲染 Web 戰情室視覺化介面...")
    sb = state.final_storyboard
    md = f"## 🎬 影片企劃：{sb.video_title}\n> ⏱️ **總時長：** {sb.total_duration_seconds} 秒\n\n"
    md += "| 鏡頭 | 時長 | 🎥 運鏡 | 👁️ 畫面 | 🗣️ 旁白 |\n|:---:|:---:|:---|:---|:---|\n"
    for s in sb.scenes:
        md += f"| **#{s.scene_number}** | `{s.duration_seconds}s` | {s.camera_movement} | {s.visual_description.replace('n', '<br>')} | {s.voiceover.replace('n', '<br>')} |\n"
    state.ui_markdown = md
    return state

# ==========================================
# 4. 編織終極工作流 (The Perfect Loop)
# ==========================================
workflow = Workflow(name="Ultimate_Video_Factory", state_schema=AppState)
# 透過 add_sequence 建立單向的流水線
workflow.add_sequence(["START", pm_node, director_node, render_node, "END"])
app = workflow.compile()

# ==========================================
# 5. 一鍵啟動 (Execution)
# ==========================================
if __name__ == "__main__":
    initial_state = AppState(user_input="幫我發想一支 15 秒的 Google ADK 宣傳影片，要好笑一點，主打工程師不用再寫 Regex。")
    
    print("🚀 啟動全自動影音腳本產線...\n" + "="*50)
    final_state = app.run(initial_state)
    
    print("\n🎉 系統執行完畢！將推送到前端介面的 Markdown 如下：\n" + "="*50)
    print(final_state.ui_markdown)

為什麼這段程式碼值得你驕傲？
當你按下執行鍵，看著 Log 一條條跳出，最後印出完美的 Markdown 表格時，這不再只是一個「呼叫 LLM 的腳本」。

你建立的是一個具備確定性 (Deterministic) 業務流程的系統。PM Agent 處理了人類語言的模糊性（發散），Director Agent 在 Pydantic 的約束下交出了強型別的資料（收斂），而 Render Node 則完美地橋接了機器資料與人類視覺體驗。

這就是 Google ADK 賦予開發者的超能力：讓 AI 不再只是聊天機器人，而是真正融入企業自動化產線的核心引擎。

小結
「複雜的底層架構，是為了造就極致簡單的使用者體驗。」

今天，我們完成了這 29 天以來所有技術的總集篇，親手打造了一個端到端的完美閉環。相信跟著實作到這裡的你，已經完全具備了開發企業級 Agent 的硬核實力。