多 Agent 協作:角色分工、辯論與 Swarm
從角色設計、通訊協議到辯論式推理與 Swarm,理解多 Agent 系統的架構模式、失敗模式與最佳實踐。
多 Agent 協作:角色分工、辯論與 Swarm
上一篇我們談了 RAG,讓單一 Agent 能擁有專業知識。
你有沒有遇過這種情況:一個任務太複雜,需要多種專業能力,單一 Agent 怎麼調 Prompt 都不夠好?
例如寫一份商業計畫書,需要市場分析、財務預測、風險評估、文案撰寫——這些能力很難用同一個 Agent 同時做好。
這一篇,我們要談的是 Agent 架構的下一個階段:多 Agent 協作(Multi-Agent Collaboration)。
我們會解釋為什麼需要多個 Agent、如何設計角色與通訊協議、辯論式推理與投票機制,以及多 Agent 系統的常見失敗模式與應對策略。
一、為什麼需要多個 Agent?
單一 Agent 已經能呼叫工具、擁有記憶、使用 RAG。
但它在面對複雜任務時,仍有幾個限制:
1. 上下文窗口有限
一個 Agent 要同時處理市場分析、財務數據、法律條文、文案風格,上下文很快就爆了。
資訊一多,模型就開始遺漏、混淆、甚至幻覺。
2. 角色衝突
同一個 Agent 既要「發散思考」又要「嚴格審查」,這兩種能力往往互相衝突。
你很難用同一個 Prompt 讓模型既天馬行空又一絲不苟。
3. 缺乏多角度視野
單一 Agent 只有一個視角。
它可能過度樂觀、可能忽略風險、可能陷入某種思維定勢。
4. 難以專精
一個 Agent 要同時擅長寫作、程式、分析、規劃,結果就是樣樣通、樣樣鬆。
不如讓每個 Agent 專精一件事,再把它們組合起來。
5. 錯誤無法互相校正
單一 Agent 錯了就是錯了,沒有第二個視角來發現問題。
多個 Agent 可以互相檢查、辯論、修正。
打個比方:
- 單一 Agent 像一個全能但精力有限的人。
- 多 Agent 像一個團隊,每個人有專長,能互相補位、互相檢查。
二、多 Agent 的常見架構模式
多 Agent 系統有很多種設計方式,以下是最常見的幾種:
模式一:順序式(Sequential / Pipeline)
Agent 按照固定順序執行,前一個的輸出是後一個的輸入。
研究 Agent → 分析 Agent → 撰寫 Agent → 審查 Agent
優點:簡單、可控、易除錯。
缺點:缺乏彈性,無法回頭修改。
適用場景:流程明確的任務,如報告生成、程式碼審查。
模式二:協調者式(Supervisor / Orchestrator)
有一個「協調者 Agent」負責分派任務給其他 Agent,並整合結果。
┌→ 研究 Agent ─┐
協調者 Agent ────┼→ 分析 Agent ─┼→ 協調者 Agent → 最終輸出
└→ 撰寫 Agent ─┘
優點:靈活、可動態分派、易於擴展。
缺點:協調者可能成為瓶頸,且需要良好的任務分解能力。
適用場景:複雜任務、需要動態決策的場景。
模式三:辯論式(Debate)
多個 Agent 針對同一問題提出不同觀點,互相辯論,最後由裁判或投票決定。
Agent A(正方)─┐
├→ 辯論 → 裁判 Agent → 最終答案
Agent B(反方)─┘
優點:能發現盲點、提升推理品質。
缺點:成本高、可能陷入僵局。
適用場景:需要嚴謹推理的任務,如事實查核、決策分析。
模式四:群體式(Swarm / Peer-to-Peer)
沒有中央協調者,多個 Agent 平等協作,透過訊息傳遞來完成任務。
Agent A ↔ Agent B
↕ ↕
Agent C ↔ Agent D
優點:高度分散、容錯性強。
缺點:難以控制、可能發散、通訊成本高。
適用場景:模擬社會行為、開放式探索任務。
模式五:階層式(Hierarchical)
多層協調結構,上層 Agent 負責策略,下層 Agent 負責執行。
策略 Agent
/ | \
研究組 分析組 執行組
/ \ / \ / \
A B C D E F
優點:可處理超大規模任務、職責分明。
缺點:架構複雜、通訊開銷大。
適用場景:大型專案、企業級自動化。
三、角色設計:讓每個 Agent 有明確的職責
多 Agent 系統的關鍵,是角色設計。
每個 Agent 應該有:
- 明確的職責:它負責做什麼?
- 專業的 Prompt:它的系統提示應該聚焦於它的專長。
- 專屬的工具:它需要哪些工具來完成任務?
- 清楚的輸入輸出格式:它接收什麼、產出什麼?
以下是一個典型的多 Agent 角色設計範例:
AGENTS = {
"researcher": {
"name": "研究員",
"system_prompt": """你是一個專業的研究員。
你的職責是蒐集、整理、驗證資訊。
你擅長使用搜尋工具、查閱文件、提取關鍵事實。
輸出時請提供具體來源與引用。""",
"tools": ["web_search", "read_document"],
},
"analyst": {
"name": "分析師",
"system_prompt": """你是一個專業的數據分析師。
你的職責是分析研究員提供的資料,找出模式、趨勢與洞見。
你擅長邏輯推理、數據解讀、風險評估。
輸出時請提供清晰的論點與支持證據。""",
"tools": ["calculate", "query_database"],
},
"writer": {
"name": "撰稿人",
"system_prompt": """你是一個專業的內容撰稿人。
你的職責是根據研究與分析結果,撰寫清晰、有說服力的文章。
你擅長結構化寫作、敘事技巧、語氣調整。
輸出時請注意邏輯連貫與可讀性。""",
"tools": [],
},
"reviewer": {
"name": "審查員",
"system_prompt": """你是一個嚴格的內容審查員。
你的職責是檢查文章的事實正確性、邏輯一致性與表達清晰度。
你會找出錯誤、矛盾、遺漏與模糊之處。
輸出時請具體指出問題,並提供修改建議。""",
"tools": ["fact_check"],
},
}
每個角色的設計要點:
- 職責單一:不要讓一個 Agent 同時做研究和寫作。
- Prompt 聚焦:系統提示要強化該角色的專業視角。
- 工具匹配:只給該角色需要的工具,避免混淆。
- 輸出格式明確:每個角色的輸出應該有固定的結構,方便下一個角色使用。
四、Agent 之間的通訊協議
多 Agent 系統要能協作,就必須有通訊協議。
常見的做法有以下幾種:
1. 結構化訊息(Structured Messages)
Agent 之間傳遞結構化的訊息,而非自由文字。
from dataclasses import dataclass
from typing import Any
@dataclass
class AgentMessage:
sender: str # 發送者
receiver: str # 接收者
task: str # 任務描述
content: Any # 具體內容
metadata: dict # 額外資訊(時間戳、優先級等)
優點:易於解析、可追蹤、可驗證。
缺點:需要事先定義格式。
2. 共享黑板(Blackboard)
所有 Agent 共享一個「黑板」(共享記憶體),各自讀寫。
class Blackboard:
def __init__(self):
self.data = {}
def write(self, key: str, value: Any, author: str):
self.data[key] = {
"value": value,
"author": author,
"timestamp": datetime.now().isoformat(),
}
def read(self, key: str) -> Any:
return self.data.get(key, {}).get("value")
def get_history(self, key: str) -> list:
# 取得某個 key 的修改歷史
pass
優點:鬆散耦合、易於擴展。
缺點:可能產生競爭條件、需要同步機制。
3. 訊息佇列(Message Queue)
用訊息佇列來傳遞任務與結果,適合非同步、分散式場景。
import queue
class MessageBus:
def __init__(self):
self.queues = {}
def register(self, agent_name: str):
self.queues[agent_name] = queue.Queue()
def send(self, receiver: str, message: dict):
if receiver in self.queues:
self.queues[receiver].put(message)
def receive(self, agent_name: str, timeout: float = 1.0):
try:
return self.queues[agent_name].get(timeout=timeout)
except queue.Empty:
return None
優點:非同步、可擴展、容錯。
缺點:需要處理順序、重複、失敗等問題。
五、實作:一個簡單的多 Agent 系統
讓我們用「協調者模式」實作一個多 Agent 系統。
步驟 1:定義 Agent 類別
class Agent:
def __init__(self, name: str, system_prompt: str, tools: list = None):
self.name = name
self.system_prompt = system_prompt
self.tools = tools or []
self.memory = []
def run(self, task: str, context: str = "") -> str:
"""執行任務"""
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"任務:{task}\n\n背景資訊:{context}"},
]
response = call_llm(messages, tools=self.tools if self.tools else None)
message = response.choices[0].message
# 如果有工具呼叫,執行工具
if message.tool_calls:
for tool_call in message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
result = execute_tool(function_name, arguments)
messages.append(message)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
# 再次呼叫 LLM 生成最終回答
response = call_llm(messages)
message = response.choices[0].message
return message.content
步驟 2:定義協調者
class Orchestrator:
def __init__(self):
self.agents = {}
self.history = []
def register(self, agent: Agent):
self.agents[agent.name] = agent
def run(self, task: str, verbose: bool = True) -> str:
"""協調多個 Agent 完成任務"""
# 1. 分解任務
subtasks = self._decompose(task)
if verbose:
print(f"[任務分解] {subtasks}")
# 2. 依序執行每個子任務
context = ""
for subtask in subtasks:
agent_name = subtask["agent"]
description = subtask["description"]
if agent_name not in self.agents:
context += f"\n[跳過] 找不到 Agent {agent_name}"
continue
if verbose:
print(f"\n[{agent_name}] 執行:{description}")
agent = self.agents[agent_name]
result = agent.run(description, context)
if verbose:
print(f"[{agent_name} 結果] {result[:200]}...")
self.history.append({
"agent": agent_name,
"task": description,
"result": result,
})
context += f"\n\n[{agent_name} 的產出]\n{result}"
# 3. 整合結果
return self._synthesize(task, context)
def _decompose(self, task: str) -> list:
"""用 LLM 把任務分解成子任務"""
prompt = f"""請將以下任務分解成子任務,並指派給適合的 Agent。
可用 Agent:
- researcher:負責蒐集與整理資訊
- analyst:負責分析與推理
- writer:負責撰寫內容
- reviewer:負責審查與修正
任務:{task}
請以 JSON 格式輸出:
{{"subtasks": [
{{"agent": "researcher", "description": "子任務描述"}},
...
]}}
"""
result = call_llm([{"role": "user", "content": prompt}])
try:
return json.loads(result)["subtasks"]
except (json.JSONDecodeError, KeyError):
return [{"agent": "writer", "description": task}]
def _synthesize(self, task: str, context: str) -> str:
"""整合所有 Agent 的結果"""
prompt = f"""請根據以下各 Agent 的產出,整合成最終回答。
原始任務:{task}
各 Agent 產出:
{context}
請輸出完整、連貫的最終回答。
"""
return call_llm([{"role": "user", "content": prompt}])
步驟 3:實際使用
# 建立 Agent
researcher = Agent(
name="researcher",
system_prompt="你是一個專業的研究員,擅長蒐集與整理資訊。",
tools=["web_search"],
)
analyst = Agent(
name="analyst",
system_prompt="你是一個專業的分析師,擅長邏輯推理與數據分析。",
)
writer = Agent(
name="writer",
system_prompt="你是一個專業的撰稿人,擅長結構化寫作。",
)
reviewer = Agent(
name="reviewer",
system_prompt="你是一個嚴格的審查員,擅長找出錯誤與改進空間。",
)
# 建立協調者
orchestrator = Orchestrator()
orchestrator.register(researcher)
orchestrator.register(analyst)
orchestrator.register(writer)
orchestrator.register(reviewer)
# 執行任務
result = orchestrator.run("寫一份關於 AI Agent 市場趨勢的簡短報告")
print(result)
執行結果:
[任務分解]
[
{"agent": "researcher", "description": "蒐集 2024 年 AI Agent 市場的相關資訊"},
{"agent": "analyst", "description": "分析市場趨勢與主要玩家"},
{"agent": "writer", "description": "撰寫報告初稿"},
{"agent": "reviewer", "description": "審查報告並提供修改建議"},
]
[researcher] 執行:蒐集 2024 年 AI Agent 市場的相關資訊
[researcher 結果] 根據搜尋結果,2024 年 AI Agent 市場快速成長...
[analyst] 執行:分析市場趨勢與主要玩家
[analyst 結果] 主要趨勢包括:多 Agent 協作、工具使用標準化...
[writer] 執行:撰寫報告初稿
[writer 結果] # AI Agent 市場趨勢報告...
[reviewer] 執行:審查報告並提供修改建議
[reviewer 結果] 報告結構清晰,但缺少具體數據支持...
六、辯論式推理:讓 Agent 互相挑戰
辯論式推理是多 Agent 系統中最有趣的模式之一。
它的核心思想是:讓多個 Agent 從不同角度思考,互相挑戰,最終得到更可靠的答案。
基本架構
class DebateSystem:
def __init__(self, debaters: list, judge: Agent, rounds: int = 3):
self.debaters = debaters
self.judge = judge
self.rounds = rounds
def run(self, question: str) -> dict:
"""執行辯論"""
# 每個辯論者先提出初始觀點
positions = {}
for debater in self.debaters:
positions[debater.name] = debater.run(
f"請針對以下問題提出你的觀點:{question}"
)
# 多輪辯論
for round_num in range(self.rounds):
print(f"\n=== 第 {round_num + 1} 輪辯論 ===")
for debater in self.debaters:
# 收集其他辯論者的觀點
others = "\n\n".join(
f"[{name} 的觀點]\n{pos}"
for name, pos in positions.items()
if name != debater.name
)
# 讓辯論者回應
response = debater.run(
f"請針對以下問題,回應其他辯論者的觀點:{question}",
context=f"其他辯論者的觀點:\n{others}",
)
positions[debater.name] = response
# 裁判裁決
all_positions = "\n\n".join(
f"[{name} 的最終觀點]\n{pos}"
for name, pos in positions.items()
)
verdict = self.judge.run(
f"請根據以下辯論,對問題「{question}」做出最終裁決。",
context=all_positions,
)
return {
"positions": positions,
"verdict": verdict,
}
使用範例
# 建立辯論者
optimist = Agent(
name="optimist",
system_prompt="你是一個樂觀的技術樂觀主義者,傾向於看到 AI 的正面潛力。",
)
skeptic = Agent(
name="skeptic",
system_prompt="你是一個謹慎的懷疑論者,傾向於指出 AI 的風險與限制。",
)
# 建立裁判
judge = Agent(
name="judge",
system_prompt="你是一個中立的裁判,負責根據雙方論點做出平衡的判斷。",
)
# 執行辯論
debate = DebateSystem(
debaters=[optimist, skeptic],
judge=judge,
rounds=2,
)
result = debate.run("AI Agent 會在五年內取代大多數軟體工程師嗎?")
print(result["verdict"])
辯論式推理的優點
- 減少確認偏誤:不同觀點互相挑戰,避免單一視角。
- 提升推理品質:多輪辯論能深入問題核心。
- 增加透明度:每個觀點都有清楚的論證過程。
辯論式推理的缺點
- 成本高:每個辯論者都要多次呼叫 LLM。
- 可能陷入僵局:雙方各執一詞,無法達成共識。
- 裁判可能偏頗:裁判的判斷品質取決於它的 Prompt。
七、Swarm:去中心化的 Agent 協作
Swarm 是一種沒有中央協調者的多 Agent 架構。
每個 Agent 都是平等的,透過訊息傳遞來協作。
核心概念
- 每個 Agent 有自己的狀態與目標
- Agent 之間透過訊息傳遞來溝通
- 沒有中央控制,行為是湧現的
實作範例
class SwarmAgent:
def __init__(self, name: str, role: str, expertise: str):
self.name = name
self.role = role
self.expertise = expertise
self.inbox = queue.Queue()
self.knowledge = []
def receive(self, message: dict):
"""接收訊息"""
self.inbox.put(message)
def process(self) -> dict:
"""處理訊息並決定下一步"""
try:
message = self.inbox.get(timeout=1.0)
except queue.Empty:
return None
# 判斷自己是否能處理這個任務
can_handle = self._can_handle(message)
if can_handle:
result = self._handle(message)
return {
"type": "result",
"from": self.name,
"content": result,
}
else:
# 轉發給其他 Agent
return {
"type": "forward",
"from": self.name,
"content": message,
}
def _can_handle(self, message: dict) -> bool:
"""判斷自己是否能處理這個任務"""
prompt = f"""你是一個 {self.role},專長是 {self.expertise}。
以下任務是否屬於你的專長?
任務:{message.get('content', '')}
回答 yes 或 no。"""
result = call_llm([{"role": "user", "content": prompt}])
return "yes" in result.lower()
def _handle(self, message: dict) -> str:
"""處理任務"""
prompt = f"""你是一個 {self.role},專長是 {self.expertise}。
請處理以下任務:
{message.get('content', '')}"""
return call_llm([{"role": "user", "content": prompt}])
class Swarm:
def __init__(self):
self.agents = {}
self.message_log = []
def register(self, agent: SwarmAgent):
self.agents[agent.name] = agent
def broadcast(self, message: dict, exclude: str = None):
"""廣播訊息給所有 Agent"""
for name, agent in self.agents.items():
if name != exclude:
agent.receive(message)
def run(self, task: str, max_steps: int = 10) -> str:
"""執行 Swarm"""
# 初始廣播
self.broadcast({
"type": "task",
"from": "user",
"content": task,
})
for step in range(max_steps):
# 每個 Agent 輪流處理
for name, agent in self.agents.items():
response = agent.process()
if response is None:
continue
self.message_log.append(response)
if response["type"] == "result":
# 有 Agent 完成了任務
return response["content"]
elif response["type"] == "forward":
# 轉發給其他 Agent
self.broadcast(response, exclude=name)
return "任務未完成(達到最大步數)"
Swarm 的優點
- 高度分散:沒有單點故障。
- 容錯性強:某個 Agent 失敗,其他 Agent 仍可運作。
- 湧現行為:可能產生預期之外的解決方案。
Swarm 的缺點
- 難以控制:行為難以預測與調試。
- 可能發散:Agent 可能一直轉發,無法收斂。
- 通訊成本高:每個 Agent 都要與其他 Agent 溝通。
八、多 Agent 系統的常見失敗模式
多 Agent 系統雖然強大,但也很容易失敗。以下是幾個常見的失敗模式:
1. 無限迴圈
Agent 之間互相轉發任務,卻沒有人真正處理。
解法:
- 設定最大步數。
- 加入「已處理」標記,避免重複處理。
- 讓每個 Agent 在無法處理時明確回報。
2. 責任擴散
每個 Agent 都以為別人會處理,結果沒人處理。
解法:
- 明確指派任務給特定 Agent。
- 使用協調者模式,由中央分派任務。
3. 訊息爆炸
Agent 之間互相廣播,訊息量指數成長。
解法:
- 限制廣播範圍。
- 使用點對點通訊而非廣播。
- 加入訊息過濾機制。
4. 觀點同質化
多個 Agent 最終都給出類似的答案,失去多樣性。
解法:
- 在 Prompt 中強化每個 Agent 的獨特視角。
- 使用不同的模型或溫度設定。
- 加入「反對者」角色,專門挑戰共識。
5. 協調者瓶頸
協調者 Agent 成為效能瓶頸,所有任務都卡在它身上。
解法:
- 使用階層式架構,分散協調工作。
- 讓協調者只負責高階決策,細節交給下層 Agent。
6. 成本失控
每個 Agent 都要呼叫 LLM,成本迅速累積。
解法:
- 用較小的模型處理簡單任務。
- 快取重複的查詢結果。
- 限制每個任務的最大 Agent 呼叫次數。
九、多 Agent 系統的最佳實踐
1. 從簡單開始
不要一開始就設計複雜的多 Agent 系統。
先從單一 Agent 開始,確定它真的不夠用,再考慮多 Agent。
2. 角色要明確
每個 Agent 的職責、專長、工具、輸出格式都應該清楚定義。
3. 通訊要結構化
用結構化的訊息格式,避免自由文字造成的誤解。
4. 加入終止條件
設定最大步數、最大成本、最大時間,避免系統失控。
5. 記錄所有互動
保留完整的訊息日誌,方便除錯與分析。
import logging
logging.basicConfig(
filename="multi_agent.log",
level=logging.INFO,
format="%(asctime)s - %(message)s",
)
def log_message(sender: str, receiver: str, content: str):
logging.info(f"{sender} → {receiver}: {content[:200]}")
6. 設計失敗處理
每個 Agent 都應該能優雅地處理失敗,並回報問題。
7. 評估整體表現
多 Agent 系統的評估比單一 Agent 更複雜。
需要評估:
- 任務完成率
- 平均步驟數
- 總成本
- 輸出品質
- 失敗模式分佈
8. 何時不該用多 Agent
不是所有任務都需要多 Agent。
如果單一 Agent 加上好的 Prompt 就能解決,就不需要多 Agent。
多 Agent 適合:
- 任務需要多種專業能力
- 任務需要多角度視野
- 任務需要互相檢查與修正
- 單一 Agent 的上下文窗口不夠用
十、總結:從單打獨鬥到團隊協作
讓我們回顧這一篇的核心:
- 為什麼需要多 Agent:單一 Agent 上下文有限、角色衝突、缺乏多角度、難以專精、錯誤無法校正。
- 常見架構模式:順序式、協調者式、辯論式、群體式、階層式。
- 角色設計:明確職責、專業 Prompt、專屬工具、清楚輸出格式。
- 通訊協議:結構化訊息、共享黑板、訊息佇列。
- 實作方式:定義 Agent 類別、協調者、任務分解與結果整合。
- 辯論式推理:多個 Agent 從不同角度思考,互相挑戰,提升推理品質。
- Swarm:去中心化的 Agent 協作,行為湧現。
- 常見失敗模式:無限迴圈、責任擴散、訊息爆炸、觀點同質化、協調者瓶頸、成本失控。
- 最佳實踐:從簡單開始、角色明確、通訊結構化、加入終止條件、記錄互動、設計失敗處理、評估整體表現。
多 Agent 協作讓 Agent 從單打獨鬥變成團隊合作。
它讓系統能處理更複雜的任務、容納更多元的視角、產生更可靠的結果。
但它也帶來新的挑戰:協調、通訊、成本、控制。
理解了多 Agent 的原理與實踐,你就掌握了打造複雜 AI 系統的關鍵能力。
不過,Agent 越強大,風險也越高。
當 Agent 能呼叫工具、存取資料、執行動作時,安全與護欄就變得至關重要。
這就是下一篇要談的主題。
下一篇預告
《Agent 的安全與護欄:Prompt Injection、權限控制與人類審核》
我們會解釋 Prompt Injection 的攻擊原理、如何設計權限控制與沙箱機制、什麼時候需要人類審核(Human-in-the-Loop),以及如何平衡自動化與安全性。