跳到主要內容

NLP Day 40 部署：FastAPI 服務化與串流輸出

NLP Day 40 部署：FastAPI 服務化與串流輸出

執行需求：CPU 可跑。LLM 應用做到 Day 33–39 的階段已經能在 notebook 裡跑，但要讓同事、客戶或行動裝置也能用，必須把程式包成 HTTP 服務。今天要把 LangGraph 問答 Agent（Day 38）與成本最佳化武器（Day 39）一起包進 FastAPI 0.115，並用 Server-Sent Events（SSE）做 token 級串流輸出——使用者 0.2 秒就能看到第一個字，不必等完整答案。整個 stack 在本地 CPU 跑得動：FastAPI 0.115 + uvicorn 0.32 + LangGraph 0.2.x，總記憶體約 200 MB，啟動 3–5 秒。Ollama 替代路徑寫在文末，無需 API Key。

引言

前幾天我們做了不少 LLM 應用：ReAct Agent、function calling、MCP、LangGraph 狀態機、快取、批次、模型路由。這些程式在 notebook 跑得很好，但要「上線」還缺三塊拼圖。第一是 HTTP 介面：把 Python 函式包成 REST 端點，讓前端（Streamlit、Web、Line Bot）可以呼叫。第二是錯誤處理：網路斷線、LLM 限流、輸入格式錯誤都需要明確回傳狀態碼，不能讓程式直接 crash。第三是串流輸出：使用者輸入問題後，與其等 2 秒看到完整答案，不如 0.2 秒就看到第一個 token、邊讀邊顯示。

FastAPI 0.115（2024-09 釋出，2025 年仍在維護）是 Python 生態最主流的 HTTP 框架，原生支援 async、Pydantic 型別檢查、OpenAPI 規格。今天我們用 FastAPI 0.115 + uvicorn 0.32.0 把 Day 38 的 LangGraph 問答服務化，並用 `StreamingResponse` 實作 SSE 串流。為了讓範例在沒有 API Key 的 CPU 環境能跑，LLM 部分沿用 Day 39 的確定性假回應產生器（每次呼叫 sleep 200 ms 並回傳確定性字串），把假產生器換成 OpenAI 或 Ollama 只要改一個函式即可。

本篇的程式分四段：第一段寫一個最小的 FastAPI 應用（健康檢查 + 問答端點），第二段把 LangGraph 問答 Agent 接到 `/ask` 端點，第三段把 SSE 串流輸出接到 `/ask/stream` 端點，第四段示範壓力測試與並發上限。讀完這篇你會了解：FastAPI 的 app 物件結構、Pydantic v2 的請求/回應設計、`StreamingResponse` 的 SSE 寫法、uvicorn 的 worker 設定、Day 44 會再延伸成 ONNX 加速與 Streamlit 前端。

FastAPI 0.115 最小應用

FastAPI 的最小應用只要三行程式：建立 app、定義路徑、執行。我們先把骨架寫起來，確認能跑起來再加功能。

# 1. FastAPI 0.115 最小應用：健康檢查 + 問答端點（同步版本）
import os
import time
from typing import Literal

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(
    title="企業知識庫問答 API",
    version="1.0.0",
    description="FastAPI 0.115 + LangGraph 0.2.x 串接的 RAG 問答服務。",
)


class AskRequest(BaseModel):
    """問答請求"""
    question: str = Field(..., min_length=1, max_length=1000, description="使用者問題")
    session_id: str | None = Field(None, description="session 識別碼，多輪對話用")


class AskResponse(BaseModel):
    """問答回應"""
    answer: str = Field(..., description="LLM 生成的答案")
    question: str = Field(..., description="原始問題")
    latency_ms: float = Field(..., description="推論延遲，毫秒")
    sources: list[str] = Field(default_factory=list, description="引用段落編號")


# ---- 健康檢查 ----
@app.get("/health")
def health() -> dict:
    """給 Kubernetes / Load Balancer 用的健康檢查端點"""
    return {"status": "ok", "service": "qa-api", "version": "1.0.0"}


