使用一個 API Key 打造 AI Chatbot：從零到正式上線只需 30 分鐘

本教學將使用 FastAPI、SSE 串流、對話記憶和模型切換功能，構建一個小型但具備生產就緒能力的聊天機器人服務。我們的目標不是一個玩具般的演示，而是一個你可以直接部署在真實產品介面後端，並能安全迭代的系統，無需在每次更換模型時重寫整合程式碼。

如果你已經將某個 OpenAI 相容的 SDK 指向 TokenLab，那麼本文將從該處繼續。如果你尚未進行 base URL 的替換，請先閱讀遷移指南。如果你主要關心的是負載下的請求整形（request shaping）與退避機制（backoff），請將本指南與 AI API 速率限制指南結合使用。

重點摘要

• 一個生產就緒的聊天機器人需要六個部分：同步端點、串流端點、伺服器端對話狀態、模型白名單、真實的錯誤處理，以及清晰的儲存升級路徑。
• 在加入串流、記憶或工具呼叫之前，先用一個小型聊天端點驗證你的 Key、base URL 和路由。
• SSE 串流足以涵蓋大多數聊天產品，且比 WebSocket 具有更低的營運開銷。
• 透過後端白名單來公開模型，而不是使用自由文字欄位，這樣前端就無法請求任意或已停用的模型 ID。
• 模型可用性和陣容經常變動。在將白名單鎖定至生產環境前，請務必查看 TokenLab 的模型目錄（觀察日期：2026-07-07）。

我們正在構建什麼

完成後的服務包含六個運作部分：

1. 用於冒煙測試的同步 /chat 端點。
2. 用於真實 UI 的串流 /chat/stream 端點。
3. 以 conversation_id 為鍵值的對話狀態。
4. 模型白名單，防止前端請求任意 ID。
5. 不會在第一次遇到 429 錯誤就崩潰的錯誤處理機制。
6. 從記憶體原型遷移到 Redis 或 PostgreSQL 的清晰路徑。

這足以支撐一個客服機器人、內部助手或嵌入式聊天小工具的第一個版本。

安裝最小技術堆疊

pip install fastapi uvicorn openai pydantic redis

你可以先跳過 redis，但現在就寫入匯入語句，可以讓後續的升級變成一個簡單的配置變更，而不是重構。

第一步：從一個小型、簡單的聊天端點開始

在聊天機器人開發中，最容易迷失方向的做法就是在基本請求路徑穩定之前，就開始處理 WebSocket、工具使用和代理編排。請先從一個能證明你的 Key、base URL 和模型路由運作正常的小型端點開始。

from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()

client = OpenAI(
    api_key="sk-tokenlab-xxx",
    base_url="https://api.tokenlab.sh/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "deepseek-v4-flash"
    conversation_id: str | None = None

@app.post("/chat")
async def chat(req: ChatRequest):
    response = client.chat.completions.create(
        model=req.model,
        messages=[{"role": "user", "content": req.message}]
    )
    return {"reply": response.choices[0].message.content}

執行一次冒煙測試。如果這都失敗了，請不要繼續在上面進行開發。

第二步：加入串流功能，因為使用者在測量延遲前就能感受到它

大多數聊天機器人產品感覺緩慢，並非因為模型慢，而是因為 UI 在完整回應抵達前一直保持空白。對於大多數聊天產品而言，SSE 已足夠，且比 WebSocket 負擔更輕。

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    def generate():
        stream = client.chat.completions.create(
            model=req.model,
            messages=[{"role": "user", "content": req.message}],
            stream=True
        )
        for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content:
                yield f"data: {delta.content}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

在前端，一個簡單的 fetch reader 就足夠了：

async function sendMessage(payload) {
  const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(payload),
  });

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

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value, { stream: true });
    console.log(chunk);
  }
}

如果你的產品已經在標準 HTTP 上運行瀏覽器客戶端，SSE 會讓架構比 WebSocket 層更簡單。

第三步：將對話狀態移出請求主體

第一個聊天機器人演示通常會將完整對話記錄保留在瀏覽器中，並在每次對話時重新發送。這對於原型來說沒問題，但一旦你需要重試、可恢復的會話或伺服器端工具時，情況就會變得混亂。

使用記憶體儲存作為開始是可以的：

from collections import defaultdict
import uuid

conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."

def build_messages(conv_id: str, user_msg: str) -> list:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    history = conversations[conv_id][-20:]
    messages.extend(history)
    messages.append({"role": "user", "content": user_msg})
    conversations[conv_id].append({"role": "user", "content": user_msg})
    return messages

升級到 Redis 的路徑主要是儲存層的調整，而非邏輯變更：

import json
import redis

redis_client = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

def load_history(conv_id: str) -> list:
    raw = redis_client.get(f"chat:{conv_id}")
    return json.loads(raw) if raw else []

def save_history(conv_id: str, history: list) -> None:
    redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))

當對話需要 TTL、可恢復性或多實例部署時，請選擇 Redis。當對話記錄本身就是你需要查詢、審計或報告的產品資料時，請選擇 PostgreSQL。

第四步：將錯誤視為產品行為，而不僅僅是異常

如果你的聊天機器人是面向客戶的，失敗路徑與成功路徑同樣重要。使用者不在乎失敗是來自速率限制、餘額耗盡還是上游模型中斷，他們只在乎 UI 是否凍結。

