NLP Day 35 多 Agent 協作與任務分解
執行需求:需 API Key(Ollama 本地替代可行)。單一 Agent 在處理單一任務時表現不錯,但企業的真實需求常常更複雜:「分析客戶的退貨資料,輸出摘要並寄給客服主管」、「讀客服信、自動分類、彙整成週報」。這種任務沒有一個 Agent 能獨立做——需要讀資料的、做分類的、寫週報的,再加上監督整個流程的協調者。多 Agent 框架正是把多個專職 Agent 串起來的工具。今天這一篇會介紹兩個主流的多 Agent 框架(LangGraph 與 AutoGen/CrewAI 概念),示範「主管—專家」這個最常見的協作模式,並用一個「合約代用賽局」把昨天的 MCP server 串進來。
引言
單一 Agent 為什麼會卡住?兩個原因:第一,prompt 一次塞太多任務會讓模型分心,品質下降;第二,工具一多,模型選擇工具的錯誤率會明顯上升。多 Agent 框架的解法是「分工」:每個 Agent 只負責一小塊任務,彼此透過訊息交換串起來。今天的範例用三個 Agent:研究 Agent(讀 MCP server 的合約資料)、分析 Agent(從合約資料萃取重點)、寫作 Agent(把分析結果整理成一段給主管看的摘要)。再加上一個主管 Agent 負責分配任務與彙整結果。
今天會帶到兩個實作框架:LangGraph(LangChain 團隊在 2024 年推出的狀態機框架,0.2 世代)和 AutoGen(微軟研究院在 2024 年公開的多 Agent 框架,活躍版本在 2025 年初)。兩個框架的設計哲學不同:LangGraph 把 Agent 之間的協作寫成「節點 + 邊」的狀態圖,適合明確流程;AutoGen 把 Agent 視為對等角色,透過訊息對話合作,適合開放式討論。今天的範例會用 LangGraph 的狀態圖,因為它對「主管—專家」模式的支援最直接。
為什麼需要多 Agent
多 Agent 的價值不是「讓更多 LLM 互相聊天」,而是「把複雜任務拆解成可以獨立驗證的子任務」。一個流程如果只有一個 Agent 出錯,你很難判斷是哪個環節錯了;拆成多 Agent 後,每個 Agent 的輸出都可以被單獨測試、單獨評估。實務上我們會對每個 Agent 寫獨立的評估腳本,確保單一 Agent 的輸出符合預期,再把它串進流程。
另一個價值是「專業化」。通用 Agent 對所有任務都「差不多會」,專職 Agent 可以針對單一任務深度打磨 prompt 的工具、試與範例。例如一個客服分類 Agent 的 prompt 可以專門針對「判斷客戶情緒嚴重程度」設計,搭配 few-shot 範例可以達到比通用 Agent 高 20% 的準確率。
LangGraph:狀態機式的多 Agent 框架
LangGraph 是 LangChain 團隊在 2024 年 8 月推出的狀態機框架,設計目標是把 LLM 應用的流程寫成一張「節點—邊」圖。每個節點是一個函式(通常呼叫 LLM),邊代表控制流向(條件式或預設式)。這個設計讓多 Agent 協作可以被明確表達與除錯,跟傳統「一連串 prompt 串接」的風格有明顯差別。
一個 LangGraph 程式通常由四個部分組成:State(狀態物件,用 TypedDict 定義每個欄位)、Node(節點函式,吃 state 吐 state)、Edge(連接兩個節點,可以是預設邊或條件邊)、Graph(把節點與邊組合成圖)。執行時,LangGraph 從起始節點出發,沿著邊走到結束節點,每次經過節點都會更新狀態。我們的「主管—專家」範例就是這個架構的標準應用。
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
class TeamState(TypedDict):
task: str
research: str
analysis: str
draft: str
final: str
這段程式定義了多 Agent 流程的狀態物件。`task` 是使用者的原始任務;`research` 是研究 Agent 的輸出;`analysis` 是分析 Agent 的輸出;`draft` 與 `final` 是寫作 Agent 的中間稿與最終稿。所有 Agent 讀寫的都是這份共享物件,這是 LangGraph 與訊息對話式框架(AutoGen)最大的差別——前者是「共享狀態」,後者是「訊息流」。
定義專家 Agent 節點
每個專家 Agent 是一個節點函式,吃進狀態、呼叫 LLM、把結果寫回狀態。我們設計三個專家:研究 Agent 負責從 MCP server 查合約資料;分析 Agent 負責從合約資料萃取重點;寫作 Agent 負責把分析結果寫成摘要。每個節點的 prompt 都針對單一任務設計,不混雜其他責任。
from openai import OpenAI
import os
def call_llm(system: str, user: str) -> str:
client = OpenAI() if os.environ.get("OPENAI_API_KEY") else None
if client:
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
temperature=0,
)
return resp.choices[0].message.content
import ollama
return ollama.chat(model="qwen2.5:7b", messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
])["message"]["content"]
def research_node(state: TeamState) -> TeamState:
"""研究 Agent:從合約系統查資料。"""
contracts = ["A123", "B456", "C789"] # 模擬從 MCP server 取得的合約清單
user = f"任務:{state['task']}\n可用合約:{contracts}"
state["research"] = call_llm(
"你是研究 Agent。任務是判斷哪些合約與使用者任務相關,並回傳合約編號。",
user,
)
return state
def analysis_node(state: TeamState) -> TeamState:
"""分析 Agent:從合約資料萃取重點。"""
return {**state, "analysis": call_llm(
"你是分析 Agent。任務是從合約資料中萃取三個關鍵數字與一段結論。",
f"任務:{state['task']}\n研究結果:{state['research']}",
)}
def writing_node(state: TeamState) -> TeamState:
"""寫作 Agent:把分析結果寫成摘要。"""
return {**state, "draft": call_llm(
"你是寫作 Agent。任務是把分析結果寫成 100 字內、給主管看的摘要。",
f"分析結果:{state['analysis']}",
)}
這段程式定義了三個專家 Agent 的節點函式。每個函式都呼叫昨天寫好的 `call_llm()` 助手(自動選 OpenAI 或 Ollama 後端),用不同的系統提示與使用者訊息產生單一任務的回應。注意每個 Agent 的系統提示都很短、很專注——「你是研究 Agent」、「你是分析 Agent」、「你是寫作 Agent」——這是專家 Agent 的核心:角色單一、責任明確。
研究中 Agent 的 `contracts` 變數是簡化版的 MCP 呼叫,實際上會用昨天的 MCP 客戶端呼叫 `list_contracts` 工具。我們在這裡硬寫清單是為了讓範例離線可執行;接上 MCP server 的版本會在文末的延伸練習給出。
主管 Agent:任務分解與結果彙整
主管 Agent 不做實際工作,只負責「任務分解」與「結果彙整」。它的角色像專案經理:接到任務後決定要叫哪幾位專家(這裡是固定流程,但實務上可以動態判斷);專家完成後彙整他們的結果、檢查品質、再決定要不要請專家重做。今天我們把主管 Agent 簡化為 LangGraph 的條件邊。
def supervisor_node(state: TeamState) -> TeamState:
"""主管 Agent:彙整最終結果並檢查品質。"""
state["final"] = call_llm(
"你是主管 Agent。任務是檢查寫作 Agent 的草稿是否符合任務需求,若不符合請重新整理。",
f"任務:{state['task']}\n草稿:{state['draft']}",
)
return state
graph = StateGraph(TeamState)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_node("writing", writing_node)
graph.add_node("supervisor", supervisor_node)
graph.add_edge(START, "research")
graph.add_edge("research", "analysis")
graph.add_edge("analysis", "writing")
graph.add_edge("writing", "supervisor")
graph.add_edge("supervisor", END)
app = graph.compile()
這段程式把主管 Agent 與三個專家 Agent 接成狀態圖:`START → research → analysis → writing → supervisor → END`。每個節點執行完都會更新狀態物件,下一個節點從狀態物件讀取自己需要的部分。`graph.compile()` 把圖編譯成可執行物件,`app.invoke()` 啟動整個流程。LangGraph 在內部會處理節點之間的訊息傳遞與狀態管理。
把昨天的 MCP server 串進來
昨天的 MCP server 提供 `list_contracts` 與 `get_contract_info` 兩個工具,今天的研究 Agent 可以呼叫它。我們用昨天示範的 `ClientSession` 改寫 `research_node`,讓研究 Agent 真的透過 MCP 通訊拿到合約資料,而不是用寫死的清單。
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def fetch_contracts() -> list[str]:
params = StdioServerParameters(command="python", args=["-m", "contracts_mcp"])
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
res = await session.call_tool("list_contracts", {})
return eval(res.content[0].text)
def research_node_mcp(state: TeamState) -> TeamState:
"""研究 Agent:透過 MCP server 取得合約清單。"""
contracts = asyncio.run(fetch_contracts())
return {**state, "research": call_llm(
"你是研究 Agent。任務是判斷哪些合約與使用者任務相關。",
f"任務:{state['task']}\n可用合約:{contracts}",
)}
這段程式用昨天寫的 MCP 客戶端從 server 取得合約清單。注意 `asyncio.run(fetch_contracts())` 把 async 函式塞進同步的節點函式,這對簡單範例夠用,但實務上 LangGraph 支援 async 節點、可以用 `async def research_node_mcp(...)` 與 `await fetch_contracts()` 直接執行。為了程式簡潔,今天用同步版本。
完整實作:跑一次多 Agent 流程
把狀態定義、四個節點、狀態圖組裝接起來,就是今天的多 Agent 完整實作。下面的程式可以整段貼進本機執行,第一次會啟動 MCP server 子行程,後續每次執行都在秒等級完成。
import asyncio
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from openai import OpenAI
import os
class TeamState(TypedDict):
task: str
research: str
analysis: str
draft: str
final: str
def call_llm(system: str, user: str) -> str:
if os.environ.get("OPENAI_API_KEY"):
client = OpenAI()
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
temperature=0,
).choices[0].message.content
import ollama
return ollama.chat(model="qwen2.5:7b", messages=[
{"role": "system", "content": system}, {"role": "user", "content": user}
])["message"]["content"]
def research(state): return {**state, "research": call_llm("你是研究 Agent", state["task"])}
def analysis(state): return {**state, "analysis": call_llm("你是分析 Agent", state["research"])}
def writing(state): return {**state, "draft": call_llm("你是寫作 Agent", state["analysis"])}
def supervisor(state): return {**state, "final": call_llm("你是主管 Agent", f"{state['task']}\n{state['draft']}")}
graph = StateGraph(TeamState)
graph.add_node("research", research)
graph.add_node("analysis", analysis)
graph.add_node("writing", writing)
graph.add_node("supervisor", supervisor)
graph.add_edge(START, "research")
graph.add_edge("research", "analysis")
graph.add_edge("analysis", "writing")
graph.add_edge("writing", "supervisor")
graph.add_edge("supervisor", END)
app = graph.compile()
result = app.invoke({"task": "列出今年到期的合約並提醒主管", "research": "", "analysis": "", "draft": "", "final": ""})
print(result["final"])
# 輸出(實際結果會略有不同):合約 B456 將於 2025-08-15 到期、合約 C789 將於 2025-06-30 到期,建議提前 60 天續約...
這段完整實作展示了 LangGraph 從狀態定義到結果輸出的完整鏈路。我們傳入任務「列出今年到期的合約並提醒主管」,流程會依序經過四個 Agent,最後印出主管 Agent 的彙整結果。輸出會包含到期合約清單與建議行動(依模型與任務而略有不同)。這是多 Agent 協作的最小可行產品,可以直接放進 Line Bot 或網頁應用。
常見錯誤與踩雷
第一個常見踩雷是「把太多東西塞進單一節點」。我們的 `supervisor_node` 同時做「品質檢查」與「彙整結果」,這在簡單場景沒問題,但任務複雜時會讓主管 Agent 的 prompt 過長、品質下降。建議主管 Agent 只做「彙整」,把品質檢查拆成另一個節點。
第二是「LangGraph 0.2 與 0.3 的 API 差異」。LangGraph 在 2024 到 2025 年之間 API 有幾次調整,`StateGraph` 與 `add_edge` 的介面大致穩定,但 `add_conditional_edges` 的條件函式簽名有變動。今天的程式鎖定 0.2.x,請在 production 升級前查閱 migration guide。
第三是「節點之間的狀態欄位忘了初始化」。LangGraph 在執行前需要把 TypedDict 的所有欄位都填好,否則會報 KeyError。我們在 `app.invoke()` 傳入完整的初始狀態,請記得補上所有欄位。
動態路由:主管 Agent 依任務分派
前面的範例主管 Agent 是固定的流程(research → analysis → writing → supervisor)。實務上更靈活的設計是「主管根據任務類型決定要叫哪些專家」。我們擴充主管節點、加上一個「分類子任務」,讓模型先判斷任務屬於研究型、分析型還是寫作型,再決定要走哪條主路徑。這種設計在 LangGraph 用 `add_conditional_edges` 實作。
from langgraph.graph import END
def classify_node(state: TeamState) -> TeamState:
"""分類 Agent:判斷任務類型。"""
state["task_type"] = call_llm(
"你是分類 Agent。請判斷任務類型,回傳 research / analysis / writing。",
state["task"],
).strip().lower()
return state
def route_by_type(state: TeamState) -> str:
"""依任務類型決定下一個節點。"""
return {
"research": "research",
"analysis": "analysis",
"writing": "writing",
}.get(state.get("task_type", "research"), "research")
graph.add_node("classify", classify_node)
graph.add_conditional_edges("classify", route_by_type)
graph.add_edge(START, "classify")
這段程式定義了 `classify_node` 與 `route_by_type` 兩個函式。`classify_node` 用 LLM 判斷任務類型並寫入狀態;`route_by_type` 依據任務類型回傳下一個節點名稱。LangGraph 的 `add_conditional_edges` 把這個動態路由掛到圖上:分類節點結束後,依 `route_by_type` 的回傳值決定走 `research`、`analysis` 或 `writing`。這個設計讓多 Agent 流程不再只有單一路徑,而是依任務特性分派。
效能與實務提醒
四個 Agent 串起來大概需要 8 至 15 秒(OpenAI gpt-4o-mini),對互動式任務偏慢。實務上可以用幾個手段加速:並行執行無相依的節點(LangGraph 支援)、把簡單任務(清單彙整)換成傳統程式(不呼叫 LLM)、快取專家 Agent 的輸出(如果任務重複性高)。
另一個常見問題是「主管 Agent 的 prompt 不好寫」。主管的 prompt 要同時涵蓋「任務理解」、「品質標準」、「彙整格式」三件事,太長會讓 LLM 分心。建議把這三件事拆成三個獨立檢查項,例如「先檢查完整性、再檢查準確度、最後檢查格式」。
另外,LangGraph 0.2.x 的 checkpointing 機制可以把每個節點的輸出序列化、存到 SQLite 或 PostgreSQL,事後可以從任何一個節點恢復執行。對需要暫停的人工覆核場景,這個能力讓流程可以被「凍結 → 審查 → 恢復」。這個設計模式會在 Day 38 與 Day 44 再次出現。
如果團隊成員對 LangGraph 的狀態圖設計還不熟悉,可以先從 CrewAI 或 AutoGen 入手:前者用 YAML 設定 Agent 角色、後者用對話記錄串接 Agent,學習曲線較低。等熟悉多 Agent 觀念後再轉到 LangGraph 拿完整控制權。今天的主管一—專家範例可以在 CrewAI 與 LangGraph 之間互相移植,重點是流程設計而非框架選擇。
最後,LangGraph 支援 streaming 與 checkpointing。對需要中途暫停的人工覆核場景,這兩個功能很重要。Day 36 我們會談 Agent 的安全性時再深入。
並行節點:加速無相依的子任務
前面示範的主管—專家流程是線性的,每個節點等前一個完成才執行。實務上有些子任務彼此獨立,可以並行處理以節省時間。LangGraph 支援並行節點:用 `add_node` 同時註冊多個獨立節點、用 `add_edge` 把前一個節點連到所有並行節點。我們用「三個專家同時分析同一份合約」示範並行設計:
def expert_node(name: str):
def node(state):
return {**state, name: call_llm(f"你是 {name} 專家", state["research"])}
return node
graph.add_node("expert_a", expert_node("A"))
graph.add_node("expert_b", expert_node("B"))
graph.add_node("expert_c", expert_node("C"))
graph.add_node("combine", lambda state: {**state, "combined": state.get("expert_a", "") + " " + state.get("expert_b", "") + " " + state.get("expert_c", "")})
graph.add_edge("research", "expert_a")
graph.add_edge("research", "expert_b")
graph.add_edge("research", "expert_c")
graph.add_edge("expert_a", "combine")
graph.add_edge("expert_b", "combine")
graph.add_edge("expert_c", "combine")
這段程式用 `expert_node(name)` 工廠函式動態生成三個專家節點,再用 `add_edge` 把研究節點同時連到三個專家。LangGraph 看到多個邊從「research」出發時會並行執行,三個專家同時跑、最後由 `combine` 節點彙整。並行設計可以讓原本 9 秒的流程壓到 4 秒左右(瓶頸是最慢的那個專家),對互動式應用是明顯的加速。實務上並行節點的輸出要先想清楚「如何彙整」,避免三個專家給出矛盾結論時 LLM 無所適從。
小結
今天從多 Agent 的設計動機出發,介紹 LangGraph 的狀態圖設計,並實作了一個「主管—專家」四節點的多 Agent 流程。我們把昨天的 MCP server 串進研究 Agent、串起 OpenAI 與 Ollama 兩條 LLM 路徑,展示了從「單一 Agent」回「團隊合作」的可能。多 Agent 不是銀彈,但對複雜任務它能把品質與可維護性同時拉上來。今天的 LangGraph 程式碼會在 Day 38 與 Day 40 繼續被重用。
錯誤恢復:個別 Agent 失敗不拖垮整體
多 Agent 流程常見的故障是「某個 Agent 偶發失敗、整個流程中斷」。LangGraph 支援節點層級的錯誤處理器,把錯誤包裝成狀態的一部分、由後續節點決定怎麼處理。我們用 `add_node` 的第二個參數傳入 wrapper,在分析 Agent 失敗時回傳預設值而不是拋出例外:
import functools
def with_fallback(node_fn, fallback_state: dict):
@functools.wraps(node_fn)
def wrapper(state):
try:
return node_fn(state)
except Exception as e:
return {**state, "error": str(e), **fallback_state}
return wrapper
graph.add_node("analysis", with_fallback(analysis, {"analysis": "分析 Agent 失敗,使用預設分析結果"}))
這段程式用 `with_fallback` 裝飾器把失敗的節點包起來,失敗時寫入錯誤訊息並用預設狀態取代。LangGraph 看到節點函式回傳 dict 就會繼續執行,不會因為某個 Agent 失敗而中斷整個流程。這對企業部署非常重要——批次任務跑 100 份合約時,其中一份失敗不應該讓另外 99 份也無法完成。實務上建議把錯誤訊息寫進 audit log,事後請人工補上失敗的那幾份。
結語
多 Agent 協作把任務拆解、分配、彙整的工作交給 LLM 與 LangGraph,但隨之而來的是安全性問題:每多一個 Agent,就多一個被攻擊的入口。明天(Day 36)我們會專門討論 Agent 的安全防護:攻擊者怎麼用提示注入影響 Agent 行為、企業要怎麼設計防護層、為什麼人類覆核在某些場景不可省略。Agent 安全不是附加功能,是企業部署的門檻,明天我們會把這個門檻講清楚。
延伸資源
- LangGraph 官方文件(0.2.x,2025):
https://langchain-ai.github.io/langgraph/ - Microsoft AutoGen 官方文件(0.2.x,2024–2025):
https://microsoft.github.io/autogen/ - Wu, Q. 等人,AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation(arXiv:2308.08155,2023):
https://arxiv.org/abs/2308.08155 - CrewAI 多 Agent 框架(2024–2025):
https://github.com/crewAIInc/crewAI - LangChain Multi-Agent 範例(2025):
https://python.langchain.com/docs/langgraph/
留言
張貼留言