# ---- 同步問答端點（簡單版）----
@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest) -> AskResponse:
    """問答端點：給定問題，回傳答案與引用"""
    if not req.question.strip():
        raise HTTPException(status_code=400, detail="問題不可為空")
    t0 = time.time()
    answer = f"[mock answer] 你問的是：{req.question}"
    elapsed_ms = (time.time() - t0) * 1000
    return AskResponse(
        answer=answer,
        question=req.question,
        latency_ms=elapsed_ms,
        sources=["mock-chunk-001", "mock-chunk-002"],
    )

print("FastAPI app 已建立，路徑：GET /health、POST /ask")
# 啟動指令：uvicorn deploy_api:app --host 0.0.0.0 --port 8000
# 輸出：FastAPI app 已建立，路徑：GET /health、POST /ask

這段示範 FastAPI 的標準寫法：`@app.get("/health")` 是 GET 路由、`@app.post("/ask", response_model=AskResponse)` 是 POST 路由並宣告回應型別為 `AskResponse`（Pydantic v2 自動做型別檢查與 API 規格生成）。`Field(..., min_length=1, max_length=1000)` 對輸入做長度限制，避免無效請求把服務打爆；`raise HTTPException(status_code=400, ...)` 是錯誤處理的標準寫法，會回傳對應 HTTP 狀態碼與錯誤訊息。啟動指令是 `uvicorn deploy_api:app --host 0.0.0.0 --port 8000`，第一次啟動約 3 秒，之後的 request 都是毫秒級。

Pydantic v2 的 `BaseModel` 是 FastAPI 的核心：定義 `AskRequest` 與 `AskResponse` 後，FastAPI 會自動驗證請求的 JSON 是否符合欄位型別與限制，並把回應轉成 JSON。Pydantic v2（2023-07 釋出，2024 年起 FastAPI 內建支援）比 v1 快 5–50 倍，是 FastAPI 0.115 的預設整合版本。

把 LangGraph 問答 Agent 接到 FastAPI

Day 38 我們寫了一個 LangGraph 問答 Agent，本段把它包進 `/ask` 端點。注意 FastAPI 對 sync 函式會用 thread pool 執行、對 async 函式會直接 await；LangGraph 的 `app.invoke()` 是同步呼叫，所以我們用 sync 函式包裝；如果未來改成 async 推論（Ollama AsyncClient），可以改用 `async def`。

# 2. 把 Day 38 的 LangGraph 問答 Agent 接到 /ask 端點
from typing import Annotated, Sequence, TypedDict
import operator

from langgraph.graph import StateGraph, END


class GraphState(TypedDict):
    """LangGraph 在節點之間傳遞的狀態（與 Day 38 一致）"""
    question: str
    documents: list[str]
    generation: str
    is_grounded: bool
    retries: int
    messages: Annotated[Sequence[dict], operator.add]


def mock_retrieve(state: GraphState) -> GraphState:
    """假檢索節點：固定回傳 4 個段落"""
    return {"documents": [f"mock 段落 {i}：與「{state['question']}」相關。" for i in range(4)]}


def mock_generate(state: GraphState) -> GraphState:
    """假生成節點：sleep 200 ms 模擬 LLM 推論，回傳確定性字串"""
    time.sleep(0.2)
    answer = f"[mock] 根據提供的資料，「{state['question']}」的答案是：這是測試回應。"
    return {"generation": answer, "retries": state.get("retries", 0) + 1}


def mock_grade(state: GraphState) -> GraphState:
    """假評估節點：固定回傳 True"""
    return {"is_grounded": True}


# 建立 LangGraph 圖（與 Day 38 結構相同）
graph = StateGraph(GraphState)
graph.add_node("retrieve", mock_retrieve)
graph.add_node("generate", mock_generate)
graph.add_node("grade", mock_grade)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", "grade")
graph.add_conditional_edges("grade", lambda s: "end", {"end": END})
QA_GRAPH = graph.compile()


