Supervisor Agent

2026 iThome 鐵人賽

DAY 24

自我挑戰組

AI Agent 從零開始系列第 24 篇

18th鐵人賽

pigbrother

2026-10-08 22:11:33

19 瀏覽

分享至

Supervisor Agent（主管代理） 是 Multi-Agent（多代理人）系統中最核心的指揮控制模式。

它的運作邏輯類似真實職場中的專案經理（Project Manager / Tech Lead）：主管本身通常不直接撰寫程式碼或深入處理底層細節，而是專注於目標拆解、任務派發、進度審查與最後成果的匯整。

一、 Supervisor Agent 的架構與運作邏輯

在傳統的網狀（Network）或自由對談 Multi-Agent 中，Agent 之間容易互相對話、推託責任，導致無限迴圈。Supervisor 模式將控制權集中，形成星狀（Star Topology）或階層式（Hierarchical）結構：

                       ┌─────────────────────────┐
                       │     User Task Goal      │
                       └────────────┬────────────┘
                                    │
                                    ▼
                       ┌─────────────────────────┐
                       │    Supervisor Agent     │ ◄─── 評估全域 State & 控制流程
                       └─┬──────────┬──────────┬─┘
                         │          │          │
         ┌───────────────┘          │          └───────────────┐
         ▼                          ▼                          ▼
【Researcher Agent】        【Coder Agent】           【Reviewer Agent】
 (子任務：資料檢索)         (子任務：撰寫代碼)         (子任務：單元測試/審查)

工作循環 (Supervisor Execution Loop)

任務分析：接收使用者的原始需求，判斷達成目標所需的步驟與目前的進度。
動態派發 (Dispatch)：選擇最適合的 Worker Agent，並將精簡後的子任務指派給它。
結果審查 (Evaluate)：檢查 Worker Agent 回傳的成果（Observation）。若未達標，附帶改進建議發回重做；若已達標，指派下一個 Worker。
終止交稿 (Finish)：確認所有子任務皆完成，彙整最終答案交付給使用者。

二、 Python 實戰：打造 Supervisor Multi-Agent 系統

我們使用純 Python + OpenAI SDK（利用 Structured Outputs 強制約束 Supervisor 的決策格式），打造一個指揮 Researcher 與 Coder 的 Supervisor Agent 系統。

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

client = OpenAI()

# ==========================================
# 1. 定義 Supervisor 決策的強型別 Schema
# ==========================================
class SupervisorDecision(BaseModel):
    thought: str = Field(description="思考分析：評估當前進度、各 Worker 的輸出與下一步決策邏輯")
    next_worker: str = Field(
        description="下一個執行的 Worker 名稱：'Researcher' | 'Coder' | 'FINISH'"
    )
    instruction: Optional[str] = Field(
        None, description="傳遞給下一個 Worker 的具體任務指令與上下文"
    )

# ==========================================
# 2. Worker 系統 (Worker Agents)
# ==========================================
def worker_researcher(instruction: str) -> str:
    """Researcher Worker：負責搜尋與資料調研"""
    print(f"🔍 [Researcher Working]: {instruction}")
    # 模擬工具呼叫與資料檢索
    return json.dumps({
        "status": "success",
        "findings": "發現 Win32 GDI 雙緩衝 (Double Buffering) 需建立 Memory DC (CreateCompatibleDC) 與 Compatible Bitmap，繪製完成後以 BitBlt 貼回目標 HWND HDC。"
    }, ensure_ascii=False)

def worker_coder(instruction: str) -> str:
    """Coder Worker：負責編寫或修復程式碼"""
    print(f"💻 [Coder Working]: {instruction}")
    # 模擬程式碼生成
    return """// Modern C++ Win32 GDI Double Buffering Snippet
HDC hdcMem = CreateCompatibleDC(hdc);
HBITMAP hbmMem = CreateCompatibleBitmap(hdc, width, height);
HANDLE hOld = SelectObject(hdcMem, hbmMem);

// 進行繪製動作...
Rectangle(hdcMem, 0, 0, width, height);

// 一次性貼回，防止閃爍 (Flicker-free)
BitBlt(hdc, 0, 0, width, height, hdcMem, 0, 0, SRCCOPY);

SelectObject(hdcMem, hOld);
DeleteObject(hbmMem);
DeleteDC(hdcMem);"""


