使用單一 API Key 打造 AI 聊天機器人：30 分鐘從零到上線

本教學將建立一個具備串流回應 (streaming responses)、對話歷史、模型切換及完善錯誤處理機制的生產級 AI 聊天機器人後端。我們將使用 Python、FastAPI 以及指向 API 聚合器的 OpenAI SDK，讓你可以使用任何模型。

前置作業

pip install fastapi uvicorn openai

步驟 1：基礎聊天端點

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()

client = OpenAI(
    api_key="sk-lemon-xxx",
    base_url="https://api.lemondata.cc/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "gpt-4.1-mini"
    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}

這段程式碼可以運作，但沒有串流、沒有歷史紀錄，也沒有錯誤處理。讓我們來改進它。

步驟 2：加入串流功能

串流功能會在 token 生成時立即傳送，而不是等待完整回應。使用者可以即時看到回覆內容逐字呈現。

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

步驟 3：對話歷史紀錄

將對話歷史儲存在記憶體中（在生產環境中請換成 Redis 或資料庫）。

from collections import defaultdict
import uuid

conversations: dict[str, list] = defaultdict(list)

SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."

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

    # 建立訊息歷史
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.extend(conversations[conv_id])
    messages.append({"role": "user", "content": req.message})

    # 儲存使用者訊息
    conversations[conv_id].append(
        {"role": "user", "content": req.message}
    )

    def generate():
        full_response = []
        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"

        # 儲存助手回應
        conversations[conv_id].append(
            {"role": "assistant", "content": "".join(full_response)}
        )
        yield f"data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={"X-Conversation-ID": conv_id}
    )

步驟 4：錯誤處理

AI API 呼叫可能會因為多種原因失敗：速率限制 (rate limits)、餘額不足、模型不可用。我們需要處理這些情況：

from openai import (
    APIError,
    RateLimitError,
    APIConnectionError
)

@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():
        try:
            full_response = []
            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"

            conversations[conv_id].append(
                {"role": "assistant", "content": "".join(full_response)}
            )

        except RateLimitError as e:
            yield f"data: [ERROR] Rate limited. Please wait a moment.\n\n"
        except APIConnectionError:
            yield f"data: [ERROR] Connection failed. Retrying...\n\n"
        except APIError as e:
            yield f"data: [ERROR] {e.message}\n\n"

        yield "data: [DONE]\n\n"

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

def build_messages(conv_id: str, user_msg: str) -> list:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    # 保留最後 10 輪對話以管理上下文長度
    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

步驟 5：模型切換

讓使用者在對話中途切換模型，根據不同需求使用不同模型：

AVAILABLE_MODELS = {
    "fast": "gpt-4.1-mini",
    "smart": "claude-sonnet-4-6",
    "reasoning": "o3",
    "budget": "deepseek-chat",
    "creative": "claude-sonnet-4-6",
}

@app.get("/models")
async def list_models():
    return {"models": AVAILABLE_MODELS}

前端可以將這些作為選項呈現。由於所有模型都透過聚合器使用相同的 OpenAI 相容格式，切換模型只需更改 model 參數即可。

步驟 6：上下文視窗管理

長對話會超過模型的上下文限制。實作一個滑動視窗 (sliding window)：

def trim_history(messages: list, max_tokens: int = 8000) -> list:
    """在 token 預算內保留系統提示詞 + 最近的訊息。"""
    # 粗略估計：1 token ≈ 4 個字元
    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

完整應用程式

# 執行指令：uvicorn main:app --reload --port 8000
# 測試指令：curl -N -X POST http://localhost:8000/chat/stream \
#   -H "Content-Type: application/json" \
#   -d '{"message": "Hello!", "model": "gpt-4.1-mini"}'

完整程式碼不到 100 行。從這裡開始，你可以加入：

• 身份驗證 (API keys 或 JWT)
• 持久化儲存 (使用 PostgreSQL 或 Redis 儲存對話)
• 針對每個使用者的速率限制
• 用量追蹤與計費
• 支援雙向串流的 WebSocket
• 前端介面 (React、Vue 或使用 EventSource 的原生 JS)

成本預估

以一個每天處理 1,000 個對話（平均每個對話 5 輪）的聊天機器人為例：

模型	每日成本	每月成本
GPT-4.1-mini	約 $2.40	約 $72
GPT-4.1	約 $12.00	約 $360
Claude Sonnet 4.6	約 $18.00	約 $540
DeepSeek V3	約 $1.68	約 $50

大多數對話使用 GPT-4.1-mini，僅在使用者要求時升級到 Claude Sonnet 4.6，對於大多數應用程式來說，可以將每月成本控制在 100 美元以下。

---

獲取你的 API key：lemondata.cc 透過單一端點提供 300 多種模型。註冊即贈送 $1 免費額度開始開發。