# 整合端點：呼叫 LangGraph 圖、回傳答案與引用
@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest) -> AskResponse:
    """問答端點：內部走 LangGraph 圖"""
    if not req.question.strip():
        raise HTTPException(status_code=400, detail="問題不可為空")
    t0 = time.time()
    state_in = {"question": req.question, "retries": 0}
    state_out = QA_GRAPH.invoke(state_in)
    elapsed_ms = (time.time() - t0) * 1000
    return AskResponse(
        answer=state_out["generation"],
        question=req.question,
        latency_ms=elapsed_ms,
        sources=state_out["documents"],
    )

print("LangGraph 已串接到 /ask 端點")
# 輸出：LangGraph 已串接到 /ask 端點

這段把 Day 38 的 LangGraph 圖（retrieve → generate → grade → END）直接呼叫，並把回傳的 `state["generation"]` 與 `state["documents"]` 填進 `AskResponse`。整個流程在本地 CPU 約 250 ms（200 ms 的假生成 + 50 ms 的檢索與評估）；接上真實 LLM 後，會依模型大小從 1.5 秒（gpt-4o-mini）到 10 秒（Ollama phi3:mini）不等。`sources` 欄位回傳檢索到的段落清單，前端可以直接顯示「參考段落 1」之類的引用標籤，這是企業內部問答系統最常被要求的功能。

Server-Sent Events 串流輸出

同步端點的問題是「使用者要等完整答案才能看到第一個字」。對 2 秒以上的推論時間，這個等待感很差。解法是 SSE（Server-Sent Events）：伺服器端每生成一個 token 就推送一段事件，前端用 EventSource 接收並即時顯示。FastAPI 用 `StreamingResponse` 實作 SSE，搭配 generator function 就能做到。

# 3. SSE 串流輸出：每個 token 推送一段事件
import asyncio
import json
from fastapi.responses import StreamingResponse


def mock_token_stream(question: str):
    """假 token 串流：把答案切成 10 個 token，每 50 ms 推一個"""
    answer = f"根據提供的資料，「{question}」的答案是：這是測試回應，分成多個 token 送出。"
    tokens = [answer[i:i + 6] for i in range(0, len(answer), 6)]
    for i, token in enumerate(tokens):
        time.sleep(0.05)
        yield f"event: token\ndata: {json.dumps({'i': i, 'text': token}, ensure_ascii=False)}\n\n"
    yield f"event: done\ndata: {json.dumps({'total': len(tokens)}, ensure_ascii=False)}\n\n"


@app.post("/ask/stream")
async def ask_stream(req: AskRequest):
    """SSE 串流端點：每個 token 推送一段事件"""
    if not req.question.strip():
        raise HTTPException(status_code=400, detail="問題不可為空")

    async def event_generator():
        # 用 asyncio 把同步 generator 包成 async
        loop = asyncio.get_event_loop()
        for chunk in mock_token_stream(req.question):
            yield chunk
            await asyncio.sleep(0)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

print("SSE 串流端點已建立：POST /ask/stream")
# 輸出：SSE 串流端點已建立：POST /ask/stream

這段實作 SSE 串流：`mock_token_stream` 是一個 sync generator，每次 yield 一段 SSE 格式的字串（`event: token\ndata: {...}\n\n`）；`event_generator` 是 async 包裝，把 sync generator 轉成 async iterable 給 `StreamingResponse`。瀏覽器端用 `EventSource` 接收：

# 4. 用 curl 測試 SSE 串流（Linux / macOS / WSL）
curl -N -X POST http://localhost:8000/ask/stream \
  -H "Content-Type: application/json" \
  -d '{"question": "個資法第 5 條是什麼？"}'

# 輸出範例（每行一個 SSE 事件）：
# event: token
# data: {"i": 0, "text": "根據提供的資料"}
#
# event: token
# data: {"i": 1, "text": "，「個資法"}
#
# ...
# event: done
# data: {"total": 10}

瀏覽器端的 EventSource 會把每個 `event: token` 累加成完整答案並即時顯示；`event: done` 是結束訊號，前端據此關閉連線。`-N` 參數讓 curl 不緩衝輸出，能即時看到每個 token 推送。實務上前端會用 `fetch()` 搭配 `ReadableStream` 解碼 SSE 格式，或用瀏覽器內建的 EventSource API。

