Hướng dẫn này sẽ xây dựng một dịch vụ chatbot nhỏ nhưng sẵn sàng cho production với FastAPI, SSE streaming, bộ nhớ hội thoại và chuyển đổi model. Mục tiêu không phải là tạo ra một bản demo đồ chơi. Mục tiêu là xây dựng một backend mà bạn thực sự có thể đưa vào sản phẩm và cải tiến một cách an toàn.
Nếu bạn đã trỏ một SDK tương thích với OpenAI tới LemonData, bài viết này sẽ tiếp nối từ đó. Nếu bạn chưa thực hiện 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à điều chỉnh request và backoff khi tải nặng, hãy kết hợp hướng dẫn này với hướng dẫn rate limiting cho AI API.
Những gì chúng ta sẽ xây dựng
Dịch vụ hoàn thiện sẽ có sáu thành phần chính:
- Một endpoint
/chatđồng bộ để smoke test. - Một endpoint
/chat/streamdạng streaming cho UI thực tế. - Trạng thái hội thoại được định danh bằng
conversation_id. - Một danh sách model cho phép (allowlist) để 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.
- Một lộ trình rõ ràng từ prototype lưu trên bộ nhớ (in-memory) sang Redis hoặc PostgreSQL.
Như vậy là đủ để vận hành một bot hỗ trợ, một trợ lý nội bộ hoặc phiên bản đầu tiên của một chat widget nhúng.
Cài đặt Stack tối thiểu
pip install fastapi uvicorn openai pydantic redis
Bạn có thể bỏ qua redis trong lượt đầu tiên, nhưng việc kết nối import ngay bây giờ là hữu ích để lộ trình nâng cấp trở nên rõ ràng.
Bước 1: Bắt đầu với một Chat Endpoint nhỏ và đơn giản
Cách nhanh nhất để bị lạc lối khi làm chatbot là bắt đầu với websockets, tool use và điều phối agent trước khi luồng request cơ bản ổn định. Hãy bắt đầu với một endpoint nhỏ để chứng minh key, base URL và định tuyến model của bạn là chính xác.
from fastapi import FastAPI
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}
Chạy một smoke test. Nếu bước này thất bại, đừng tiếp tục thêm các tính năng khác lên trên.
Bước 2: Thêm Streaming vì người dùng cảm nhận độ trễ trước khi họ đo lường nó
Hầu hết các sản phẩm chatbot mang lại cảm giác chậm chạp không phải vì model chậm, mà vì UI trống trơn cho đến khi toàn bộ phản hồi được trả về. SSE là đủ cho nhiều sản phẩm chat và có gánh nặng vận hành thấp hơn 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")
Ở phía frontend, một client trình duyệt đơn giản nhất vẫn đủ dùng:
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 đã sử dụng browser client và HTTP tiêu chuẩn, SSE sẽ giúp kiến trúc đơn giản hơn.
Bước 3: Đưa 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ộ lịch sử trò chuyện trong trình duyệt và gửi nó đi trong mỗi lượt. Cách này hoạt động tốt cho các bản prototype. Nhưng nó sẽ trở nên rắc rối ngay khi bạn cần tính năng retry, phiên làm việc có thể tiếp tục (resumable sessions) hoặc các công cụ phía server.
Một bộ lưu trữ 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à việc thiết lập lưu trữ:
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))
Sử dụng Redis nếu hội thoại cần TTL, khả năng tiếp tục hoặc triển khai trên nhiều instance. Sử dụng PostgreSQL nếu bản thân lịch sử trò chuyện là dữ liệu sản phẩm.
Bước 4: Coi lỗi là một hành vi của sản phẩm, không chỉ là các Exception
Nếu chatbot của bạn hướng đến khách hàng, luồng xử lý thất bại cũng quan trọng như luồng thành công. Người dùng không quan tâm thất bại đến từ việc bị rate limiting, hết số dư hay model bị lỗi. Họ chỉ quan tâm liệu UI 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] The model is busy. Please retry in a few seconds.\n\n"
except APIConnectionError:
yield "data: [ERROR] Temporary network issue. Please retry.\n\n"
except APIError as error:
yield f"data: [ERROR] {error.message}\n\n"
else:
conversations[conv_id].append(
{"role": "assistant", "content": "".join(full_response)}
)
finally:
yield "data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={"X-Conversation-ID": conv_id}
)
Nếu bạn đang phục vụ lượng tải lớn, bạn cũng nên điều chỉnh request trước khi chúng đến upstream. Các pattern chi tiết có trong hướng dẫn rate limiting, nhưng tóm tắt là: sử dụng retry có giới hạn, sử dụng jitter và đừng bao giờ dùng except Exception: time.sleep(1).
Bước 5: Chuyển đổi Model cần một Allowlist, không phải một ô nhập văn bản tự do
Một API key có thể tiếp cận hàng trăm model. Điều đó không có nghĩa là UI của bạn nên hiển thị hàng trăm model đó. Backend nên công bố một allowlist nhỏ phù hợp với trường hợp sử dụng của bạn.
AVAILABLE_MODELS = {
"fast": "gpt-4.1-mini",
"balanced": "claude-sonnet-4-6",
"reasoning": "o3",
"budget": "deepseek-chat",
}
@app.get("/models")
async def list_models():
return {"models": AVAILABLE_MODELS}
Việc này mang lại ba lợi ích:
- Ngăn chặn frontend yêu cầu các model ID không hợp lệ hoặc đã lỗi thời.
- Cho phép bạn ánh xạ lại một phân cấp (tier) sau này mà không cần triển khai lại mọi client.
- Cung cấp một nơi duy nhất để thực thi kiểm soát chi phí.
Nếu team của bạn vẫn đang phân vân nên tiêu chuẩn hóa nhà cung cấp nào, so sánh giá và so sánh OpenRouter và LemonData là hai trang đáng đọc trước khi bạn chốt allowlist.
Bước 6: Thêm các yếu tố Production trước khi có traffic
Một backend chatbot trở nên chuyên nghiệp khi các yếu tố xung quanh được xử lý, chứ 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 số lượng kết nối đồng thời (concurrency) và kích thước request trên mỗi người dùng.
- Cắt tỉa lịch sử dài trước khi chúng làm bùng nổ ngân sách token của bạn.
- Log lại model, độ trễ, kích thước input và lý do kết thúc (finish reason).
- 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ế để bạn biết cơ chế dự phòng (fallback) hoạt động trước khi xảy ra sự cố đầu tiên.
Việc cắt tỉa lịch sử có thể thực hiện đơ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
Vấn đề không phải là tính toán token chính xác tuyệt đối. Vấn đề là ngăn chặn việc bùng nổ context một cách rõ ràng.
Từ Demo đến Sản phẩm
Khi backend này đã ổn định, bước nâng cấp tiếp theo hiếm khi là “thêm AI.” Nó thường là các hạ tầng cơ bản:
- Auth để người dùng này không thể đọc hội thoại của người dùng khác.
- Lưu trữ bền vững (persistence) để các phiên làm việc tồn tại qua các lần deploy.
- Rate limiting để một người dùng hoạt động quá mức không làm cháy hạn mức (quota) của bạn.
- Thanh toán hoặc phân bổ mức sử dụng nếu chatbot hướng đến khách hàng.
- Tóm tắt chạy ngầm (background summarization) nếu hội thoại cần bộ nhớ dài hạn.
Đó là lý do tại sao một gateway thống nhất lại hữu ích. Một khi bạn đã hoàn thành việc di chuyển base URL, việc thay đổi model sẽ không còn là viết lại nền tảng mà chỉ là thay đổi 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": "gpt-4.1-mini"}'
Nếu bạn có thể stream một lượt, lưu giữ một hội thoại và trả về một lỗi sạch sẽ khi giả lập thất bại, bạn đã có một nền tảng đúng đắn.
Ước tính chi phí
Tạo một API key tại LemonData, trỏ OpenAI SDK của bạn tới https://api.lemondata.cc/v1, và bạn có thể triển khai 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 nhà cung cấp riêng biệt.
| Model | Chi phí hàng ngày | Chi phí hàng tháng |
|---|---|---|
| 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 |
Việc sử dụng GPT-4.1-mini cho hầu hết các cuộc hội thoại và chỉ nâng cấp lên Claude Sonnet 4.6 khi người dùng yêu cầu sẽ giúp giữ chi phí dưới $100/tháng cho hầu hết các ứng dụng.
Nhận API key của bạn: lemondata.cc 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.