Dec 17, 2025

Google Agent Development Kit (ADK) Python 架構深度解析

0. 為什麼要關注 Google ADK？

在 AI Agent 框架的競爭中，Google 推出了 Agent Development Kit (ADK)——一個以代碼優先為理念的企業級代理開發框架。與其他框架相比，ADK 有幾個顯著特點：

無狀態編排引擎：Runner 完全無狀態，極易水平擴展
代理樹結構：清晰的多代理協作模式，支援順序、並行、循環執行
豐富的工具生態：50+ 內建工具，涵蓋 Google Cloud 全系列服務
插件系統：橫切關注點的清晰分離，類似中間件概念
生產就緒：原生支援 Vertex AI、Cloud Run、GKE 部署

整體架構可以用一張圖概括：

flowchart TD
  classDef pad fill:transparent,stroke:transparent,color:transparent;

  subgraph App["App 容器"]
    direction TB
    app_pad[" "]:::pad
    Plugins["Plugins 插件層"]
    RootAgent["Root Agent"]
    SubAgents["Sub Agents"]
  end

  subgraph Runner["Runner 無狀態編排引擎"]
    direction TB
    runner_pad[" "]:::pad
    SessionSvc["SessionService"]
    MemorySvc["MemoryService"]
    ArtifactSvc["ArtifactService"]
  end

  subgraph LLMs["LLM Models"]
    direction TB
    llm_pad[" "]:::pad
    Gemini["Gemini"]
    Claude["Claude"]
    GPT["GPT-4"]
    Others["其他"]
  end

  App --> Runner
  Runner --> LLMs

接下來深入探討 ADK 的核心架構與設計理念。

1. 專案結構與規模

ADK Python 的代碼庫規模相當龐大，約有 79,500 行核心代碼：

adk-python/
├── src/google/adk/              # 核心源代碼 (~79.5K 行)
│   ├── __init__.py              # 主包導出
│   ├── version.py               # 版本信息
│   ├── runners.py               # 核心編排引擎 (1,400+ 行)
│   ├── agents/                  # 代理實現 (28 個文件)
│   ├── tools/                   # 工具生態系統 (50+ 子模塊)
│   ├── models/                  # LLM 集成 (13 個文件)
│   ├── sessions/                # 會話管理 (10 個文件)
│   ├── memory/                  # 長期內存 (5 個文件)
│   ├── artifacts/               # 工件管理 (5 個文件)
│   ├── flows/llm_flows/         # 執行流程 (22 個文件)
│   ├── evaluation/              # 評估框架 (44 個文件)
│   ├── cli/                     # CLI 和 Web UI (18 個文件)
│   ├── plugins/                 # 插件系統 (9 個文件)
│   ├── code_executors/          # 代碼執行器 (9 個文件)
│   ├── auth/                    # 認證系統 (14 個文件)
│   ├── apps/                    # 應用容器 (4 個文件)
│   ├── a2a/                     # Agent-to-Agent 協議
│   ├── events/                  # 事件系統
│   ├── planners/                # 規劃器
│   ├── telemetry/               # 遠程觀測
│   └── utils/                   # 工具函數
├── tests/                       # 測試套件 (346 個文件)
├── contributing/samples/        # 示例代碼 (60+ 示例)
└── pyproject.toml               # 項目配置

版本資訊：

當前版本：1.21.0
許可證：Apache 2.0
Python 要求：>= 3.10

2. 核心架構：無狀態編排引擎

2.1 架構層次

ADK 的架構分為四個主要層次：

flowchart TD
  subgraph Layer1["應用層"]
    App["App 容器"]
    Plugins["Plugins"]
  end

  subgraph Layer2["代理層"]
    RootAgent["Root Agent"]
    SubAgent1["SubAgent 1"]
    SubAgent2["SubAgent 2"]
    SubAgent3["SubAgent 3"]
    RootAgent --> SubAgent1
    RootAgent --> SubAgent2
    RootAgent --> SubAgent3
  end

  subgraph Layer3["編排層"]
    Runner["Runner"]
    SessionSvc["Session Service"]
    MemorySvc["Memory Service"]
    ArtifactSvc["Artifact Service"]
  end

  subgraph Layer4["模型層"]
    Gemini["Gemini"]
    Claude["Claude"]
    GPT["GPT-4"]
    LiteLLM["LiteLLM"]
  end

  Layer1 --> Layer2
  Layer2 --> Layer3
  Layer3 --> Layer4

