iT邦幫忙

2026 iThome 鐵人賽

DAY 5
0
Build on Google AI

將考國際證照的應用程式變成開源系列 第 5

運用 Google Skills 新推出的 ADK 2.0 在應用程式內

  • 分享至 

  • xImage
  •  

在8月5日進行了 Build Multi-Agent Systems with ADK ( https://www.skills.google/course_templates/1445/labs/629096 ) 並且實際運用的目前正在開發的應用程式上,以下是我將整個過程記載下來並且運用 Google Gemini 整理出來的筆記.

📖 學習筆記:使用 ADK 建立多智能體系統 (Build Multi-Agent Systems with ADK)

🌟 核心概念 (Key Concepts)

Google Agent Development Kit (ADK) 讓開發者能將複雜的任務拆解,交給多個「專職」的智能體 (Agents) 協作完成。比起寫一個超長、超複雜的 Prompt,多智能體架構具有易於設計表現更穩定易於維護與除錯等優點。

  1. 階層式架構 (The Hierarchical Agent Tree)
  • 系統由一個 root_agent (根智能體) 開始對話。
  • 透過設定 sub_agents,可以建立「父 (Parent)」與「子 (Sub-agent)」的關係。
  • 父智能體會根據子智能體的 description (描述) 自動判斷何時該把對話轉交給哪個子智能體。
  1. 會話狀態 (Session State)
  • 所有參與對話的智能體共用同一個 Session。
  • 透過建立特定的 Tool (如 save_attractions_to_state),可以將重要的資料(如使用者的選擇、生成的草稿)寫入 State 字典中。
  • 智能體可以在 instruction 中使用 {變數名稱?} 的語法(Key templating),直接讀取 State 裡的資料作為上下文。
  1. 工作流智能體 (Workflow Agents)
  • 用於讓智能體在背景「自動連續工作」,不需等待使用者介入。
  • SequentialAgent (循序智能體): 讓子智能體依序執行,一個接一個(例如:研究員查完資料 -> 編劇寫劇本 -> 存檔)。
  • LoopAgent (迴圈智能體): 讓子智能體重複循環執行,直到達到指定次數或被觸發退出(例如:研究員 -> 編劇 -> 評論家給建議 -> 再次研究...)。
  • ParallelAgent (平行智能體): 讓多個子智能體「同時」平行處理不同任務,節省時間(例如:同時進行「票房預測」與「選角建議」)。

💻 專案一:旅遊規劃助手 (涵蓋 Task 2 & 3)

學習目標: 練習設定父子智能體轉接 (Transfers),以及使用 Session State 記憶使用者選擇的景點。

核心程式碼 (parent_and_subagents/agent.py)

這個專案包含一個負責引導的 steering,以及兩個子智能體:負責構思國家的 travel_brainstormer 與負責規劃景點的 attractions_planner

import os
import sys
import logging
sys.path.append("..")
from callback_logging import log_query_to_model, log_model_response
from dotenv import load_dotenv
import google.cloud.logging
from google.adk import Agent
from google.adk.models import Gemini
from google.genai import types
from typing import Optional, List, Dict
from google.adk.tools.tool_context import ToolContext
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from adk_utils.plugins import Graceful429Plugin
from google.adk.apps.app import App

load_dotenv()
cloud_logging_client = google.cloud.logging.Client()
cloud_logging_client.setup_logging()
RETRY_OPTIONS = types.HttpRetryOptions(initial_delay=1, max_delay=3, attempts=30)

# ==========================================
# 工具定義 (Tools) - 用於操作 Session State
# ==========================================
def save_attractions_to_state(tool_context: ToolContext, attractions: List[str]) -> dict[str, str]:
    """將景點清單儲存到 state['attractions'] 中"""
    existing_attractions = tool_context.state.get("attractions", [])
    tool_context.state["attractions"] = existing_attractions + attractions
    return {"status": "success"}

# ==========================================
# 智能體定義 (Agents)
# ==========================================
attractions_planner = Agent(
    name="attractions_planner",
    model=Gemini(model=os.getenv("MODEL"), retry_options=RETRY_OPTIONS),
    description="Build a list of attractions to visit in a country.",
    instruction="""
        - Provide the user options for attractions to visit within their selected country.
        - When they reply, use your tool to save their selected attraction and then provide more possible attractions.
        - If they ask to view the list, provide a bulleted list of {attractions?} and then suggest some more.
        """,
    tools=[save_attractions_to_state] # 賦予存取 State 的工具
)

travel_brainstormer = Agent(
    name="travel_brainstormer",
    model=Gemini(model=os.getenv("MODEL"), retry_options=RETRY_OPTIONS),
    description="Help a user decide what country to visit.",
    instruction="""
        Provide a few suggestions of popular countries for travelers.
        Help a user identify their primary goals of travel: adventure, leisure, learning, shopping, or viewing art
        Identify countries that would make great destinations based on their priorities.
        """
)

root_agent = Agent(
    name="steering",
    model=Gemini(model=os.getenv("MODEL"), retry_options=RETRY_OPTIONS),
    description="Start a user on a travel adventure.",
    instruction="""
        Ask the user if they know where they'd like to travel or if they need some help deciding.
        If they need help deciding, send them to 'travel_brainstormer'.
        If they know what country they'd like to visit, send them to the 'attractions_planner'.
        """,
    sub_agents=[travel_brainstormer, attractions_planner] # 設定子智能體
)

# ... (Plugin 與 App 註冊省略,請參考原始檔) ...


🎬 專案二:電影提案產生器 (涵蓋 Task 4, 5, 6)

學習目標: 組合使用 SequentialAgentLoopAgentParallelAgent,打造一個全自動的「編劇室」工作流。

工作流程解構:

  1. root_agent (greeter): 詢問使用者想要哪位歷史人物,將輸入存入 State。
  2. SequentialAgent (film_concept_team): 控制整體大流程 (Loop -> Parallel -> 存檔)。
  • 2a. LoopAgent (writers_room): 迴圈執行(研究員查維基 -> 編劇寫草稿 -> 評論家審查並決定是否跳出迴圈)。
  • 2b. ParallelAgent (preproduction_team): 同時執行(預測票房 + 尋找選角)。
  • 2c. 寫檔智能體 (file_writer): 將 State 中的所有產出統整並寫入實體檔案。

核心程式碼 (workflow_agents/agent.py)

(此為最終完成版,包含了 Task 4 到 Task 6 的所有邏輯)

import os
import logging
import google.cloud.logging
import sys
sys.path.append("..")
from callback_logging import log_query_to_model, log_model_response
from dotenv import load_dotenv

from google.adk import Agent
from google.adk.agents import SequentialAgent, LoopAgent, ParallelAgent
from google.adk.tools.tool_context import ToolContext
from google.adk.tools.langchain_tool import LangchainTool
from google.adk.tools import exit_loop # 匯入跳出迴圈工具
from google.adk.models import Gemini
from google.genai import types

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from adk_utils.plugins import Graceful429Plugin
from google.adk.apps.app import App
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper

# ... (初始化與 Tools: append_to_state, write_file 略,見原始檔) ...

# ==========================================
# Agents: 基礎工作單元
# ==========================================
file_writer = Agent(
    name="file_writer",
    model=Gemini(model=model_name, retry_options=RETRY_OPTIONS),
    description="Creates marketing details and saves a pitch document.",
    instruction="""
    INSTRUCTIONS:
    - Create a marketable, contemporary movie title...
    - Use your 'write_file' tool to create a new txt file...
    - For the 'content' to write, include :
        - The PLOT_OUTLINE
        - The BOX_OFFICE_REPORT
        - The CASTING_REPORT

    PLOT_OUTLINE: { PLOT_OUTLINE? }
    BOX_OFFICE_REPORT: { box_office_report? }
    CASTING_REPORT: { casting_report? }
    """,
    tools=[write_file],
)

screenwriter = Agent(
    name="screenwriter",
    model=Gemini(model=model_name, retry_options=RETRY_OPTIONS),
    description="As a screenwriter, write a logline and plot outline...",
    instruction="""
    Your goal is to write a logline and three-act plot outline...
    - Use the 'append_to_state' tool to write your logline...
    PLOT_OUTLINE: { PLOT_OUTLINE? }
    RESEARCH: { research? }
    CRITICAL_FEEDBACK: { CRITICAL_FEEDBACK? }
    """,
    tools=[append_to_state],
)

researcher = Agent(
    name="researcher",
    model=Gemini(model=model_name, retry_options=RETRY_OPTIONS),
    description="Answer research questions using Wikipedia.",
    instruction="""
    PROMPT: { PROMPT? }
    PLOT_OUTLINE: { PLOT_OUTLINE? }
    CRITICAL_FEEDBACK: { CRITICAL_FEEDBACK? }
    ... (使用 wikipedia 工具並儲存到 state) ...
    """,
    tools=[LangchainTool(tool=WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper(), handle_tool_error=True)), append_to_state],
)

