Day 18 | 打造一支 Agent 團隊：Collaborative Workflows 與 Agent Modes

主張：讓 agent 互相委派任務只是第一步，真正決定系統穩不穩的是「委派出去的那個 agent，任務做完之後控制權怎麼交回來」。
讀完能做到：用漸進的方式組出一支會互相委派的 agent 團隊，並替每個 subagent 選對 chat／task／single_turn 三種 mode 之一。

從天氣機器人開始

我今天帶你做一個「Weather Bot」練習，從單一 agent 一路組到一支團隊。這個練習延伸自更早的 Multi-tool agent 入門，我挑其中跟多 agent 協作最直接相關的部分講。整體節奏是：先有一個能查天氣的單一 agent，再逐步加上多模型支援、專職委派、跨輪記憶、安全防護——走完這一輪，你會摸過建構一個真實多 agent 系統要用到的大多數 ADK 元件。

今天的程式碼都疊在同一個檔案裡（weather_team.py），一步一步往下加，用 python weather_team.py 跑；我原本是照 Colab notebook 一格一格跑的，如果你偏好那個節奏，也可以直接開 notebook 照著做。

開始前先裝好套件（本文實測環境：google-adk 2.8.0、Python 3.13）：

pip install google-adk
pip install "litellm>=1.84"   # 第二步換模型才會用到

然後在檔案最上面把今天會用到的東西一次匯入，並設定好 Gemini API 金鑰——去 AI Studio 申請一把免費的就行，不需要開 GCP 專案或裝 gcloud：

import os
import asyncio
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import InMemoryRunner
from google.genai import types

os.environ["GOOGLE_API_KEY"] = "YOUR_GOOGLE_API_KEY"  # 去 AI Studio 申請
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" 
MODEL_GEMINI_FLASH = "gemini-flash-latest"   # 用 -latest 別名，避免具名版本過期下線
MODEL_GPT_5 = "openai/gpt-5-mini"              # LiteLLM 呼叫 OpenAI 模型需要 provider 前綴

模型常數這裡刻意用 -latest 別名而不是 gemini-2.0-flash 這種具名版本——具名版本會隨時間下線，-latest 由 Google 保證永遠指向當時最新的 flash 模型，今天的範例都照這個寫法。

第一步：先讓單一 agent 動起來

一切從一個工具跟一個 agent 開始。工具是一個回傳 mock 天氣資料的 Python 函式，重點在 docstring——這是 Day 6 講過的「docstring 就是介面定義」原則的具體示範：

def get_weather(city: str) -> dict:
    """Retrieves the current weather report for a specified city.

    Args:
        city (str): The name of the city (e.g., "New York", "London", "Tokyo").

    Returns:
        dict: A dictionary containing the weather information.
              Includes a 'status' key ('success' or 'error').
              If 'success', includes a 'report' key with weather details.
              If 'error', includes an 'error_message' key.
    """
    city_normalized = city.lower().replace(" ", "")
    mock_weather_db = {
        "newyork": {"status": "success", "report": "The weather in New York is sunny with a temperature of 25°C."},
        "london": {"status": "success", "report": "It's cloudy in London with a temperature of 15°C."},
        "tokyo": {"status": "success", "report": "Tokyo is experiencing light rain and a temperature of 18°C."},
    }
    if city_normalized in mock_weather_db:
        return mock_weather_db[city_normalized]
    return {"status": "error", "error_message": f"Sorry, I don't have weather information for '{city}'."}

weather_agent = Agent(
    name="weather_agent_v1",
    model=MODEL_GEMINI_FLASH,
    description="Provides weather information for specific cities.",
    instruction="You are a helpful weather assistant. "
                "When the user asks for the weather in a specific city, "
                "use the 'get_weather' tool to find the information. "
                "If the tool returns an error, inform the user politely. "
                "If the tool is successful, present the weather report clearly.",
    tools=[get_weather],
)

