在 昨天 Day 23,
我們讓單一 Agent 學會了「反思」與「記憶」——
它能在任務結束後回顧、修正、並保留經驗。
但在真實應用中,任務往往不止一個面向。
就像規劃旅行,不只是要考慮天氣,還需要同時處理交通、景點、預算、飲食偏好等資訊。
若交給單一 Agent 負責所有任務,常會出現:
為了解決這些問題,就需要 分工合作。
其實在 基礎篇 Day 11,
我們曾介紹過最基本的 Multi-Agent 與 Supervisor(統籌型) 模式:
由中央的主控 Agent 負責協調多位專家 Agent 各司其職,
如同指揮與樂手,共同完成複雜任務。
今天,我們將延續維也納旅遊助理的情境,
透過 LangChain v1.0 來實現 Multi-Agent 架構,
打造一個能「協作」的系統:
由一位 統籌 Supervisor 協調三位專家 Agent,
共同完成任務決策。
即使具備記憶與反思,單一 Agent 仍容易陷入「多工超負荷」的狀況:
它要同時推理、查詢、分析與規劃,結果效率下降、品質不穩。
Multi-Agent 的關鍵理念是:
讓每個 Agent 只專注一件事,並透過協作完成複雜任務。
問題情境 | 單一 Agent 的限制 | Multi-Agent 解法 |
---|---|---|
工具過多、選擇混亂 | 無法決定何時使用何種工具 | 由 Supervisor 控制呼叫流程 |
記憶過大 | 所有資訊堆疊在同一模型中 | 每位 Agent 僅保留必要上下文 |
缺乏專業化 | 同一模型需同時處理不同領域 | 以角色分工(氣象 / 規劃 / 顧問) |
LangChain v1.0 將 Multi-Agent 的實作歸納為兩種主要模式(官方文件說明):
今天我們聚焦第一種——Supervisor × Tool Calling,
透過一個多角色協作的 維也納旅遊助理,展示如何讓多位 Agent 分工合作,產出整合性決策。
今天採用 Tool Calling Pattern,我們將系統分為四個角色,
由 Supervisor(統籌 Agent) 控制整個任務流程,
依需求呼叫三位專家:
圖:Supervisor 採用集中式協作模式,負責統籌三位專家 Agent。每位子 Agent 具備不同任務與工具,最終由 Supervisor 整合所有回應形成完整決策。
接下來的程式示範,
展示如何在 LangChain v1.0 中,以 Supervisor 控制多位專家 Agent 的協作。
import asyncio
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langchain_mcp_adapters.client import MultiServerMCPClient
import os
# 初始化 MCP 工具
async def init_tools():
accuweather_api_key = os.getenv("ACCUWEATHER_API_KEY")
if not accuweather_api_key:
raise ValueError("請設定 ACCUWEATHER_API_KEY 環境變數")
client = MultiServerMCPClient(
{
"accuweather": {
"transport": "stdio",
"command": "uv",
"args": [
"--directory",
"{MCP 程式路徑}",
"run",
"weather_mcp_server.py"
],
"env": {"ACCUWEATHER_API_KEY": accuweather_api_key}
},
"attractions": {
"transport": "stdio",
"command": "uv",
"args": [
"--directory",
"{MCP 程式路徑}",
"run",
"attractions_mcp_server.py"
]
}
}
)
return await client.get_tools()
# 建立三位專家 Agent
async def build_subagents(tools):
model = init_chat_model("gemini-2.5-flash", model_provider="google_genai")
checkpointer = InMemorySaver()
weather_agent = create_agent(
model=model,
tools=[t for t in tools if "weather" in t.name.lower()],
prompt="你是氣象專家,負責分析目的地天氣狀況與對行程的影響。",
checkpointer=checkpointer,
)
planner_agent = create_agent(
model=model,
tools=[t for t in tools if "attractions" in t.name.lower()],
prompt="你是旅遊規劃師,根據天氣與景點資料設計一日行程。",
checkpointer=checkpointer,
)
advisor_agent = create_agent(
model=model,
tools=[],
prompt="你是旅遊顧問,負責整合天氣與行程結果,提出最終建議。",
checkpointer=checkpointer,
)
return weather_agent, planner_agent, advisor_agent
# Supervisor 呼叫子 Agent
def make_supervisor(weather_agent, planner_agent, advisor_agent):
@tool("WeatherAgent", description="查詢目的地天氣並分析其對行程的影響")
async def call_weather(query: str):
result = await weather_agent.ainvoke(
{"messages": [{"role": "user", "content": query}]},
{"configurable": {"thread_id": "1"}}
)
print("\n[WeatherAgent] 輸出:")
print(result["messages"][-1].content)
return result["messages"][-1].content
@tool("PlannerAgent", description="規劃維也納一日行程,考慮天氣與景點")
async def call_planner(query: str):
result = await planner_agent.ainvoke(
{"messages": [{"role": "user", "content": query}]},
{"configurable": {"thread_id": "1"}}
)
print("\n[PlannerAgent] 輸出:")
print(result["messages"][-1].content)
return result["messages"][-1].content
@tool("AdvisorAgent", description="整合各 Agent 輸出,提供最終建議")
async def call_advisor(query: str):
result = await advisor_agent.ainvoke(
{"messages": [{"role": "user", "content": query}]},
{"configurable": {"thread_id": "1"}}
)
print("\n[AdvisorAgent] 輸出:")
print(result["messages"][-1].content)
return result["messages"][-1].content
model = init_chat_model("gemini-2.5-flash", model_provider="google_genai")
supervisor = create_agent(
model=model,
tools=[call_weather, call_planner, call_advisor],
prompt=(
"你是旅行統籌助理。根據使用者需求,自行決定何時呼叫氣象專家、旅遊規劃師與顧問。"
"最後整合所有子 Agent 結果,輸出完整建議。"
),
checkpointer=InMemorySaver(),
)
return supervisor
# 主程式
async def main():
tools = await init_tools()
weather_agent, planner_agent, advisor_agent = await build_subagents(tools)
supervisor = make_supervisor(weather_agent, planner_agent, advisor_agent)
result = await supervisor.ainvoke(
{
"messages": [{
"role": "user",
"content": "幫我規劃維也納一日遊,請考慮天氣與適合的行程安排。"
}]
},
{"configurable": {"thread_id": "1"}}
)
print("\n=== 最終整合輸出 ===")
print(result["messages"][-1].content)
if __name__ == "__main__":
asyncio.run(main())
在這個範例中,我們可以清楚觀察到 Supervisor × Multi-Agent 協作流程 的實際運作方式。
整個系統的決策分為三個層次:
首先,三位專家 Agent 各自根據職能完成任務:
WeatherAgent 負責解析天氣狀況並提供氣候建議;
PlannerAgent 根據天氣與景點資料規劃路線;
AdvisorAgent 則整合前兩者輸出,產出具體建議。
Supervisor 並不直接行動,而是像導演般掌握流程,
自動決定應先諮詢哪位專家、何時串聯結果,
讓整個決策過程更有條理與透明度。
圖:三位專家 Agent 並行協作的執行過程。Supervisor 根據使用者請求動態呼叫 WeatherAgent、PlannerAgent 與 AdvisorAgent,形成分工明確的多角色決策流程。
接著,Supervisor 將三位專家的回覆整合後,
生成一份完整且具彈性的「維也納一日遊行程建議」。
內容不僅包含天氣分析與景點順序,
還提出替代方案與行程優化原則——
顯示模型具備「理解 → 分析 → 整合 → 建議」的多層推理能力。
圖:Supervisor 整合各專家輸出後生成最終行程建議。模型綜合天氣、開放時間與景點資訊,輸出具備彈性與可操作性的完整決策結果。
透過這樣的分工協作機制,
整個系統就像一支具備「氣象顧問 × 行程設計師 × 旅遊顧問」的智慧團隊:
專家各司其職,Supervisor 負責決策統籌,
最終輸出更合理、更具上下文一致性的結果。
回顧前幾天的進化脈絡:
Reflection 讓 Agent 學會「自我檢討」,
Memory 讓它「記得經驗」,
而 Multi-Agent 則讓它「學會與他人協作」。
這是智慧體從「單一思考者」進化為「協作系統」的關鍵一步。
今天,我們讓多位 Agent 學會了「合作」。
透過 Supervisor × Multi-Agent 架構,
模型不再單打獨鬥,而是能:
這不僅是技術架構的升級,更是 Agent 思維模式的轉變:
從「單一思考的 AI」邁向「協作運作的團隊式 AI」,
讓模型不只是能思考、能學習,更能 協同工作、共同決策。
圖:維也納金色大廳(Großer Musikvereinssaal)。金碧輝煌的天花與懸掛的水晶吊燈,構成協奏的視覺節奏。每一場演出都由不同樂手在同一份樂譜下協力完成,如同 Multi-Agent 架構中各個專家協作於同一任務,由 Supervisor 統籌指揮,最終奏出和諧一致的智慧交響。(攝影:作者自攝)