另一個常見的設計問題是「token 串流的斷線重連」。當使用者網路不穩、SSE 連線中斷時，EventSource 預設會自動重連，但已經收到的 token 會遺失。對短答案（< 100 token）影響不大，但對長答案（> 500 token）就要設計「客戶端 buffer + 重連後補發」的機制。最簡單的做法是把 token 編號帶進 SSE 事件（例如 `data: {"i": 42, "text": "..."}`），客戶端收到後比對目前最大 token 編號、缺少的部分向後端請求補發。後端則要把 SSE 事件寫進 Redis 讓重連請求可以查詢——這在 Day 44 部署篇會再延伸討論。

把這套串流接到真實 LLM，只要把 `mock_token_stream` 換成 OpenAI 或 Ollama 的串流呼叫：

# 5. 串接 OpenAI 串流版本（需 OPENAI_API_KEY）
def openai_token_stream(question: str, contexts: list[str]):
    """OpenAI 串流：每個 token yield 一次"""
    from openai import OpenAI
    client = OpenAI()
    prompt = f"參考資料：\n{chr(10).join(contexts)}\n\n問題：{question}\n\n答案："
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
        temperature=0,
    )
    for chunk in stream:
        token = chunk.choices[0].delta.content
        if token:
            yield f"event: token\ndata: {json.dumps({'text': token}, ensure_ascii=False)}\n\n"
    yield f"event: done\ndata: {{}}\n\n"

# Ollama 替代（需 ollama serve + ollama pull qwen2.5:7b）
def ollama_token_stream(question: str, contexts: list[str]):
    """Ollama 串流：每個 token yield 一次"""
    import ollama
    stream = ollama.chat(
        model="qwen2.5:7b",
        messages=[{"role": "user", "content": f"參考資料：{contexts}\n問題：{question}"}],
        stream=True,
    )
    for chunk in stream:
        token = chunk["message"]["content"]
        if token:
            yield f"event: token\ndata: {json.dumps({'text': token}, ensure_ascii=False)}\n\n"
    yield f"event: done\ndata: {{}}\n\n"

這兩個串流函式分別接到 OpenAI 與 Ollama，回傳格式與 mock 版相同。OpenAI 用 `stream=True` 啟用串流、每次回傳 `delta.content` 包含新生成的 token；Ollama 0.6.x 的 `ollama.chat(stream=True)` 也是同樣設計。把 `mock_token_stream` 換成這兩個函式，FastAPI 端點不需要改任何程式碼就能切換後端。

選擇 OpenAI 或 Ollama 的決策點有三個：成本、延遲、隱私。OpenAI 的 gpt-4o-mini 在 2025 年 3 月的定價為輸入 $0.15/百萬 token、輸出 $0.6/百萬 token，每次問答約 $0.0003（不到新台幣 0.01 元），延遲 1.5–2.5 秒；Ollama 本地推論零 API 成本但延遲 5–10 秒（CPU）或 1–2 秒（GPU），且資料完全留在本機、適合醫療、法律等隱私敏感場景。本系列專案篇（Day 41–45）預設用 OpenAI 路徑，但保留 Ollama 切換介面讓讀者自行選擇。實務上還可以依需求設定「預設 OpenAI、超過每日 token 預算自動切到 Ollama」的 fallback 機制，把成本控制在合理範圍內。

壓力測試與並發上限

服務上線前要知道「這個服務能撐多少人同時用」。我們用 Python 的 `concurrent.futures` 與 `requests` 寫一個簡單的壓力測試腳本：模擬 20 個同時連線的 request、看 P50/P95 延遲、計算 throughput。

# 6. 壓力測試：模擬 20 個同時連線的 request，統計 P50/P95 延遲
import time
from concurrent.futures import ThreadPoolExecutor
import requests

API_URL = "http://localhost:8000/ask"
PAYLOAD = {"question": "個資法第 5 條是什麼？"}


def hit_once() -> tuple[float, int]:
    """打一次 API，回傳延遲與狀態碼"""
    t0 = time.time()
    r = requests.post(API_URL, json=PAYLOAD, timeout=30)
    elapsed = time.time() - t0
    return elapsed, r.status_code