這一步我要特別強調兩個原則，它們在後面的委派機制裡會變得更重要：instruction 要具體到連錯誤處理都寫清楚；name 跟 description 要精準，因為它們是 ADK 內部用來判斷「該不該把任務轉交給這個 agent」的依據。

第二步：多模型彈性（LiteLLM）

ADK 不綁死單一模型供應商。透過 LiteLlm wrapper，同一支 agent 可以換上 GPT-4o 或 Claude Sonnet，不用改動核心邏輯：

from google.adk.models.lite_llm import LiteLlm

weather_agent_gpt = Agent(
    name="weather_agent_gpt",
    model=LiteLlm(model=MODEL_GPT_5),
    description="Provides weather information (using GPT-5 mini).",
    instruction="You are a helpful weather assistant powered by GPT-5 mini. ...",
    tools=[get_weather],
)

三個模型背後用的是完全相同的 get_weather 工具邏輯，差異只在於最終回應的措辭與語氣——這個對照示範了 ADK 的模型層抽象做得多乾淨：核心的 agent 結構、工具、instruction 邏輯完全不用因為換供應商而重寫。

第三步：組一支團隊，學會委派

單一 agent 一直加工具、加複雜指令，很快就會變得難以維護。更穩健的做法是拆成多個專職 agent，再指定一個 root agent 當 coordinator。我整理成一張清單方便對照：

• Modularity（模組化）：每個 agent 易於獨立開發測試
• Specialization（專職化）：可以針對任務微調 instruction 和模型選擇
• Scalability（可擴展）：加新能力只要加新 agent
• Efficiency（效率）：簡單任務可以配便宜的模型

先定義兩個簡單的 sub-agent——問候與道別，各自配一個工具：

def say_hello(name: str = None) -> str:
    """Provides a simple greeting. If a name is provided, it will be used."""
    return f"Hello, {name}!" if name else "Hello there!"

def say_goodbye() -> str:
    """Provides a simple farewell message to conclude the conversation."""
    return "Goodbye! Have a great day."

greeting_agent = Agent(
    model=MODEL_GEMINI_FLASH,
    name="greeting_agent",
    instruction="You are the Greeting Agent. Your ONLY task is to provide a friendly greeting to the user. "
                "Use the 'say_hello' tool to generate the greeting. "
                "Do not engage in any other conversation or tasks.",
    description="Handles simple greetings and hellos using the 'say_hello' tool.",
    tools=[say_hello],
)

farewell_agent = Agent(
    model=MODEL_GEMINI_FLASH,
    name="farewell_agent",
    instruction="You are the Farewell Agent. Your ONLY task is to provide a polite goodbye message. "
                "Use the 'say_goodbye' tool when the user indicates they are leaving. "
                "Do not perform any other actions.",
    description="Handles simple farewells and goodbyes using the 'say_goodbye' tool.",
    tools=[say_goodbye],
)

再把它們接到 root agent 的 sub_agents：

weather_agent_team = Agent(
    name="weather_agent_v2",
    model=MODEL_GEMINI_FLASH,
    description="The main coordinator agent. Handles weather requests and delegates greetings/farewells to specialists.",
    instruction="You are the main Weather Agent coordinating a team. "
                "Use the 'get_weather' tool ONLY for specific weather requests. "
                "You have specialized sub-agents: "
                "1. 'greeting_agent': Handles simple greetings. Delegate to it for these. "
                "2. 'farewell_agent': Handles simple farewells. Delegate to it for these. "
                "Analyze the user's query and delegate or handle accordingly.",
    tools=[get_weather],
    sub_agents=[greeting_agent, farewell_agent]
)

這個機制在 ADK 裡叫 Automatic Delegation（Auto Flow）：只要提供 sub_agents 清單，root agent 的 LLM 在推理時，不只看自己的 instruction 跟工具，還會一併考慮每個 sub-agent 的 description。一旦判斷某個查詢更符合某個 sub-agent 的描述，就會自動產生一個內部動作把控制權轉移過去——這跟 Day 6 提過的 AgentTool（把另一個 agent 包成一個普通工具來呼叫，呼叫完會回到原 agent）行為模式不同，transfer 是真正把控制權交出去，不是呼叫完就回來。