2.2 核心執行流程

Runner 是 ADK 的核心編排引擎，負責協調代理、工具和 LLM 之間的交互：

flowchart TD
  Start([開始])
  Load["1. SessionService 加載會話歷史"]
  Context["2. 創建 InvocationContext"]
  AgentRun["3. agent.run_async(ctx)"]

  subgraph Loop["推理-行動循環"]
    LLM["調用 LLM"]
    Tool["執行工具"]
    Text["生成文本/音頻"]
    Transfer["轉移控制"]
  end

  Event["4. 生成 Event 對象"]
  Stream["5. 事件流回調用者"]
  Persist["6. 持久化到會話"]
  Compress["7. 可選：事件壓縮"]
  Return["8. 返回給用戶"]
  End([結束])

  Start --> Load --> Context --> AgentRun
  AgentRun --> Loop
  Loop --> Event --> Stream --> Persist --> Compress --> Return --> End

執行流程詳解：

會話加載：SessionService 從存儲中加載完整的會話歷史
上下文創建：創建 InvocationContext，包含會話狀態、工具列表、模型配置
代理執行：進入「推理-行動」循環
事件生成：每個步驟都會生成 Event 對象
事件流：事件即時流回調用者，支持 streaming
持久化：事件自動持久化到會話存儲
壓縮：可選的事件壓縮以管理上下文大小

3. 代理系統：層次化代理樹

3.1 代理類別層次

ADK 提供了豐富的代理類型：

BaseAgent (基類)
├── LlmAgent               # LLM 驅動的代理 (主要使用)
│   └── Agent (別名)       # 簡化版本
├── LoopAgent              # 循環執行代理
├── SequentialAgent        # 順序執行代理
├── ParallelAgent          # 並行執行代理
├── LangGraphAgent         # LangGraph 集成
└── RemoteA2aAgent         # Agent-to-Agent 遠程代理

3.2 LlmAgent：核心代理實現

LlmAgent 是最常用的代理類型，由 LLM 驅動決策：

from google.adk.agents import Agent

agent = Agent(
    name="assistant",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant.",
    description="An assistant that can search the web.",
    tools=[google_search, custom_tool],
    sub_agents=[child_agent1, child_agent2],
    generate_content_config=types.GenerateContentConfig(
        temperature=0.7,
        safety_settings=[...]
    )
)

主要屬性說明：

屬性	類型	用途
`name`	`str`	代理唯一名稱，用於標識和轉移
`model`	`str`	LLM 模型標識符
`instruction`	`str`	系統指令，定義代理行為
`description`	`str`	代理描述，用於代理間轉移決策
`tools`	`list`	可用工具列表
`sub_agents`	`list`	子代理列表，形成代理樹

3.3 多代理協作模式

ADK 支援多種多代理協作模式：

順序執行 (Sequential)

from google.adk.agents import SequentialAgent

pipeline = SequentialAgent(
    name="analytics_pipeline",
    sub_agents=[
        data_retrieval_agent,  # 步驟 1：數據獲取
        transform_agent,       # 步驟 2：數據轉換
        analysis_agent         # 步驟 3：數據分析
    ]
)

flowchart LR
  Input([輸入]) --> A["Data Retrieval"]
  A --> B["Transform"]
  B --> C["Analysis"]
  C --> Output([輸出])

並行執行 (Parallel)

from google.adk.agents import ParallelAgent

coordinator = ParallelAgent(
    name="coordinator",
    sub_agents=[agent_a, agent_b, agent_c]  # 同時運行
)

flowchart TD
  Input([輸入]) --> Split{並行分發}
  Split --> A["Agent A"]
  Split --> B["Agent B"]
  Split --> C["Agent C"]
  A --> Merge{結果合併}
  B --> Merge
  C --> Merge
  Merge --> Output([輸出])

