使用一个 API Key 构建 AI 聊天机器人：从零到生产环境仅需 30 分钟

本教程将使用 FastAPI、SSE 流式传输、对话记忆和模型切换功能，构建一个小型但可用于生产环境的聊天机器人服务。我们的目标不是一个玩具演示，而是构建一个你可以直接部署在真实产品界面之后，并且在更换模型时无需重写集成代码，即可安全迭代的后端。

如果你已经将某个兼容 OpenAI 的 SDK 指向了 TokenLab，那么本文将从该步骤继续。如果你尚未进行 base URL 替换，请先阅读迁移指南。如果你主要关心负载下的请求整形和退避策略，请将本指南与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、工具调用和 Agent 编排，很容易迷失方向。先从一个能证明你的 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”，而是枯燥的基础设施：

• 身份验证，防止一个用户读取另一个用户的对话
• 持久化，确保会话在部署后依然存在
• 速率限制，防止一个嘈杂的用户耗尽你的配额
• 计费或使用归因，如果聊天机器人是面向客户的
• 后台摘要，如果对话需要长期记忆

统一网关有助于解决大部分问题。一旦完成了 base 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 费率。

常见问题

构建 AI 聊天机器人需要 WebSocket 吗？ 不需要。第 2 步中展示的 SSE 流式传输涵盖了绝大多数聊天产品。当你需要在请求/响应之外进行双向推送（如实时协作或服务器发起的事件）时，WebSocket 才有真正的价值。对于标准的聊天 UI，SSE 更易于部署、调试和扩展。

我该如何决定默认使用哪个模型？ 从 DeepSeek V4 Flash 或 Gemini 3.5 Flash 等快速、低成本的模型作为默认层级开始，并在第 5 步展示的白名单后面添加基于 Claude Sonnet 5 或 GPT-5.5 的均衡或推理层级。请查看模型目录（观察日期：2026-07-07）了解当前选项，因为新模型会发布，旧模型会按照你无法控制的时间表被弃用。

当聊天机器人从演示走向真实流量时，什么最先崩溃？ 几乎总是错误路径，而不是成功路径。无限制的重试、缺失的每用户并发上限以及无限制的对话历史，是聊天机器人后端在真实负载下崩溃的三个最常见原因。上述第 4 步和第 6 步直接解决了这三个问题。

---

使用你的 API Key 开始：tokenlab.sh 通过一个端点提供 300 多个模型。提供 $1 免费额度助你开始构建。