critic = Agent(
    name="critic",
    model=Gemini(model=model_name, retry_options=RETRY_OPTIONS),
    description="Reviews the outline so that it can be improved.",
    instruction="""
    If the PLOT_OUTLINE does a good job... exit the writing loop with your 'exit_loop' tool.
    If significant improvements can be made, use the 'append_to_state' tool to add your feedback...
    PLOT_OUTLINE: { PLOT_OUTLINE? }
    RESEARCH: { research? }
    """,
    tools=[append_to_state, exit_loop]
)

box_office_researcher = Agent(
    name="box_office_researcher",
    model=Gemini(model=model_name, retry_options=RETRY_OPTIONS),
    description="Considers the box office potential of this film",
    instruction="Write a report on the box office potential... based on PLOT_OUTLINE: { PLOT_OUTLINE? }",
    output_key="box_office_report"
)

casting_agent = Agent(
    name="casting_agent",
    model=Gemini(model=model_name, retry_options=RETRY_OPTIONS),
    description="Generates casting ideas for this film",
    instruction="Generate ideas for casting... based on PLOT_OUTLINE: { PLOT_OUTLINE? }",
    output_key="casting_report"
)

# ==========================================
# Workflow Agents: 組合與控制流程
# ==========================================

# 1. 迴圈智能體:反覆研究、寫作與評論
writers_room = LoopAgent(
    name="writers_room",
    description="Iterates through research and writing to improve a movie plot outline.",
    sub_agents=[researcher, screenwriter, critic],
    max_iterations=5,
)

# 2. 平行智能體:同時進行票房分析與選角
preproduction_team = ParallelAgent(
    name="preproduction_team",
    sub_agents=[box_office_researcher, casting_agent]
)

# 3. 循序智能體:大流程 (編劇室 -> 前置作業團隊 -> 存檔)
film_concept_team = SequentialAgent(
    name="film_concept_team",
    description="Write a film plot outline and save it as a text file.",
    sub_agents=[writers_room, preproduction_team, file_writer],
)