代理樹結構

root = Agent(
    name="root",
    sub_agents=[
        Agent(name="middle", sub_agents=[leaf1, leaf2]),
        sibling
    ]
)

flowchart TD
  Root["Root Agent"]
  Middle["Middle Agent"]
  Sibling["Sibling Agent"]
  Leaf1["Leaf 1"]
  Leaf2["Leaf 2"]

  Root --> Middle
  Root --> Sibling
  Middle --> Leaf1
  Middle --> Leaf2

4. 工具生態系統

4.1 工具類型層次

ADK 提供了豐富的工具類型：

BaseTool (基類)
├── FunctionTool            # Python 函數工具
├── AgentTool              # 代理工具 (包裝子代理)
├── GoogleSearchTool       # Google 搜索
├── VertexAiSearchTool     # Vertex AI 搜索
├── GoogleMapsTool         # Google Maps
├── MCPToolset             # Model Context Protocol
├── OpenAPITool            # OpenAPI 規範工具
├── GoogleApiToolset       # Google API 工具集
├── BigQueryToolset        # BigQuery 工具集
├── BigTableToolset        # BigTable 工具集
├── SpannerToolset         # Spanner 工具集
├── CrewAiTool             # CrewAI 集成
├── LangChainTool          # LangChain 集成
└── LongRunningFunctionTool # 長運行操作

4.2 函數工具定義

最常用的是 FunctionTool，將 Python 函數轉換為代理可用的工具：

from google.adk.tools import FunctionTool
from google.adk.tools import ToolContext

def fetch_weather(
    city: str,
    unit: str = "celsius"
) -> str:
    """Fetch weather information for a city.

    Args:
        city: The city name
        unit: Temperature unit (celsius or fahrenheit)

    Returns:
        Weather information as a string
    """
    # 實際實現會調用天氣 API
    return f"Weather in {city}: 20°{unit[0].upper()}"

def analyze_sentiment(
    text: str,
    tool_context: ToolContext  # 特殊參數：獲取會話上下文
) -> dict:
    """Analyze sentiment with session state access."""
    result = {"text": text, "sentiment": "positive"}
    # 可以訪問和修改會話狀態
    tool_context.state["last_sentiment"] = result
    return result

# 自動轉換為工具
weather_tool = FunctionTool(func=fetch_weather)
sentiment_tool = FunctionTool(func=analyze_sentiment)

關鍵設計：

函數的 docstring 會自動提取為工具描述
類型註解用於生成工具的參數 schema
ToolContext 參數提供會話狀態訪問能力

4.3 MCP 工具集成

ADK 原生支援 Model Context Protocol (MCP)：

from google.adk.tools import MCPToolset
from google.adk.tools.mcp_toolset import StdioConnectionParams, StdioServerParameters

mcp = MCPToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command='python',
            args=['mcp_server.py']
        )
    ),
    tool_filter=['tool1', 'tool2']  # 可選：過濾特定工具
)

# 在代理中使用
agent = Agent(
    name="mcp_agent",
    model="gemini-2.5-flash",
    tools=[mcp]  # MCPToolset 會自動展開為多個工具
)

5. Runner 編排引擎詳解

5.1 Runner 配置

Runner 是 ADK 的核心編排引擎：

from google.adk import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.memory import VertexAiRagMemoryService
from google.adk.artifacts import GcsArtifactService

runner = Runner(
    app_name="my_app",
    agent=root_agent,
    session_service=InMemorySessionService(),  # 會話存儲
    memory_service=memory_service,              # 可選：長期記憶
    artifact_service=artifact_service,          # 可選：工件存儲
)

5.2 三種執行模式

同步執行（本地測試）

from google.genai import types

response = runner.run(
    user_id="user123",
    session_id="session456",
    new_message=types.Content(
        role='user',
        parts=[types.Part(text="What is Python?")]
    )
)
print(response.events[-1].parts)  # 獲取最後響應

異步執行（生產環境）

async for event in runner.run_async(
    user_id="user123",
    session_id="session456",
    new_message=content
):
    print(event.parts)  # 流式輸出，即時響應

