LangGraph 多模型實戰：從零到 Production 的完整教學

這篇文章是給誰看的？

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

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

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

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

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

面向	Claude Code / Cursor / Copilot	LangGraph
本質	開發者工具 — 你用它寫 code	開發框架 — 你用它造產品
使用者是誰	你自己（開發者）	你的客戶 / 你的系統 / 你的團隊
使用場景	日常寫程式、debug、重構	建 chatbot、自動化流程、API 服務
互動方式	人 ↔ AI 即時對話	程式碼自動跑，可以無人值守
模型選擇	工具幫你選好（通常固定一個）	你自己決定哪個步驟用哪個模型
計費方式	月費訂閱（工具包了）	純 API 按量計費，你自己控制成本

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

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

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

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

能力	Claude Sonnet	Groq Llama 70B	Groq Llama 8B
架構設計 / Code Review	⭐ 最強	普通	弱
程式碼生成	強	⭐ 強且快	普通
簡單問答	大材小用	大材小用	⭐ 夠用且極便宜
生成速度	~50 tok/s	~300 tok/s	~800 tok/s
費用 (Output/1M tokens)	$15.00	$0.79	$0.08

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

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)")

實測結果

階段	模型	結果
🧠 Design	Claude Sonnet 4.6	產出 4 個參數、10 個 edge case、9 步 pseudocode 的完整規格
⚡ Implement	Groq Llama 70B	完整 Python 函數，含 type hints、docstring、error handling
🔍 Review	Claude Sonnet 4.6	VERDICT: APPROVED — 第一次就通過，沒有重試

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

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

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

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

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

分類器怎麼運作？

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

分類	路由到	觸發條件	費用 (Output/1M)
🧠 深度思考	Claude Sonnet	架構分析、比較權衡、code review、規劃	$15.00
⚡ 寫程式	Llama 70B	實作函數、生成腳本、重構、修 bug	$0.79
💨 快速回答	Llama 8B	打招呼、簡單問答、定義、基礎算數	$0.08

核心程式碼

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

問題	路由結果	正確？
Hi!	💨 Llama 8B	✅
1+1=?	💨 Llama 8B	✅
Write a Python quicksort	⚡ Llama 70B	✅
Compare microservices vs monolith	🧠 Sonnet	✅
What is Docker?	💨 Llama 8B	✅
幫我寫一個 REST API	⚡ Llama 70B	✅
分析 Redis cache vs CDN 優缺點	🧠 Sonnet	✅

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

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

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

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

Streaming：體感延遲降 25 倍

方式	使用者體驗	程式碼差別
invoke()（非串流）	等 5 秒 → 突然出現整段文字	response = model.invoke(messages)
stream()（串流）	0.2 秒開始出字 → 像打字一樣流暢	for chunk in model.stream(messages)

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

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

Fallback：模型掛了自動切換

主要模型	Fallback 模型	切換代價
🧠 Sonnet	⚡ Llama 70B	分析品質略降，速度更快
⚡ Llama 70B	🧠 Sonnet	速度略慢，品質更高
💨 Llama 8B	⚡ Llama 70B	稍慢稍貴，但一定能回答

# 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 端點設計

端點	方式	適合場景
POST /chat	非串流，回傳完整 JSON	Slack/Line bot、後端呼叫、批次處理
POST /chat/stream	SSE 串流，逐 token 推送	網頁聊天窗、需要即時體感的 UI
GET /health	健康檢查	負載平衡器、監控系統
GET /sessions/{id}	取得對話歷史	Debug、對話紀錄查詢
DELETE /sessions/{id}	清除對話	使用者開始新對話

啟動與測試

# 安裝額外套件
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 推

四個階段的完整對照

階段	檔案	新增能力	適合誰	可以停在這嗎？
1. Duo 流水線	duo.py	設計→實作→審查	批量生成程式碼	✅
2. Smart Router	router.py	自動選模型	個人實驗、學習	✅
3. Streaming + Fallback	router_stream.py	逐字輸出 + 自動容錯	團隊內部使用	✅
4. FastAPI 部署	server.py	HTTP API + SSE + Session	對外服務、產品整合	✅

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

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

方案	月成本估算	品質
全部用 Claude Sonnet	~$54	最高，但簡單題大材小用
Smart Router（自動切換）	~$13.50	深度題用 Sonnet，其餘用 Llama
全部用 Groq Llama 70B	~$4.20	最便宜，但分析品質弱

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。

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

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

面向	做什麼	不做的後果
RAG（檢索增強）	接向量資料庫，讓 AI 查你的文件回答	AI 只能回答通用知識，不懂你的業務
評估（LangSmith）	追蹤每次呼叫的路由、延遲、成本	不知道 Router 分類準不準，成本失控
Session 持久化	用 Redis 存對話歷史（目前在記憶體）	Server 重啟，所有對話消失
認證	API key 或 JWT 驗證	任何人都能呼叫你的 API，幫你花錢
Prompt Injection 防護	驗證使用者輸入，防止惡意 prompt	使用者讓 AI 做不該做的事
Docker 容器化	打包成 Docker image 部署	環境不一致，部署困難
Subgraph 嵌套	Router 判斷「寫程式」→ 啟動整個 Duo 流水線	只能單步回答，沒有品質把關

完整專案結構

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 的差距