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

你可以在一个下午的时间里构建出一个支持流式回复的 AI 聊天机器人。而真正的挑战在于如何处理第一次 429 错误之后的情况。我们将构建一个小型 FastAPI 服务，包含 SSE 流式传输、服务端对话记忆以及模型白名单，所有这些都由一个 TokenLab Key 驱动。我们的目标是打造一个后端，你可以将其置于真实的产品界面之后，并在不重写集成代码的情况下更换模型。如果你还没有完成基础 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 聊天机器人：设置与核心服务

成品服务包含六个动态部分：

1. 用于冒烟测试的同步 /chat 端点。
2. 用于真实 UI 的流式 /chat/stream 端点。
3. 以 conversation_id 为键的对话状态。
4. 模型白名单，防止前端请求任意 ID。
5. 不会在第一次 429 错误时崩溃的错误处理机制。
6. 从内存原型迁移到 Redis 或 PostgreSQL 的路径。

这对于支持机器人、内部助手或嵌入式聊天小部件的第一个版本来说已经足够了。下面的每个代码块都按所示顺序附加到同一个文件 main.py 中。

第 1 步：设置 Key 并安装技术栈

在你的工作区中通过 API Keys 创建一个 Key。完整的值仅显示一次，Key 列表中的前缀无法验证请求。所有 TokenLab Key 都以 sk- 开头。同时请在 账单 中检查你的余额，因为第一次请求将按模型的当前价格计费（快速入门，观察于 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 和基础 URL，官方的 OpenAI SDK 即可工作（SDK 指南，观察于 2026-10-03）。

第 2 步：定义模式、白名单和记忆

我们从共享部分开始。请求模式使用 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，因为它涵盖了大多数聊天产品，并且比 Websockets 承载的操作压力更小。如果你的产品已经在纯 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 一起保存。文档没有在我们要使用的页面上命名该标头，因此在连接日志之前请对照 故障排除路径 进行确认。

生产清单的其余部分很短：

• 添加你自己的请求 ID，以便将前端故障连接到后端日志。
• 限制每个用户的并发和请求大小。message 上的 max_length=4000 是一个你应该调整的起点。
• 像 MAX_HISTORY 那样按消息计数限制历史记录。账单文档称 Token 化因模型、语言和媒体输入而异，因此字符计数公式不是可靠的计费估算。
• 记录每一轮的模型、延迟、使用情况和完成原因。
• 将用户可见的消息与内部错误详细信息分开。
• 向备用层级发送一个测试轮次，以便在第一次真正中断之前知道回退是否有效。

选择模型并估算 AI 聊天机器人的成本

永远不要让前端将任意模型字符串传递给 API。自由文本字段会引发拼写错误、已退役的模型以及你从未打算暴露的模型。ALLOWED_MODELS 映射为你提供了一个在提供商弃用模型时更换模型的地方，而无需更新客户端。未知的模型 ID 会返回 model_not_found，文档会将你指向 /v1/models 以获取当前 ID。在锁定白名单之前，请检查 模型目录。

当前每 100 万 Token 的价格

这些是来自每个模型实时 API 条目的 TokenLab 美元价格，观察于 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 的最低价格并不总是完成任务的最低成本。在两个层级上运行相同的提示，并在 使用情况 中比较最终成本。

发布前进行两次检查。账单文档称不要硬编码复制的价格表，因此如果你的应用显示成本，请读取 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。然后打开 请求 以确认成功调用的模型、状态和使用情况。如果你能流式传输一轮、保留一个对话，并在强制失败时返回一个干净的错误，你就拥有了正确的基础。

之后，接下来的升级是无聊的基础设施，而不是更多的 AI：

• 身份验证，这样一个用户就无法读取另一个用户的对话。
• 持久性，这样会话可以在部署后存活。
• 按用户速率限制，这样吵闹的用户就不会耗尽你的配额。
• 使用情况归因，如果聊天机器人是面向客户的。
• 后台摘要，如果对话需要长期记忆。

一旦 基础 URL 迁移 完成，模型更改就是对 ALLOWED_MODELS 的配置编辑。

常见问题解答

我需要 Websockets 来构建 AI 聊天机器人吗？

不需要。第 4 步中展示的 SSE 涵盖了大多数聊天产品。当你需要在请求和响应之外进行双向推送（例如实时协作或服务器发起的事件）时，Websockets 才有价值。对于标准的聊天 UI，SSE 更易于部署和调试。

我的聊天机器人默认应该使用哪个模型？

从低价层级（如 deepseek-v4-flash）开始，并在白名单后面为更难的轮次添加 claude-sonnet-5 或 gpt-5.5。上述价格（观察于 2026-10-03）显示了差距。然后在两个层级上测试你自己的提示，并在使用情况中比较最终成本。在提交之前检查 模型目录，因为阵容会发生变化。

为什么我的聊天机器人返回 402 或 429？

402 表示账户余额或 Key 本身的使用限制过低。代码是 insufficient_balance 和 quota_exceeded，重复请求无济于事。429 表示你达到了请求限制，因此等待 Retry-After 然后重试。我们的同步端点重试一次，流式端点将错误报告给客户端。

浏览器可以直接调用 TokenLab 吗？

不安全。身份验证文档称切勿在客户端代码中暴露 API Key。将 Key 保留在服务器环境中，并让浏览器调用你自己的已验证后端，就像前端示例所做的那样。

---

在 tokenlab.sh 创建 API Key，添加资金，并运行上面的冒烟测试。