實時雙向（Gemini Live API）

from google.adk.agents import RunConfig, StreamingMode

run_config = RunConfig(
    stream=True,
    streaming_mode=StreamingMode.BIDIRECTIONAL,
    input_speech_config=InputSpeechConfig(...),
    output_speech_config=OutputSpeechConfig(...)
)

async with runner.run_live(
    user_id="user1",
    session_id="session1",
    run_config=run_config
) as live_session:
    # 發送音頻輸入
    audio_data = await microphone.read()
    await live_session.send(audio_data)

    # 接收音頻輸出
    async for response in live_session.stream():
        await speaker.play(response.audio_bytes)

6. 會話與存儲服務

6.1 會話服務層次

ADK 提供多種會話存儲後端：

BaseSessionService (抽象)
├── InMemorySessionService      # 內存存儲 (開發)
├── SqliteSessionService        # SQLite 數據庫
├── DatabaseSessionService      # 通用 SQL (Spanner 等)
└── VertexAiSessionService      # Vertex AI 集成

6.2 Session 數據模型

class Session:
    id: str                    # 唯一標識符
    app_name: str              # 應用名稱
    user_id: str               # 用戶標識
    state: dict                # 會話狀態字典
    events: list[Event]        # 事件列表 (完整歷史)
    last_update_time: datetime # 最後更新時間戳

6.3 長期記憶服務

from google.adk.memory import VertexAiRagMemoryService

memory_service = VertexAiRagMemoryService(
    project="my-project",
    location="us-central1"
)

6.4 工件存儲服務

from google.adk.artifacts import GcsArtifactService, FileArtifactService

# 本地文件系統（開發環境）
artifact_service = FileArtifactService(base_path="/tmp/artifacts")

# Google Cloud Storage（生產環境）
artifact_service = GcsArtifactService(bucket_name="my-bucket")

7. 插件系統：橫切關注點分離

7.1 插件回調點

插件系統提供了豐富的回調點，類似於中間件概念：

class BasePlugin(ABC):
    # 代理生命週期回調
    async def before_agent_callback(ctx: CallbackContext) -> Optional[Content]
    async def after_agent_callback(ctx: CallbackContext) -> Optional[Content]

    # 模型調用回調
    async def before_model_callback(ctx: CallbackContext, req: LlmRequest) -> Optional[LlmResponse]
    async def after_model_callback(ctx: CallbackContext, resp: LlmResponse) -> Optional[LlmResponse]
    async def on_model_error_callback(ctx: CallbackContext, req: LlmRequest, err: Exception)

    # 工具調用回調
    async def before_tool_callback(tool: BaseTool, args: dict, ctx: ToolContext)
    async def after_tool_callback(tool: BaseTool, args: dict, ctx: ToolContext, result: dict)
    async def on_tool_error_callback(tool: BaseTool, args: dict, ctx: ToolContext, err: Exception)

flowchart TD
  subgraph PluginHooks["插件回調點"]
    BA["before_agent"]
    BM["before_model"]
    Model["模型調用"]
    AM["after_model"]
    BT["before_tool"]
    Tool["工具執行"]
    AT["after_tool"]
    AA["after_agent"]
  end

  BA --> BM --> Model --> AM
  AM -->|有工具調用| BT --> Tool --> AT --> BM
  AM -->|無工具調用| AA

7.2 內建插件

插件	功能	典型用途
`LoggingPlugin`	結構化日誌	調試與監控
`GlobalInstructionPlugin`	全局指令注入	統一系統提示
`ContextFilterPlugin`	上下文過濾	敏感信息過濾
`ReflectRetryToolPlugin`	工具重試邏輯	失敗自動重試
`SaveFilesAsArtifactsPlugin`	文件保存	文件管理
`BigQueryAnalyticsPlugin`	分析跟蹤	使用統計

7.3 自定義插件示例

from google.adk.plugins import BasePlugin
from google.adk.agents import CallbackContext
from typing import Optional