# ==========================================
# Root Agent: 對話起點
# ==========================================
root_agent = Agent(
    name="greeter",
    model=Gemini(model=model_name, retry_options=RETRY_OPTIONS),
    description="Guides the user in crafting a movie plot.",
    instruction="""
    - Ask them for a historical figure...
    - When they respond, use the 'append_to_state' tool to store the user's response in the 'PROMPT' state key and transfer to the 'film_concept_team' agent.
    """,
    tools=[append_to_state],
    sub_agents=[film_concept_team],
)
# ... (Plugin 與 App 註冊省略) ...


🛠️ 實用指令備忘錄 (Cheatsheet)

  • 啟動終端機對話 (CLI Mode):
adk run <資料夾名稱>

  • 啟動 Web UI (具有熱更新功能):
adk web --allow_origins "regex:https://.*\.cloudshell\.dev" --reload_agents

(若 port 8000 被佔用,可加上 --port 8001 等指定新的通訊埠)

  • 在 Web UI 中觀察狀態與除錯:

  • 點擊智能體回覆旁邊的圖示,可以進入 Event View 觀看智能體呼叫的樹狀圖 (Graph)、請求 (Request) 與回應 (Response)。

  • 切換到左側欄位的 State 頁籤,可以隨時監控目前 Session 變數中儲存了哪些資料。

  • 將 ADK 2.0 實際運用在我的應用程式上的過程記錄(摘錄)
    MockExam-CCNA × ADK 2.0 整合實作計畫.md

MockExam-CCNA × ADK 2.0 整合實作計畫

將 Google Agent Development Kit (ADK) 2.0 的 Workflow Runtime、Graph Execution Engine 與 Multi-Agent 架構整合至 MockExam-CCNA 應用程式,涵蓋 AI 備考助教、RAG 智慧知識庫、考題解析、弱點分析、子網路計算及 Admin 考題協作等所有 AI 功能。


設計決策總覽

# 設計決策 選定方案
1 部署架構 雲端 ADK 2.0 微服務 + 保留端側 Gemma 4 離線 Fallback
2 雲端平台 Google Cloud Run + ADK Python 2.0(Serverless 按用量計費)
3 Agent 架構 5 個專職 Agent + 1 個 Coordinator Agent
4 RAG 向量資料庫 複用現有 Supabase pgvector (768 維)
5 Workflow 設計 完整多步驟 Graph Workflow(5 節點考題解析流程)
6 State 管理 三層式 State(Session / User / App)
7 HITL 範圍 僅 Admin 考題 AI 分類後的人工審核
8 通訊協定 REST API + SSE 串流混合式
9 MCP 工具 Supabase pgvector 資料庫工具(優先)
10 離線策略 連網走 ADK 2.0;離線走現有 Gemma 4 端側推理
11 LLM 模型 混合模型 + 使用者可自選模型(參考 ai.google.dev 模型清單)
12 API 認證 保留自帶 Key + Firebase Auth Token 後端代理雙軌
13 遷移策略 第一階段:RAG + 考題解析;第二階段:弱點分析/子網路/Admin
14 可觀測性 ADK Graph Event Tracing + Cloud Logging + Crashlytics + Shake

系統架構圖

graph TB
    subgraph "Flutter 前端 (Android & Web)"
        UI["Flutter UI Layer"]
        GLS["GemmaLocalService\n(離線 Gemma 4 Fallback)"]
        RAG_LOCAL["本地 RAG\n(Inverted Index)"]
        PS["PowerSync SQLite\n(離線題庫)"]
        ADK_CLIENT["ADK API Client\n(REST + SSE)"]
    end

    subgraph "Google Cloud Run (ADK 2.0 後端)"
        COORD["🎯 Coordinator Agent\n(路由分發 + 模型選擇器)"]
        
        subgraph "5 個專職 Agent"
            RAG_AGENT["📚 RAG 檢索 Agent\n(Gemini Flash)"]
            EXAM_AGENT["📝 題目解析 Agent\n(Gemini Pro / 使用者選擇)"]
            WEAK_AGENT["📊 弱點分析 Agent\n(Gemini Flash)"]
            SUBNET_AGENT["🔢 子網路計算 Agent\n(Gemini Flash)"]
            ADMIN_AGENT["⚙️ 考題編輯協作 Agent\n(Gemini Flash + HITL)"]
        end

        subgraph "ADK 2.0 Graph Workflows"
            WF_EXAM["考題解析 Workflow\n(5 節點 Graph)"]
            WF_RAG["RAG 問答 Workflow\n(3 節點 Graph)"]
            WF_ADMIN["考題分類 Workflow\n(含 HITL 審核節點)"]
        end

        subgraph "ADK 2.0 State Management"
            SS["Session State\n(當次作答)"]
            US["User State\n(弱點/進度/歷史)"]
            AS["App State\n(全域設定/模型清單)"]
        end
    end

    subgraph "外部資料源"
        SUPA["Supabase\n(pgvector 768D + Questions)"]
        RTDB["Firebase RTDB\n(examsMetadata + approvedKeys)"]
        AUTH["Firebase Auth\n(身份驗證 Token)"]
    end

    UI --> ADK_CLIENT
    UI -.->|離線時| GLS
    UI -.->|離線時| RAG_LOCAL
    UI -.->|離線時| PS
    
    ADK_CLIENT -->|REST + SSE| COORD
    COORD --> RAG_AGENT
    COORD --> EXAM_AGENT
    COORD --> WEAK_AGENT
    COORD --> SUBNET_AGENT
    COORD --> ADMIN_AGENT

    RAG_AGENT -->|MCP Tool| SUPA
    EXAM_AGENT --> WF_EXAM
    RAG_AGENT --> WF_RAG
    ADMIN_AGENT --> WF_ADMIN
    
    WF_EXAM --> SS
    WEAK_AGENT --> US
    COORD --> AS

    ADK_CLIENT -->|Firebase Auth Token| AUTH

