【模式 A：Orchestrator-Worker（指揮官-執行者）】

              ┌─────────────┐
              │ Orchestrator │  ← 接收任務、分解、分派、彙整
              └──────┬──────┘
         ┌───────────┼───────────┐
    ┌────▼────┐ ┌────▼────┐ ┌───▼─────┐
    │Worker A │ │Worker B │ │Worker C │
    │(爬蟲)   │ │(分析)   │ │(發送)   │
    └─────────┘ └─────────┘ └─────────┘

【模式 B：Pipeline（流水線）】

  Agent 1 → Agent 2 → Agent 3 → Agent 4
  (搜尋)   (摘要)   (翻譯)   (發布)
  每個 Agent 處理前一個的輸出，依序傳遞

【模式 C：Debate / Peer Review（辯論/同儕審查）】

  Agent A ──提案──▶ Agent B（審查）
     ▲                    │
     └────────修正────────┘
  多個 Agent 互相挑戰，直到達成高品質共識

PR 提交
  → Agent 1（靜態分析）：找語法錯誤、安全漏洞
  → Agent 2（邏輯審查）：理解業務邏輯，評估實作正確性
  → Agent 3（文件審查）：確認註解、README、CHANGELOG 是否更新
  → Agent 4（彙整）：合併所有審查意見，生成結構化 Review 報告

python
# 方式 A：Orchestrator 主動分派（適合動態任務）
class Orchestrator:
    def run(self, task: str):
        subtasks = self.decompose(task)         # LLM 分解任務
        results = asyncio.gather(*[             # 並行執行
            worker.execute(t) for t in subtasks
        ])
        return self.synthesize(results)         # LLM 彙整結果

# 方式 B：靜態 Pipeline（適合固定流程）
result = (
    SearchAgent().run(query)
    |> SummaryAgent().run
    |> TranslationAgent().run
    |> PublishAgent().run
)

python
class AgentMessage(BaseModel):
    agent_id: str
    task_id: str
    status: Literal["success", "partial", "failed"]
    output: dict          # 符合預定 Schema 的輸出
    confidence: float     # 0.0 ~ 1.0，供 Orchestrator 判斷是否需要重試
    metadata: dict        # 執行時間、token 用量等診斷資訊

python
async def safe_execute(agent, task, fallback=None):
    try:
        result = await asyncio.wait_for(agent.run(task), timeout=30)
        if result.confidence < 0.6:
            result = await agent.run(task, temperature=0.2)
        return result
    except asyncio.TimeoutError:
        return fallback or AgentMessage(status="failed", output={})