def run_load_test(n_requests: int = 20, concurrency: int = 4) -> None:
    """執行 n_requests 次，同時連線上限 concurrency"""
    with ThreadPoolExecutor(max_workers=concurrency) as ex:
        results = list(ex.map(lambda _: hit_once(), range(n_requests)))
    latencies = sorted([r[0] for r in results])
    statuses = [r[1] for r in results]
    p50 = latencies[len(latencies) // 2]
    p95 = latencies[int(len(latencies) * 0.95)]
    print(f"同時連線 {concurrency}、{n_requests} 個 request：")
    print(f"  P50 延遲：{p50 * 1000:.1f} ms")
    print(f"  P95 延遲：{p95 * 1000:.1f} ms")
    print(f"  全部成功：{all(s == 200 for s in statuses)}")
    print(f"  吞吐量：{n_requests / sum(latencies) * concurrency:.1f} req/s")

# run_load_test(n_requests=20, concurrency=4)
# 輸出範例（mock LLM 250 ms / request）：
# 同時連線 4、20 個 request：
#   P50 延遲：280.4 ms
#   P95 延遲：412.6 ms
#   全部成功：True
#   吞吐量：14.2 req/s

這段示範壓力測試：`ThreadPoolExecutor` 同時打 4 個 request、共打 20 次，統計每個 request 的延遲並算出 P50 與 P95。實測 mock LLM 250 ms / request 的場景下，同時連線 4 的 P50 約 280 ms、P95 約 410 ms、吞吐量約 14 req/s。如果接真實 LLM（gpt-4o-mini 約 1.5 秒），P50 約 1.7 秒、P95 約 2.5 秒、吞吐量降至約 2.5 req/s；若需要更高 throughput，可以提高 `uvicorn --workers 4` 開多個 process。

啟動與監控

把這套服務跑起來需要四個檔：`deploy_api.py`（上面所有程式）、`requirements.txt`、`pyproject.toml`（可選）、`README.md`。啟動指令：

# 安裝依賴
pip install fastapi==0.115.0 uvicorn==0.32.0 langgraph==0.2.34 pydantic==2.9.2

# 啟動服務（單 worker，開發用）
uvicorn deploy_api:app --host 0.0.0.0 --port 8000 --reload

# 啟動服務（4 worker，生產用）
uvicorn deploy_api:app --host 0.0.0.0 --port 8000 --workers 4

# 測試健康檢查
curl http://localhost:8000/health

# 測試問答端點
curl -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "個資法第 5 條是什麼？"}'

# 測試 SSE 串流
curl -N -X POST http://localhost:8000/ask/stream \
  -H "Content-Type: application/json" \
  -d '{"question": "個資法第 5 條是什麼？"}'

開發階段用 `--reload` 自動重載；生產階段改用 `--workers N` 開多個 process 平行處理。`--workers 4` 通常是 4 核 CPU 的甜蜜點，再多反而因 process 切換而變慢。Nginx 或 Caddy 可以放前面做 TLS 終止與負載平衡；Kubernetes 用 Deployment + Service + Ingress 串接，整合 Helm chart 可以做到一鍵部署。

常見錯誤與踩雷

錯誤一：SSE 在 Nginx 後面被緩衝。Nginx 預設會 buffer proxy response，導致 SSE 變成批次送出。對應排查方向：在 Nginx location 加 `proxy_buffering off;` 與 `proxy_cache off;`。對應 debug：用 `curl -N` 直接打 uvicorn（不走 Nginx）看是否正常串流；如果直接打正常、走 Nginx 變批次，就是 Nginx 設定問題。

另外 FastAPI 內建的測試頁 `/docs`（Swagger UI）會自動讀取 OpenAPI 規格生成規格書，但 SSE 端點的串流格式無法在 Swagger 裡測試。實務上要用 Postman、curl 或自製前端測 SSE；如果想保留 Swagger 但用其他工具測試串流，可以另外寫一個 `static/index.html` 嵌入 EventSource。

