你可以在一個下午內構建出一個能夠串流回覆的 AI 聊天機器人。但在遇到第一次 429 錯誤後還能讓它保持運作,才是真正的挑戰。我們將構建一個小型 FastAPI 服務,具備 SSE 串流、伺服器端對話記憶以及模型白名單功能,所有這些都由一個 TokenLab Key 驅動。我們的目標是打造一個可以放在真實產品介面後端的服務,並在不重寫整合程式碼的情況下更換模型。如果你尚未執行 Base URL 切換,請先閱讀遷移指南。如果你的主要考量是負載下的重試機制(backoff),請將本指南與AI API 速率限制指南搭配閱讀。
重點摘要
- 一個生產就緒的聊天機器人需要六個部分:同步端點、串流端點、伺服器端對話狀態、模型白名單、真實的錯誤處理以及儲存升級路徑。
- 在每個程序中從
TOKENLAB_API_KEY載入 Key,並將其保留在伺服器端。身份驗證文件(觀察於 2026-10-03)指出,切勿在客戶端程式碼中暴露 Key 或將其提交到版本控制中。 GET /v1/models是公開的,因此成功的響應並不能證明你的 Key 有效。請透過一次真實的聊天請求來驗證(快速入門,觀察於 2026-10-03)。- SSE 適用於大多數聊天產品。請將輸出後中斷的串流視為不完整,而非已完成的回答。
- 本指南中的白名單將層級映射至
deepseek-v4-flash、gpt-5.5、claude-sonnet-5和gemini-3.5-flash。每個模型都將openai_chat_completions列為可接受的格式(觀察於 2026-10-03)。 - 根據下方的價格估算,一輪 2,000 個輸入 Token 和 500 個輸出 Token 的成本約為 $0.0006 到 $0.0075,具體取決於模型。
構建 AI 聊天機器人:設定與核心服務
完成後的服務包含六個運作部分:
- 用於冒煙測試的同步
/chat端點。 - 用於真實 UI 的串流
/chat/stream端點。 - 以
conversation_id為鍵的對話狀態。 - 模型白名單,防止前端請求任意 ID。
- 不會在第一次 429 錯誤時崩潰的錯誤處理機制。
- 從記憶體原型遷移至 Redis 或 PostgreSQL 的路徑。
這對於支援機器人、內部助理或嵌入式聊天小工具的第一個版本來說已經足夠。下方的每個程式碼區塊都將按順序附加到 main.py 檔案中。
步驟 1:設定 Key 並安裝堆疊
在你的工作區中透過 API Keys 建立一個 Key。完整數值僅顯示一次,且 Key 列表中的前綴無法驗證請求。所有 TokenLab Key 皆以 sk- 開頭。同時請在 帳單 (Billing) 中檢查你的餘額,因為第一次請求將按模型的當前價格計費(快速入門,觀察於 2026-10-03)。
將 Key 儲存在環境變數中,絕不要寫入原始碼。下方的提示指令可避免其出現在命令列和 Shell 歷史記錄中,且僅適用於當前終端及其子程序。對於部署的應用程式,請使用密鑰管理服務。快速入門文件也提供了 PowerShell 版本。
printf 'TokenLab API key: '
read -r -s TOKENLAB_API_KEY
printf '\n'
export TOKENLAB_API_KEY
pip install fastapi uvicorn openai pydantic
你不需要 TokenLab 專用的 SDK。一旦你更改了 Key 和 Base URL,官方的 OpenAI SDK 即可運作(SDK 指南,觀察於 2026-10-03)。
步驟 2:定義 Schema、白名單與記憶體
我們從共享部分開始。請求 Schema 使用 tier 而非原始的 model 字串,本指南後續的所有 curl 範例都將傳送相同的欄位。
第一個聊天機器人範例通常會將對話記錄保存在瀏覽器中,並在每一輪重新傳送。這對於原型來說可行,但一旦你需要重試、可恢復的會話或伺服器端工具時,情況就會變得混亂。因此,我們從一開始就讓伺服器擁有對話歷史。
import json
import logging
import os
import time
import uuid
from collections import defaultdict
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from openai import APIConnectionError, APIStatusError, OpenAI, RateLimitError
from pydantic import BaseModel, Field
log = logging.getLogger("chatbot")
app = FastAPI()
client = OpenAI(
api_key=os.environ["TOKENLAB_API_KEY"],
base_url="https://api.tokenlab.sh/v1",
timeout=30.0,
max_retries=0,
)
ALLOWED_MODELS = {
"default": "deepseek-v4-flash",
"flagship": "gpt-5.5",
"balanced": "claude-sonnet-5",
"budget": "gemini-3.5-flash",
}
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
MAX_HISTORY = 20 # 每個對話保留的訊息數量(可自行設定)
MAX_REPLY_TOKENS = 1000 # 每輪輸出的上限(可自行設定)
class ChatRequest(BaseModel):
message: str = Field(min_length=1, max_length=4000)
tier: str = "default"
conversation_id: str | None = None
def resolve_model(tier: str) -> str:
model = ALLOWED_MODELS.get(tier)
if model is None:
raise HTTPException(status_code=400, detail=f"unknown tier: {tier}")
return model
conversations: dict[str, list[dict]] = defaultdict(list)
def build_messages(conv_id: str, user_msg: str) -> list[dict]:
history = conversations[conv_id][-MAX_HISTORY:]
return [
{"role": "system", "content": SYSTEM_PROMPT},
*history,
{"role": "user", "content": user_msg},
]
def remember(conv_id: str, user_msg: str, reply: str) -> None:
conversations[conv_id].append({"role": "user", "content": user_msg})
conversations[conv_id].append({"role": "assistant", "content": reply})
客戶端設定 max_retries=0 和 30 秒逾時,與快速入門一致。我們自行決定重試策略,因為在串流過程中盲目使用 SDK 重試可能會導致重複計費。
步驟 3:映射錯誤並新增同步端點
我們保留一個小型函式,將任何 SDK 例外轉換為狀態碼和使用者友善的訊息。詳細資訊會進入你的日誌,而非顯示給使用者。
def describe_error(exc: Exception) -> tuple[int, str]:
if isinstance(exc, RateLimitError): # 先檢查:它是 APIStatusError 的子類
return 429, "The assistant is busy. Please retry shortly."
if isinstance(exc, APIConnectionError):
return 502, "Connection problem. Please retry."
if isinstance(exc, APIStatusError):
log.error("upstream status=%s code=%s", exc.status_code, exc.code)
if exc.status_code in (401, 402, 403): # Key、餘額或權限問題:操作員問題
return 503, "The assistant is unavailable right now."
if exc.status_code in (400, 404): # 錯誤的欄位或模型 ID:我們的 Bug
return 500, "Something went wrong on our side."
return 502, "The assistant is unavailable right now."
return 500, "Something went wrong."
def retry_delay(exc: RateLimitError) -> float:
try:
return min(float(exc.response.headers.get("retry-after", "2")), 10.0)
except ValueError:
return 2.0
def create_with_one_retry(**kwargs):
try:
return client.chat.completions.create(**kwargs)
except RateLimitError as exc:
time.sleep(retry_delay(exc))
return client.chat.completions.create(**kwargs)
@app.post("/chat")
def chat(req: ChatRequest):
model = resolve_model(req.tier)
conv_id = req.conversation_id or str(uuid.uuid4())
messages = build_messages(conv_id, req.message)
try:
response = create_with_one_retry(
model=model, messages=messages, max_tokens=MAX_REPLY_TOKENS
)
except (APIConnectionError, APIStatusError) as exc:
status, message = describe_error(exc)
raise HTTPException(status_code=status, detail=message)
reply = response.choices[0].message.content or ""
log.info("model=%s usage=%s", model, response.usage)
remember(conv_id, req.message, reply)
return {"conversation_id": conv_id, "reply": reply}
在構建其他功能之前,請先執行一次請求。如果失敗,請勿繼續在其上構建。
串流回覆並保持 AI 聊天機器人的響應速度
大多數聊天機器人產品感覺緩慢並非因為模型慢,而是因為 UI 在完整回應到達前保持空白。我們加入 SSE,因為它涵蓋了大多數聊天產品,且比 Websocket 負擔更輕。如果你的產品已經透過純 HTTP 執行瀏覽器客戶端,SSE 能讓架構更簡單。
步驟 4:新增串流端點
串流文件(串流指南,觀察於 2026-10-03)保持 Chat Completions 串流對於已預期來自 /v1/chat/completions 的 SSE 區塊的框架有效。我們堅持該路徑,因為所有四個白名單模型都接受它。我們將每個事件編碼為 JSON,因此回覆內部的換行符號不會破壞框架。
def sse(payload: dict) -> str:
return f"data: {json.dumps(payload)}\n\n"
@app.post("/chat/stream")
def chat_stream(req: ChatRequest):
model = resolve_model(req.tier)
conv_id = req.conversation_id or str(uuid.uuid4())
messages = build_messages(conv_id, req.message)
def generate():
parts: list[str] = []
finish_reason = None
try:
with client.chat.completions.create(
model=model,
messages=messages,
max_tokens=MAX_REPLY_TOKENS,
stream=True,
stream_options={"include_usage": True},
) as stream:
for chunk in stream:
if chunk.usage:
log.info("model=%s usage=%s", model, chunk.usage)
if not chunk.choices:
continue
choice = chunk.choices[0]
if choice.delta.content:
parts.append(choice.delta.content)
yield sse({"delta": choice.delta.content})
if choice.finish_reason:
finish_reason = choice.finish_reason
except (APIConnectionError, APIStatusError) as exc:
_, message = describe_error(exc)
yield sse({"error": message})
return
if finish_reason != "stop":
yield sse({"error": f"Reply incomplete ({finish_reason})."})
return
remember(conv_id, req.message, "".join(parts))
yield sse({"done": True, "conversation_id": conv_id})
return StreamingResponse(generate(), media_type="text/event-stream")
這裡有兩個細節很重要。首先,include_usage: true 會將 Token 使用量加入串流區塊,而最終的使用量區塊沒有 choices,因此迴圈會跳過它。其次,我們僅在 finish_reason 為 stop 時才儲存該輪對話。length 完成原因表示 Token 上限截斷了回覆,文件將兩者都列為正常的結束條件。
由於錯誤發生在產生器內部,當失敗發生時,HTTP 狀態碼已經是 200。客戶端必須讀取 error 事件。with 區塊會在產生器退出時關閉上游串流。文件建議在使用者停止或斷開連線時取消上游請求,因此請測試你的 ASGI 伺服器是否確實會在斷開連線時停止產生器。
步驟 5:在瀏覽器中讀取串流
瀏覽器應呼叫你自己的已驗證後端,絕不要直接呼叫 TokenLab。請按事件邊界進行解析,因為一個網路區塊不一定是一個完整的 SSE 事件。
async function sendMessage(payload, onDelta) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(payload), // { message, tier, conversation_id }
});
if (!response.ok || !response.body) throw new Error(`HTTP ${response.status}`);
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const events = buffer.split('\n\n');
buffer = events.pop();
for (const raw of events) {
if (!raw.startsWith('data: ')) continue;
const event = JSON.parse(raw.slice(6));
if (event.error) throw new Error(event.error);
if (event.delta) onDelta(event.delta);
}
}
}
當你需要時將記憶體升級至 Redis
記憶體字典對於單一程序來說沒問題。當對話需要 TTL、可恢復性或多個實例時,請使用 Redis。當對話記錄是必須查詢、審計或報告的產品資料時,請使用 PostgreSQL。Redis 的變更屬於儲存層規劃,而非邏輯。執行 pip install redis,然後讓 build_messages 和 remember 透過這些輔助函式進行讀寫。
import redis
redis_client = redis.Redis(
host=os.environ.get("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))
為生產環境構建 AI 聊天機器人時處理錯誤
面對客戶的使用者不在乎失敗是來自速率限制、餘額不足還是上游中斷。他們在乎的是 UI 是否凍結。想像一個支援小工具的串流在四十個字後中斷。TokenLab 文件指出,在輸出後中斷的串流並非完整回應,重複請求可能會產生不同的輸出並導致第二次計費(處理 API 錯誤,觀察於 2026-10-03)。這就是為什麼步驟 4 從不儲存未完成的對話輪次。
步驟 6:應用重試規則
根據 HTTP 狀態碼和 code 處理錯誤,而非人類可讀的 message,因為後者可能會在未經通知的情況下變更。下表結合了文件與我們服務的做法。
| 狀態碼 | 含義(根據文件) | 本服務的做法 |
|---|---|---|
400 |
欄位、模型 ID 或輸入無效 | 回傳 500,記錄日誌,絕不重試 |
401 |
Key 缺失、無效、過期或已撤銷 | 回傳 503,警示操作員 |
402 |
餘額或 API-key 限制過低 (insufficient_balance, quota_exceeded) |
回傳 503,警示操作員 |
403 |
Key 無法使用此模型 | 回傳 503,檢查 Key 的模型列表 |
404 |
資源或模型未找到 | 回傳 500,對照 /v1/models 檢查 ID |
429 |
達到請求限制 | 等待 Retry-After,重試一次(僅限同步) |
500–504 |
服務不可用或網路故障 | 僅在 retryable 為 true 時重試;遵守 retry_after |
文件中有兩個警告。503 all_channels_failed 並不總是意味著暫時性中斷。如果 retryable 為 false,請勿重複相同的請求,並在選擇其他模型前檢查 GET /v1/models。此外,Chat Completions 和 Responses 使用 OpenAI 風格的 error 物件,但 Anthropic Messages 和 Gemini 保留各自的格式。我們的服務在所有白名單模型上呼叫 Chat Completions,因此一個解析器即可運作。如果你將某個層級遷移至原生格式,則需要第二個解析器。
將回應標頭中的 Request ID 與端點、模型、時間以及你自己的使用者或工作 ID 一起儲存。文件在我們使用的頁面上未命名該標頭,因此請在連接日誌前對照故障排除路徑進行確認。
其餘的生產清單很簡短:
- 新增你自己的 Request ID,以便將前端失敗與後端日誌連結。
- 限制每個使用者的併發數和請求大小。
message上的max_length=4000是一個你可以調整的起點。 - 像
MAX_HISTORY一樣按訊息數量限制歷史記錄。帳單文件指出 Token 化方式因模型、語言和媒體輸入而異,因此字元數公式並非可靠的計費估算。 - 在每一輪記錄模型、延遲、使用量和完成原因。
- 將使用者可見的訊息與內部錯誤細節分開。
- 向替代層級發送一次測試請求,確保在第一次真實中斷前備用機制運作正常。
選擇模型並估算 AI 聊天機器人的成本
絕不要讓前端將任意模型字串傳遞給 API。自由文字欄位會導致拼字錯誤、已退役的模型以及你從未打算暴露的模型。ALLOWED_MODELS 對映表讓你在供應商棄用模型時,無需更新客戶端即可更換模型。未知的模型 ID 會回傳 model_not_found,文件會引導你至 /v1/models 獲取當前 ID。在鎖定白名單前,請檢查模型目錄。
當前每 100 萬 Token 價格
這些是 TokenLab 以美元計算的價格,來自每個模型的即時 API 條目,觀察於 2026-10-03。定價最後更新於 2026-10-02T16:53:30.068Z。所有四個模型皆顯示生命週期階段為 active。
| 層級 | 模型 ID | 輸入 / 1M | 輸出 / 1M | 快取讀取 / 1M | 來源 |
|---|---|---|---|---|---|
| default | deepseek-v4-flash (離峰) |
$0.15 | $0.60 | $0.003 | 模型 API |
| default | deepseek-v4-flash (尖峰) |
$0.30 | $1.20 | $0.006 | 模型 API |
| flagship | gpt-5.5 |
$1.50 | $9.00 | $0.15 | 模型 API |
| balanced | claude-sonnet-5 |
$0.90 | $4.50 | $0.09 | 模型 API |
| budget | gemini-3.5-flash |
$0.75 | $4.50 | $0.075 | 模型 API |
對於 deepseek-v4-flash,條目列出了平日的尖峰時段,不含中國公眾假期。離峰時段包含週末和這些假期。claude-sonnet-5 也列出了每 100 萬 Token $1.125 的快取寫入價格。
一次成功的請求範例
假設一輪對話有 2,000 個輸入 Token 和 500 個輸出 Token。該大小是我們的假設,而非測量值。對於 gpt-5.5:
- 輸入:2,000 × $1.50 / 1,000,000 = $0.0030
- 輸出:500 × $9.00 / 1,000,000 = $0.0045
- 總計:每輪 $0.0075(估算值,無快取命中)
使用相同的算術,我們得到這些估算值:deepseek-v4-flash 離峰時為 $0.0006,尖峰時為 $0.0012;claude-sonnet-5 為 $0.00405;gemini-3.5-flash 為 $0.00375。超過 1,000 輪這樣的對話,gpt-5.5 約為 $7.50,而 deepseek-v4-flash 則為 $0.60 到 $1.20。真實對話會隨著歷史累積而增長,因此請從回應或最終串流區塊讀取 usage,而非信任此估算值。
在實務上,我們將大多數對話路由至預設層級,並為需要更高能力的輪次保留 gpt-5.5 或 claude-sonnet-5。帳單文件警告說,最低的 Token 單價並不總是最低的任務完成成本。請在兩個層級上執行相同的提示詞,並在 使用量 (Usage) 中比較最終成本。
發佈前進行兩項檢查。帳單文件指出不要硬編碼複製的價格表,因此如果你的應用程式顯示成本,請讀取 GET /v1/models/:model/pricing(帳單與定價,觀察於 2026-10-03)。此外,deepseek-v4-flash 條目列出了 json-mode、prompt-cache 和 tool-use,但沒有 vision。如果你的聊天機器人接受圖片,請將這些輪次路由至條目中列出 vision 的層級。
冒煙測試並發佈你的 AI 聊天機器人
步驟 7:端到端執行
在已匯出 TOKENLAB_API_KEY 的終端中啟動伺服器。curl 負載使用與 ChatRequest 相同的 message、tier 和 conversation_id 欄位。
uvicorn main:app --reload --port 8000
# 1. 同步端點:預期簡短回覆和 conversation_id
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "Reply only with OK.", "tier": "default"}'
# 2. 串流:預期 data: {"delta": ...} 行,然後是 {"done": true, ...}
curl -N -X POST http://localhost:8000/chat/stream \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "tier": "default"}'
# 3. 記憶體:重複使用上述回傳的 conversation_id
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "What did I just say?", "tier": "default", "conversation_id": "PASTE_ID_FROM_STEP_1"}'
# 4. 強制失敗:預期 HTTP 400 和 "unknown tier: nope"
curl -i -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "Hello!", "tier": "nope"}'
若要強制上游失敗,請使用錯誤的 TOKENLAB_API_KEY 重啟伺服器。你的客戶端應收到乾淨的 503 訊息,且你的日誌應顯示帶有 invalid_api_key 的上游 401 錯誤。然後開啟 請求 (Requests) 以確認成功呼叫的模型、狀態和使用量。如果你能串流一輪對話、保留一個對話狀態,並在強制失敗時回傳乾淨的錯誤,你就擁有了正確的基礎。
之後,下一個升級是無聊的基礎設施,而非更多的 AI 功能:
- 身份驗證,防止一個使用者讀取另一個使用者的對話。
- 持久化,確保會話在部署後依然存在。
- 按使用者的速率限制,防止吵鬧的使用者耗盡你的配額。
- 使用量歸屬,如果聊天機器人是面向客戶的。
- 背景摘要,如果對話需要長期記憶。
一旦Base URL 遷移完成,模型變更僅需編輯 ALLOWED_MODELS 設定檔即可。
常見問題
我需要 Websocket 來構建 AI 聊天機器人嗎?
不需要。步驟 4 展示的 SSE 涵蓋了大多數聊天產品。Websocket 在你需要請求與回應之外的雙向推送時才有價值,例如即時協作或伺服器發起的事件。對於標準聊天 UI,SSE 更易於部署和除錯。
我的聊天機器人預設應該使用哪個模型?
從低價層級(如 deepseek-v4-flash)開始,並在白名單後方加入 claude-sonnet-5 或 gpt-5.5 以處理更困難的輪次。上述觀察於 2026-10-03 的價格顯示了差距。然後在兩個層級上測試你自己的提示詞,並在 Usage 中比較最終成本。在提交前檢查模型目錄,因為陣容會變更。
為什麼我的聊天機器人回傳 402 或 429?
402 意味著帳戶餘額或 Key 本身的使用限制過低。代碼為 insufficient_balance 和 quota_exceeded,重複請求無濟於事。429 意味著你達到了請求限制,請等待 Retry-After 後再重試。我們的同步端點會重試一次,串流端點則會將錯誤報告給客戶端。
瀏覽器可以直接呼叫 TokenLab 嗎?
不安全。身份驗證文件指出絕不要在客戶端程式碼中暴露 API Key。請將 Key 保留在伺服器環境中,並讓瀏覽器呼叫你自己的已驗證後端,如前端範例所示。
在 tokenlab.sh 建立 API Key,儲值並執行上述冒煙測試。
來源
價格觀測於 2026-10-03
- TokenLab Docs: Quickstart觀測於 2026-10-03
- TokenLab Docs: Authentication觀測於 2026-10-03
- TokenLab Docs: Streaming觀測於 2026-10-03
- TokenLab Docs: Handle API errors觀測於 2026-10-03
- TokenLab Docs: Billing and pricing觀測於 2026-10-03
- TokenLab Docs: SDKs & Libraries觀測於 2026-10-03
- TokenLab Docs: Create Chat Completion觀測於 2026-10-03
- TokenLab Docs: List Models觀測於 2026-10-03