class CustomLoggingPlugin(BasePlugin):
    async def before_model_callback(
        self,
        ctx: CallbackContext,
        llm_request: LlmRequest
    ) -> Optional[LlmResponse]:
        print(f"[LOG] Agent: {ctx.agent.name}")
        print(f"[LOG] Request tokens: {len(llm_request.contents)}")
        return None  # 返回 None 表示不修改請求

    async def after_model_callback(
        self,
        ctx: CallbackContext,
        llm_response: LlmResponse
    ) -> Optional[LlmResponse]:
        print(f"[LOG] Response parts: {len(llm_response.parts)}")
        return None  # 返回 None 表示不修改響應

    async def on_model_error_callback(
        self,
        ctx: CallbackContext,
        llm_request: LlmRequest,
        error: Exception
    ):
        print(f"[ERROR] Model call failed: {error}")
        # 可以在這裡實現錯誤報告、告警等

8. 事件系統：完整的執行審計

8.1 Event 數據模型

每個執行步驟都會生成 Event 對象：

class Event(LlmResponse):
    id: str                    # 事件唯一 ID
    invocation_id: str         # 調用 ID
    author: str                # "user" 或代理名
    actions: EventActions      # 執行的動作
    branch: Optional[str]      # 多代理分支標識
    long_running_tool_ids: Optional[set[str]]

8.2 事件類型

ADK 的事件系統涵蓋完整的執行生命週期：

flowchart LR
  subgraph Events["事件類型"]
    UserInput["用戶輸入事件"]
    LLMReq["LLM 請求事件"]
    LLMResp["LLM 響應事件"]
    ToolCall["工具調用事件"]
    ToolResult["工具結果事件"]
    Transfer["代理轉移事件"]
    Compress["壓縮事件"]
  end

  UserInput --> LLMReq --> LLMResp
  LLMResp --> ToolCall --> ToolResult --> LLMReq
  LLMResp --> Transfer
  LLMResp --> Compress

事件溯源的優勢：

審計：完整記錄每個決策和操作
回放：可以重放任意執行過程
調試：精確定位問題發生的步驟

9. 代碼執行器：安全的代碼執行

ADK 提供多種代碼執行環境：

from google.adk.code_executors import (
    BuiltInCodeExecutor,         # 內置沙箱
    ContainerCodeExecutor,       # Docker 容器
    GkeCodeExecutor,             # Kubernetes
    VertexAiCodeExecutor,        # Vertex AI
    UnsafeLocalCodeExecutor      # 本地 (危險)
)

# 生產環境推薦使用 Vertex AI
executor = VertexAiCodeExecutor(
    project="my-project",
    location="us-central1"
)

# 開發環境可以使用 Docker
executor = ContainerCodeExecutor(
    image="python:3.11-slim",
    timeout=30
)

10. 支持的 LLM 模型

ADK 支援多種 LLM 提供商：

模型	類	特性
Gemini	`GoogleLLM`	推薦，原生支持，最佳整合
Claude	`AnthropicLLM`	Anthropic 官方支持
Gemma	`GemmaLLM`	開源模型，本地部署
OpenAI/多模型	`LiteLLM`	通過 LiteLLM 支持 100+ 模型
企業 API	`ApiegeeLLM`	企業 API 網關整合

from google.adk.agents import Agent

# 使用 Gemini
gemini_agent = Agent(
    name="gemini_agent",
    model="gemini-2.5-flash",
    instruction="..."
)

# 使用 Claude (需要安裝 anthropic 依賴)
claude_agent = Agent(
    name="claude_agent",
    model="claude-sonnet-4-5-20250929",
    instruction="..."
)

# 使用 LiteLLM (支持 OpenAI, Azure, 等)
openai_agent = Agent(
    name="openai_agent",
    model="litellm:gpt-4o",
    instruction="..."
)

11. CLI 和部署

11.1 本地開發

# 啟動 Web UI (包含聊天界面和調試工具)
adk web my_agent_dir
# 打開 http://localhost:8000

# 啟動 API 服務器
adk api_server my_agent_dir --port 8080

11.2 評估測試

# 運行評估測試集
adk eval my_agent_dir eval_set.json

11.3 生產部署

# 部署到 Cloud Run
adk deploy my_agent_dir \
  --platform cloudrun \
  --project my-gcp-project