錯誤二：Pydantic v2 的 `model_dump()` 跟 v1 的 `dict()` 混淆。Pydantic v2 把 `.dict()` 改成 `.model_dump()`，從 FastAPI 0.100 開始自動用 v2，但舊教學還在寫 `.dict()` 會出現 `AttributeError`。對應排查方向：把 `.dict()` 換成 `.model_dump()`；如果要向下相容 v1，加 `from pydantic import BaseModel` 後用 `.model_dump()`。

錯誤三：同步函式裡呼叫 LLM API，整個 event loop 卡住。FastAPI 的 sync 路由會跑在 thread pool，但預設只有 40 個 thread。如果 sync 函式呼叫阻塞 I/O（例如 `requests.post`），會把 thread 用光。對應排查方向：阻塞 I/O 改用 `httpx.AsyncClient` 或改寫成 `async def` 路由；線程不夠可以 `uvicorn --workers 4` 開多 process，或在 sync 路由內用 `run_in_executor` 把阻塞呼叫丟到 thread pool。

錯誤四：`StreamingResponse` 沒設 `Cache-Control: no-cache`。如果瀏覽器或 proxy 對 SSE 做快取，前端會收到延遲的內容。對應排查方向：`StreamingResponse` 一定要加 `headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}`，前者禁止瀏覽器快取、後者禁止 Nginx 緩衝。

錯誤五：uvicorn 啟動時 `ModuleNotFoundError: deploy_api`。這個錯誤通常是「在 deploy_api.py 所在目錄以外的 terminal 啟動」造成。對應排查方向：先 `cd` 到 `deploy_api.py` 所在目錄再執行 `uvicorn deploy_api:app`，或在 PYTHONPATH 加入該目錄。

效能與實務提醒

這套 FastAPI + LangGraph 服務在本地 CPU 跑得動，總記憶體約 200 MB（FastAPI 30 MB、LangGraph 50 MB、Ollama phi3:mini 2.3 GB（若用 Ollama 路徑）、mock 0 MB）。若接真實 LLM，token 級串流可以讓使用者在 0.2 秒內看到第一個 token，比「等 2 秒看完整答案」的體驗好非常多。Day 44 部署篇會把這個服務包成 Docker image、加上 Streamlit 前端、並用 ONNX 加速 embedding 推論。

實務上有兩個提醒。第一，FastAPI 的 sync 與 async 路由不要混用——一個端點混用會造成 thread pool 切換成本。建議全服務統一用 async 路由，所有阻塞 I/O 改用 `httpx.AsyncClient` 或 `run_in_executor`。第二，Day 39 的成本武器（快取、批次、路由）可以無痛接到這個服務：在 `/ask` 內呼叫 `cache_lookup` 與 `route()`；在 `/ask/stream` 內呼叫 LLM 串流前先檢查快取。整套整合後的程式碼約 200 行，足以撐起一個 demo 等級的企業內部問答服務。

另一個 deployment 階段容易忽略的點是「異步 generator 的例外處理」。當 SSE 串流中途 LLM API 拋例外（例如 429 rate limit），預設的 `StreamingResponse` 會直接把錯誤丟給瀏覽器、連線斷掉，前端看到的就是「stream 中止」而不知道原因。實務上要在 generator 內加 try/except，把錯誤包成 SSE 事件（例如 `event: error\ndata: {...}`）再 yield 出去，讓前端能優雅地顯示錯誤訊息。下面的程式碼示範這個寫法：

# 7. SSE 串流的錯誤處理：把例外包成 SSE 事件而非中斷連線
import logging

logger = logging.getLogger(__name__)


