Hướng dẫn này sẽ giúp bạn xây dựng một dịch vụ chatbot nhỏ gọn nhưng sẵn sàng cho môi trường production với FastAPI, SSE streaming, bộ nhớ hội thoại và khả năng chuyển đổi model. Mục tiêu không phải là một bản demo chơi thử, mà là một backend bạn có thể đưa vào sử dụng thực tế và lặp lại (iterate) một cách an toàn mà không cần viết lại mã tích hợp mỗi khi thay đổi model.
Nếu bạn đã trỏ một SDK tương thích với OpenAI vào TokenLab, bài viết này sẽ tiếp nối từ bước đó. Nếu bạn chưa thực hiện việc thay đổi base URL, hãy đọc hướng dẫn di chuyển trước. Nếu mối quan tâm chính của bạn là định hình yêu cầu (request shaping) và cơ chế backoff khi chịu tải, hãy kết hợp hướng dẫn này với hướng dẫn giới hạn tốc độ AI API.
Những điểm chính
- Một chatbot sẵn sàng cho production cần sáu phần: endpoint đồng bộ, endpoint streaming, trạng thái hội thoại phía server, danh sách cho phép (allowlist) model, xử lý lỗi thực tế và lộ trình nâng cấp lưu trữ rõ ràng.
- Hãy kiểm chứng key, base URL và định tuyến của bạn với một endpoint chat nhỏ trước khi thêm vào streaming, bộ nhớ hoặc gọi công cụ (tool calls).
- SSE streaming đáp ứng hầu hết các sản phẩm chat và ít tốn kém chi phí vận hành hơn so với websockets.
- Hãy hiển thị các model thông qua một danh sách cho phép ở backend, thay vì một trường văn bản tự do, để frontend không thể yêu cầu các ID model tùy ý hoặc đã ngừng hỗ trợ.
- Tính khả dụng và danh mục model thay đổi thường xuyên. Hãy kiểm tra danh mục model của TokenLab (cập nhật ngày 07/07/2026) trước khi khóa danh sách cho phép của bạn vào môi trường production.
Chúng ta đang xây dựng những gì
Dịch vụ hoàn thiện sẽ có sáu phần chuyển động:
- Một endpoint
/chatđồng bộ để kiểm tra nhanh (smoke test). - Một endpoint
/chat/streamstreaming cho giao diện người dùng thực tế. - Trạng thái hội thoại được định danh bởi
conversation_id. - Một danh sách cho phép model để frontend không thể yêu cầu các ID tùy ý.
- Xử lý lỗi không bị sụp đổ ngay khi gặp lỗi 429 đầu tiên.
- Lộ trình rõ ràng từ bản mẫu lưu trong bộ nhớ (in-memory) sang Redis hoặc PostgreSQL.
Đó là đủ để vận hành một bot hỗ trợ, trợ lý nội bộ hoặc phiên bản đầu tiên của widget chat nhúng.
Cài đặt ngăn xếp tối thiểu
pip install fastapi uvicorn openai pydantic redis
Bạn có thể bỏ qua redis cho lần chạy đầu tiên, nhưng việc thêm import ngay từ bây giờ sẽ giúp việc nâng cấp sau này trở nên dễ dàng thay vì phải refactor lại toàn bộ.
Bước 1: Bắt đầu với một endpoint Chat nhỏ, đơn giản
Cách nhanh nhất để bị lạc lối khi làm chatbot là bắt đầu với websockets, sử dụng công cụ và điều phối tác nhân (agent orchestration) trước khi đường dẫn yêu cầu cơ bản ổn định. Hãy bắt đầu với một endpoint nhỏ để chứng minh rằng key, base URL và định tuyến model của bạn hoạt động tốt.
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}
Hãy chạy một bài kiểm tra nhanh. Nếu bước này thất bại, đừng tiếp tục xây dựng thêm trên đó.
Bước 2: Thêm Streaming vì người dùng cảm nhận được độ trễ trước khi họ đo lường nó
Hầu hết các sản phẩm chatbot cảm thấy chậm không phải vì model chậm, mà vì giao diện người dùng vẫn trống trơn cho đến khi toàn bộ phản hồi được tải xong. SSE là đủ cho hầu hết các sản phẩm chat và có gánh nặng vận hành thấp hơn so với websockets.
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")
Trên frontend, một trình đọc fetch cơ bản là đủ:
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);
}
}
Nếu sản phẩm của bạn đã chạy client trình duyệt qua HTTP tiêu chuẩn, SSE giúp kiến trúc đơn giản hơn so với lớp websocket.
Bước 3: Di chuyển trạng thái hội thoại ra khỏi Request Body
Bản demo chatbot đầu tiên thường giữ toàn bộ bản ghi trong trình duyệt và gửi lại mỗi khi có lượt chat mới. Điều đó hiệu quả với bản mẫu, nhưng sẽ trở nên lộn xộn nhanh chóng khi bạn cần thử lại (retry), các phiên có thể tiếp tục hoặc công cụ phía server.
Lưu trữ trong bộ nhớ (in-memory) là ổn để bắt đầu:
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
Lộ trình nâng cấp lên Redis chủ yếu là về kỹ thuật lưu trữ, không phải thay đổi logic:
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))
Hãy sử dụng Redis khi các cuộc hội thoại cần TTL, khả năng tiếp tục phiên hoặc triển khai đa instance. Hãy sử dụng PostgreSQL khi bản ghi hội thoại là dữ liệu sản phẩm mà bạn cần truy vấn, kiểm toán hoặc báo cáo sau này.
Bước 4: Xử lý lỗi như một hành vi sản phẩm, không chỉ là ngoại lệ
Nếu chatbot của bạn hướng tới khách hàng, đường dẫn lỗi cũng quan trọng như đường dẫn thành công. Người dùng không quan tâm liệu lỗi đến từ giới hạn tốc độ, hết số dư hay sự cố model ở phía upstream. Họ chỉ quan tâm giao diện có bị treo hay không.
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")
Bước 5: Khóa các model mà frontend có thể yêu cầu
Đừng bao giờ để frontend truyền một chuỗi model tùy ý trực tiếp vào API. Một trường văn bản tự do sẽ dẫn đến các yêu cầu cho các model đã ngừng hỗ trợ, sai chính tả hoặc các model bạn không bao giờ định cung cấp. Thay vào đó, hãy định tuyến thông qua một danh sách cho phép ở backend.
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"])
# phần còn lại của logic streaming sử dụng `model` thay vì chuỗi do client cung cấp
Điều này cho phép bạn có một nơi duy nhất để thay đổi model khi nhà cung cấp ngừng hỗ trợ một model nào đó, mà không cần đụng vào mã frontend hoặc phát hành bản cập nhật client.
Bước 6: Xử lý phần còn lại của production
Một backend chatbot được coi là sẵn sàng cho production khi các khía cạnh xung quanh được xử lý, không phải khi lệnh gọi chat cốt lõi trở nên thông minh.
Danh sách kiểm tra rất ngắn gọn:
- thêm request ID để bạn có thể kết nối lỗi frontend với log backend
- giới hạn concurrency và kích thước yêu cầu cho mỗi người dùng
- cắt bớt lịch sử dài trước khi chúng làm nổ ngân sách token của bạn
- ghi log model, độ trễ, kích thước đầu vào và lý do kết thúc
- tách biệt thông báo lỗi hiển thị cho người dùng với chi tiết lỗi nội bộ
- kiểm tra một model thay thế để đảm bảo cơ chế dự phòng hoạt động trước khi sự cố thực sự xảy ra
Việc cắt bớt lịch sử có thể giữ ở mức đơn giản:
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
Mục tiêu không phải là tính toán chính xác từng token, mà là ngăn chặn việc bùng nổ ngữ cảnh trước khi nó ảnh hưởng đến hóa đơn hoặc độ trễ của bạn.
Từ Demo đến Sản phẩm
Khi backend này đã ổn định, bản nâng cấp tiếp theo hiếm khi là "thêm AI". Thường đó là cơ sở hạ tầng nhàm chán:
- xác thực để một người dùng không thể đọc cuộc hội thoại của người khác
- lưu trữ bền vững để các phiên làm việc tồn tại sau khi triển khai
- giới hạn tốc độ để một người dùng gây nhiễu không thể đốt cháy hạn ngạch của bạn
- thanh toán hoặc phân bổ mức sử dụng nếu chatbot hướng tới khách hàng
- tóm tắt nền nếu các cuộc hội thoại cần bộ nhớ dài hạn
Một cổng kết nối thống nhất (unified gateway) giúp ích cho hầu hết các việc này. Khi việc di chuyển base URL đã hoàn tất, việc thay đổi model không còn là viết lại nền tảng mà chỉ là chỉnh sửa cấu hình.
Smoke Test
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"}'
Nếu bạn có thể stream một lượt chat, bảo toàn một cuộc hội thoại và trả về lỗi sạch sẽ khi ép buộc lỗi, bạn đã có nền tảng đúng đắn.
Ước tính chi phí
Tạo một API key tại TokenLab, trỏ SDK OpenAI của bạn vào https://api.tokenlab.sh/v1, và bạn có thể phát hành phiên bản production đầu tiên của chatbot mà không cần quản lý các tài khoản riêng biệt giữa các nhà cung cấp.
| Model | Tier điển hình | Ghi chú |
|---|---|---|
| DeepSeek V4 Flash | Nhanh / mặc định | Lựa chọn tốt cho các lượt chat khối lượng lớn, độ trễ thấp |
| GPT-5.5 | Flagship | Sử dụng cho các lượt chat cần khả năng suy luận mạnh hơn |
| Claude Sonnet 5 | Cân bằng | Lựa chọn mạnh mẽ cho lập trình và phản hồi kiểu đánh giá |
| Gemini 3.5 Flash | Ngân sách / thay thế nhanh | Chi phí thấp, thay thế nhanh cho định tuyến khối lượng lớn |
Giá chính xác cho mỗi token thay đổi thường xuyên giữa các nhà cung cấp và không được liệt kê ở đây dưới dạng con số cố định. Hãy kiểm tra giá hiện tại trên danh mục model (cập nhật ngày 07/07/2026) trước khi lập ngân sách. Trong thực tế, việc định tuyến hầu hết các cuộc hội thoại vào tier nhanh/mặc định như DeepSeek V4 Flash hoặc Gemini 3.5 Flash và dành riêng GPT-5.5 hoặc Claude Sonnet 5 cho các lượt cần thiết sẽ giúp giữ hóa đơn hàng tháng ở mức thấp, nhưng hãy xác nhận giá thực tế cho mỗi triệu token cho tài khoản của bạn trước khi cam kết ngân sách.
Câu hỏi thường gặp
Tôi có cần websockets để xây dựng AI chatbot không? Không. SSE streaming, được trình bày ở Bước 2, đáp ứng đại đa số các sản phẩm chat. Websockets mang lại giá trị thực khi bạn cần đẩy dữ liệu hai chiều bên ngoài request/response, như cộng tác trực tiếp hoặc các sự kiện do server khởi tạo. Đối với giao diện chat tiêu chuẩn, SSE đơn giản hơn để triển khai, gỡ lỗi và mở rộng.
Làm thế nào để biết nên mặc định dùng model nào? Hãy bắt đầu với một model nhanh, chi phí thấp như DeepSeek V4 Flash hoặc Gemini 3.5 Flash cho tier mặc định, và thêm một tier cân bằng hoặc suy luận trên Claude Sonnet 5 hoặc GPT-5.5 đằng sau danh sách cho phép được trình bày ở Bước 5. Kiểm tra danh mục model (cập nhật ngày 07/07/2026) để biết các tùy chọn hiện tại, vì các model mới ra mắt và các model cũ sẽ bị ngừng hỗ trợ theo lịch trình nằm ngoài tầm kiểm soát của bạn.
Điều gì hỏng đầu tiên khi chatbot chuyển từ demo sang lưu lượng thực tế? Hầu như luôn là đường dẫn lỗi, không phải đường dẫn thành công. Việc thử lại không giới hạn, thiếu giới hạn concurrency cho mỗi người dùng và lịch sử hội thoại không giới hạn là ba nguyên nhân phổ biến nhất khiến backend chatbot sụp đổ dưới tải thực tế. Các bước 4 và 6 ở trên giải quyết trực tiếp cả ba vấn đề này.
Bắt đầu với API key của bạn: tokenlab.sh cung cấp hơn 300 model thông qua một endpoint duy nhất. Tặng $1 credit miễn phí để bắt đầu xây dựng.
Nguồn
Giá quan sát ngày 2026-07-07
- TokenLab model directoryQuan sát ngày 2026-07-07
