Chào các bạn! Mình là Minh, một lập trình viên backend tại HolySheep AI. Hôm nay mình sẽ chia sẻ với các bạn một chủ đề mà nhiều người mới rất hay hỏi: "Làm sao để tạo chatbot AI real-time như ChatGPT, có phản hồi tức thì và không bị gián đoạn khi mất mạng?"
Trong bài viết này, mình sẽ hướng dẫn các bạn từ con số 0, không cần biết gì về API hay WebSocket trước đó. Tất cả code đều có thể copy-paste và chạy ngay!
WebSocket Là Gì? Tại Sao Cần Nó?
Trước khi code, mình muốn giải thích bằng ngôn ngữ đời thường:
HTTP truyền thống giống như bạn gọi điện cho nhà hàng đặt đồ ăn: bạn gọi → nhà hàng trả lời → kết thúc. Mỗi lần muốn hỏi gì, bạn phải gọi lại từ đầu.
WebSocket giống như có đường dây nóng 24/7 với đầu bếp: bạn cứ nói chuyện liên tục, đầu bếp trả lời liền mạch, không cần ngắt kết nối. Đây là lý do AI chat cần WebSocket — để có trải nghiệm real-time, phản hồi từng từ như đang chat thật.
Với HolySheep AI, độ trễ chỉ dưới 50ms — nhanh hơn nhiều so với các provider khác, giúp trải nghiệm chat mượt mà hơn bao giờ hết.
Kiến Trúc Hai Chiều (Bidirectional Communication)
Trong mô hình WebSocket, có hai luồng dữ liệu chạy song song:
- Client → Server: Gửi tin nhắn người dùng, sự kiện typing, heartbeat
- Server → Client: Phản hồi AI theo thời gian thực (streaming), trạng thái kết nối
Mình đã test trên nhiều provider và nhận thấy HolySheep cho tốc độ phản hồi streaming ấn tượng với chi phí chỉ từ $0.42/1M tokens (DeepSeek V3.2) — tiết kiệm đến 85% so với OpenAI!
Code Mẫu Hoàn Chỉnh - Phía Client (JavaScript)
Đây là code mà mình đã dùng trong production, xử lý đầy đủ các trường hợp kết nối, mất kết nối, và retry:
// ws-client.js - Client WebSocket cho HolySheep AI Chat
// Copy nguyên file này và chạy!
class HolySheepWebSocket {
constructor(apiKey) {
this.apiKey = apiKey;
this.baseUrl = 'wss://api.holysheep.ai/v1/realtime/chat';
this.ws = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 5;
this.reconnectDelay = 1000; // Bắt đầu với 1 giây
this.isConnected = false;
this.messageQueue = [];
this.heartbeatInterval = null;
this.onMessageCallback = null;
this.onStatusChangeCallback = null;
}
// Kết nối đến WebSocket server
connect() {
return new Promise((resolve, reject) => {
try {
// Tạo URL với auth token
const url = ${this.baseUrl}?api_key=${this.apiKey};
this.ws = new WebSocket(url);
this.ws.onopen = () => {
console.log('✅ Kết nối WebSocket thành công!');
this.isConnected = true;
this.reconnectAttempts = 0;
this.reconnectDelay = 1000;
this.startHeartbeat();
// Gửi các message đang chờ trong queue
this.flushMessageQueue();
this.updateStatus('connected');
resolve();
};
this.ws.onmessage = (event) => {
const data = JSON.parse(event.data);
this.handleMessage(data);
};
this.ws.onerror = (error) => {
console.error('❌ WebSocket Error:', error);
this.updateStatus('error');
};
this.ws.onclose = (event) => {
console.log(⚠️ WebSocket đóng: Code=${event.code}, Reason=${event.reason});
this.isConnected = false;
this.stopHeartbeat();
this.updateStatus('disconnected');
// Tự động reconnect nếu không phải đóng chủ động
if (event.code !== 1000) {
this.scheduleReconnect();
}
};
} catch (error) {
reject(error);
}
});
}
// Xử lý message từ server
handleMessage(data) {
console.log('📨 Message nhận được:', data);
switch (data.type) {
case 'text':
if (this.onMessageCallback) {
this.onMessageCallback({
content: data.content,
isComplete: data.is_complete || false,
usage: data.usage || null
});
}
break;
case 'streaming_chunk':
// Xử lý streaming - hiển thị từng phần
if (this.onMessageCallback) {
this.onMessageCallback({
content: data.delta,
isComplete: false
});
}
break;
case 'error':
console.error('❌ Server Error:', data.message);
break;
case 'pong':
console.log('💓 Heartbeat acknowledged');
break;
default:
console.log('📦 Unknown message type:', data.type);
}
}
// Gửi tin nhắn đến server
sendMessage(content, options = {}) {
const message = {
type: 'user_message',
content: content,
model: options.model || 'gpt-4.1',
temperature: options.temperature || 0.7,
max_tokens: options.max_tokens || 2048,
stream: options.stream !== false // Mặc định bật streaming
};
if (!this.isConnected) {
console.log('⏳ Chưa kết nối, đưa message vào queue...');
this.messageQueue.push(message);
this.connect(); // Tự động kết nối lại
return;
}
this.ws.send(JSON.stringify(message));
console.log('📤 Đã gửi message:', content.substring(0, 50) + '...');
}
// Gửi typing indicator
sendTyping() {
if (this.isConnected) {
this.ws.send(JSON.stringify({ type: 'typing', status: 'started' }));
}
}
// Heartbeat để giữ kết nối alive
startHeartbeat() {
this.heartbeatInterval = setInterval(() => {
if (this.isConnected && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify({ type: 'ping' }));
}
}, 30000); // Mỗi 30 giây
}
stopHeartbeat() {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.heartbeatInterval = null;
}
}
// Reconnect với exponential backoff
scheduleReconnect() {
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.error('❌ Đã vượt quá số lần reconnect tối đa!');
this.updateStatus('failed');
return;
}
this.reconnectAttempts++;
const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
console.log(🔄 Sẽ reconnect sau ${delay}ms (lần ${this.reconnectAttempts}/${this.maxReconnectAttempts}));
setTimeout(() => {
console.log('🔄 Đang thử kết nối lại...');
this.connect().catch(err => {
console.error('❌ Reconnect thất bại:', err);
});
}, delay);
}
// Gửi các message đang chờ trong queue
flushMessageQueue() {
while (this.messageQueue.length > 0 && this.isConnected) {
const msg = this.messageQueue.shift();
this.ws.send(JSON.stringify(msg));
}
}
updateStatus(status) {
if (this.onStatusChangeCallback) {
this.onStatusChangeCallback(status);
}
}
// Đóng kết nối
disconnect() {
this.stopHeartbeat();
if (this.ws) {
this.ws.close(1000, 'Client initiated close');
}
this.isConnected = false;
}
}
// === CÁCH SỬ DỤNG ===
async function demo() {
// Khởi tạo với API key của bạn
const client = new HolySheepWebSocket('YOUR_HOLYSHEEP_API_KEY');
// Callback khi nhận được phản hồi từ AI
client.onMessageCallback = (data) => {
process.stdout.write(data.content); // In từng phần khi streaming
};
// Callback khi trạng thái thay đổi
client.onStatusChangeCallback = (status) => {
console.log('🔔 Trạng thái:', status);
};
try {
// Kết nối
await client.connect();
// Gửi tin nhắn đầu tiên
client.sendMessage('Xin chào, bạn là AI nào?');
// Đợi một chút rồi gửi tiếp
setTimeout(() => {
client.sendMessage('Giải thích WebSocket là gì?');
}, 3000);
} catch (error) {
console.error('❌ Lỗi:', error);
}
}
// Chạy demo
demo();
Code Mẫu Hoàn Chỉnh - Phía Server (Python)
Nếu bạn muốn tự host WebSocket server để xử lý logic riêng, đây là code Python với FastAPI:
server.py - WebSocket Server với HolySheep AI Integration
Chạy: uvicorn server:app --reload
import asyncio
import json
import logging
from datetime import datetime
from typing import Optional
import httpx
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.middleware.cors import CORSMiddleware
Cấu hình logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="HolySheep AI WebSocket Server")
CORS cho phép frontend truy cập
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
Cấu hình HolySheep API
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HTTP client dùng chung cho async requests
http_client: Optional[httpx.AsyncClient] = None
class ConnectionManager:
"""Quản lý kết nối WebSocket với reconnection support"""
def __init__(self):
self.active_connections: dict[str, WebSocket] = {}
self.client_metadata: dict[str, dict] = {}
async def connect(self, websocket: WebSocket, client_id: str):
await websocket.accept()
self.active_connections[client_id] = websocket
self.client_metadata[client_id] = {
"connected_at": datetime.now().isoformat(),
"message_count": 0,
"last_activity": datetime.now().isoformat()
}
logger.info(f"✅ Client {client_id} kết nối thành công")
def disconnect(self, client_id: str):
if client_id in self.active_connections:
del self.active_connections[client_id]
if client_id in self.client_metadata:
del self.client_metadata[client_id]
logger.info(f"🔌 Client {client_id} đã ngắt kết nối")
async def send_to_client(self, client_id: str, data: dict):
if client_id in self.active_connections:
websocket = self.active_connections[client_id]
try:
await websocket.send_json(data)
except Exception as e:
logger.error(f"Lỗi gửi đến {client_id}: {e}")
self.disconnect(client_id)
async def broadcast(self, data: dict):
"""Gửi message đến tất cả clients"""
for client_id in list(self.active_connections.keys()):
await self.send_to_client(client_id, data)
manager = ConnectionManager()
async def call_holysheep_stream(messages: list, websocket: WebSocket, client_id: str):
"""
Gọi HolySheep AI API với streaming response
Chi phí cực rẻ: DeepSeek V3.2 chỉ $0.42/1M tokens
"""
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # Model giá rẻ, độ trễ thấp
"messages": messages,
"stream": True,
"temperature": 0.7
}
full_response = ""
try:
async with httpx.AsyncClient(timeout=120.0) as client:
async with client.stream(
"POST",
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status_code != 200:
error_body = await response.aread()
logger.error(f"API Error: {response.status_code} - {error_body}")
await manager.send_to_client(client_id, {
"type": "error",
"message": f"API Error: {response.status_code}"
})
return
# Xử lý streaming response
async for line in response.aiter_lines():
if line.startswith("data: "):
data_str = line[6:] # Bỏ "data: "
if data_str == "[DONE]":
break
try:
chunk = json.loads(data_str)
delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
if delta:
full_response += delta
# Gửi từng chunk đến client
await manager.send_to_client(client_id, {
"type": "streaming_chunk",
"delta": delta
})
except json.JSONDecodeError:
continue
# Gửi message hoàn chỉnh
await manager.send_to_client(client_id, {
"type": "text",
"content": full_response,
"is_complete": True,
"usage": {
"model": "deepseek-v3.2",
"timestamp": datetime.now().isoformat()
}
})
logger.info(f"✅ Hoàn thành response cho {client_id}: {len(full_response)} chars")
except httpx.TimeoutException:
logger.error(f"⏱️ Timeout khi gọi HolySheep API cho {client_id}")
await manager.send_to_client(client_id, {
"type": "error",
"message": "Request timeout - vui lòng thử lại"
})
except Exception as e:
logger.error(f"❌ Lỗi khi gọi HolySheep: {e}")
await manager.send_to_client(client_id, {
"type": "error",
"message": str(e)
})
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket, client_id: str = "default"):
"""Endpoint WebSocket cho real-time chat"""
# Xác thực API key từ query params
api_key = websocket.query_params.get("api_key")
if not api_key or api_key != HOLYSHEEP_API_KEY:
await websocket.close(code=4001, reason="Invalid API key")
return
await manager.connect(websocket, client_id)
# Lịch sử hội thoại cho context
conversation_history = []
try:
while True:
# Nhận message từ client
data = await websocket.receive_text()
try:
message = json.loads(data)
except json.JSONDecodeError:
await manager.send_to_client(client_id, {
"type": "error",
"message": "Invalid JSON format"
})
continue
# Cập nhật metadata
if client_id in manager.client_metadata:
manager.client_metadata[client_id]["message_count"] += 1
manager.client_metadata[client_id]["last_activity"] = datetime.now().isoformat()
# Xử lý các loại message
msg_type = message.get("type")
if msg_type == "user_message":
user_content = message.get("content", "")
# Thêm vào lịch sử
conversation_history.append({
"role": "user",
"content": user_content
})
# Gửi typing indicator
await manager.send_to_client(client_id, {
"type": "typing",
"status": "started"
})
# Gọi HolySheep AI
await call_holysheep_stream(conversation_history, websocket, client_id)
# Gửi typing stopped
await manager.send_to_client(client_id, {
"type": "typing",
"status": "stopped"
})
elif msg_type == "ping":
# Heartbeat response
await manager.send_to_client(client_id, {
"type": "pong",
"timestamp": datetime.now().isoformat()
})
elif msg_type == "clear_history":
# Xóa lịch sử hội thoại
conversation_history = []
await manager.send_to_client(client_id, {
"type": "history_cleared"
})
except WebSocketDisconnect:
manager.disconnect(client_id)
except Exception as e:
logger.error(f"❌ Lỗi WebSocket cho {client_id}: {e}")
manager.disconnect(client_id)
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"active_connections": len(manager.active_connections),
"timestamp": datetime.now().isoformat()
}
@app