三個 agent 之間的關係畫成圖會更直觀——注意 greeting_agent、farewell_agent 都沒設 mode，所以走的是「LLM 自己決定要不要 transfer」這條手動路線：

把團隊跑起來

sub_agents 接好之後，用一個最小的 driver 就能看到委派實際發生：

async def call_agent_async(query: str, runner, user_id, session_id):
    content = types.Content(role="user", parts=[types.Part(text=query)])
    async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
        if event.is_final_response() and event.content and event.content.parts:
            print(f"<<< {event.author}: {event.content.parts[0].text}")

async def main():
    runner = InMemoryRunner(agent=weather_agent_team, app_name="weather_team")
    session = await runner.session_service.create_session(app_name="weather_team", user_id="u1")
    await call_agent_async("Hello there!", runner, "u1", session.id)
    await call_agent_async("What is the weather in New York?", runner, "u1", session.id)
    await call_agent_async("Thanks, bye!", runner, "u1", session.id)

asyncio.run(main())

跑起來會看到三行回應分別來自 greeting_agent、weather_agent_v2、farewell_agent——event.author 就是實際處理這句話的 agent，這是驗證委派有沒有生效最直接的方法。

第四步：用 Session State 做記憶

到目前為止，每次互動都是從零開始，agent 不記得任何跨輪的東西。ADK 用 session.state（一個綁定在特定 session 的 dict）解決這個問題，提供兩種存取方式：

• ToolContext：工具函式的參數只要標注 ToolContext 型別，ADK 就會自動注入，讓工具在執行期間讀寫 state（不管這個參數放在簽章的哪個位置都行，且對 LLM 看到的 schema 是隱藏的）。
• output_key：agent 設定這個參數後，ADK 會自動把該輪的最終文字回應存進 session.state[your_key]。

一個會依使用者偏好調整攝氏／華氏的天氣工具，就是靠 ToolContext 讀取 state 裡的偏好值：

def get_weather_stateful(city: str, tool_context: ToolContext) -> dict:
    preferred_unit = tool_context.state.get("user_preference_temperature_unit", "Celsius")
    ...
    if preferred_unit == "Fahrenheit":
        temp_value = (temp_c * 9/5) + 32
        temp_unit = "°F"
    else:
        temp_value = temp_c
        temp_unit = "°C"
    ...
    tool_context.state["last_city_checked_stateful"] = city
    return result

讀取 state 時我會一律用 dict.get(key, default_value)，避免 key 還不存在時整個工具崩潰。

教學接下來的第五步會用 before_model_callback 跟 before_tool_callback 加安全防護——這部分本質上就是 Day 12 已經講過的 Callback 機制，套用在多 agent 團隊的場景裡，今天不重複展開。

什麼是 Collaborative Workflow

上面的天氣機器人示範的，正是 ADK 裡的 Collaborative workflows：一個 coordinator agent 帶著多個 subagent，自動處理任務委派，任務做完會自動返回父 agent。

但光是「能委派」還不夠，ADK 2.0 引入了 mode 設定，用來限制每個 subagent 的行為與工作範圍，讓多 agent 系統的行為更可預測：

from google.adk import Agent

weather_agent = Agent(
    name="weather_checker",
    mode="single_turn",         # no user interaction
    tools=[get_weather, user_info, geocode_address],
)
flight_agent = Agent(
    name="flight_booker",
    mode="task",                # can ask user questions
    input_schema=FlightInput,
    output_schema=FlightResult,
    tools=[search_flights, book_flight],
)
root = Agent(
    name="travel_planner",      # coordinator agent
    sub_agents=[weather_agent, flight_agent],
    # Auto-injects delegation tools named after each subagent:
    # weather_checker, flight_booker
)

（get_weather、user_info、geocode_address、FlightInput 等請自行替換成你的工具與 schema，這段重點是示範 mode 的宣告方式。）

