分類: 💻 技術實踐

這篇文章是給誰看的？

如果你符合以下任一情況，這篇文章就是為你寫的：

• 你想做一個 AI chatbot（客服、內部助手、產品功能），但不知道怎麼開始
• 你已經在用 Claude / GPT，但覺得全部用最貴的模型太浪費錢
• 你聽過「多模型協作」但不確定具體怎麼實作
• 你想知道 LangGraph 跟你平常用的 Claude Code / Cursor / Copilot 到底差在哪

本文從零開始，帶你走完四個階段，每個階段都有完整可執行的程式碼。你可以在任何一個階段停下來——不是每個人都需要走到 production。

先搞清楚：這跟 Claude Code / Cursor / Copilot 完全不同

在往下讀之前，先釐清最容易搞混的一件事：

用一個比喻：Claude Code 是你請了一個很強的工程師坐在旁邊幫你寫 code；LangGraph 是你在蓋一條自動化產線，產線上有不同的機器人各司其職。

如果你只是日常寫程式，用 Claude Code 就好，不需要 LangGraph。往下讀之前，確認你的需求是「造一個東西給別人用」，而不是「讓自己寫 code 更快」。

為什麼不同的 AI 模型要分工合作？

沒有一個模型什麼都最好。每個模型有不同的強項和定價：

核心邏輯：讓擅長「想」的模型去想，讓擅長「做」的模型去做，讓便宜的模型處理瑣事。就像軟體團隊裡，架構師出規格、工程師寫程式、實習生回答簡單問題一樣。

LangGraph 是什麼？一分鐘看懂

LangGraph 是 LangChain 團隊開發的有狀態流程編排框架。三個核心概念：

1. Node（節點）— 一個步驟，比如「用 Sonnet 設計」或「用 Llama 實作」
2. Edge（邊）— 步驟之間的連線，決定「做完 A 接著做 B」
3. State（狀態）— 所有節點共享的資料，比如設計規格、程式碼、審查結果

把這三個組合起來，就是一個可以自動跑的流水線。你定義「什麼步驟做什麼、什麼條件走什麼路」，框架負責執行。

環境準備（所有階段共用）

開始之前，你需要準備兩個 API key 和安裝套件。這一步做完，後面四個階段都不用再設定。

1. 取得 API Key（兩個都有免費額度）

• Anthropic API：console.anthropic.com/settings/keys
• Groq API：console.groq.com/keys

⚠️ 注意：這是 API key，跟 Claude.ai 的月費訂閱、GitHub Copilot 的訂閱完全無關。API 是按用量計費的。

2. 建立專案

mkdir langgraph-duo && cd langgraph-duo

# 安裝套件
pip install langgraph langchain-anthropic langchain-groq python-dotenv

# 建立 .env 檔案（填入你的 key）
cat > .env << 'EOF'
ANTHROPIC_API_KEY=sk-ant-xxxx
GROQ_API_KEY=gsk_xxxx
EOF

3. 驗證連線

python3 -c "
from dotenv import load_dotenv; load_dotenv()
from langchain_anthropic import ChatAnthropic
from langchain_groq import ChatGroq

sonnet = ChatAnthropic(model='claude-sonnet-4-6', max_tokens=50)
llama = ChatGroq(model='llama-3.3-70b-versatile', max_tokens=50)

print('Sonnet:', sonnet.invoke('Say OK').content)
print('Llama:', llama.invoke('Say OK').content)
"

兩行都印出 OK，就可以開始了。

第一階段：Duo 流水線（設計 → 實作 → 審查）

目標：讓 Claude Sonnet 設計規格，Groq Llama 寫程式碼，Sonnet 再審查品質。審查不通過自動重做，最多 3 次。

適合場景：批量生成程式碼、自動化 code review、需要品質把關的程式碼生成。

Task → [Sonnet 設計] → [Llama 實作] → [Sonnet 審查]
                            ↑               |
                            └── 未通過 ──────┘  (最多 3 次)
                                  ↓ 通過
                                 END

完整程式碼：duo.py