第一階段:RAG 知識庫 + 考題解析 ADK 2.0 整合

組件一:Cloud Run ADK 2.0 後端服務

[NEW] adk-backend/main.py

  • ADK 2.0 後端入口,使用 FastAPI 包裝
  • 註冊所有 Agent 與 Workflow
  • 配置 Cloud Run 服務端口與健康檢查
  • 整合 Firebase Auth Token 驗證中間件

[NEW] adk-backend/agents/coordinator_agent.py

  • Coordinator Agent 主路由
  • 根據請求類型分發至對應的專職 Agent
  • 管理模型選擇器邏輯(讀取 App State 中的可用模型清單)
  • Key Templating Prompt 範例:
你是 CCNA 認證備考系統的總導師。
【學員資訊】姓名:{user:display_name?},語系:{user:locale?}
【弱點 Domain】{user:weak_domains?}
【當前模式】{session:mode?}
請將學員的問題路由至最適合的專家 Agent。

[NEW] adk-backend/agents/rag_search_agent.py

  • RAG 檢索 Agent,使用 Gemini Flash
  • 透過 MCP Tool 連接 Supabase pgvector 進行語意向量檢索
  • 支援多模態圖表標記識別([多模態圖表/拓撲圖教材]
  • Key Templating Prompt 範例:
你是 CCNA 教材知識庫搜尋專家。
【使用者查詢】{session:user_query?}
【語系約束】{user:terminology_guidelines?}
請從知識庫中檢索最相關的講義片段。

[NEW] adk-backend/agents/exam_analysis_agent.py

  • 題目解析 Agent,使用 Gemini Pro 或使用者自選模型
  • 支援 5 種解析模式:基礎解析、從考題學英文、說文解字、刷題高手、心智圖
  • Key Templating Prompt 範例:
你是 CCNA 認證考試的資深分析師。
【考題】{session:question_text?}
【學員選擇】{session:user_selected_option?}
【正確答案】{session:correct_option?}
【解析模式】{session:analysis_mode?}
【RAG 參考資料】{session:rag_retrieved_context?}
【術語規範】{user:terminology_guidelines?}

[NEW] adk-backend/tools/supabase_vector_tool.py

  • MCP Tool:Supabase pgvector 向量檢索
  • 接受查詢文字,呼叫 Supabase match_documents RPC
  • 返回 Top-K 相關講義片段與分數

[NEW] adk-backend/tools/subnet_calculator_tool.py

  • 確定性 Tool(非 AI):子網路計算
  • 移植 subnet_validator_service.dart 的計算邏輯至 Python
  • IPv4 轉整數、CIDR 遮罩計算、子網路劃分校驗

組件二:ADK 2.0 Graph Workflow 定義

[NEW] adk-backend/workflows/exam_analysis_workflow.py

考題解析 5 節點 Graph Workflow:

graph TD
    N1["Node 1: 答案比對\n(確定性節點)"]
    N2["Node 2: RAG 檢索\n(RAG Agent)"]
    N3["Node 3: 弱點分析\n(讀取 User State)"]
    N4["Node 4: AI 解析生成\n(Exam Agent + Key Templating)"]
    N5["Node 5: 推薦下一題\n(確定性節點 + User State 更新)"]

    N1 -->|答對| N4_LITE["Node 4-Lite: 簡短備註"]
    N1 -->|答錯| N2
    N2 --> N3
    N3 --> N4
    N4 --> N5
    N4_LITE --> N5
  • Node 1 (確定性):比對使用者答案與正確答案
  • Node 2 (AI):呼叫 RAG Agent 檢索相關講義
  • Node 3 (確定性):從 User State 讀取歷史弱點資料,計算 Domain 權重
  • Node 4 (AI):呼叫 Exam Agent 使用 Key Templating 生成解析
  • Node 5 (確定性):更新 User State 弱點分佈,推薦下 3 題同類題目

[NEW] adk-backend/workflows/rag_qa_workflow.py

RAG 問答 3 節點 Graph Workflow:

graph TD
    R1["Node 1: 查詢預處理\n(確定性:斷詞/CJK)"]
    R2["Node 2: 向量檢索\n(MCP Tool: pgvector)"]
    R3["Node 3: 答案生成\n(RAG Agent + 串流 SSE)"]

    R1 --> R2
    R2 --> R3

組件三:ADK 2.0 Session State 與 Key Templating 配置

[NEW] adk-backend/state/state_schema.py

三層式 State Schema 定義:

# Session State (每次作答對話)
session_state = {
    "session:mode": "exam_analysis",          # exam_analysis | rag_qa | mock_exam
    "session:question_text": "",               # 當前考題題幹
    "session:question_domain": "",             # Domain 1~6
    "session:user_selected_option": "",        # 學員選擇
    "session:correct_option": "",              # 正確答案
    "session:analysis_mode": "basic",          # basic | english | vocab | master | mindmap
    "session:rag_retrieved_context": "",        # RAG 檢索結果
    "session:user_query": "",                  # RAG 問答查詢
}

# User State (跨對話持久化)
user_state = {
    "user:display_name": "",                   # 使用者名稱
    "user:locale": "zh_TW",                    # 語系
    "user:weak_domains": "",                   # 弱點 Domain 列表
    "user:total_questions_answered": 0,        # 累計答題數
    "user:accuracy_by_domain": {},             # 各 Domain 正確率
    "user:terminology_guidelines": "",         # 術語規範 (依語系動態注入)
    "user:preferred_model": "gemini-2.5-flash", # 使用者偏好模型
}

# App State (全域共享)
app_state = {
    "app:available_models": [],                # 可選 LLM 模型清單 (from ai.google.dev)
    "app:ccna_domains": [...],                 # CCNA Domain 1~6 定義
    "app:default_model_coordinator": "gemini-2.5-flash",
    "app:default_model_analysis": "gemini-2.5-pro",
}

組件四:Flutter 前端 ADK Client 整合

[NEW] lib/services/adk_client_service.dart

  • ADK 2.0 後端 REST + SSE 客戶端
  • 自動偵測網路狀態,連網時路由至 ADK 後端,離線時降級至 GemmaLocalService
  • Firebase Auth Token 自動附加
  • SSE 串流解析,支援逐字打出 AI 回應
  • 模型選擇器 API(從 App State 取得可用模型清單)

[MODIFY] gemma_local_service.dart

  • 修改 getExplanationTutor 方法
  • 連網時委派給 AdkClientService 呼叫 ADK 2.0 後端
  • 離線時保留現有 Gemma 4 LiteRT 端側推理邏輯
  • 對外 API 介面不變,確保 ai_tutor_panel.dart 無需修改

[MODIFY] rag_knowledge_service.dart

  • 新增 queryWithAdk() 方法
  • 連網時透過 AdkClientService 呼叫 ADK RAG Workflow (SSE 串流)
  • 離線時保留現有本地倒排索引搜尋 + Gemma 4 生成

[MODIFY] exam_screen.dart

  • 新增「AI 模型選擇器」下拉選單(從 ADK App State 動態載入可用模型)
  • 將模型選擇結果傳入 ADK Session State

第二階段:弱點分析 / 子網路計算 / Admin 協作

組件五:第二階段 Agent 與 Workflow

[NEW] adk-backend/agents/weakness_analysis_agent.py

  • 弱點分析 Agent,使用 Gemini Flash
  • 分析 User State 中的歷史作答數據
  • 計算 Domain 1~6 弱點分佈與學習建議
  • Key Templating:
【學員歷史】
- 總答題數:{user:total_questions_answered?}
- 各 Domain 正確率:{user:accuracy_by_domain?}
- 已知弱點:{user:weak_domains?}
請生成個人化學習建議報告。

[NEW] adk-backend/agents/subnet_calculator_agent.py

  • 子網路計算 Agent,使用 Gemini Flash
  • 整合確定性 Tool(subnet_calculator_tool.py
  • 先用 Tool 計算精確結果,再由 AI 生成台灣繁體中文解說

[NEW] adk-backend/agents/admin_collaboration_agent.py

  • 考題編輯協作 Agent,使用 Gemini Flash
  • 整合 HITL 節點:AI 分類 Domain 1~6 後暫停等待 Admin 確認
  • Key Templating:
請將以下考題分類至 CCNA Domain 1~6:
{session:question_to_classify?}
分類結果將提交給管理員審核。

[NEW] adk-backend/workflows/admin_classify_workflow.py

Admin 考題分類 Workflow(含 HITL):

graph TD
    AC1["Node 1: 解析考題內容\n(確定性)"]
    AC2["Node 2: AI 自動分類\n(Admin Agent)"]
    AC3["Node 3: HITL 審核\n(等待 Admin 確認)"]
    AC4["Node 4: 寫入資料庫\n(Supabase + RTDB)"]

    AC1 --> AC2
    AC2 --> AC3
    AC3 -->|Admin 確認| AC4
    AC3 -->|Admin 修改| AC2

部署與基礎建設

Cloud Run 配置

[NEW] adk-backend/Dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

[NEW] adk-backend/requirements.txt

google-adk>=2.0.0
fastapi>=0.115.0
uvicorn>=0.34.0
supabase>=2.0.0
firebase-admin>=6.0.0
google-cloud-logging>=3.0.0

[NEW] adk-backend/cloudbuild.yaml

  • Cloud Build CI/CD 配置
  • 自動部署至 Cloud Run
  • 設定環境變數(Supabase URL/Key、Firebase Config)

安全與成本控制

[!IMPORTANT]

  • Cloud Run 設定 最大並發請求數 = 80最小實例 = 0(完全 Serverless,閒置零成本)
  • 設定 請求逾時 = 300 秒(考慮 AI 串流生成時間)
  • 所有 API 端點必須驗證 Firebase Auth Token,防止未授權呼叫

驗證計畫

自動化測試

# ADK 後端 Python 測試
cd adk-backend && python -m pytest tests/ -v

# Flutter 前端測試
cd .. && flutter test

# Flutter 靜態分析
flutter analyze lib

手動驗證

  1. 連網模式驗證:確認 Flutter App 連網時,AI 解析/RAG 問答走 ADK 2.0 Cloud Run 後端
  2. 離線模式驗證:開啟飛航模式,確認 App 降級至 Gemma 4 端側推理
  3. SSE 串流驗證:確認 AI 回應逐字打出,無卡頓
  4. 模型選擇器驗證:確認使用者可切換模型,且對應 Agent 使用正確模型
  5. HITL 驗證:確認 Admin 分類流程在 AI 分類後暫停等待確認
  6. Key Templating 驗證:確認 Prompt 中的 {key?} 佔位符正確替換為 State 數值
  7. Graph Event Tracing 驗證:確認 Cloud Logging 中可追蹤每個 Workflow 節點的執行狀態

Open Questions

[!IMPORTANT]
Q1:Cloud Run 的區域 (Region) 選擇?
建議使用 asia-east1(台灣彰化)以獲得最低延遲。請確認您的 GCP 專案 exam-realtime-database-b882c 是否已在該區域開通 Cloud Run。

[!IMPORTANT]
Q2:使用者自選模型的免費額度與計費?
如果使用者選擇 Gemini Pro 等較貴的模型,是由使用者自帶 API Key 支付,還是由您的後端 Service Account 統一支付?建議:自帶 Key 使用者用自己的額度,無 Key 使用者限制只能使用 Flash 模型。

[!WARNING]
Q3:ADK Python 2.0 GA 版本相容性
ADK Python 2.0 已於 2026/05/19 GA。請確認 google-adk>=2.0.0 的 pip 套件已穩定可用,且 Graph Workflow API 未有 Breaking Changes。建議在開發初期鎖定特定版本(如 google-adk==2.0.1)。

[!NOTE]
Q4:第二階段 Timeline
第一階段(RAG + 考題解析)預計需要約 2~3 週開發。第二階段(弱點分析/子網路/Admin)可在第一階段穩定後再啟動,預計額外 1~2 週。

  • 將運用 ADK 2.0 的內容加入到所有 SDD 等文件(摘錄)

📚 繁體中文文件索引

歡迎來到 MockExam 文件套件 — 一套完整、專業級的規格文件,旨在讓任何開發者或 AI 代理都能從零開始建構一個完整的、可投入正式環境的考試練習應用程式。

🎯 什麼是 MockExam?

MockExam 是一個開源、AI 驅動、支援離線使用的考試練習平台,使用 Flutter 建構。它支援多種題型、智慧輔導、多資料庫同步、企業級安全防護,以及多語系在地化。

🗺️ 文件導航地圖

依角色快速導航

您的角色 從這裡開始 接著閱讀
🤖 AI 代理(從零建構) AI 代理指南 PRD架構設計BDD 場景
👔 產品經理 PRD UI/UX 規格BDD 場景
👩‍💻 開發者(有經驗) 架構設計 資料模型資料庫設計
🧪 QA 工程師 測試策略 BDD 場景測試矩陣
🔧 DevOps 工程師 部署指南 API 整合密鑰管理
📖 非技術人員 PRD §1-2 UI/UX 畫面規格

📁 完整文件清冊

1. 產品需求文件 (PRD)

文件 說明
PRD.md 產品願景、使用者故事、功能與非功能需求

2. 系統設計文件 (SDD)

文件 說明
01-系統總覽 技術棧、系統上下文圖、目錄結構
02-架構設計 分層架構、設計模式、導航與路由
03-資料模型 Question、User、Exam、Transaction 模型定義
04-資料庫設計 四大資料庫 Schema、安全規則、同步策略
05-安全架構 10+ 安全層、威脅模型、角色安全矩陣
06-AI 引擎 雲端/端側雙 AI、Prompt 設計、在地化策略
07-離線同步 PowerSync 架構、同步規則、衝突解決
08-付費系統 雙付款提供商、訂閱模型、交易審計
09-狀態管理 Provider/ChangeNotifier、控制器、效能防護
10-UI 組件 Widget 庫、考試組件、安全組件
11-自動化腳本 Python/PowerShell/Node.js 自動化腳本

3. 行為驅動開發 (BDD)

Feature 檔案 場景數 涵蓋範圍
auth.feature ~15 身份驗證與授權
exam-practice.feature ~20 考試練習模式
mock-exam.feature ~12 全真模擬考
wrong-questions.feature ~10 錯題消滅複習
ai-tutor.feature ~15 AI 助教對話
payment-subscription.feature ~12 付費與訂閱
admin-panel.feature ~15 管理後台
security.feature ~18 安全防護
search.feature ~8 考題搜尋
localization.feature ~8 多語系支援
tts-voice.feature ~8 語音與 TTS
notes.feature ~6 學習筆記

4. 測試規格

文件 說明
01-測試策略 測試金字塔、工具鏈、覆蓋率目標
02-單元測試 Models、Services、Controllers 測試案例
03-Widget 測試 UI 組件測試案例
04-整合測試 整合測試流程
05-E2E 測試 端對端測試場景
06-測試矩陣 功能 × 測試類型 × 平台 × 角色矩陣

5. UI/UX 設計規格

文件 說明
01-設計系統 色彩、字型、間距、組件庫
02-畫面規格 26 個畫面的佈局與互動規格
03-互動模式 導航、載入狀態、動畫、手勢

6. API 與服務整合

文件 說明
01-Firebase 設定 Firebase 專案、Auth、RTDB、Crashlytics
02-Supabase 設定 Supabase 專案、PostgreSQL、Edge Functions
03-MongoDB 設定 MongoDB Atlas、Cloudflare Worker 代理
04-PowerSync 設定 PowerSync Cloud、同步規則、Connector
05-RevenueCat 設定 RevenueCat 專案、權益、產品
06-AdMob 設定 AdMob 廣告單元、測試裝置
07-Crashlytics 設定 崩潰回報、使用者回饋
08-AI API 設定 AI API 金鑰、端側模型下載

7. 部署與運維

文件 說明
01-環境設定 開發環境、首次 Clone 設定
02-編譯配置 Gradle、NDK、R8、簽章
03-Android 部署 Play Store 軌道、版本管理
04-Web 部署 Web 編譯、Cloudflare 部署
05-密鑰管理 必要憑證、安全儲存
06-CI/CD 自動化 自動化腳本、部署流水線

8. 資料庫遷移與種子資料

文件 說明
01-Schema 建置 四大資料庫 Schema 建立步驟
02-種子資料 種子資料格式、匯入腳本
03-AI 分類 AI 考題分類流水線
04-同步操作 跨資料庫同步操作

9. AI 代理指南

文件 說明
AGENTS.md AI 代理約束、工作流程、文件導航

10. 架構決策記錄 (ADR)

ADR 決策
ADR-001 多資料庫架構
ADR-002 離線優先同步策略
ADR-003 雙 AI 引擎設計
ADR-004 雙付款提供商
ADR-005 安全優先設計
ADR-006 角色基礎存取控制

11. 開源專案配套

檔案 說明
README.md 專案介紹、功能列表、快速開始
QUICKSTART.md 5 分鐘快速上手指南
CONTRIBUTING.md 貢獻指南
CODE_OF_CONDUCT.md 社群行為準則
SECURITY.md 安全漏洞通報流程
CHANGELOG.md 版本更新日誌模板
LICENSE Apache License 2.0

📊 技術棧總覽

graph TB
    subgraph "前端 Frontend"
        A["Flutter / Dart"]
    end
    
    subgraph "資料庫 Databases"
        B["Firebase RTDB"]
        C["Supabase PostgreSQL"]
        D["MongoDB Atlas"]
        E["PowerSync SQLite"]
    end
    
    subgraph "AI 引擎 AI Engine"
        F["雲端 API Cloud API"]
        G["端側模型 On-Device Model"]
    end
    
    subgraph "服務 Services"
        H["Firebase Auth"]
        I["RevenueCat / IAP"]
        J["AdMob"]
        K["Crashlytics / Shake"]
    end
    
    A --> B & C & D & E
    A --> F & G
    A --> H & I & J & K

📝 License: This documentation is licensed under Apache License 2.0.

📅 Last Updated: 2026-08-02

現今早上(2026年8月6日)繼續開發加上新功能

Viewed main.py:30-60
Viewed main.py:60-90

這是一系列非常好的維運與架構管理問題!關於您的 Cloud Run 微服務,以下是針對您的問題所整理的詳細運作機制與說明:

1. 是否可以在 App 內由 Admin 進行開啟或關閉?使用者會知道嗎?

  • Admin 開關功能:目前我們已經具備「雙軌 AI(雲端優先、端側備援)」的底層架構,但 App 的 UI 後台尚未實作這顆「開關按鈕」。若您需要,我們只要在 Firebase Remote Config 或 RTDB 設定一個參數(例如 enable_cloud_ai: true/false),Flutter 客戶端在發送請求前先讀取此開關,一旦 Admin 將其關閉,App 就會完全阻斷向 Cloud Run 發送請求,直接交由本機端的 Gemma 4 處理。
  • 使用者感知:在目前的設計下,Cloud Run 若無法連線(或被關閉),系統會靜默且無縫地 fallback 到本機端 AI,使用者不會察覺到錯誤,只會覺得回應風格有些不同。但如果您希望讓使用者知道,我們可以在聊天介面上方加入一個小徽章(Badge),動態顯示目前是連線到 「☁️ 雲端導師」 還是 「📱 離線導師」,這樣當 Admin 關閉雲端服務時,使用者就會立刻知道。

2. Cloud Run 的費用是如何計算的?

  • 零流量=零費用 (Scale to Zero):Cloud Run 是真正的無伺服器 (Serverless) 架構。我們將 min-instances 設定為 0,這代表當半夜或沒有人在詢問 AI 時,容器會完全休眠,這時候不會產生任何主機租賃費用
  • 依請求時間計費:您只需為「實際處理請求的當下」付費,計費單位精確到 100 毫秒(包含分配的 CPU 與記憶體時間)。
  • 免費額度:Google Cloud 每個月都會提供高達 200 萬次請求、18 萬 vCPU 秒與 36 萬 GB 秒的免費額度。對於非超大型連線數的應用來說,初期的 Cloud Run 費用極低,甚至長期都可能落在免費額度內(註:呼叫 Gemini API 的 Token 費用是另外計費)。

3. 是否能傳送訊息,讓 Admin 即時知道誰正在使用 Cloud Run?

  • 絕對可以! 因為 Cloud Run 後端是我們自己掌控的 Python 微服務,我們隨時可以在裡面插入追蹤與通知邏輯
  • 實作方式
    1. 記錄到資料庫:當有人發送問題,Python 後端可將該用戶的 ID、名稱與詢問內容寫入 Firestore/RTDB 的 admin_logs 節點,Admin 隨時可以從後台面板查看。
    2. 即時推播:我們可以在 Python 端串接 Telegram Bot、Discord Webhook 或 Slack API。當有使用者觸發複雜的 RAG 查詢時,Admin 的手機會立刻收到推播通知:「使用者 OOO 剛剛發起了 CCNA 相關查詢:[問題內容]」。

4. 這個 Cloud Run 是否有安全或資安上的管控要求?

  • 這點您可以完全放心,我們已經實作了嚴格的軍規級資安管控。
  • adk-backend/main.py 的架構中,我已經在所有核心端點(分析、聊天、RAG 檢索)加入了 Depends(verify_token) 中介軟體攔截器。
  • 這意味著,發送給 Cloud Run 的每一筆 HTTP 請求,都必須強制夾帶由您 App 產生的 Firebase Auth Token (身分驗證權杖)。任何人即使知道了這個網址,只要沒有登入您的 App 並取得合法的 Token,伺服器就會直接攔截並回傳 401 Unauthorized,絕無可能被惡意盜用資源。

5. 如果要在其他地方(或其他人)使用這個 Cloud Run,該如何做?

目前該服務受到 Firebase 嚴格保護,只允許您 App 內的合法註冊用戶使用。如果您希望開放給其他平台(例如公司內網的網站、自動化腳本或是特定的商業夥伴)使用,我們有幾種做法可以擴充:

  1. 開放 API Key 機制:我們可以在 Python 後端加入一個「白名單 API Key」驗證機制,除了 Firebase Token 外,只要 Header 帶有您專屬核發的 API Key,就允許呼叫。
  2. 跨平台共用 Auth:如果是您的 Web 版或其他 App,只要同樣接上這組 Firebase 專案,它們產生的 Token 就能直接拿來呼叫這組 Cloud Run,完美達成跨平台微服務共用。

如果您想要立刻實作 「Admin 介面手動開關 Cloud Run 功能」「即時推播管理員通知功能」,隨時跟我說,我們可以馬上規劃實作!

☁️ Cloud Run 管理與即時通知實作完成報告

我們已經成功將「手動開關 Cloud Run」與「即時使用紀錄推播」的功能,無縫整合至 MockExam-CCNA 的前端與後端!

✨ 實作功能摘要

1. App 內建 Admin 控制面板

  • [NEW] CloudRunManagementScreen: 專為 Admin 打造的全新管理介面。
  • 即時開關: 面板上方提供了一個動態開關 (Toggle)。只要將其關閉,Firebase RTDB 中的 system_config/cloud_run_enabled 狀態會立即切換。
  • 智慧 Fallback 機制: App 內的 RagKnowledgeServiceGemmaLocalService 現在會先檢查此開關狀態。一旦關閉,App 將完全不發送 HTTP 請求到 Cloud Run,並瞬間切換回本機端的 Gemma 4 離線 AI 模型,確保考試與發問不中斷!

2. 即時使用紀錄追蹤

  • Python 後端整合: 在 adk-backend/main.py 的 FastAPI 中,加入了背景執行任務 (BackgroundTasks)。
  • 當使用者成功呼叫 /api/v1/exam/analyze/api/v1/rag/query/api/v1/chat 任何一個端點時,後端會解析他們的 Firebase Auth Token,並將使用者的 Email、動作類型與時間戳記,以極低的延遲非同步寫入至 Firebase RTDB 的 admin_logs/cloud_run_usage 節點。

3. Dashboard 選單入口

  • Admin 使用者登入後,打開左側選單 (Drawer),會看到全新的 「☁️ Cloud Run 管理」 選項。
  • 點擊進入後,下方會即時監聽 RTDB,列出最新的 50 筆雲端 AI 呼叫紀錄,達到「免外掛、免離開 App」即可完成即時監控的目標。

🚀 後續部署步驟

因為我們修改了 Python 後端 (adk-backend/main.py) 與 Flutter 前端,請您執行以下步驟將新功能上線:

1. 更新 Cloud Run 服務

請透過 PowerShell 在 adk-backend 資料夾中執行重新部署指令:

cd adk-backend
gcloud run deploy mockexam-adk --source . --region asia-east1

2. 發布新版 Android App

由於 Flutter 程式碼已新增 CloudRunManagementScreen,您可以透過既有的自動化腳本打包 Build 186 並上傳至內部測試軌道:

.\scripts\build_aab_internal.ps1 -AutoY

[!NOTE]
目前採用的方案是「App 內建管理面板」。這代表您隨時只要打開 App,就能一覽無遺所有人的使用情況,無需依賴其他通訊軟體。若未來營運規模擴大,我們隨時可以再把 Telegram Bot 加進去!

Ran command: gcloud run deploy mockexam-adk --source . --region asia-east1 --quiet
Ran command: .\scripts\build_aab_internal.ps1 -AutoY
Listed directory MockExam20260417
Ran command: .\build_aab_internal.ps1 -AutoY
Used tool: manage_task

已經為您啟動了自動化部署程序,兩個工作正在背景同時進行中:

  1. Cloud Run 後端微服務部署 (task-701):正在將最新的 main.py 打包並發布至 GCP 的 asia-east1 區域。
  2. Android App 內部測試版本編譯 (task-710):正在執行 build_aab_internal.ps1 -AutoY,它會自動打包含有新版 CloudRunManagementScreen 與相關邏輯的 App,並上傳至 Google Play Internal Testing。

這兩個程序都需要幾分鐘的時間(尤其是打包 Android AAB 與上傳)。我會在背景為您監控,一旦有結果(無論成功或發生錯誤),系統會自動喚醒我為您報告進度。您可以稍微休息一下!

進度回報

Cloud Run 部署已成功完成!
最新的 Python 後端(包含 Admin 授權檢查與使用量 Firebase Logging)已經順利發布至 mockexam-adk-448229238513.asia-east1.run.app。現在只要 App 有開啟 Cloud Run 選項,使用者的 RAG 查詢紀錄就會自動寫入您的 RTDB。

目前 Android App 的打包與上傳作業(build_aab_internal.ps1)仍在背景持續進行中(通常需要 5~10 分鐘)。完成後我會立刻通知您!


上一篇
第4天 sdd , database-design , security-architecture , ai-engine
系列文
將考國際證照的應用程式變成開源5
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言