from openai import APIConnectionError, APIError, RateLimitError

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)

    def generate():
        full_response = []
        try:
            stream = client.chat.completions.create(
                model=req.model,
                messages=messages,
                stream=True
            )
            for chunk in stream:
                delta = chunk.choices[0].delta
                if delta.content:
                    full_response.append(delta.content)
                    yield f"data: {delta.content}\n\n"
        except RateLimitError:
            yield "data: [error: rate limited, please retry shortly]\n\n"
        except APIConnectionError:
            yield "data: [error: connection issue, please retry]\n\n"
        except APIError:
            yield "data: [error: something went wrong on our end]\n\n"
        finally:
            if full_response:
                conversations[conv_id].append(
                    {"role": "assistant", "content": "".join(full_response)}
                )
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

第五步：鎖定前端可請求的模型

永遠不要讓前端將任意模型字串直接傳遞給 API。自由文字欄位會引發對已停用模型、拼字錯誤或你從未打算公開的模型的請求。請改為透過後端白名單進行路由。

ALLOWED_MODELS = {
    "default": "deepseek-v4-flash",
    "flagship": "gpt-5.5",
    "balanced": "claude-sonnet-5",
}

class ChatRequest(BaseModel):
    message: str
    tier: str = "default"
    conversation_id: str | None = None

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    model = ALLOWED_MODELS.get(req.tier, ALLOWED_MODELS["default"])
    # 其餘的串流邏輯使用 `model` 而不是客戶端提供的原始字串

這為你提供了一個統一的地方來更換模型，當供應商棄用某個模型時，無需觸動前端程式碼或發布客戶端更新。

第六步：處理生產環境的其餘部分，而不僅僅是成功路徑

當周邊邊緣情況得到處理時，聊天機器人後端才被視為生產就緒，而不是當核心聊天呼叫變得聰明時。

檢查清單很短：

• 添加請求 ID，以便將前端失敗與後端日誌連接起來
• 限制每個使用者的併發數和請求大小
• 在長對話耗盡你的 Token 預算前進行修剪
• 記錄模型、延遲、輸入大小和完成原因
• 將使用者可見的錯誤訊息與內部錯誤細節分開
• 測試一個替代模型，確保在第一次真正中斷前，後備機制是有效的

歷史記錄修剪可以保持簡單：

def trim_history(messages: list, max_tokens: int = 8000) -> list:
    system = messages[0]
    history = messages[1:]
    total_chars = len(system["content"])
    trimmed = []

    for msg in reversed(history):
        msg_chars = len(msg["content"])
        if total_chars + msg_chars > max_tokens * 4:
            break
        trimmed.insert(0, msg)
        total_chars += msg_chars

    return [system] + trimmed

重點不在於精確的 Token 計算，而在於在上下文爆炸影響你的帳單或延遲之前阻止它。

從演示到產品

一旦這個後端穩定下來，下一次升級通常很少是「更多的 AI」，而通常是無聊的基礎設施：

• 身份驗證，確保一個使用者無法讀取另一個使用者的對話
• 持久化，確保會話在部署後依然存在
• 速率限制，防止一個吵鬧的使用者耗盡你的配額
• 如果聊天機器人是面向客戶的，則需要計費或使用量歸因
• 如果對話需要長期記憶，則需要後台總結

統一的閘道器有助於解決大部分問題。一旦完成了基礎 URL 遷移，模型變更就不再是平台重寫，而變成了配置編輯。

冒煙測試

uvicorn main:app --reload --port 8000

curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "model": "deepseek-v4-flash"}'

如果你能串流一次對話、保留一個會話，並在強制失敗時返回乾淨的錯誤，你就擁有了正確的基礎。

成本估算

在 TokenLab 建立一個 API Key，將你的 OpenAI SDK 指向 https://api.tokenlab.sh/v1，你就可以在無需跨供應商管理多個帳戶的情況下，發布你的第一個聊天機器人生產版本。

模型	典型層級	備註
DeepSeek V4 Flash	快速 / 預設	適合高流量、低延遲對話的良好預設
GPT-5.5	旗艦	用於需要更強推理能力的對話
Claude Sonnet 5	平衡	編碼和審閱風格回覆的強大選擇
Gemini 3.5 Flash	預算 / 快速替代	高流量路由的低成本、快速替代方案

確切的每 Token 定價在各供應商之間頻繁變動，此處不作為固定數字列出。在編列預算前，請查看模型目錄（觀察日期：2026-07-07）上的當前費率。實際上，將大多數對話路由到快速/預設層級（如 DeepSeek V4 Flash 或 Gemini 3.5 Flash），並為需要它的對話保留 GPT-5.5 或 Claude Sonnet 5，可以讓大多數應用程式保持較低的每月帳單，但在承諾預算前，請確認你帳戶實際的每百萬 Token 費率。

常見問題

我需要 WebSocket 來構建 AI 聊天機器人嗎？ 不需要。第二步中展示的 SSE 串流涵蓋了絕大多數聊天產品。當你需要請求/回應之外的雙向推送（如即時協作或伺服器發起的事件）時，WebSocket 才有真正的價值。對於標準聊天 UI，SSE 更易於部署、除錯和擴展。

我該如何決定預設模型？ 從快速、低成本的模型（如 DeepSeek V4 Flash 或 Gemini 3.5 Flash）作為預設層級開始，並在第五步展示的白名單後方加入 Claude Sonnet 5 或 GPT-5.5 作為平衡或推理層級。請查看模型目錄（觀察日期：2026-07-07）以獲取當前選項，因為新模型會發布，舊模型會按你無法控制的時間表被棄用。

當聊天機器人從演示轉向真實流量時，什麼最先崩潰？ 幾乎總是錯誤路徑，而不是成功路徑。無限制的重試、缺失的每使用者併發限制以及無限制的對話歷史，是聊天機器人後端在真實負載下崩潰的三個最常見原因。上述第四步和第六步直接解決了這三個問題。

---

開始使用你的 API Key：tokenlab.sh 透過單一端點提供 300 多個模型。提供 $1 免費額度供你開始構建。