from typing import TypedDict
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage, SystemMessage

load_dotenv()

# 模型分工：Sonnet 想，Llama 做
sonnet = ChatAnthropic(model="claude-sonnet-4-6", max_tokens=4096)
llama = ChatGroq(model="llama-3.3-70b-versatile", max_tokens=4096)

# 所有節點共享的狀態
class AgentState(TypedDict):
    task: str             # 原始需求
    design: str           # Sonnet 的設計規格
    code: str             # Llama 的實作
    review: str           # Sonnet 的審查結果
    revision_notes: str   # 修改指示（給重試用）
    approved: bool        # 是否通過
    attempt: int          # 重試次數

MAX_ATTEMPTS = 3

# Node 1: Sonnet 設計
def design_node(state):
    response = sonnet.invoke([
        SystemMessage(content="You are a senior architect. Produce a precise technical spec with function signatures, edge cases, and pseudocode."),
        HumanMessage(content=f"Request: {state['task']}")
    ])
    return {"design": response.content}

# Node 2: Llama 實作
def implement_node(state):
    prompt = f"Spec:\n{state['design']}"
    if state.get("revision_notes"):
        prompt += f"\n\nFix these issues:\n{state['revision_notes']}"
    response = llama.invoke([
        SystemMessage(content="Implement per spec. Output only Python code."),
        HumanMessage(content=prompt)
    ])
    return {"code": response.content, "attempt": state.get("attempt", 0) + 1}

# Node 3: Sonnet 審查
def review_node(state):
    response = sonnet.invoke([
        SystemMessage(content="Review code vs spec. Reply VERDICT: APPROVED or REJECTED with details."),
        HumanMessage(content=f"Spec:\n{state['design']}\n\nCode:\n{state['code']}")
    ])
    review = response.content
    approved = "APPROVED" in review.upper()
    return {"review": review, "approved": approved,
            "revision_notes": "" if approved else review}

# 條件路由：通過 → 結束，未通過 → 回去重做
def should_continue(state):
    if state["approved"] or state["attempt"] >= MAX_ATTEMPTS:
        return "end"
    return "revise"

# 組裝 Graph
workflow = StateGraph(AgentState)
workflow.add_node("design", design_node)
workflow.add_node("implement", implement_node)
workflow.add_node("review", review_node)
workflow.set_entry_point("design")
workflow.add_edge("design", "implement")
workflow.add_edge("implement", "review")
workflow.add_conditional_edges("review", should_continue,
                               {"end": END, "revise": "implement"})
app = workflow.compile()

# 跑！
result = app.invoke({
    "task": "Write a Python function that reads a CSV and returns column averages as a dict",
    "design": "", "code": "", "review": "",
    "revision_notes": "", "approved": False, "attempt": 0,
})

print("=== CODE ===")
print(result["code"])
print(f"\n{'✅ APPROVED' if result['approved'] else '⚠️ BEST EFFORT'} after {result['attempt']} attempt(s)")

實測結果

到這裡你已經有一個能跑的多模型流水線了。如果你的需求是「批量生成程式碼 + 自動品質把關」，可以停在這個階段。

第二階段：Smart Router（自動選模型的聊天機器人）

目標：做一個像 ChatGPT 一樣的對話介面，但底下不是固定一個模型，而是自動根據問題類型選最適合的模型。

適合場景：客服 chatbot、團隊內部 AI 助手、需要控制 API 成本的聊天服務。

你的問題 → [Llama 8B 分類器] → 判斷類型
                                  |
                  ┌────────────────┼────────────────┐
                  ↓                ↓                ↓
            [Claude Sonnet]  [Llama 70B]      [Llama 8B]
             深度分析          寫程式            簡單問答
                  ↓                ↓                ↓
                  └────────────────┼────────────────┘
                                   ↓
                                回答你

分類器怎麼運作？

分類器用最便宜的 Llama 8B（每次呼叫不到 $0.0001）讀取問題，輸出一個 JSON 判斷結果。分類規則：

核心程式碼

import json
from langgraph.graph import StateGraph, END

