在 Multi-Agent 系統中，「通訊（Agent Communication）」是協同架構的骨幹。如果說 Single Agent 的重點是 LLM 與 Tool 之間的互動，那麼 Multi-Agent 的成敗則完全取決於 Agent 之間如何傳遞意圖、同步狀態與協調行動。

缺少合理的通訊機制，Multi-Agent 系統會迅速退化為「互相說廢話」或「無限轉發對話」的混亂系統。

一、 Multi-Agent 通訊的四大核心架構（Topologies & Patterns）

Agent 之間的訊息流動方式，決定了系統的控制權歸屬與擴充上限：

1. 階層主從 (Supervisor / Router)     2. 點對點 / 鏈式 (Pipeline / Peer-to-Peer)
         [ Supervisor ]                          [Agent A]
        ╱      │      ╲                             │
   [Agent A] [Agent B] [Agent C]                 [Agent B]
                                                    │
                                                 [Agent C]

3. 匯流排 / 訊息佇列 (Event Bus)        4. 黑板模式 (Blackboard / Shared Memory)
   [Agent A] ──► [ Event ] ──► [Agent B]         [Agent A] ──┐      ┌──► [Agent C]
                    │                                        ├──► [Shared State] ┤
                 [ Bus ] ──► [Agent C]          [Agent B] ──┘      └──► [Agent D]

1. 階層主從式（Supervisor / Router Pattern）

• 運作機制：有一個專職的 Supervisor Agent 充當「專案經理（PM）」。中央控管所有子 Agent，決定下一個由誰發言，子 Agent 執行完畢後必須將結果回報給 Supervisor，子 Agent 之間通常不直接對話。
• 優點：流程高度可控、不容易發生 Agent 之間的無限聊天死迴圈。
• 缺點：Supervisor 成爲效能與記憶體瓶頸。

2. 鏈式 / 管道式（Pipeline / Peer-to-Peer Pattern）

• 運作機制：Agent A 的 Output 直連 Agent B 的 Input（$A \rightarrow B \rightarrow C$）。每個 Agent 只需要關心上游傳來的資料與傳給下游的目標。
• 優點：結構極度乾淨、適合確定性高的工作流（如：需求分析 $\rightarrow$ 架構設計 $\rightarrow$ 寫 Code $\rightarrow$ Code Review）。
• 缺點：缺乏靈活性，中途遭遇 Exception 時很難回溯修復。

3. 訊息匯流排式（Pub/Sub / Event Bus Pattern）

• 運作機制：Agent 不直接呼叫其他 Agent，而是發送 Event 到訊息佇列（如 Kafka / Redis PubSub）。訂閱該 Event 的 Agent 收到通知後觸發動作。
• 優點：鬆散耦合（Decoupled），支援高併發與非同步平行處理。
• 缺點：除錯難度極高，訊息追蹤（Tracing）不易。

4. 共用狀態 / 黑板模式（Blackboard / Shared State Pattern）

• 運作機制：所有 Agent 不進行直接的對話點對點傳輸，而是共同讀寫一個全域共享的狀態集（Shared State Object）。
• 優點：現代框架（如 LangGraph）的首選。徹底解決 Context Window 爆炸問題，Agent 之間只讀寫自己關心的欄位。

二、 通訊協定：Agent 之間到底在傳什麼？

Agent 之間的溝通不能只有自然語言（Plain Text），否則解析成本過高。現代通訊協定通常分為三個層級：

FIPA-ACL (Agent Communication Language) 語意範例

在學術與高階 Agent 系統中，常見的溝通模型包含：

• REQUEST：請求對方執行某項作業。
• INFORM：單向告知對方某項事實/數據。
• **PROPOSE / ACCEPT / REJECT**：兩 Agent 間進行條件談判或計畫協商。

三、 Python 實戰：以 LangGraph 理念打造「黑板架構 (Shared State)」Multi-Agent 通訊

以下實作示範現代 Multi-Agent 最標準的通訊方式：共享狀態匯流排（Shared State Bus）。Supervisor 負責訊息路由，Researcher 與 Writer 透過讀寫同一份 State 來完成溝通，避免將整個對話歷史盲目塞給所有人。

import json
from typing import Dict, Any, List, Optional
from pydantic import BaseModel, Field
from openai import OpenAI

client = OpenAI()

# ==========================================
# 1. 共享狀態 (Shared Blackboard State)
# ==========================================
class AgentState(BaseModel):
    task: str                                  # 初始任務目標
    next_step: str = "Supervisor"              # 當前控制權歸屬
    research_data: Optional[str] = None        # Researcher 寫入的資料
    draft_article: Optional[str] = None       # Writer 寫入的草稿
    review_feedback: Optional[str] = None     # Supervisor 寫入的修改建議
    is_finished: bool = False                  # 終止條件

# ==========================================
# 2. 通訊路由器 Schema (Router)
# ==========================================
class RouterDecision(BaseModel):
    thought: str = Field(description="評估當前全域狀態與各 Agent 產出的邏輯分析")
    next_agent: str = Field(description="下一個指派的 Agent 名稱：'Researcher' | 'Writer' | 'FINISH'")
    feedback: Optional[str] = Field(None, description="傳遞給下一個 Agent 的具體指引或修改建議")


