在昨天的文章中，我們探討了 Multi-Agent 協同架構的核心概念與主管模式（Supervisor Pattern）的設計理念。我們知道，將複雜任務拆解給多個專業 Agent，能有效解決單一 Agent 面臨的 Prompt 膨脹、Context 污染與工具調用失準問題。

今天我們進入實戰篇：如何使用 Python 與 LangGraph，從零實作一個具備 Supervisor 主管節點、Research 專家節點與 Writer 專家節點的完整 Multi-Agent 系統！

一、系統架構與拓撲設計

我們要打造的工作流如下：

                  ┌───────────────────┐
                  │     __start__     │
                  └─────────┬─────────┘
                            │
                            ▼
               ┌─────────────────────────┐
               │    Supervisor Node      │ ◄─────────────┐
               │       (主管節點)        │               │
               └────────────┬────────────┘               │
                            │ (條件路由派工)              │
         ┌──────────────────┴──────────────────┐         │ (回傳執行結果)
         ▼                                     ▼         │
┌─────────────────┐                   ┌─────────────────┐│
│  research_node  │                   │   writer_node   ││
│   (調研專家)    │                   │   (文案專家)    ││
└────────┬────────┘                   └────────┬────────┘│
         │                                     │         │
         └──────────────────┬──────────────────┘─────────┘
                            │ (當 Supervisor 判定 FINISH)
                            ▼
                     ┌─────────────┐
                     │   __end__   │
                     └─────────────┘

團隊角色分工：

1. Supervisor（主管）：負責審視當前對話與各專家回報的進度，決定下一步由誰執行（Researcher、Writer 或 FINISH）。
2. Researcher（調研專家）：專門掛載檢索工具，負責蒐集外部事實資料，並將整理後的重點寫回全域狀態。
3. Writer（文案專家）：專門負責將調研專家蒐集的原始事實，轉化為結構清晰、專業流暢的最終 Markdown 報告。

二、步驟 1：定義全域狀態（State）與 Supervisor 路由契約

我們定義共享狀態 TeamState，並使用 Pydantic 定義 Supervisor 的結構化輸出契約。

from typing import Annotated, Literal
from typing_extensions import TypedDict
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. 定義 Multi-Agent 全域狀態
class TeamState(TypedDict):
    messages: Annotated[list, add_messages]  # 團隊對話軌跡
    next_worker: str                         # 主管指派的下一個工作者

# 2. 定義 Supervisor 路由結構化契約
class RouterOutput(BaseModel):
    next_worker: Literal["Researcher", "Writer", "FINISH"] = Field(
        description="決定下一步應指派的工作者。若需要查資料選 'Researcher'，若資料足夠且需撰寫報告選 'Writer'，若已全部完成選 'FINISH'。"
    )
    reasoning: str = Field(
        description="做出此派工決策的簡要原因"
    )

三、步驟 2：定義專家節點（Researcher & Writer）

我們為 Researcher 節點配備專屬工具，並為 Writer 節點定義專屬的文案提示詞。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 初始化共用 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)

# 1. 定義調研工具
@tool
def search_tech_database(query: str) -> str:
    """查詢技術資料庫與前沿資訊。"""
    mock_db = {
        "LangGraph": "LangGraph 是 LangChain 推出的狀態圖架構，原生支援循環、狀態持久化與 Human-in-the-loop。",
        "Agent": "AI Agent 是結合 LLM、規劃推理、記憶機制與工具調用能力的自主系統。"
    }
    for key, value in mock_db.items():
        if key.lower() in query.lower():
            return value
    return "已檢索最新資料：LangGraph 主管模式透過結構化路由實現多 Agent 狀態協同。"

# 2. 建立 Researcher 節點
researcher_llm = llm.bind_tools([search_tech_database])

def research_node(state: TeamState):
    """調研專家：負責查詢技術事實"""
    system_prompt = SystemMessage(
        content="你是專屬的【調研專家】。請根據主管需求查詢精準事實，並簡潔回報核心要點。不要擅自撰寫長篇文案。"
    )
    # 將系統提示詞與既有對話合併
    messages = [system_prompt] + state["messages"]
    
    # 執行調研推理與工具調用
    response = researcher_llm.invoke(messages)
    
    # 若有工具調用，在此進行簡易執行處理
    if response.tool_calls:
        tool_call = response.tool_calls[0]
        tool_result = search_tech_database.invoke(tool_call["args"])
        return {
            "messages": [
                AIMessage(content=f"【調研專家回報】\n檢索結果：{tool_result}", name="Researcher")
            ]
        }
    
    return {"messages": [AIMessage(content=f"【調研專家回報】\n{response.content}", name="Researcher")]}

