多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、姓名、電子郵件檢索",
            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))]

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 langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt

with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/agentdb") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "user-session-12345"}}
    result = graph.invoke({"query": "分析 Q2 財報"}, config)

def high_risk_action_agent(state):
    proposed_action = plan_action(state)
    human_decision = interrupt({
        "proposed_action": proposed_action,
        "risk_level": "HIGH",
        "message": "此操作將修改正式環境資料庫，請確認是否執行"
    })
    if human_decision["approved"]:
        return execute_action(proposed_action)
    return {"status": "cancelled", "reason": human_decision.get("reason")}

import time
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = "CLOSED"
        self.last_failure_time = None

    def __call__(self, func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if self.state == "OPEN":
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = "HALF_OPEN"
                else:
                    raise Exception("Circuit breaker OPEN - Agent 暫時無法使用")
            try:
                result = await func(*args, **kwargs)
                if self.state == "HALF_OPEN":
                    self.state = "CLOSED"
                    self.failure_count = 0
                return result
            except Exception:
                self.failure_count += 1
                self.last_failure_time = time.time()
                if self.failure_count >= self.failure_threshold:
                    self.state = "OPEN"
                raise
        return wrapper

class TokenBudgetManager:
    def __init__(self, total_budget: int = 100_000):
        self.total_budget = total_budget
        self.used_tokens = 0
        self.agent_usage = {}

    def check_budget(self, agent_name: str, estimated_tokens: int) -> bool:
        remaining = self.total_budget - self.used_tokens
        if estimated_tokens > remaining:
            raise BudgetExceededException(
                f"Agent {agent_name} 請求 {estimated_tokens} tokens，"
                f"但剩餘預算僅 {remaining} tokens"
            )
        return True

    def record_usage(self, agent_name: str, actual_tokens: int):
        self.used_tokens += actual_tokens
        self.agent_usage[agent_name] = self.agent_usage.get(agent_name, 0) + actual_tokens

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

MONITORING_METRICS = {
    "task_success_rate": "端到端任務完成率（目標：>85%）",
    "e2e_latency_p95": "P95 端到端延遲（目標：<30s）",
    "total_cost_per_task": "每次任務平均 Token 成本",
    "agent_error_rate": "各 Agent 錯誤率（目標：<5%）",
    "agent_retry_count": "重試次數（高重試 = 需要調查）",
    "tool_call_budget_usage": "工具呼叫次數／預算比",
    "output_quality_score": "輸出品質評分",
    "goal_alignment_score": "目標一致性評分",
    "hallucination_rate": "幻覺偵測率",
}

def evaluate_agent_output(original_task: str, agent_output: str) -> dict:
    evaluation_prompt = f"""
    原始任務：{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)

def validate_agent_output(output: dict, schema: dict) -> bool:
    jsonschema.validate(output, schema)
    if output.get("confidence_score", 1.0) < 0.7:
        raise LowConfidenceError(f"Agent 輸出置信度過低: {output['confidence_score']}")
    required_fields = schema.get("required", [])
    missing = [f for f in required_fields if not output.get(f)]
    if missing:
        raise MissingFieldsError(f"輸出缺少必填欄位: {missing}")
    return True

MAX_ITERATIONS = 10
MAX_TOOL_CALLS_PER_AGENT = 20
MAX_TOTAL_TOKENS = 50_000

class ProductionGuardrails:
    def validate_input(self, user_input: str) -> str:
        if len(user_input) > 10000:
            raise InputTooLongError("輸入超過 10000 字元限制")
        injection_patterns = ["ignore previous instructions", "forget everything"]
        for pattern in injection_patterns:
            if pattern.lower() in user_input.lower():
                raise PromptInjectionError("偵測到潛在的提示注入攻擊")
        return user_input.strip()

    def validate_output(self, output: str) -> str:
        output = self.pii_filter.redact(output)
        if self.content_classifier.is_harmful(output):
            raise HarmfulContentError("輸出包含有害內容")
        return output

builder.add_node("supervisor", supervisor_node, defer=True)

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