AG Day 29 多代理架構:supervisor 與 worker
執行需求:CPU 可跑。在 AG Day 28(原文連結)我們把 research-agent 接上了 MCP client,讓代理可以透過 Model Context Protocol 呼叫外部 MCP server 提供的工具與資源,工具生態一下子從「我們自己寫的幾個函式」擴大到「任何遵循 MCP 規格的伺服器」。這件事帶來一個新問題:當可用工具、可查詢的知識來源越來越多,單一代理要在同一次思考裡同時扮演「搜尋專家」「資料整理員」「報告撰寫者」,決策負擔會明顯上升,工具選錯、選重複、選了不相關工具的機率也跟著提高。今天我們要把 research-agent 從單一代理拆成「supervisor(主管)+ worker(工作者)」的多代理架構:由一個 supervisor 節點負責判斷目前該交給哪一個 worker 處理,worker 各自專心做一件事,做完就把結果交回。這個模式很適合在本機 CPU 上先跑起來,因為我們可以用規則式路由與離線模擬資料把整個骨架搭好,不需要任何 API 金鑰就能看到多代理協作的雛形。
引言
回顧一下 research-agent 目前的樣子:從 AG Day 6 手刻的第一個代理迴圈,到 AG Day 14 換成 create_react_agent、AG Day 18 拆出子圖、AG Day 23 加上引用、AG Day 28 接上 MCP,這條線一直是「一個代理、一份工具箱」的思路,只是工具箱越裝越滿。這種做法在工具數量不多時很好用:模型的系統提示詞裡列出所有工具說明,模型自己判斷該呼叫哪一個。但當工具箱裡同時有搜尋、檢索、寫報告、呼叫 MCP server 上的各種資源時,單一次決策要考慮的選項太多,容易出現「該搜尋卻跑去檢索」「該收尾卻又搜尋一輪」的抖動行為,而且每一次思考都要把落落長的工具說明整段塞進上下文,成本與延遲都會上升。
多代理架構的核心想法很單純:把「決定接下來做什麼」跟「實際去做」拆成兩種角色。supervisor 只負責路由決策,它看到的狀態是精簡過的摘要,不用背下所有工具的細節;worker 只負責把分配到的一件事做好,它的提示詞可以窄而深,不必知道其他 worker 在幹什麼。今天的目標是把這個分工骨架用 LangGraph 搭出來,並且讓它能在完全沒有 API 金鑰的離線模擬模式下跑完一輪「搜尋 → 檢索 → 寫報告」的迴圈,驗證架構本身是通的,之後再逐步把真實的模型呼叫接進去。
原理/觀念
為什麼要拆成多代理
單一代理的擴展性瓶頸主要來自三個地方。第一是決策負擔:工具越多,模型每次要在更大的選項空間裡挑一個,錯誤率會隨選項數量上升,這是提示工程裡常見的現象,並不是模型不夠聰明,而是選擇本身變難了。第二是上下文膨脹:所有工具的說明、所有歷史訊息都堆在同一個對話裡,隨著任務進行,上下文視窗越吃越滿,回應速度與品質都會受影響。第三是單一失敗點:只要這一個代理在某一步卡住或誤判,整條任務就停擺,沒有其他角色可以介入或分攤。多代理架構把這三個問題各自拆開處理:路由決策交給 supervisor,一次只想「接下來要哪一種能力」;實際工作交給對應的 worker,各自只需要知道自己那一份的工具與提示詞;就算某個 worker 出錯,supervisor 也還能決定重試、換一個 worker,或直接進入下一步。
Supervisor-Worker 模式:集中式路由
今天採用的是「集中式」的多代理拓樸:所有控制權都經過 supervisor,worker 做完事一定先回到 supervisor,再由它決定下一步,不會有 worker 直接把工作丟給另一個 worker。這種寫法的好處是流程好推理、好除錯——你永遠可以從 supervisor 這一個節點看懂「現在為什麼往這裡走」,也方便加上步數上限、逾時保護等統一的守門邏輯。缺點是 supervisor 會成為流量的必經之路,worker 之間沒辦法直接交換細節,只能透過共用狀態傳遞。AG Day 30(原文連結)會介紹另一種「去中心化交接」的通訊協議,讓 worker 之間可以帶著結構化的訊息互相交棒,屆時你會更清楚兩種拓樸的取捨。
用 LangGraph 表示多代理圖
在 LangGraph 裡,多代理系統其實就是「每個代理都是一個節點」的 StateGraph:supervisor 是一個節點,每個 worker 也是一個節點,它們共用同一份狀態(state)。supervisor 執行完之後,用條件邊(conditional edge)依照它寫入狀態的決策值,動態決定走到哪一個 worker 節點;每個 worker 執行完之後,固定的邊會把控制權送回 supervisor。整張圖因此長得像一個以 supervisor 為中心的星狀結構,這跟 AG Day 13 學過的條件邊、AG Day 18 學過的子圖是同一套機制的延伸應用,只是這次每個分支節點本身就是一個具備獨立職責的「代理」。
完整實作
我們在 research-agent/src/research_agent/ 底下新增一個 agents/ 子套件,專門放多代理相關的程式碼,跟既有的 graph.py(單一代理 ReAct 流程)分開,互不干擾:
mkdir -p research-agent/src/research_agent/agents
mkdir -p research-agent/scripts
touch research-agent/src/research_agent/agents/__init__.py
touch research-agent/src/research_agent/agents/state.py
touch research-agent/src/research_agent/agents/supervisor.py
touch research-agent/src/research_agent/agents/workers.py
touch research-agent/src/research_agent/agents/graph.py
第一步:定義所有代理共用的狀態結構 agents/state.py。這份狀態會在 supervisor 與三個 worker 之間傳遞,每個節點回傳一個部分更新的字典,LangGraph 會把它併入整體狀態:
# research-agent/src/research_agent/agents/state.py
from typing import TypedDict, Optional
from typing_extensions import Annotated
from langgraph.graph.message import add_messages
class TeamState(TypedDict):
"""多代理共用狀態:supervisor 與所有 worker 都讀寫同一份狀態。"""
messages: Annotated[list, add_messages]
topic: str
search_findings: list[dict]
retrieved_chunks: list[dict]
final_report: Optional[str]
active_worker: Optional[str]
step_count: int
第二步:實作 agents/supervisor.py。supervisor 每一輪要回答一個結構化的路由決策,這裡沿用 AG Day 8 建立的 Pydantic 結構化輸出習慣,並讓它在離線模擬模式下退回規則式路由,確保沒有金鑰也能跑:
# research-agent/src/research_agent/agents/supervisor.py
from typing import Literal
from pydantic import BaseModel, Field
from research_agent.config import load_settings
from research_agent.agents.state import TeamState
WorkerName = Literal["search_worker", "retrieval_worker", "writer_worker", "FINISH"]
class RouteDecision(BaseModel):
"""supervisor 每一輪要回答的結構化決策。"""
next_worker: WorkerName = Field(description="下一個要執行的 worker,或 FINISH 代表任務完成")
reason: str = Field(description="為什麼選這個 worker 的簡短理由")
def supervisor_node(state: TeamState) -> dict:
"""根據目前狀態決定下一個 worker;沒有 LLM 金鑰時退回規則式路由。"""
settings = load_settings()
step = state.get("step_count", 0)
decision = _rule_based_route(state) if settings.is_dry_run else _llm_route(state)
return {
"active_worker": decision.next_worker,
"step_count": step + 1,
"messages": [
{"role": "system", "content": f"supervisor 決策:{decision.next_worker}({decision.reason})"}
],
}
def _rule_based_route(state: TeamState) -> RouteDecision:
"""離線模擬模式下的規則路由:先搜尋、再檢索、最後寫報告。"""
if not state.get("search_findings"):
return RouteDecision(next_worker="search_worker", reason="尚未有搜尋結果,先派 search_worker")
if not state.get("retrieved_chunks"):
return RouteDecision(next_worker="retrieval_worker", reason="已有搜尋結果,交給 retrieval_worker 整理成可引用片段")
if not state.get("final_report"):
return RouteDecision(next_worker="writer_worker", reason="素材齊備,交給 writer_worker 寫報告")
return RouteDecision(next_worker="FINISH", reason="報告已完成,結束多代理流程")
def _llm_route(state: TeamState) -> RouteDecision:
"""正式模式:交由 RESEARCH_AGENT_MODEL 用結構化輸出判斷分工,銜接 AG Day 3-8 的 llm.chat。"""
raise NotImplementedError("需搭配既有 llm.chat(response_model=RouteDecision) 實作;今天先驗證離線骨架")
第三步:實作三個 worker,放在 agents/workers.py。每個 worker 都遵守同一個約定:讀取自己需要的狀態欄位,做事,回傳部分更新的字典。離線模擬模式下用固定的示範資料,讓整條流程可以在沒有金鑰、沒有網路的情況下跑完:
# research-agent/src/research_agent/agents/workers.py(1/3:search_worker)
from research_agent.config import load_settings
from research_agent.agents.state import TeamState
def search_worker(state: TeamState) -> dict:
"""負責對外搜尋,正式模式呼叫 AG Day 24 建立的 web_search 工具。"""
settings = load_settings()
topic = state["topic"]
if settings.is_dry_run:
findings = [
{"title": f"{topic} 示範資料 1", "url": "https://example.com/demo-1", "snippet": "示範搜尋結果,非真實網路資料。"},
{"title": f"{topic} 示範資料 2", "url": "https://example.com/demo-2", "snippet": "示範搜尋結果,非真實網路資料。"},
]
else:
from research_agent.tools import web_search # AG Day 24 建立的搜尋工具
findings = web_search(topic, max_results=5)
return {
"search_findings": findings,
"messages": [{"role": "system", "content": f"search_worker 完成,取得 {len(findings)} 筆資料"}],
}
# research-agent/src/research_agent/agents/workers.py(2/3:retrieval_worker)
def retrieval_worker(state: TeamState) -> dict:
"""把 search_worker 的原始資料整理成帶引用資訊的片段,呼應 AG Day 20-23 的檢索與引用設計。"""
settings = load_settings()
findings = state.get("search_findings", [])
if settings.is_dry_run:
chunks = [
{"text": item["snippet"], "source_url": item["url"], "source_title": item["title"]}
for item in findings
]
else:
from research_agent.retrieval import retrieve # AG Day 20-23 建立的檢索函式
chunks = retrieve(state["topic"], top_k=5)
return {
"retrieved_chunks": chunks,
"messages": [{"role": "system", "content": f"retrieval_worker 完成,整理出 {len(chunks)} 個可引用片段"}],
}
# research-agent/src/research_agent/agents/workers.py(3/3:writer_worker)
def writer_worker(state: TeamState) -> dict:
"""把檢索片段組成帶引用的研究報告,呼應 AG Day 23 的引用格式。"""
settings = load_settings()
chunks = state.get("retrieved_chunks", [])
topic = state["topic"]
if settings.is_dry_run:
lines = [f"# {topic}(示範報告,離線模擬模式)", ""]
for i, chunk in enumerate(chunks, start=1):
lines.append(f"{i}. {chunk['text']}[來源:{chunk['source_title']}]")
report = "\n".join(lines)
else:
from research_agent.report import compose_report # AG Day 23 建立的報告產生函式
report = compose_report(topic, chunks)
return {
"final_report": report,
"messages": [{"role": "system", "content": "writer_worker 完成,報告已寫入 final_report"}],
}
第四步:在 agents/graph.py 把 supervisor 與三個 worker 組成一張圖。supervisor 之後接條件邊,依照它剛寫入的 active_worker 動態決定走向;每個 worker 執行完固定回到 supervisor,形成一個以 supervisor 為中心的迴圈:
# research-agent/src/research_agent/agents/graph.py
from langgraph.graph import StateGraph, START, END
from research_agent.agents.state import TeamState
from research_agent.agents.supervisor import supervisor_node
from research_agent.agents.workers import search_worker, retrieval_worker, writer_worker
MAX_TEAM_STEPS = 8
def _route_from_supervisor(state: TeamState) -> str:
"""讀取 supervisor 剛寫入的 active_worker,決定走哪一條邊;步數超過上限強制收尾。"""
if state.get("step_count", 0) >= MAX_TEAM_STEPS:
return "FINISH"
return state["active_worker"]
def build_team_graph():
"""組出 supervisor -> worker -> supervisor 的多代理圖。"""
graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("search_worker", search_worker)
graph.add_node("retrieval_worker", retrieval_worker)
graph.add_node("writer_worker", writer_worker)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
"supervisor",
_route_from_supervisor,
{
"search_worker": "search_worker",
"retrieval_worker": "retrieval_worker",
"writer_worker": "writer_worker",
"FINISH": END,
},
)
for worker in ("search_worker", "retrieval_worker", "writer_worker"):
graph.add_edge(worker, "supervisor")
return graph.compile()
第五步:寫一個示範腳本 scripts/run_team.py,跑一次完整的多代理流程並印出最終報告:
# research-agent/scripts/run_team.py
from research_agent.agents.graph import build_team_graph
def main():
team = build_team_graph()
initial_state = {
"messages": [],
"topic": "多代理架構的優勢與限制",
"search_findings": [],
"retrieved_chunks": [],
"final_report": None,
"active_worker": None,
"step_count": 0,
}
result = team.invoke(initial_state)
print("=== 多代理流程結束 ===")
print(f"總共執行 {result['step_count']} 輪 supervisor 決策")
print("--- 最終報告(離線模擬模式示範輸出) ---")
print(result["final_report"])
if __name__ == "__main__":
main()
在專案根目錄執行:
cd research-agent
uv run python scripts/run_team.py
示範輸出(離線模擬模式,實際字句會依模擬資料而定):
=== 多代理流程結束 ===
總共執行 4 輪 supervisor 決策
--- 最終報告(離線模擬模式示範輸出) ---
# 多代理架構的優勢與限制(示範報告,離線模擬模式)
1. 示範搜尋結果,非真實網路資料。[來源:多代理架構的優勢與限制 示範資料 1]
2. 示範搜尋結果,非真實網路資料。[來源:多代理架構的優勢與限制 示範資料 2]
這個示範輸出雖然簡陋,但已經完整走了一輪「supervisor 派工 → search_worker → supervisor 派工 → retrieval_worker → supervisor 派工 → writer_worker → supervisor 判定完成」的迴圈,而且全程不需要任何 API 金鑰。等到把 _llm_route 與各 worker 裡的正式呼叫接上既有的 llm.py、tools.py、retrieval.py、report.py,只要環境變數裡有 OPENAI_API_KEY 或 ANTHROPIC_API_KEY,同一張圖就能無縫切換成真實模式,不需要改動圖的拓樸。
常見錯誤與踩雷
第一個常見錯誤是 supervisor 陷入無限迴圈:如果路由邏輯有漏洞、或某個 worker 沒有正確更新它該負責的狀態欄位,supervisor 可能永遠判斷不到「已完成」,一直派同一個 worker。今天的 _route_from_supervisor 特別加了 MAX_TEAM_STEPS 步數上限,這跟 AG Day 7 學過的錢包保護邏輯是同一個精神:任何會反覆執行的迴圈都要有一個絕對會觸發的終止條件,不能只依賴「正常情況下」的邏輯判斷。
第二個常見錯誤是多個節點同時寫入同一個狀態欄位卻沒有指定合併方式。messages 欄位用了 add_messages reducer,所以多次寫入會自動累加而不是互相覆蓋;但像 search_findings、retrieved_chunks 這類欄位如果之後改成讓多個 worker 平行寫入,就必須額外設計合併邏輯,否則後寫入的節點會把前面的結果整段蓋掉,這個坑會在 AG Day 30 討論狀態交接協議時進一步展開。
第三個常見錯誤是把整份對話歷史一股腦塞給每個 worker。因為所有節點共用同一份 messages,很容易誤以為 worker 應該讀取全部歷史再做事,但這會讓 worker 的提示詞跟單一代理時一樣肥大,失去拆分的意義。正確做法是讓每個 worker 只讀它真正需要的欄位(例如 topic、search_findings),messages 留給人類除錯與後續的可觀測性用途,不當作 worker 決策的輸入。
第四個常見錯誤是忘記在離線模擬模式的輸出上標示清楚。今天所有示範資料都明確寫了「示範」「非真實網路資料」字樣,這一點在正式專案裡務必保留,避免讀者或使用者把模擬輸出誤認為真實的搜尋結果或研究結論。
效能與實務提醒
supervisor 每一輪決策如果是真實呼叫模型,就會多一次網路往返與一次計費;worker 數量越多、流程繞的圈數越多,成本與延遲就疊得越高。實務上建議先把 worker 數量控制在三到五個以內,把每個 worker 的職責切得夠清楚(一個 worker 只做一件明確的事),這樣 supervisor 的路由決策才會單純,也比較容易在 AG Day 32 之後導入評估時看出哪一段在拖累整體表現。
另外,規則式路由不只是離線模擬的替代方案,在正式環境裡也很有價值:如果任務的流程本身是固定的(例如永遠是「搜尋 → 檢索 → 寫報告」),用規則式路由取代每一輪都呼叫模型的路由,可以省下大量不必要的 API 呼叫,只有在流程真的需要動態判斷分支時才交給模型決策。這是多代理系統裡常被忽略的一個省錢技巧:不是所有決策都值得用大型語言模型去做。
最後,多代理圖的除錯難度比單一代理高,因為問題可能出現在 supervisor 的路由邏輯、也可能出現在某個 worker 內部。建議善用今天 messages 裡累積的系統訊息當作最基本的執行軌跡,搭配 AG Day 9 建立的 token 與延遲紀錄,先確認「走的路徑對不對」,再深入某個 worker 內部除錯,會比一開始就鑽進單一節點更有效率。
小結
今天我們把 research-agent 從單一代理拆成 supervisor 與三個 worker 的多代理架構:state.py 定義共用狀態,supervisor.py 用結構化的 RouteDecision 決定下一步(離線模擬時退回規則式路由),workers.py 實作 search_worker、retrieval_worker、writer_worker 三個各自專心做一件事的節點,graph.py 把它們組成以 supervisor 為中心的星狀圖。整條流程在完全沒有 API 金鑰的情況下也能跑完,方便先驗證架構、之後再逐步接上真實模型呼叫。
幾個今天新增的術語:supervisor(主管代理,負責路由決策)、worker(工作代理,負責執行單一職責)、集中式路由(所有控制權經過同一個節點)。這些概念會在接下來幾天持續出現,值得記在筆記裡。
結語
今天 worker 之間傳遞資料的方式相當粗糙:全部塞進同一份大狀態,靠欄位名稱約定該讀哪裡、該寫哪裡,沒有明確的訊息格式,也沒有處理「worker 失敗了該怎麼回報」這種情況。這在骨架驗證階段沒問題,但放到真實專案裡遲早會出狀況——欄位一多就容易對不齊,錯誤也無從追查起。
明天,我們會進入「AG Day 30 多代理通訊:狀態交接與訊息協議」,設計一套結構化的訊息協議,讓 worker 回報結果時帶著明確的狀態、摘要與後續建議,而不是隨手塞幾個字典欄位;也會談談去中心化的交接模式,跟今天集中式的 supervisor 路由做個對比。
延伸資源
- LangGraph 官方文件:
https://langchain-ai.github.io/langgraph/。多代理系統(Multi-agent systems)章節說明了 supervisor 與 handoff 兩種常見拓樸,本篇的節點與條件邊寫法以此為準。 - LangGraph 官方文件的
StateGraph與條件邊(Conditional edges)章節:本篇add_conditional_edges的用法與參數以官方文件為準。 - Pydantic 官方文件:
https://docs.pydantic.dev/。RouteDecision的欄位型別與Literal用法可在此查閱完整規格。
留言
張貼留言