# 部署到 Vertex AI Agent Engine
adk deploy my_agent_dir \
  --platform vertex-ai \
  --project my-gcp-project

12. 完整示例

12.1 快速開始

from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types

# 1. 創建代理
agent = Agent(
    name="assistant",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant that can search the web.",
    tools=[google_search]
)

# 2. 創建運行器
runner = Runner(
    app_name="my_app",
    agent=agent,
    session_service=InMemorySessionService()
)

# 3. 執行對話
session = runner.run(
    user_id="user1",
    session_id="session1",
    new_message=types.Content(
        role='user',
        parts=[types.Part(text="What is the weather in Tokyo?")]
    )
)

# 4. 獲取響應
print(session.events[-1].parts)

12.2 多代理數據管道

from google.adk.agents import Agent, SequentialAgent

# 定義各階段代理
data_retrieval = Agent(
    name="retriever",
    model="gemini-2.5-flash",
    instruction="You retrieve data from databases and APIs.",
    tools=[db_query, api_fetch]
)

transformer = Agent(
    name="transformer",
    model="gemini-2.5-flash",
    instruction="You transform and clean the data.",
    tools=[data_transform, data_validate]
)

analyzer = Agent(
    name="analyzer",
    model="gemini-2.5-flash",
    instruction="You analyze data and generate insights.",
    tools=[statistical_analysis, generate_report]
)

# 組合為順序管道
pipeline = SequentialAgent(
    name="analytics_pipeline",
    sub_agents=[data_retrieval, transformer, analyzer]
)

12.3 帶插件的企業應用

from google.adk.apps import App
from google.adk.plugins import LoggingPlugin

app = App(
    name="enterprise_assistant",
    root_agent=root_agent,
    plugins=[
        LoggingPlugin(),
        CustomSecurityPlugin(),
        AuditTrailPlugin(),
    ],
    context_cache_config=ContextCacheConfig(
        max_tokens=100000,
        ttl_seconds=3600
    ),
    resumability_config=ResumabilityConfig(
        is_resumable=True  # 支持會話恢復
    )
)

13. 設計特點總結

特性	描述	優勢
代碼優先	所有配置都在 Python 代碼中	版本控制、IDE 支持、類型安全
無狀態編排	Runner 完全無狀態	水平伸縮、高並發、容錯
代理樹結構	清晰的多代理協作模式	LLM 自主路由、責任分離
事件溯源	完整的執行審計跟蹤	調試、回放、合規
豐富工具	50+ 內建工具	快速整合 Google Cloud 服務
插件系統	橫切關注點分離	可組合、可重用
多模式支持	同步、異步、實時雙向流	適應各種場景
生產就緒	Vertex AI、Cloud Run、GKE	企業級部署

14. 架構決策分析

決策	理由	影響
無狀態 Runner	水平伸縮需求	高並發處理、易於容錯恢復
事件溯源	審計和調試需求	完整追蹤、可回放執行
代理樹	複雜任務分解	責任清晰、LLM 自主路由
插件系統	關注點分離	功能模塊化、可組合可擴展
依賴注入	服務可替換	測試友好、環境適應性

15. 與其他框架比較

特性	Google ADK	LangGraph	Claude SDK
核心理念	代碼優先	圖優先	簡潔優先
狀態管理	無狀態 Runner	BSP/Channel	Loop Runner
多代理	代理樹	StateGraph	原生 Handoffs
工具生態	50+ Google 工具	通用工具	MCP 支持
部署	Vertex AI	LangGraph Cloud	自行部署

16. 結語

Google ADK 是一個設計精良的企業級 AI Agent 框架，特別適合：

Google Cloud 用戶：無縫整合 Vertex AI、BigQuery、Cloud Run 等服務
企業級應用：完整的審計、安全、可觀測性支持
複雜多代理系統：清晰的代理樹結構和多種協作模式
需要快速迭代的團隊：豐富的內建工具和插件生態

如果你正在構建需要與 Google Cloud 深度整合的 AI Agent 應用，ADK 是一個非常值得考慮的選擇。