# 分類器：Llama 8B 讀問題，判斷該走哪條路
def router_node(state):
    response = llama_8b.invoke([
        SystemMessage(content="""Classify into one category:
- "sonnet": complex reasoning, analysis, architecture
- "llama_70b": write code, implement, fix bugs
- "llama_8b": greetings, simple facts, casual chat
Reply ONLY JSON: {"route": "...", "reason": "..."}"""),
        HumanMessage(content=state["question"])
    ])
    parsed = json.loads(response.content)
    return {"route": parsed["route"]}

# 條件路由：根據分類結果，走不同的回答節點
workflow = StateGraph(RouterState)
workflow.add_node("router", router_node)
workflow.add_node("sonnet", answer_sonnet)
workflow.add_node("llama_70b", answer_llama_70b)
workflow.add_node("llama_8b", answer_llama_8b)

workflow.set_entry_point("router")
workflow.add_conditional_edges("router", lambda s: s["route"], {
    "sonnet": "sonnet",
    "llama_70b": "llama_70b",
    "llama_8b": "llama_8b",
})
app = workflow.compile()

實測分類準確度：7/7

到這裡你有一個能自動選模型的聊天機器人了。但它還缺兩個東西：回答會一次全部吐出（不是逐字顯示），而且 Groq 掛了就整個壞掉。第三階段解決這兩個問題。

第三階段：Streaming + Fallback（讓它不會掛）

目標：回答逐字出現（像 ChatGPT 一樣流暢），而且模型掛了自動切換備用模型。

為什麼這一步很重要：沒有 Streaming 的 chatbot，使用者體驗像 2010 年的網頁——按下送出，等 5 秒，突然一大段文字出現。沒有 Fallback 的服務，Groq 一限速（免費版每分鐘 30 次），你的整個服務就掛了。

Streaming：體感延遲降 25 倍

# 只需要把 .invoke() 改成 .stream()
# 然後迭代每個 chunk 即時輸出

for chunk in model.stream(messages):
    print(chunk.content, end="", flush=True)  # flush=True 強制即時顯示

Fallback：模型掛了自動切換

# Fallback 模式：try 主要模型，失敗自動切備用
try:
    for chunk in primary_model.stream(messages):
        print(chunk.content, end="", flush=True)
except Exception:
    print("⚠️ Primary failed, switching to fallback...")
    for chunk in fallback_model.stream(messages):
        print(chunk.content, end="", flush=True)

完整的 router_stream.py 把 Streaming + Fallback + 對話歷史 + 使用統計整合在一起，不到 200 行。跑起來就是一個帶有自動模型切換的 terminal 聊天機器人。

到這裡你有一個穩定的、體驗流暢的聊天機器人了。但它還是跑在你的 terminal 裡，只有你能用。第四階段讓它變成任何人都能呼叫的 API 服務。

第四階段：FastAPI 部署（讓別人能用）

目標：把 chatbot 包成 HTTP API，讓網頁、App、Line bot、Slack bot 都能呼叫。

為什麼是 FastAPI：Python 生態最主流的 API 框架，原生支援 async、自動生成 API 文件、社群龐大。

API 端點設計

啟動與測試

# 安裝額外套件
pip install fastapi uvicorn sse-starlette

# 啟動 server
python server.py
# → 🤖 Smart Router API running on http://localhost:8900

# 測試非串流
curl -X POST http://localhost:8900/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "什麼是 Docker?", "session_id": "test-user"}'

# 回應範例：
# {
#   "answer": "Docker 是一個容器化平台...",
#   "model_used": "llama_8b",
#   "route_reason": "simple definition question",
#   "session_id": "test-user",
#   "fallback_used": false,
#   "elapsed_seconds": 0.85
# }

前端怎麼接 SSE 串流？

如果你要做網頁聊天窗，前端只需要幾行 JavaScript：

// 瀏覽器原生 EventSource API
const response = await fetch('/chat/stream', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({message: '寫一個排序函數', session_id: 'user-1'})
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const {done, value} = await reader.read();
  if (done) break;
  const text = decoder.decode(value);
  // 每個 chunk 到了就即時顯示在畫面上
  document.getElementById('chat').innerHTML += text;
}

