ในยุคที่การสื่อสารข้ามภาษาต้องเป็นไปอย่างรวดเร็วและลื่นไหล การสร้างระบบแปลภาษาแบบเรียลไทม์ผ่าน WebSocket Streaming ถือเป็นทักษะที่ Developer ทุกคนควรมี ในบทความนี้ผมจะพาคุณสร้างระบบ Multilingual Real-time Translation ตั้งแต่เริ่มต้นจนใช้งานจริง พร้อมวิธีประหยัดค่าใช้จ่ายได้ถึง 85%+ ผ่าน HolySheep AI
ทำไมต้องใช้ WebSocket Streaming สำหรับการแปลภาษา
การแปลภาษาแบบดั้งเดิมที่รอให้พิมพ์เสร็จก่อนแล้วค่อยส่ง Request จะมีความล่าช้าและประสบการณ์ผู้ใช้ไม่ดี WebSocket Streaming ช่วยให้ได้ผลลัพธ์การแปลทีละส่วน เหมาะสำหรับ:
- แอปพลิเคชันสนทนาข้ามภาษาแบบเรียลไทม์
- ระบบ Live Caption และ Subtitle
- เว็บไซต์ E-commerce ที่รองรับหลายภาษา
- เครื่องมือเรียนภาษาที่ต้องการ Feedback แบบทันที
- ระบบ Support ลูกค้าข้ามชาติ
การเปรียบเทียบต้นทุน AI API ปี 2026
ก่อนเริ่มพัฒนา มาดูค่าใช้จ่ายจริงของแต่ละ Provider สำหรับ 10 ล้าน Tokens ต่อเดือน:
┌─────────────────────────────────────────────────────────────────────┐
│ เปรียบเทียบต้นทุน 10M Tokens/เดือน │
├─────────────────────────┬──────────────┬─────────────────────────────┤
│ Provider │ ราคา/MTok │ ค่าใช้จ่าย 10M Tokens │
├─────────────────────────┼──────────────┼─────────────────────────────┤
│ Claude Sonnet 4.5 │ $15.00 │ $150.00 │
│ GPT-4.1 │ $8.00 │ $80.00 │
│ Gemini 2.5 Flash │ $2.50 │ $25.00 │
│ DeepSeek V3.2 │ $0.42 │ $4.20 │
├─────────────────────────┼──────────────┼─────────────────────────────┤
│ HolySheep AI (DeepSeek) │ ¥3.00 ($3.00)│ $30.00 (ประหยัด 85%+) │
└─────────────────────────┴──────────────┴─────────────────────────────┘
* อัตราแลกเปลี่ยน HolySheep: ¥1 = $1 (ดีกว่า Official 85%+)
* Gemini 2.5 Flash เหมาะกับงานแปลทั่วไป
* DeepSeek V3.2 เหมาะกับงานแปลเชิงเทคนิคที่ต้องการความแม่นยำสูง
ติดตั้งและตั้งค่า Environment
# สร้าง Project ใหม่
mkdir multilingual-translation-ws
cd multilingual-translation-ws
สร้าง Virtual Environment
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
ติดตั้ง Dependencies
pip install websockets>=14.0
pip install websockets[Speedups]>=14.0
pip install python-dotenv>=1.0.0
สร้างไฟล์ .env
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
EOF
echo "Setup completed!"Client-side: WebSocket Client สำหรับ Translation Streaming
# translation_client.py
import asyncio
import json
from websockets.client import connect
from typing import AsyncGenerator, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TranslationWebSocketClient:
"""
WebSocket Client สำหรับรับ Translation Stream แบบเรียลไทม์
รองรับการแปลหลายภาษา: ไทย, อังกฤษ, จีน, ญี่ปุ่น, เกาหลี, เวียดนาม
"""
SUPPORTED_LANGUAGES = {
"th": "ไทย",
"en": "อังกฤษ",
"zh": "จีน",
"ja": "ญี่ปุ่น",
"ko": "เกาหลี",
"vi": "เวียดนาม",
"fr": "ฝรั่งเศส",
"de": "เยอรมัน",
"es": "สเปน"
}
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.websocket_url = base_url.replace("https://", "wss://").replace("http://", "ws://")
async def translate_stream(
self,
text: str,
source_lang: str = "en",
target_lang: str = "th",
model: str = "gpt-4.1"
) -> AsyncGenerator[str, None]:
"""
รับผลการแปลแบบ Streaming
Args:
text: ข้อความต้นฉบับ
source_lang: ภาษาต้นทาง (ISO 639-1)
target_lang: ภาษาเป้าหมาย (ISO 639-1)
model: โมเดลที่ใช้ (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
Yields:
ข้อความที่แปลแล้วทีละส่วน
"""
ws_url = f"{self.websocket_url}/chat/completions"
# สร้าง System Prompt สำหรับการแปล
system_prompt = f"""คุณเป็นนักแปลมืออาชีพ
แปลข้อความจากภาษา{self.SUPPORTED_LANGUAGES.get(source_lang, source_lang)}เป็นภาษา{self.SUPPORTED_LANGUAGES.get(target_lang, target_lang)}
แปลให้เป็นธรรมชาติ กรอกเฉพาะข้อความที่แปลเท่านั้น ไม่ต้องมีคำอธิบาย"""
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": text}
],
"stream": True,
"stream_options": {"include_usage": True}
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
async with connect(ws_url, extra_headers=headers) as websocket:
await websocket.send(json.dumps(payload))
accumulated_text = ""
while True:
try:
response = await websocket.recv()
data = json.loads(response)
# ตรวจสอบว่าเป็น Stream Chunk
if "choices" in data and len(data["choices"]) > 0:
delta = data["choices"][0].get("delta", {})
if "content" in delta:
chunk = delta["content"]
accumulated_text += chunk
yield chunk
# ตรวจสอบว่าจบการ streaming แล้วหรือยัง
finish_reason = data["choices"][0].get("finish_reason")
if finish_reason:
break
# ตรวจสอบ usage stats (ถ้ามี)
if "usage" in data:
logger.info(f"Token usage: {data['usage']}")
except json.JSONDecodeError:
continue
except websockets.exceptions.ConnectionClosed:
logger.info("Connection closed by server")
break
logger.info(f"Translation completed. Total: {accumulated_text}")
except Exception as e:
logger.error(f"WebSocket error: {e}")
raise
async def translate_full(
self,
text: str,
source_lang: str = "en",
target_lang: str = "th",
model: str = "deepseek-v3.2"
) -> str:
"""
แปลข้อความเต็ม (รอจนครบ) แล้วคืนค่าทั้งหมด
"""
chunks = []
async for chunk in self.translate_stream(text, source_lang, target_lang, model):
chunks.append(chunk)
return "".join(chunks)
ทดสอบการใช้งาน
async def main():
client = TranslationWebSocketClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
test_text = "Hello, how are you today? I would like to discuss about the new project requirements."
print(f"ต้นฉบับ (EN): {test_text}")
print(f"กำลังแปลเป็น TH (Streaming)...\n")
print("ผลการแปล: ", end="", flush=True)
async for chunk in client.translate_stream(
text=test_text,
source_lang="en",
target_lang="th",
model="deepseek-v3.2"
):
print(chunk, end="", flush=True)
print("\n\n✅ การทดสอบเสร็จสมบูรณ์")
if __name__ == "__main__":
asyncio.run(main())Server-side: FastAPI WebSocket Server พร้อม Connection Pooling
# translation_server.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Dict, Optional, List
import asyncio
import json
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="Multilingual Translation WebSocket Server")
CORS Configuration
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class TranslationRequest(BaseModel):
text: str
source_lang: str = "en"
target_lang: str = "th"
model: str = "deepseek-v3.2"
class ConnectionManager:
"""จัดการ WebSocket Connections พร้อม Room Support"""
def __init__(self):
# connection_id -> WebSocket
self.active_connections: Dict[str, WebSocket] = {}
# room_id -> Set[connection_id]
self.rooms: Dict[str, set] = {}
# connection_id -> metadata
self.connection_metadata: Dict[str, dict] = {}
# Rate limiting
self.message_counts: Dict[str, List[datetime]] = {}
self.RATE_LIMIT = 100 # ข้อความต่อนาที
self.RATE_WINDOW = 60 # วินาที
async def connect(self, websocket: WebSocket, connection_id: str, metadata: dict = None):
await websocket.accept()
self.active_connections[connection_id] = websocket
self.connection_metadata[connection_id] = metadata or {}
self.message_counts[connection_id] = []
logger.info(f"Client connected: {connection_id}")
def disconnect(self, connection_id: str):
if connection_id in self.active_connections:
del self.active_connections[connection_id]
if connection_id in self.connection_metadata:
del self.connection_metadata[connection_id]
if connection_id in self.message_counts:
del self.message_counts[connection_id]
# ลบออกจากทุก rooms
for room_id in list(self.rooms.keys()):
if connection_id in self.rooms[room_id]:
self.rooms[room_id].discard(connection_id)
if not self.rooms[room_id]:
del self.rooms[room_id]
logger.info(f"Client disconnected: {connection_id}")
def check_rate_limit(self, connection_id: str) -> bool:
"""ตรวจสอบ Rate Limit"""
now = datetime.now()
cutoff = now.timestamp() - self.RATE_WINDOW
# ลบข้อความเก่าออก
self.message_counts[connection_id] = [
ts for ts in self.message_counts.get(connection_id, [])
if ts.timestamp() > cutoff
]
if len(self.message_counts[connection_id]) >= self.RATE_LIMIT:
return False
self.message_counts[connection_id].append(now)
return True
async def send_personal_message(self, message: str, connection_id: str):
if connection_id in self.active_connections:
await self.active_connections[connection_id].send_text(message)
async def broadcast_to_room(self, message: str, room_id: str, sender_id: str):
if room_id in self.rooms:
for connection_id in self.rooms[room_id]:
if connection_id != sender_id: # ไม่ส่งให้ผู้ส่ง
await self.send_personal_message(message, connection_id)
def join_room(self, connection_id: str, room_id: str):
if room_id not in self.rooms:
self.rooms[room_id] = set()
self.rooms[room_id].add(connection_id)
def leave_room(self, connection_id: str, room_id: str):
if room_id in self.rooms:
self.rooms[room_id].discard(connection_id)
manager = ConnectionManager()
@app.get("/")
async def root():
return {"status": "ok", "service": "Multilingual Translation WebSocket"}
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"active_connections": len(manager.active_connections),
"active_rooms": len(manager.rooms)
}
@app.get("/stats")
async def get_stats():
return {
"total_connections": len(manager.active_connections),
"rooms": {
room_id: len(members)
for room_id, members in manager.rooms.items()
}
}
@app.websocket("/ws/translate/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await manager.connect(websocket, client_id, {"connected_at": datetime.now().isoformat()})
try:
while True:
# รับข้อความจาก Client
data = await websocket.receive_text()
# ตรวจสอบ Rate Limit
if not manager.check_rate_limit(client_id):
await websocket.send_json({
"type": "error",
"error": "Rate limit exceeded. Maximum 100 messages/minute."
})
continue
try:
message = json.loads(data)
msg_type = message.get("type", "translate")
if msg_type == "translate":
# ส่งต่อไปยัง Translation Service
await websocket.send_json({
"type": "processing",
"original": message.get("text", ""),
"status": "translating"
})
# TODO: ส่งไปยัง AI API จริง
# ใน production จะใช้ HolySheep AI ผ่าน WebSocket
await websocket.send_json({
"type": "translated",
"original": message.get("text", ""),
"translated": f"[Translated]: {message.get('text', '')}",
"source_lang": message.get("source_lang", "en"),
"target_lang": message.get("target_lang", "th")
})
elif msg_type == "join_room":
room_id = message.get("room_id")
manager.join_room(client_id, room_id)
await websocket.send_json({
"type": "system",
"message": f"Joined room: {room_id}"
})
elif msg_type == "leave_room":
room_id = message.get("room_id")
manager.leave_room(client_id, room_id)
await websocket.send_json({
"type": "system",
"message": f"Left room: {room_id}"
})
elif msg_type == "broadcast":
room_id = message.get("room_id")
translation = message.get("text", "")
await manager.broadcast_to_room(json.dumps({
"type": "message",
"sender": client_id,
"text": translation,
"timestamp": datetime.now().isoformat()
}), room_id, client_id)
elif msg_type == "ping":
await websocket.send_json({"type": "pong"})
except json.JSONDecodeError:
await websocket.send_json({
"type": "error",
"error": "Invalid JSON format"
})
except WebSocketDisconnect:
manager.disconnect(client_id)
logger.info(f"Client {client_id} disconnected gracefully")
ทดสอบ WebSocket Client
import websockets
async def test_websocket():
uri = "ws://localhost:8000/ws/translate/test_client"
async with websockets.connect(uri) as websocket:
# Test translation
await websocket.send(json.dumps({
"type": "translate",
"text": "Hello, World!",
"source_lang": "en",
"target_lang": "th"
}))
while True:
response = await websocket.recv()
data = json.loads(response)
print(f"Received: {data}")
if data.get("type") in ["translated", "error"]:
break
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)Frontend: React Component สำหรับ Translation UI
// TranslationInterface.tsx
import React, { useState, useRef, useEffect, useCallback } from 'react';
interface Message {
id: string;
text: string;
translated: string;
sourceLang: string;
targetLang: string;
timestamp: Date;
isStreaming?: boolean;
}
interface TranslationInterfaceProps {
apiKey: string;
baseUrl?: string;
}
const LANGUAGES = [
{ code: 'en', name: '🇬🇧 อังกฤษ' },
{ code: 'th', name: '🇹🇭 ไทย' },
{ code: 'zh', name: '🇨🇳 จีน' },
{ code: 'ja', name: '🇯🇵 ญี่ปุ่น' },
{ code: 'ko', name: '🇰🇷 เกาหลี' },
{ code: 'vi', name: '🇻🇳 เวียดนาม' },
{ code: 'fr', name: '🇫🇷 ฝรั่งเศส' },
{ code: 'de', name: '🇩🇪 เยอรมัน' },
];
const MODELS = [
{ id: 'deepseek-v3.2', name: 'DeepSeek V3.2', price: '$0.42/MTok' },
{ id: 'gemini-2.5-flash', name: 'Gemini 2.5 Flash', price: '$2.50/MTok' },
{ id: 'gpt-4.1', name: 'GPT-4.1', price: '$8/MTok' },
{ id: 'claude-sonnet-4.5', name: 'Claude Sonnet 4.5', price: '$15/MTok' },
];
export default function TranslationInterface({
apiKey,
baseUrl = 'https://api.holysheep.ai/v1'
}: TranslationInterfaceProps) {
const [messages, setMessages] = useState([]);
const [inputText, setInputText] = useState('');
const [sourceLang, setSourceLang] = useState('en');
const [targetLang, setTargetLang] = useState('th');
const [selectedModel, setSelectedModel] = useState('deepseek-v3.2');
const [isConnected, setIsConnected] = useState(false);
const [isTranslating, setIsTranslating] = useState(false);
const websocketRef = useRef(null);
const messagesEndRef = useRef(null);
const scrollToBottom = useCallback(() => {
messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
}, []);
useEffect(() => {
scrollToBottom();
}, [messages, scrollToBottom]);
// Connect to WebSocket
useEffect(() => {
if (!apiKey) return;
const wsUrl = baseUrl
.replace('https://', 'wss://')
.replace('http://', 'ws://') + '/chat/completions';
const ws = new WebSocket(wsUrl);
websocketRef.current = ws;
ws.onopen = () => {
console.log('WebSocket Connected');
setIsConnected(true);
};
ws.onclose = () => {
console.log('WebSocket Disconnected');
setIsConnected(false);
};
ws.onerror = (error) => {
console.error('WebSocket Error:', error);
};
return () => {
ws.close();
};
}, [apiKey, baseUrl]);
const translateText = async () => {
if (!inputText.trim() || isTranslating) return;
const messageId = Date.now().toString();
const newMessage: Message = {
id: messageId,
text: inputText,
translated: '',
sourceLang,
targetLang,
timestamp: new Date(),
isStreaming: true,
};
setMessages(prev => [...prev, newMessage]);
setInputText('');
setIsTranslating(true);
try {
// ส่ง request ไปยัง WebSocket
const response = await fetch(${baseUrl}/chat/completions, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': Bearer ${apiKey},
},
body: JSON.stringify({
model: selectedModel,
messages: [
{
role: 'system',
content: แปลข้อความจากภาษา${LANGUAGES.find(l => l.code === sourceLang)?.name}เป็นภาษา${LANGUAGES.find(l => l.code === targetLang)?.name} ตอบเฉพาะข้อความที่แปลเท่านั้น
},
{
role: 'user',
content: inputText
}
],
stream: true,
stream_options: { include_usage: true }
}),
});
if (!response.ok) {
throw new Error(HTTP error! status: ${response.status});
}
const reader = response.body?.getReader();
const decoder = new TextDecoder();
let fullTranslation = '';
while (reader) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value);
const lines = chunk.split('\n');
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = line.slice(6);
if (data === '[DONE]') continue;
try {
const parsed = JSON.parse(data);
if (parsed.choices?.[0]?.delta?.content) {
const content = parsed.choices[0].delta.content;
fullTranslation += content;
// Update message with streaming content
setMessages(prev => prev.map(msg =>
msg.id === messageId
? { ...msg, translated: fullTranslation }
: msg
));
}
// Check for usage stats
if (parsed.usage) {
console.log('Token usage:', parsed.usage);
}
} catch (e) {
// Skip invalid JSON
}
}
}
}
// Mark streaming as complete
setMessages(prev => prev.map(msg =>
msg.id === messageId
? { ...msg, translated: fullTranslation, isStreaming: false }
: msg
));
} catch (error) {
console.error('Translation error:', error);
setMessages(prev => prev.map(msg =>
msg.id === messageId
? { ...msg, translated: '❌ เกิดข้อผิดพลาดในการแปล', isStreaming: false }
: msg
));
} finally {
setIsTranslating(false);
}
};
const swapLanguages = () => {
const temp = sourceLang;
setSourceLang(targetLang);
setTargetLang(temp);
};
return (
🌐 ระบบแปลภาษาแบบเรียลไทม์
status-dot ${isConnected ? 'connected' : 'disconnected'}}>
{isConnected ? 'เชื่อมต่อแล้ว' : 'ไม่ได้เชื่อมต่อ'}
{messages.length === 0 && (
เริ่มพิมพ์ข้อความเพื่อแปลภาษา...
)}
{messages.map(msg => (
ต้นฉบับ ({msg.sourceLang}):
{msg.text}
แปลแล้ว ({msg.targetLang}):
{msg.translated}
{msg.isStreaming && |}
))}