這裡有個關鍵行為值得注意：當 subagent 設定了 task 或 single_turn mode，ADK 會自動注入以各 subagent 命名的委派工具（在這個例子就是 weather_checker、flight_booker），你不需要手動幫 coordinator 寫委派邏輯的工具定義。沒有設 mode（也就是預設走 chat）的 subagent 走的是另一條路——像第三步天氣機器人那種寫法，coordinator 拿到的其實是單一的 transfer_to_agent 工具，由 LLM 自己填入目標 agent 的名字。這也是為什麼下面表格裡 chat 那一欄的「Return to parent」寫的是「手動（transfer）」。

同一個 travel_planner 團隊換成圖示，可以清楚看到跟上面天氣機器人的關鍵差異：coordinator 拿到的是兩個具名工具，而不是一個泛用的 transfer_to_agent，而且兩個 subagent 的返回路徑也不一樣：

三種 Mode 的差異

表格最後一列的「手動」與「自動」，指的都是 ADK 自動掛在該 agent 工具清單裡的內建工具，由 LLM 在推理時自行決定何時呼叫——不是你要 import 或寫進 tools=[] 的東西。task mode 的 agent 完成任務後由模型自己呼叫 finish_task，控制權自動交還；chat mode 的 agent 則要靠模型自己決定呼叫 transfer_to_agent，所以叫「手動」——手動的是模型的決策，不是你的 Python 程式碼。

⚠️ mode 只給 subagent 用，切記不要在 root agent 上設定 mode。這個限制在開發 ADK2.0 建構期不會報錯也不會警告——Agent(name="root", mode="task", ...) 一樣能建構成功，違反了只會在執行時出現非預期的控制權行為，得自己把關，框架不會替你擋。

mode 欄位本身的預設值其實是 None，不是字面上的 'chat'——只有當這個 agent 被放進另一個 agent 的 sub_agents 時，ADK 才會把它正規化成 'chat'；被當成 workflow graph 節點時則會正規化成 'single_turn'。所以表格標的「chat（預設）」，準確地說是「作為 LlmAgent 的 subagent 時的預設」。

三種 mode 的選用直覺：天氣查詢這種一次性、不需要跟使用者來回確認的任務用 single_turn，做完立刻返回，還能平行跑多個；訂機票這種可能需要跟使用者澄清幾句（「你要哪一天？」「經濟艙還是商務艙？」）但終究會走到一個明確終點的任務用 task；需要開放式、持續對話的角色才留給預設的 chat。

同一個 task agent，在不同脈絡下行為不同

這是一個容易被忽略、但影響很大的設計細節。同一個 task 或 single_turn 模式的 agent，被放進不同的呼叫脈絡裡，完成後的行為是不一樣的：

同一個 task agent 擺進這兩種情境，圖示如下——注意兩邊都是同一個 agent 定義，差別只在誰呼叫它、以及完成後控制權去哪：

放進 SequentialAgent、ParallelAgent 這類 workflow graph 節點，這個 task agent 執行完就照 workflow 圖的邏輯往下走；但如果是被一個 LlmAgent 用委派工具轉交過來的，它執行到呼叫 finish_task 為止，然後控制權自動返回發起委派的那個 agent——不像預設的 chat mode agent 需要顯式呼叫 transfer_to_agent 才能交還控制權。這個差異讓同一個 task agent 可以在兩種脈絡下重用，不需要為了不同用法寫兩份程式碼。

Context 隔離：平行執行時彼此看不到對方

task 與 single_turn 模式的 agent 各自在獨立的 session branch 運作。當這些 agent 平行執行時，每個 agent 在建構要送給模型的 context 時，只看得到自己分支裡的事件，完全看不到同儕 agent 在做什麼。等所有平行分支都完成，parent agent 才會拿到彙整後的結果，繼續往下走——這個隔離機制正是 single_turn mode 能安全支援平行執行的前提。

已知限制