為什麼用 SSE 而不是 WebSocket？

聊天場景是「使用者送一次訊息、AI 回一次」的單向推送。SSE 比 WebSocket 更適合：

• 更簡單 — 單向傳輸，不需要管雙向連線
• 瀏覽器原生 — EventSource API 內建自動重連
• 穿透力好 — 走標準 HTTP，能通過代理和 CDN
• 夠用 — 使用者的訊息用 POST 送，AI 的回應用 SSE 推

四個階段的完整對照

費用比較：到底能省多少？

假設一天 100 個問題，其中 20% 深度分析、30% 寫程式、50% 簡單問答：

Smart Router 方案比全用 Sonnet 省 75%，同時保留了深度分析任務的 Sonnet 品質。規模越大差距越明顯——1000 個使用者的 SaaS 產品，月省 $3000+。

什麼場景適合？什麼場景不適合？

✅ 適合的場景

1. 客服 Chatbot — 70% 簡單題用 Llama 8B 秒回，複雜題自動升級到 Sonnet
2. 團隊 AI 助手 — 接 Slack，PM 問策略用 Sonnet，工程師要 code 用 Llama 70B
3. 自動化 Pipeline — CI/CD 中的 AI code review，PR 提交自動跑
4. SaaS 產品 — 加 AI 功能但要控成本，簡單摘要用 Llama，深度分析用 Sonnet
5. 批量內容生成 — 50 篇產品描述：Sonnet 定規範 → Llama 批量寫 → Sonnet 抽檢

❌ 不適合的場景

• 日常寫程式 — 用 Claude Code 或 Cursor 就好，不需要 LangGraph
• 一次性分析 — 直接貼給 Claude 問就好，不需要搭 pipeline
• 不需要控成本 — 個人使用月花不到 $10，Smart Router 省下的錢不值得建置成本

三個問題判斷法

在決定要不要用之前，問自己：

1. 使用者是誰？ — 你自己 → 不需要。別人（客戶/團隊/系統）→ 繼續看
2. 會跑多少次？ — 幾次 → 直接呼叫 API。幾百幾千次 → LangGraph 有意義
3. 需要品質分級嗎？ — 所有問題都要最高品質 → 用最強模型。不同問題可以不同品質 → Smart Router

三個都答「後者」才值得用 LangGraph。

走完四個階段之後，還有什麼？

如果你的服務要上正式商業環境，還有幾個面向需要處理：

完整專案結構

langgraph-duo/
├── .env                  # API keys（不要 commit）
├── .gitignore            # 排除 .env
├── requirements.txt      # 所有套件
├── duo.py                # 階段 1：設計→實作→審查 流水線
├── duo_notebook.ipynb    # 階段 1 的 Jupyter 版
├── router.py             # 階段 2：Smart Router 基本版
├── router_stream.py      # 階段 3：+ Streaming + Fallback
└── server.py             # 階段 4：FastAPI HTTP API

所有程式碼都有詳細的中英文註解，說明每個模型的選用原因和適用場景。從哪個階段開始都可以，每個檔案都是獨立可執行的。

總結

LangGraph 多模型流水線的核心價值不是「用 AI 寫程式更快」——如果只是速度，直接用一個模型最快。它的價值在於把 AI 當成團隊來管理：讓擅長設計的去設計、擅長實作的去實作、擅長審查的去審查。

從 Duo 流水線到 Smart Router，再到帶有 Streaming、Fallback、API 部署的生產版本，每一步都是從「能跑」走向「能上線」的必經之路。你不需要一次走完四個階段——先跑通第一階段，確認這個架構對你有用，再往下走。

相關閱讀：LangChain/LangGraph 深度分析：架構師、顧問、個人公司的實戰指南 | Claude Code Agent Teams：從穩定執行到自動化代碼審查的完整指南 | 舊系統整合場景下，會用 vs 不會用 Claude Code 的差距