# 3. 建立 Writer 節點
def writer_node(state: TeamState):
    """文案專家：負責統整調研結果並撰寫最終報告"""
    system_prompt = SystemMessage(
        content="你是專屬的【文案專家】。請根據對話中調研專家提供的所有事實，整理出一份條理分明的 Markdown 報告。"
    )
    messages = [system_prompt] + state["messages"]
    response = llm.invoke(messages)
    
    return {
        "messages": [
            AIMessage(content=f"【文案專家產出】\n{response.content}", name="Writer")
        ]
    }

四、步驟 3：定義 Supervisor 主管節點與條件路由

主管節點透過結構化輸出（Structured Output）決定下一步該派工給誰。

# 1. 建立綁定路由契約的主管 LLM
supervisor_chain = llm.with_structured_output(RouterOutput)

# 2. 定義 Supervisor 節點
def supervisor_node(state: TeamState):
    """主管節點：審視對話進度並進行派工決策"""
    system_prompt = SystemMessage(
        content="""你是團隊的【主管 (Supervisor)】。
你的團隊有兩位專家：
1. Researcher：負責查找技術資料與事實。
2. Writer：負責將蒐集到的資訊撰寫成完整的技術文案。

請評估目前的對話進度：
- 若使用者提出新主題且尚未調研，指派給 'Researcher'。
- 若調研專家已回報事實，但尚未產生正式報告，指派給 'Writer'。
- 若文案專家已完成報告輸出，選擇 'FINISH'。
"""
    )
    messages = [system_prompt] + state["messages"]
    decision: RouterOutput = supervisor_chain.invoke(messages)
    
    return {"next_worker": decision.next_worker}

# 3. 定義條件路由函數
def route_next_step(state: TeamState) -> str:
    """根據主管的決策轉發到對應節點或結束"""
    next_worker = state.get("next_worker")
    if next_worker == "FINISH":
        return END
    elif next_worker == "Researcher":
        return "researcher"
    elif next_worker == "Writer":
        return "writer"
    return END

五、步驟 4：組裝 Multi-Agent 狀態圖並編譯

我們將三個節點連線成完整的協作網絡：

# 1. 初始化狀態圖
workflow = StateGraph(TeamState)

# 2. 加入節點
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", research_node)
workflow.add_node("writer", writer_node)

# 3. 建立連線架構
# 起點直達主管
workflow.add_edge(START, "supervisor")

# 主管依據條件邊分流派工
workflow.add_conditional_edges(
    "supervisor",
    route_next_step,
    {
        "researcher": "researcher",
        "writer": "writer",
        END: END
    }
)

# 各專家執行完畢後，一律回報給主管進行下一輪評估
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("writer", "supervisor")

# 4. 編譯狀態圖
app = workflow.compile()

六、步驟 5：執行與端到端協同驗證

我們丟入一個需要先查資料再寫作的複合任務，觀察狀態圖的派工軌跡：

# 啟動任務
user_input = "請幫我調研 LangGraph 的核心特性，並整理成一份簡明技術介紹。"

initial_state = {
    "messages": [HumanMessage(content=user_input)],
    "next_worker": ""
}

print("=== Multi-Agent 團隊開始協同作業 ===\n")
for step in app.stream(initial_state, stream_mode="updates"):
    for node_name, node_output in step.items():
        print(f"--- 節點執行: [{node_name}] ---")
        if "next_worker" in node_output:
            print(f"主管決策 ➔ 下一步指派給: {node_output['next_worker']}")
        if "messages" in node_output:
            for msg in node_output["messages"]:
                print(f"{msg.content}\n")

執行軌跡解析：

1. START ➔ supervisor：主管審視初始需求，判定需要先調研，輸出 next_worker="Researcher"。
2. supervisor ➔ researcher：調研專家被喚醒，調用 search_tech_database 工具取得 LangGraph 核心特性，回傳精簡事實。
3. researcher ➔ supervisor：控制權回到主管，主管發現已具備事實但未排版報告，輸出 next_worker="Writer"。
4. supervisor ➔ writer：文案專家被喚醒，讀取調研結果並產出結構完整的 Markdown 技術介紹。
5. writer ➔ supervisor：控制權再次回到主管，主管判定報告已完成，輸出 next_worker="FINISH" ➔ 順利抵達 END。

七、實戰工程總結：Supervisor 模式的設計心法

小結與下集預告

透過 Supervisor 模式，我們成功將單一複雜的任務拆解為「主管指揮 ➔ 調研 ➔ 撰寫」的模組化流水線。每個 Agent 職責單一、Context 純淨，系統具備極高的穩定度與擴展性。

然而，在涉及金錢交易、敏感資料修改或重要郵件發送的場景中，完全放手讓 Agent 自主決策依然存在風險。我們必須在關鍵節點引入**人工審批（Human-in-the-Loop）**機制。

明天 【Day 14】LangGraph Human-in-the-Loop：斷點中斷、人工審批與狀態修改實戰，我們將探討如何讓 Agent 在關鍵步驟安全暫停、等待人類審查並動態修改狀態！