多 Agent 協作架構實戰：從設計模式到生產落地（2026 完整指南）

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class PipelineState(TypedDict):
    query: str
    retrieved_docs: str
    analysis: str
    final_report: str

def retrieval_agent(state: PipelineState):
    docs = search_knowledge_base(state["query"])
    return {"retrieved_docs": docs}

def analysis_agent(state: PipelineState):
    result = llm.invoke(f"分析以下內容：{state['retrieved_docs']}")
    return {"analysis": result.content}

def writer_agent(state: PipelineState):
    report = llm.invoke(f"根據分析撰寫報告：{state['analysis']}")
    return {"final_report": report.content}

builder = StateGraph(PipelineState)
builder.add_node("retriever", retrieval_agent)
builder.add_node("analyzer", analysis_agent)
builder.add_node("writer", writer_agent)
builder.add_edge(START, "retriever")
builder.add_edge("retriever", "analyzer")
builder.add_edge("analyzer", "writer")
builder.add_edge("writer", END)
pipeline = builder.compile()

from langgraph.types import Send
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class ResearchState(TypedDict):
    query: str
    research_results: Annotated[list, operator.add]
    final_synthesis: str

def supervisor(state: ResearchState):
    subtasks = [
        {"query": state["query"], "source": "academic"},
        {"query": state["query"], "source": "industry"},
        {"query": state["query"], "source": "news"},
    ]
    return [Send("research_worker", task) for task in subtasks]

def research_worker(state: dict):
    result = search_by_source(state["query"], state["source"])
    return {"research_results": [result]}

def synthesizer(state: ResearchState):
    combined = "\n".join(state["research_results"])
    synthesis = llm.invoke(f"綜合以下研究結果：{combined}")
    return {"final_synthesis": synthesis.content}

builder = StateGraph(ResearchState)
builder.add_node("research_worker", research_worker)
builder.add_node("synthesizer", synthesizer)
builder.add_conditional_edges(START, supervisor, ["research_worker"])
builder.add_edge("research_worker", "synthesizer")
builder.add_edge("synthesizer", END)
graph = builder.compile()

KEYWORD_ROUTING = {
    "程式碼": "code_agent",
    "code": "code_agent",
    "搜尋": "search_agent",
    "查詢": "search_agent",
    "資料": "data_agent",
}

def supervisor_with_fast_path(state):
    query = state["query"].lower()
    for keyword, agent_name in KEYWORD_ROUTING.items():
        if keyword in query:
            return {"next": agent_name}
    routing_prompt = f"""
    使用者請求：{state['query']}
    可用 Agent：code_agent, search_agent, data_agent
    請返回最合適的 Agent 名稱，只返回名稱。
    """
    decision = llm.invoke(routing_prompt)
    return {"next": decision.content.strip()}

import autogen

reviewer_1 = autogen.AssistantAgent(
    name="SecurityReviewer",
    system_message="你是一位安全專家，專注於程式碼中的安全漏洞。"
)
reviewer_2 = autogen.AssistantAgent(
    name="PerformanceReviewer",
    system_message="你是一位效能專家，專注於程式碼的效率和資源使用。"
)
human_proxy = autogen.UserProxyAgent(
    name="CodeAuthor",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=2,
    is_termination_msg=lambda x: "APPROVED" in x.get("content", "")
)
groupchat = autogen.GroupChat(
    agents=[human_proxy, reviewer_1, reviewer_2],
    messages=[],
    max_round=6
)
manager = autogen.GroupChatManager(groupchat=groupchat)

from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("data-agent-mcp")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_customer_db",
            description="查詢客戶資料庫，支援按 ID、姓名、Email 檢索",
            inputSchema={
                "type": "object",
                "properties": {
                    "field": {"type": "string", "enum": ["id", "name", "email"]},
                    "value": {"type": "string"}
                },
                "required": ["field", "value"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_customer_db":
        result = db.query(arguments["field"], arguments["value"])
        return [TextContent(type="text", text=str(result))]

{
  "name": "ResearchAgent",
  "version": "1.0",
  "description": "專業資訊檢索與摘要 Agent",
  "url": "https://research-agent.internal/a2a",
  "capabilities": {
    "streaming": true,
    "async": true
  },
  "skills": [
    {
      "id": "web_research",
      "name": "網路資訊檢索",
      "description": "從網際網路檢索並摘要最新資訊",
      "tags": ["research", "summarization", "web"]
    },
    {
      "id": "academic_search",
      "name": "學術文獻檢索",
      "description": "檢索 arXiv、Google Scholar 等學術資料庫"
    }
  ]
}

import httpx

async def discover_and_delegate(agent_url: str, task: str):
    card_response = await httpx.get(f"{agent_url}/.well-known/agent.json")
    agent_card = card_response.json()
    available_skills = [s["id"] for s in agent_card["skills"]]
    if "web_research" not in available_skills:
        raise ValueError(f"Agent {agent_card['name']} 不支援 web_research 技能")
    payload = {
        "jsonrpc": "2.0",
        "method": "message/send",
        "id": "task-001",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": task}]
            }
        }
    }
    response = await httpx.post(agent_card["url"], json=payload)
    return response.json()

from opentelemetry import trace
import uuid

tracer = trace.get_tracer("multi-agent-system")

def traced_agent_call(agent_name: str, task: dict, correlation_id: str = None):
    if not correlation_id:
        correlation_id = str(uuid.uuid4())
    with tracer.start_as_current_span(f"agent.{agent_name}") as span:
        span.set_attribute("agent.name", agent_name)
        span.set_attribute("correlation.id", correlation_id)
        span.set_attribute("task.type", task.get("type", "unknown"))
        try:
            result = agent_registry[agent_name].run(task)
            span.set_attribute("agent.tokens_used", result.get("tokens", 0))
            span.set_attribute("agent.status", "success")
            return result
        except Exception as e:
            span.set_attribute("agent.status", "error")
            span.set_attribute("error.message", str(e))
            raise

import json

def evaluate_agent_output(original_task: str, agent_output: str) -> dict:
    evaluation_prompt = f"""
    你是一位嚴格的品質評審專家。請評估以下 AI Agent 的輸出品質。
    原始任務：{original_task}
    Agent 輸出：{agent_output}
    請從完成度、準確性、相關性、幻覺檢測四個維度評分（1-5 分）。
    以 JSON 格式返回：
    {{"completeness": x, "accuracy": x, "relevance": x, "hallucination_detected": true/false, "comments": "..."}}
    """
    evaluation = llm.invoke(evaluation_prompt)
    return json.loads(evaluation.content)

你的任務是否有明確的線性依賴步驟？
├─ 是 → 子任務是否可以並行執行？
│        ├─ 否 → 【順序流水線】
│        └─ 是 → 【並行扇出 + 順序流水線 混合】
│
└─ 否 → 是否有一個 Agent 具有決策權威？
         ├─ 是 → 規模是否足夠大需要子團隊？
         │        ├─ 否 → 【Supervisor-Worker 層級模式】
         │        └─ 是 → 【層級式（Supervisors of Supervisors）】
         │
         └─ 否 → 任務是否是長時間非同步的？
                  ├─ 是 → 【黑板架構】
                  └─ 否 → Agent 數量是否 ≤ 5？
                           ├─ 是 → 【Swarm（注意設定終止條件）】
                           └─ 否 → 【考慮重新拆分為層級模式】