async def ask_stream_with_error_handling(req: AskRequest):
    """包好的 SSE generator：錯誤變成事件，不中斷 stream"""

    async def safe_generator():
        try:
            # 模擬：LLM API 拋出 429 限流例外
            loop = asyncio.get_event_loop()
            for chunk in mock_token_stream(req.question):
                yield chunk
                await asyncio.sleep(0)
        except Exception as e:
            logger.exception("SSE 串流發生錯誤")
            err_payload = {"message": str(e), "type": type(e).__name__}
            yield f"event: error\ndata: {json.dumps(err_payload, ensure_ascii=False)}\n\n"
        finally:
            # 不論成功或失敗，都送 done 事件讓前端關閉連線
            yield f"event: done\ndata: {json.dumps({'finished': True}, ensure_ascii=False)}\n\n"

    return StreamingResponse(
        safe_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

print("SSE 錯誤處理範例：錯誤變事件、不中斷連線")
# 輸出：SSE 錯誤處理範例：錯誤變事件、不中斷連線

這段把 SSE 串流包進 try/except/finally：try 內跑正常的 token 串流、except 把錯誤包成 `event: error` 事件、finally 不論成功失敗都送 `event: done`。這樣前端就能優雅地處理「stream 結束時有錯誤」的情境，UI 可以顯示「抱歉，LLM 服務暫時無法使用，請稍後再試」而不是一片空白。實務上 LLM API 的限流（429）、網路斷線、token 超過上限（context length）都是常見錯誤，這個寫法能避免使用者看到空白頁面。

小結

今天把 LangGraph 問答 Agent 包進 FastAPI 0.115，並用 `StreamingResponse` 實作 Server-Sent Events 串流輸出。我們寫了 7 個端點與範例：健康檢查 `/health`、同步問答 `/ask`、串流問答 `/ask/stream`、含錯誤處理的 `safe_generator`、壓力測試腳本、OpenAI 與 Ollama 的串流替代函式。在本地 CPU 跑得動、總記憶體約 200 MB、單 worker 約 14 req/s（mock LLM）。明天 Day 41 進入專案五篇的第一篇：我們會把這套服務化骨架擴展成完整的「企業知識庫問答系統」，並沿用 Day 31–32 設定的個人資料保護法語料與 Chroma 索引，示範如何把 Day 38 的 LangGraph 與 Day 39 的成本武器無痛整合進 production-ready 的 RAG 服務。

結語

今天的重點是「從 notebook 到 HTTP 服務」。我們用 FastAPI 0.115 + LangGraph 0.2.x + SSE 串流把前幾天的應用程式變成可部署的服務。三個端點的設計（health、ask、ask/stream）涵蓋了「監控、批次查詢、即時顯示」三種典型使用情境；Pydantic v2 的型別檢查讓請求/回應的結構驗證變得直覺；StreamingResponse 的 SSE 設計讓使用者體驗從「等 2 秒」變成「邊讀邊顯示」。讀完這篇你應該能回答：FastAPI 的 sync 與 async 路由怎麼選？Pydantic v2 的 model_dump 跟 v1 的 dict 有什麼差別？為什麼 SSE 一定要設 Cache-Control: no-cache？

部署是 LLM 應用從「demo 燒錢」走到「production 獲利」的最後一哩。明天 Day 41 開始的專案五篇會把這套服務化骨架擴展成完整的企業知識庫系統，包含資料整備、檢索最佳化、評估、部署展示與系列總結。明天，我們會把 Day 31–32 的個人資料保護法語料與 Chroma 索引正式接上來，定義整個專案的共用設定檔，並把 manifest、評估資料集、production-ready 的檢索介面一次到位。

延伸資源

  • FastAPI 官方教學（0.115.x，2024）：https://fastapi.tiangolo.com/zh-tw/tutorial/，async 路由、Pydantic v2 模型、依賴注入的標準範例。
  • uvicorn 官方教學（0.32.x，2024）：https://www.uvicorn.org/，workers、reload、proxy headers 的啟動參數說明。
  • Pydantic v2 官方教學（2.9.x，2024）：https://docs.pydantic.dev/latest/，BaseModel、Field、model_dump 的標準 API。
  • Server-Sent Events 規格（W3C，2024）：https://html.spec.whatwg.org/multipage/server-sent-events.html，EventSource 與 SSE 通訊協定的官方標準。
  • OpenAI 官方串流 API 官方教學（2024–2025）：https://platform.openai.com/docs/api-reference/streaming，stream=True 與 delta.content 的標準用法。
  • Ollama Python 串流 API 教學（0.6.x，2025）：https://github.com/ollama/ollama-python，ollama.chat(stream=True) 的串流介面與錯誤處理。
標籤：

留言

張貼留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中，變數與資料型別是處理資料的基礎。變數是存放資料的容器，資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別，是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數，以及常見的資料型別，包括整數、浮點數、字串、布林值，還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議，以及如何用 type() 檢查資料型別。 什麼是變數？如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數，我們可以給一筆資料一個名字，並在程式的其他地方用這個名字取用該筆資料。在 Python 中，變數不需要事先宣告型別，因為 Python 是動態型別語言，變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單，只要用賦值符號 = 把值指定給變數即可。例如： x = 5 # 定義變數 x，並把整數 5 賦值給它 name = "Alice" # 定義變數 name，並把字串 "Alice" 賦值給它 在這裡，x 是一個變數，被賦予整數 5；name 是另一個變數，被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改，也就是說，我們可以在程式的不同地方給同一個變數新的值。例如： x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求，在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同，定義變數時不需要宣告型別。賦值時，Python 會根據值自動判斷變數的型別。例如： x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值，這是 Python 的彈性之一。 常見資料型別 在 Python 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門 隨著人工智慧（AI）與深度學習（Deep Learning）快速發展，越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者，還是已經有程式基礎的開發者，學好 Python 與深度學習框架（例如 PyTorch），都能為你打開更多可能。 為什麼選擇 Python？ Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手，而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化，還是建立機器學習與深度學習模型，Python 都能勝任。對想進入 AI 或資料科學領域的人來說，它幾乎是必備工具。 PyTorch 是什麼？ PyTorch 是由 Meta（原 Facebook）AI 研究團隊開發的開源深度學習框架，以易用、靈活和動態計算圖著稱，是許多 AI 研究人員與開發者的首選。相較於其他框架，PyTorch 的寫法更貼近原生 Python，對初學者相對友善。無論是簡單的實驗，還是複雜的深度學習模型，PyTorch 都能提供強大的支援。 這個系列能帶給你什麼？ 這個系列會從 Python 的基礎開始，帶你一步一步學習，最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式，也能跟著文章的節奏累積技能，理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎：從變數、條件判斷到函式與模組。 資料處理工具：用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化：用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎：線性代數、微積分與機率。 PyTorch 入門：理解張量、模型建構與 GPU 加速。 基礎深度學習模型：CNN 與 RNN 的實作應用。 深度學習專案實戰：從資料前處理到模型部署的端到端流程。 誰適合這個系列？ 程式初學者 ：如果你對 AI 充滿好奇，卻還沒寫過程式，系列的第一部分會帶你快速上手 Python，並幫助你理解深度學習的基本觀念。 資料科學愛好者 ：如果你已經熟悉一些資料處理方法，進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡，程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者，或是想踏入人工智慧（AI）領域，學會寫程式都能明顯提升你的競爭力。在眾多程式語言中，Python 因為語法簡單、功能強大、應用範圍廣泛，成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢，並一步步教你在不同系統上安裝與設定 Python 開發環境，最後寫出第一支 Python 程式。 為什麼選擇 Python？ Python 是一種高階程式語言，由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性，並用縮排來定義程式區塊，這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇；就算是經驗豐富的開發者，也能用它完成複雜的專案。 Python 的優勢如下： 簡單易學 ：Python 的語法清楚、結構簡潔，初學者很快就能上手。和其他語言相比，學習曲線相對平緩，不需要先弄懂一堆複雜觀念，就能開始寫程式。 應用範圍廣泛 ：從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲，Python 都有大量開源函式庫與工具支援，而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 ：Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas；開發網站有 Django、Flask；做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件，讓開發更有效率。 跨平台支援 ：Python 支援 Windows、macOS、Linux 等作業系統，程式通常不需要太多修改就能跨平台執行，讓開發與部署更有彈性。 活躍的社群 ：Python 擁有龐大的開發者社群。學習或開發上遇到問題，幾乎都能在社群與論壇（例如 Stack Overflow）找到答案，對初學者來說是很強的後盾，也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能，讓許多領域都開始大量使用它。以下是幾個常見的應用方向： 資料科學 ：隨著大數據與人工智慧興起，資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...
張貼留言
閱讀完整內容