# ==========================================
# 3. Supervisor Agent Engine
# ==========================================
class SupervisorSystem:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.model = model

    def _call_supervisor(self, goal: str, execution_logs: List[Dict[str, str]]) -> SupervisorDecision:
        """Supervisor 的決策核心：決定 Next Worker 或 FINISH"""
        logs_text = "\n".join([f"[{log['worker']}]: {log['output']}" for log in execution_logs])
        
        prompt = (
            f"你是一位高階 Supervisor Agent（主管代理）。\n"
            f"任務目標：【{goal}】\n\n"
            f"【過往執行日誌】:\n{logs_text if logs_text else '尚無執行紀錄'}\n\n"
            f"可用的 Worker 列表：\n"
            f"1. Researcher: 負責技術調研、資料與演算法檢索。\n"
            f"2. Coder: 負責撰寫高品質程式碼。\n\n"
            f"請評估目標達成進度：\n"
            f"- 如果缺乏關鍵資訊，指派 'Researcher' 並給予指令。\n"
            f"- 如果資訊充足但尚未寫出程式碼，指派 'Coder' 並給予指令。\n"
            f"- 如果程式碼已順利撰寫完成且符合需求，指派 'FINISH'。"
        )

        completion = client.beta.chat.completions.parse(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            response_format=SupervisorDecision,
        )
        return completion.choices[0].message.parsed

    def run(self, goal: str, max_rounds: int = 5) -> str:
        print(f"🎯 [Task Goal]: {goal}\n" + "="*60)
        execution_logs = []

        for round_num in range(1, max_rounds + 1):
            print(f"\n🧠 [Round {round_num}] Supervisor 評估中...")
            
            # 1. Supervisor 決策
            decision = self._call_supervisor(goal, execution_logs)
            print(f"💭 [Supervisor Thought]: {decision.thought}")
            print(f"👉 [Decision]: Next -> {decision.next_worker}")

            # 2. 終止條件檢查
            if decision.next_worker == "FINISH":
                print("\n✅ [Task Finished] Supervisor 確認任務完成！")
                last_code = execution_logs[-1]['output'] if execution_logs else "無產出"
                return last_code

            # 3. 根據 Supervisor 決策派發給對應的 Worker
            worker_output = ""
            if decision.next_worker == "Researcher":
                worker_output = worker_researcher(decision.instruction)
            elif decision.next_worker == "Coder":
                worker_output = worker_coder(decision.instruction)

            # 4. 將 Worker 產出的結果紀錄回日誌，提供給下一輪 Supervisor 參考
            execution_logs.append({
                "worker": decision.next_worker,
                "output": worker_output
            })
            print(f"📥 [Worker Observation Recorded]")

        return "任務超過最大輪次，流程中斷。"


# ==========================================
# 4. 測試執行
# ==========================================
if __name__ == "__main__":
    system = SupervisorSystem()
    task = "幫我查一下 Win32 GDI 防閃爍 double buffering 繪製機制，並用 C++ 寫出範例程式碼。"
    final_output = system.run(task)
    print("\n📄 【最終交付成果】:\n")
    print(final_output)

三、 Supervisor 模式的執行軌跡

執行上述程式碼時，系統呈現標準的主管分配 $\rightarrow$ 子 Agent 執行 $\rightarrow$ 主管驗收流程：

🎯 [Task Goal]: 幫我查一下 Win32 GDI 防閃爍 double buffering 繪製機制，並用 C++ 寫出範例程式碼。

🧠 [Round 1] Supervisor 評估中...
💭 [Supervisor Thought]: 目前尚未進行技術調研，需要先請 Researcher 查詢 Win32 GDI 雙緩衝的繪製機制與相關 API。
👉 [Decision]: Next -> Researcher
🔍 [Researcher Working]: 查詢 Win32 GDI 防閃爍 double buffering 繪製機制與必備 API
📥 [Worker Observation Recorded]

🧠 [Round 2] Supervisor 評估中...
💭 [Supervisor Thought]: 已拿到 Researcher 提供關於 CreateCompatibleDC 與 BitBlt 的技術細節，現在可以請 Coder 撰寫 C++ 範例程式碼。
👉 [Decision]: Next -> Coder
💻 [Coder Working]: 使用 Modern C++ 撰寫包含 CreateCompatibleDC、CreateCompatibleBitmap 與 BitBlt 的雙緩衝繪製程式碼
📥 [Worker Observation Recorded]

🧠 [Round 3] Supervisor 評估中...
💭 [Supervisor Thought]: Coder 已經寫出完整的雙緩衝 C++ 範例程式碼，技術要點均已涵蓋，任務已完全達成。
👉 [Decision]: Next -> FINISH

✅ [Task Finished] Supervisor 確認任務完成！

四、 Supervisor Agent 的工程優缺點與最佳實踐

維度	優點 (Pros)	局限與挑戰 (Cons)
可控性與穩定度	控制權高度集中，防止 Worker Agents 之間私下聊廢話或陷入死迴圈。	Supervisor 成為單點瓶頸（Single Point of Failure），若 Supervisor 評估失誤，全盤皆錯。
Context Window 最佳化	子 Agent 只處理專屬子任務，Context 乾淨；Supervisor 只需要看 Summarized Logs。	訊息過度摘要時，Supervisor 可能遺忘子 Agent 回傳的微小細節。
可擴充性 (Scalability)	新增功能只需註冊新的 Worker 名字與 Prompt，不影響原有的 Worker 邏輯。	當 Worker 數量超過 10 個以上時，Supervisor 選錯 Worker 的機率會增加。

💡 生產環境最佳實踐（Best Practices）：

巢狀 Supervisor (Hierarchical Supervisors)：若 Worker 超過 10 個，改採用多層級架構（例如：Top Supervisor $\rightarrow$ 管理 Dev Team Lead 與 Research Lead $\rightarrow$ 下掛各自的工程師 Worker）。
與 LangGraph 結合：LangGraph 原生提供了 Supervisor Pattern 的圖架構實現（StateGraph），能更簡潔地管理帶有條件邊（Conditional Edges）的狀態轉移。