WebSocket สำหรับ AI Streaming: คู่มือ Full-Duplex Communication ฉบับสมบูรณ์

การสร้างระบบสนทนา AI แบบ Real-time ในปัจจุบันต้องการการสื่อสารแบบ Full-Duplex ที่รองรับการรับ-ส่งข้อมูลพร้อมกัน WebSocket คือเทคโนโลยีที่ทำให้สิ่งนี้เป็นไปได้ โดยเฉพาะเมื่อใช้ร่วมกับ AI API ที่มี Latency ต่ำ บทความนี้จะสอนทุกขั้นตอนตั้งแต่พื้นฐานจนถึง Production-ready Implementation

สรุปคำตอบ: ทำไมต้องใช้ WebSocket สำหรับ AI Streaming

• Latency ต่ำกว่า: รับ Token ทีละตัวเมื่อ AI สร้าง ไม่ต้องรอจนเสร็จทั้งหมด
• ประสบการณ์ผู้ใช้ดีกว่า: เห็นคำตอบเกิดขึ้นทีละคำ สไตล์ ChatGPT/Claude
• ประหยัด Bandwidth: HTTP Long-Polling ใช้ Header ซ้ำผ่าน WebSocket Upgrade
• เหมาะกับ Multi-turn Conversation: รองรับการสนทนาต่อเนื่องหลายรอบ

WebSocket vs Server-Sent Events (SSE) vs HTTP Polling

เปรียบเทียบราคาและ Specs: HolySheep AI vs Official API vs คู่แข่ง

หมายเหตุ: HolySheep AI มีอัตราแลกเปลี่ยน ¥1=$1 ประหยัดได้มากกว่า 85% เมื่อเทียบกับ Official API พร้อมรับเครดิตฟรีเมื่อลงทะเบียน

พื้นฐาน WebSocket: Concept และการทำงาน

WebSocket คือ Protocol ที่สร้าง Connection แบบ Persistent ระหว่าง Client กับ Server ต่างจาก HTTP ที่เป็น Request-Response แบบ Stateless

WebSocket Handshake Process

┌─────────────────────────────────────────────────────────┐
│                    HTTP Upgrade Request                   │
├─────────────────────────────────────────────────────────┤
│ GET /v1/chat/completions HTTP/1.1                        │
│ Host: api.holysheep.ai                                   │
│ Upgrade: websocket                                       │
│ Connection: Upgrade                                       │
│ Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==             │
│ Sec-WebSocket-Version: 13                                │
│ Authorization: Bearer YOUR_HOLYSHEEP_API_KEY            │
└─────────────────────────────────────────────────────────┘
                         ↓
┌─────────────────────────────────────────────────────────┐
│                  HTTP 101 Switching Protocols            │
├─────────────────────────────────────────────────────────┤
│ Upgrade: websocket                                       │
│ Connection: Upgrade                                      │
│ Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=     │
└─────────────────────────────────────────────────────────┘
                         ↓
┌─────────────────────────────────────────────────────────┐
│              ✅ WebSocket Connection Established          │
│         พร้อมส่งข้อมูลแบบ Full-Duplex ทั้งสองทาง        │
└─────────────────────────────────────────────────────────┘

Implementation ด้วย Python (FastAPI + WebSocket)

จากประสบการณ์ในการสร้าง Production System หลายตัว ผมพบว่า FastAPI เป็น Framework ที่เหมาะที่สุดสำหรับงานนี้เนื่องจาก Built-in WebSocket Support และ Async Capability

# server.py - FastAPI WebSocket Server สำหรับ AI Streaming
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import asyncio
import json
import httpx

app = FastAPI(title="AI WebSocket Streaming Server")

กำหนด Configuration
HOLYSHEEP_CONFIG = {
    "base_url": "https://api.holysheep.ai/v1",  # ✅ Base URL ที่ถูกต้อง
    "api_key": "YOUR_HOLYSHEEP_API_KEY",  # แทนที่ด้วย API Key จริง
    "model": "gpt-4.1"  # หรือ claude-3-5-sonnet-20241022, gemini-2.0-flash-exp
}

class AIStreamHandler:
    def __init__(self, config: dict):
        self.config = config
        self.client = httpx.AsyncClient(timeout=60.0)
    
    async def stream_chat(self, messages: list, websocket: WebSocket):
        """Stream คำตอบจาก AI ไปยัง WebSocket Client"""
        
        headers = {
            "Authorization": f"Bearer {self.config['api_key']}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": self.config["model"],
            "messages": messages,
            "stream": True  # ✅ สำคัญ: เปิด Streaming Mode
        }
        
        async with self.client.stream(
            "POST",
            f"{self.config['base_url']}/chat/completions",
            headers=headers,
            json=payload
        ) as response:
            
            if response.status_code != 200:
                error_text = await response.aread()
                await websocket.send_json({
                    "error": f"API Error: {response.status_code}",
                    "detail": error_text.decode()
                })
                return
            
            # Parse Server-Sent Events (SSE) from stream
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    data = line[6:]  # ตัด "data: " ออก
                    
                    if data == "[DONE]":
                        break
                    
                    try:
                        chunk = json.loads(data)
                        # ส่ง Token ทีละตัวไปยัง Client
                        if "choices" in chunk and len(chunk["choices"]) > 0:
                            delta = chunk["choices"][0].get("delta", {})
                            if "content" in delta:
                                await websocket.send_json({
                                    "type": "token",
                                    "content": delta["content"]
                                })
                    except json.JSONDecodeError:
                        continue