task mode 的 agent 必須是 leaf agent，不能再有自己的 subagent——換句話說，委派鏈條在遇到 task mode 的節點時就必須終止，不能再往下分支。這個限制跟前面「mode 只能設在 subagent 上」一樣，建構期不會報錯：task mode 的 agent 照樣可以掛 sub_agents 而不觸發任何錯誤，只是這麼做違反了框架假設，執行期行為不保證正確，得自己守住這條線。

另外我在 Day 13 提過的那個限制在這裡要更新一下：task mode 一度在 graph-based workflow 中於 ADK Python v2.0.0 被停用，但這個限制沒有撐太久——我翻了 ADK Python 2.8.0 的原始碼，已經明確把 task 列為 workflow graph 節點的合法 mode，跟 single_turn、chat 平起平坐，不再是停用狀態。

第四種團隊成員：不用自己寫、Google 幫你養的 Agent

到目前為止，團隊裡的每個成員都是你自己寫的：chat、task、single_turn，差別只在控制權怎麼交還。但如果團隊裡有一個位置，你根本不想自己維護——例如「一個能上網查資料、能跑程式碼解計算題」的通用助手，你不想為了它自己顧一套 sandbox、自己寫一堆 function declaration——這時可以讓 ManagedAgent 頂上：這是 ADK 目前還在 Preview 階段的第四條路，把整個 agent 的推理、工具、執行環境都交給 Google 代管，你只要接上 sub_agents 就能用。

它一樣實作了 BaseAgent 介面，所以放進 sub_agents 的寫法跟今天教的其他 agent 一模一樣，唯一不同的是背後跑的地方。我拿來對照的範例是一個能查即時資訊、一個能跑程式解題的兩人隊友，都指到同一個 agent_id（範例用 antigravity-preview-05-2026）：

import os
from google.adk.agents import ManagedAgent
from google.adk.tools import google_search
from google.genai import types

_AGENT_ID = os.environ.get('MANAGED_AGENT_ID', 'antigravity-preview-05-2026')

managed_search_agent = ManagedAgent(
    name='managed_search_agent',
    description='Answers questions that need fresh, grounded information from the web.',
    agent_id=_AGENT_ID,
    environment={'type': 'remote'},
    tools=[google_search],
)

managed_code_execution_agent = ManagedAgent(
    name='managed_code_execution_agent',
    description='Solves computational questions by running code server-side.',
    agent_id=_AGENT_ID,
    environment={'type': 'remote'},
    tools=[types.Tool(code_execution=types.ToolCodeExecution())],
)

拿它跟今天的三種 mode 對比會更清楚它的定位：local sub-agent 是「你自己養的隊友，你能看到、能改它的每一步」；ManagedAgent 是「Google 幫你養的隊友，你只負責派任務、看結果」。畫成圖就是同一份 sub_agents 清單裡，有一條線跨出了你的流程邊界：

這樣的分工省事，但麻煩的是：

• 只能用伺服器端內建的工具，不支援自訂 function tool 或 MCP 工具（硬塞進去會直接 raise NotImplementedError）
• 目前只有串流一種互動方式

這也是為什麼它得跟本地 sub-agent 分開講：同一份 sub_agents 清單裡，你可能同時放著自己完全掌控的委派隊友，跟一個你只租用能力、不碰內部實作的外包隊友。

銜接

Collaborative workflow、還有今天新登場的 ManagedAgent，都是「同一個流程裡」的多 agent 協作——不管隊友的推理是跑在你的程式碼裡還是 Google 的伺服器上，對你的 agent 來說都還是一次函式呼叫等級的互動。但如果你要合作的對象是一個完全獨立的服務、由不同團隊維護、甚至用不同程式語言寫的呢？Day 19 開始進入 A2A（Agent2Agent）協定——先講「什麼時候該用、什麼時候不該用」，這比協定本身的技術細節更值得先弄清楚。

Google ADK 官方網站
GitHub - Agent Development Kit (ADK) 2.0

GitHub 開源實作：https://github.com/SeanLinH/adk_tutor