# ==========================================
# 3. 具備狀態通訊能力的 Agents
# ==========================================
class MultiAgentSystem:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model

    def supervisor_node(self, state: AgentState) -> RouterDecision:
        """Supervisor: 審視共享狀態，評估控制權轉移"""
        prompt = (
            f"你是一個專案總監 (Supervisor)。請審視目前的共享狀態並決定下一步由誰執行。\n\n"
            f"【全域狀態】:\n"
            f"- 任務: {state.task}\n"
            f"- 研究資料: {state.research_data or '無'}\n"
            f"- 文章草稿: {state.draft_article or '無'}\n"
            f"- 修改建議: {state.review_feedback or '無'}\n\n"
            f"規則：\n"
            f"1. 若缺乏研究資料，指派給 'Researcher'。\n"
            f"2. 若已有研究資料但無草稿，指派給 'Writer'。\n"
            f"3. 若已有草稿且品質良好，指派 'FINISH'；若品質不佳，給予 feedback 並重新指派 'Writer'。"
        )
        completion = client.beta.chat.completions.parse(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format=RouterDecision,
        )
        return completion.choices[0].message.parsed

    def researcher_node(self, state: AgentState) -> str:
        """Researcher Agent: 讀取 task，寫入 research_data"""
        print("🔍 [Researcher]: 正在檢索與整理資料...")
        prompt = f"請針對任務：'{state.task}'，提供 3 個關鍵的事實與數據架構。"
        res = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        return res.choices[0].message.content

    def writer_node(self, state: AgentState) -> str:
        """Writer Agent: 讀取 research_data 與 review_feedback，寫入 draft_article"""
        print("✍️ [Writer]: 正在根據研究資料撰寫內文...")
        prompt = (
            f"任務：{state.task}\n"
            f"參考研究資料：\n{state.research_data}\n"
        )
        if state.review_feedback:
            prompt += f"\n主管修改建議 (Feedback)：\n{state.review_feedback}\n請依據建議調整文章。"

        res = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )
        return res.choices[0].message.content

    def run(self, task: str) -> AgentState:
        # 1. 初始化全域共享狀態
        state = AgentState(task=task)
        print(f"🎬 [System Start] 任務: {task}\n" + "="*50)

        step_count = 0
        max_steps = 6

        # 2. 通訊與狀態轉移迴圈
        while not state.is_finished and step_count < max_steps:
            step_count += 1
            print(f"\n🔄 --- Round {step_count} (Current Node: {state.next_step}) ---")

            if state.next_step == "Supervisor":
                decision = self.supervisor_node(state)
                print(f"🧠 [Supervisor Decision]: {decision.thought}")
                print(f"👉 [Next Target]: {decision.next_agent}")

                if decision.next_agent == "FINISH":
                    state.is_finished = True
                else:
                    state.next_step = decision.next_agent
                    state.review_feedback = decision.feedback

            elif state.next_step == "Researcher":
                # 執行點對點狀態更新
                state.research_data = self.researcher_node(state)
                print(f"📥 [State Update]: research_data 已寫入共享狀態。")
                state.next_step = "Supervisor"  # 控制權交還給 Supervisor

            elif state.next_step == "Writer":
                # 執行點對點狀態更新
                state.draft_article = self.writer_node(state)
                print(f"📥 [State Update]: draft_article 已寫入共享狀態。")
                state.next_step = "Supervisor"  # 控制權交還給 Supervisor

        print("\n✅ [Multi-Agent Workflow Finished]")
        return state


# ==========================================
# 4. 測試執行
# ==========================================
if __name__ == "__main__":
    system = MultiAgentSystem()
    final_state = system.run("撰寫一份關於 2026 AI Agent 通訊架構的簡短科技快報")
    print("\n📄 【最終產出文章】:\n")
    print(final_state.draft_article)

四、 通訊工程常遇到的 4 大陷阱與解法

1. 無限廣播 (Broadcast Storm)：

• 問題：Agent A 發出一個 Event，被 Agent B 和 C 收到，B 和 C 又發出新 Event，導致系統訊息量呈指數級暴增。
• 解法：限制通訊拓撲，使用 Explicit Routing（明確路由），非必要不上 Pub/Sub。

2. Context Window 連鎖爆滿 (Context Contamination)：

• 問題：將 Agent A 與 B 的完整對話 Log 無腦複製給 Agent C。
• 解法：State Projection（狀態投影）——傳遞給 Agent C 的 Prompt 僅包含其決策必需的 Key-Value，不傳送歷史對話列表。

3. 語意協議失配 (Schema Drift)：

• 問題：Agent A 回傳的 JSON 格式與 Agent B 預期讀取的欄位名稱不符。
• 解法：使用 Pydantic 或 JSON Schema 在接收端加上 Strict Validation 防護。

4. 死鎖與競爭條件 (Deadlock & Race Conditions)：

• 問題：Agent A 在等待 Agent B 回覆，而 Agent B 又在等待 Agent A 提供進一步參數。
• 解法：每個 Agent 通訊點設定 Timeout 限制與 Fallback 降級機制。