สร้าง Singleton Instance
ai_handler = AIStreamHandler(HOLYSHEEP_CONFIG)

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    """WebSocket Endpoint สำหรับ AI Chat"""
    await websocket.accept()
    
    # เก็บ History สำหรับ Multi-turn Conversation
    conversation_history = []
    
    try:
        while True:
            # รอข้อความจาก Client
            data = await websocket.receive_text()
            message_data = json.loads(data)
            
            user_message = message_data.get("content", "")
            system_prompt = message_data.get("system", "คุณเป็นผู้ช่วย AI ที่เป็นประโยชน์")
            
            # เพิ่ม System และ User Message เข้า History
            conversation_history.append({"role": "system", "content": system_prompt})
            conversation_history.append({"role": "user", "content": user_message})
            
            # แจ้ง Client ว่าเริ่ม Stream
            await websocket.send_json({"type": "start"})
            
            # Stream คำตอบ
            await ai_handler.stream_chat(conversation_history, websocket)
            
            # เพิ่ม Assistant Response เข้า History
            # (ใน Production ควรดึงข้อความที่ส่งจริงมาเก็บ)
            conversation_history.append({"role": "assistant", "content": ""})
            
            # แจ้ง Client ว่า Stream เสร็จ
            await websocket.send_json({"type": "end"})
            
    except WebSocketDisconnect:
        print("Client disconnected")
    except Exception as e:
        await websocket.send_json({"type": "error", "message": str(e)})

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Client Implementation: JavaScript WebSocket

<!-- index.html -->
<!DOCTYPE html>
<html lang="th">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>AI Chat Streaming Demo</title>
    <style>
        body { font-family: 'Sarabun', sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
        #chat-container { border: 1px solid #ddd; height: 400px; overflow-y: auto; padding: 15px; margin-bottom: 10px; }
        .user-message { color: #2196F3; margin-bottom: 10px; }
        .ai-message { color: #4CAF50; margin-bottom: 10px; }
        #typing-indicator { display: none; color: #999; font-style: italic; }
        input, button { padding: 10px; margin: 5px 0; }
        #message-input { width: 70%; }
        #send-btn { width: 25%; }
    </style>
</head>
<body>
    <h1>AI Streaming Chat Demo</h1>
    
    <div id="chat-container"></div>
    <div id="typing-indicator">AI กำลังพิมพ์...</div>
    
    <input type="text" id="message-input" placeholder="พิมพ์ข้อความของคุณ..." />
    <button id="send-btn" onclick="sendMessage()">ส่ง</button>
    
    <script>
        class AIChatClient {
            constructor(wsUrl = "ws://localhost:8000/ws/chat") {
                this.wsUrl = wsUrl;
                this.ws = null;
                this.reconnectAttempts = 0;
                this.maxReconnectAttempts = 5;
                this.currentAiElement = null;
            }
            
            connect() {
                return new Promise((resolve, reject) => {
                    this.ws = new WebSocket(this.wsUrl);
                    
                    this.ws.onopen = () => {
                        console.log("✅ WebSocket Connected");
                        this.reconnectAttempts = 0;
                        resolve();
                    };
                    
                    this.ws.onmessage = (event) => {
                        const data = JSON.parse(event.data);
                        this.handleMessage(data);
                    };
                    
                    this.ws.onerror = (error) => {
                        console.error("❌ WebSocket Error:", error);
                        reject(error);
                    };
                    
                    this.ws.onclose = () => {
                        console.log("⚠️ WebSocket Closed");
                        this.attemptReconnect();
                    };
                });
            }
            
            handleMessage(data) {
                switch(data.type) {
                    case "start":
                        // เริ่มแสดงผล AI Message
                        this.currentAiElement = document.createElement("div");
                        this.currentAiElement.className = "ai-message";
                        this.currentAiElement.textContent = "";
                        document.getElementById("chat-container").appendChild(this.currentAiElement);
                        document.getElementById("typing-indicator").style.display = "block";
                        break;
                        
                    case "token":
                        // เพิ่ม Token ทีละตัว
                        if (this.currentAiElement) {
                            this.currentAiElement.textContent += data.content;
                            // Auto-scroll
                            document.getElementById("chat-container").scrollTop = 
                                document.getElementById("chat-container").scrollHeight;
                        }
                        break;
                        
                    case "end":
                        // จบการแสดงผล
                        document.getElementById("typing-indicator").style.display = "none";
                        this.currentAiElement = null;
                        break;
                        
                    case "error":
                        alert("❌ Error: " + data.message);
                        break;
                }
            }
            
            async sendMessage(content, systemPrompt = "คุณเป็นผู้ช่วย AI ที่เป็นประโยชน์") {
                if (this.ws && this.ws.readyState === WebSocket.OPEN) {
                    // แสดง User Message
                    const userDiv = document.createElement("div");
                    userDiv.className = "user-message";
                    userDiv.textContent = "คุณ: " + content;
                    document.getElementById("chat-container").appendChild(userDiv);
                    
                    // ส่งไปยัง Server
                    this.ws.send(JSON.stringify({
                        content: content,
                        system: systemPrompt
                    }));
                }
            }
            
            attemptReconnect() {
                if (this.reconnectAttempts < this.maxReconnectAttempts) {
                    this.reconnectAttempts++;
                    console.log(`🔄 Reconnecting... Attempt ${this
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
Gemini 2.5 Flash Thinking API คู่มือย้ายระบบสู่ HolySheep AI
Gradio AI Demo Deployment: HuggingFace Spaces และการเชื่อมต่
คู่มือย้ายระบบ Prompt Version Management และ A/B Testing Fra