ในฐานะวิศวกรที่พัฒนาแชทบอท AI มาหลายปี ผมเคยเจอปัญหา latency สูงจนผู้ใช้งานบ่นว่า "รอนานมาก" กว่าจะได้คำตอบ นี่คือที่มาของบทความนี้ ผมจะสอนวิธี implement Streaming ด้วย Server-Sent Events (SSE) อย่างละเอียด พร้อม benchmark จริงจาก สมัครที่นี่ ซึ่งให้บริการ API ที่ราคาประหยัดกว่า OpenAI ถึง 85%+ ราคาเพียง ¥1=$1 และมีความหน่วงต่ำกว่า 50ms

SSE คืออะไร และทำไมต้องใช้ Streaming

Server-Sent Events เป็นเทคโนโลยีที่ช่วยให้ server ส่งข้อมูลไปยัง browser ได้แบบ real-time ผ่าน HTTP connection เดียว เมื่อเทียบกับการรอ response ทั้งหมดแล้วค่อยแสดงผล Streaming ช่วยให้ผู้ใช้เห็นคำตอบเกิดขึ้นทีละส่วน ลด perceived latency ได้อย่างมาก

จากการทดสอบของผม การใช้ Streaming ช่วยลดเวลารอรับรู้ของผู้ใช้ลงถึง 60-70% เมื่อเทียบกับ traditional request-response เพราะผู้ใช้เริ่มเห็นตัวอักษรภายใน 200-500ms แรก แทนที่จะรอ 3-5 วินาทีแบบเดิม

สถาปัตยกรรม Streaming Request

┌─────────────┐     SSE Stream      ┌──────────────────┐
│   Browser   │ ◄────────────────── │   Backend API    │
│             │                     │                  │
│  EventSource│     POST Request    │  openai.chat.    │
│   .onmessage│ ──────────────────► │  completions     │
└─────────────┘                     │  stream: true    │
                                     └────────┬─────────┘
                                              │
                                              ▼
                                     ┌──────────────────┐
                                     │  HolySheep AI    │
                                     │  api.holysheep   │
                                     │  .ai/v1          │
                                     └──────────────────┘

สถาปัตยกรรมพื้นฐานประกอบด้วย 3 ส่วนหลัก: Browser รับ stream ผ่าน EventSource หรือ Fetch API, Backend รับ request และส่งต่อไปยัง AI provider โดยใช้ chunked transfer encoding, และ AI Provider (ในที่นี้คือ HolySheep AI) ประมวลผลและส่ง token กลับมาทีละส่วน

Backend Implementation (Python FastAPI)

ผมจะเริ่มจาก backend ก่อน เพราะต้องจัดการ SSE headers และ connection lifecycle อย่างถูกต้อง

import FastAPI
import uvicorn
import httpx
import asyncio
from typing import AsyncGenerator
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

API_KEY = "YOUR_HOLYSHEEP_API_KEY"
BASE_URL = "https://api.holysheep.ai/v1"

@app.post("/v1/chat/stream")
async def chat_stream(messages: list[dict]):
    """
    Streaming endpoint ที่ proxy ไปยัง HolySheep AI
    ราคา GPT-4.1: $8/MTok, DeepSeek V3.2: $0.42/MTok
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    
    payload = {
        "model": "gpt-4.1",
        "messages": messages,
        "stream": True,
        "max_tokens": 2048,
        "temperature": 0.7,
    }
    
    async def event_generator() -> AsyncGenerator[str, None]:
        async with httpx.AsyncClient(timeout=120.0) as client:
            async with client.stream(
                "POST",
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
            ) as response:
                # ตรวจสอบ HTTP status
                if response.status_code != 200:
                    error_text = await response.aread()
                    yield f"data: Error: {error_text}\n\n"
                    return
                
                # parse SSE stream
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]  # ตัด "data: " ออก
                        if data == "[DONE]":
                            break
                        yield f"data: {data}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # ปิด nginx buffering
        },
    )

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

จุดสำคัญในโค้ดนี้คือการใช้ httpx.AsyncClient กับ aiter_lines() เพื่อ parse SSE format ทีละบรรทัด การตรวจสอบ status code 200 ก่อนประมวลผล และการ yield ข้อมูลพร้อม prefix "data: " ตามมาตรฐาน SSE

Frontend Implementation (JavaScript/TypeScript)

Frontend มี 2 วิธีหลักในการรับ streaming: EventSource สำหรับ GET requests และ Fetch API สำหรับ POST requests ผมแนะนำ Fetch API เพราะรองรับ POST method และสามารถส่ง request body ได้

interface Message {
  role: 'user' | 'assistant';
  content: string;
}

interface StreamChunk {
  id: string;
  choices: Array<{
    delta: { content?: string };
    finish_reason: string | null;
  }>;
}

class AIStreamClient {
  private baseUrl: string;
  private apiKey: string;

  constructor(baseUrl: string, apiKey: string) {
    this.baseUrl = baseUrl;
    this.apiKey = apiKey;
  }

  async *streamChat(
    messages: Message[],
    onToken: (token: string) => void,
    onError: (error: Error) => void
  ): AsyncGenerator<string> {
    try {
      const response = await fetch(${this.baseUrl}/v1/chat/stream, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': Bearer ${this.apiKey},
        },
        body: JSON.stringify({ messages }),
      });

      if (!response.ok) {
        const error = await response.text();
        throw new Error(HTTP ${response.status}: ${error});
      }

      const reader = response.body?.getReader();
      if (!reader) throw new Error('No response body');

      const decoder = new TextDecoder();
      let buffer = '';

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = line.slice(6);
            if (data === '[DONE]') return;

            try {
              const chunk: StreamChunk = JSON.parse(data);
              const content = chunk.choices[0]?.delta?.content;
              if (content) {
                onToken(content);
                yield content;
              }
            } catch (e) {
              // Skip malformed JSON
            }
          }
        }
      }
    } catch (error) {
      onError(error as Error);
    }
  }
}

// ตัวอย่างการใช้งาน
const client = new AIStreamClient(
  'https://your-backend.com',
  'user-api-key'
);

const messages: Message[] = [
  { role: 'user', content: 'อธิบายเรื่อง quantum computing' }
];

let fullResponse = '';

for await (const token of client.streamChat(
  messages,
  (token) => {
    fullResponse += token;
    document.getElementById('output')!.textContent = fullResponse;
  },
  (error) => console.error('Error:', error)
)) {
  // token by token processing
}

โค้ดนี้ใช้ async generator pattern ทำให้สามารถ iterate ผ่าน token ได้ทีละตัว มี callback onToken สำหรับอัพเดท UI แบบ real-time และ onError สำหรับจัดการ error กรณี connection หลุดหรือ API error

การ Implement ด้วย React Hook

สำหรับ React application ผมสร้าง custom hook ที่จัดการ streaming state ทั้งหมดให้

import { useState, useCallback, useRef } from 'react';

interface UseChatStreamOptions {
  apiEndpoint: string;
  apiKey: string;
}

interface ChatState {
  messages: Array<{ role: string; content: string }>;
  isStreaming: boolean;
  error: string | null;
  currentResponse: string;
}

export function useChatStream({ apiEndpoint, apiKey }: UseChatStreamOptions) {
  const [state, setState] = useState<ChatState>({
    messages: [],
    isStreaming: false,
    error: null,
    currentResponse: '',
  });
  
  const abortControllerRef = useRef<AbortController | null>(null);

  const sendMessage = useCallback(async (content: string) => {
    // Cancel previous stream if exists
    if (abortControllerRef.current) {
      abortControllerRef.current.abort();
    }

    const abortController = new AbortController();
    abortControllerRef.current = abortController;

    const userMessage = { role: 'user' as const, content };
    
    setState(prev => ({
      ...prev,
      messages: [...prev.messages, userMessage],
      isStreaming: true,
      error: null,
      currentResponse: '',
    }));

    try {
      const response = await fetch(apiEndpoint, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': Bearer ${apiKey},
        },
        body: JSON.stringify({
          messages: [...state.messages, userMessage],
        }),
        signal: abortController.signal,
      });

      if (!response.ok) {
        throw new Error(Error: ${response.status});
      }

      const reader = response.body?.getReader();
      if (!reader) throw new Error('No response body');

      const decoder = new TextDecoder();
      let fullResponse = '';

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        const chunk = decoder.decode(value, { stream: true });
        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);
              const content = parsed.choices?.[0]?.delta?.content;
              if (content) {
                fullResponse += content;
                setState(prev => ({
                  ...prev,
                  currentResponse: fullResponse,
                }));
              }
            } catch (e) {
              // Skip invalid JSON
            }
          }
        }
      }

      // เพิ่ม assistant message เมื่อ stream เสร็จ
      setState(prev => ({
        ...prev,
        messages: [
          ...prev.messages,
          { role: 'assistant', content: fullResponse },
        ],
        isStreaming: false,
        currentResponse: '',
      }));

    } catch (error) {
      if ((error as Error).name === 'AbortError') {
        setState(prev => ({ ...prev, isStreaming: false }));
        return;
      }
      
      setState(prev => ({
        ...prev,
        isStreaming: false,
        error: (error as Error).message,
      }));
    }
  }, [apiEndpoint, apiKey, state.messages]);

  const cancelStream = useCallback(() => {
    abortControllerRef.current?.abort();
    setState(prev => ({
      ...prev,
      isStreaming: false,
      currentResponse: '',
    }));
  }, []);

  return {
    ...state,
    sendMessage,
    cancelStream,
  };
}

การ Optimize ประสิทธิภาพและต้นทุน

จากประสบการณ์ของผม มีหลายวิธีในการ optimize streaming ทั้งด้าน latency และ cost

การวัดผล benchmark บน HolySheep AI ซึ่งมี <50ms latency:

# Benchmark script สำหรับทดสอบ streaming performance
import asyncio
import httpx
import time
import statistics

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

async def benchmark_stream():
    """ทดสอบ time-to-first-token และ throughput"""
    
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    
    payload = {
        "model": "gpt-4.1",
        "messages": [{"role": "user", "content": "นับ 1-100"}],
        "stream": True,
        "max_tokens": 500,
    }
    
    results = {
        "time_to_first_token_ms": [],
        "total_time_ms": [],
        "tokens_received": [],
    }
    
    # ทดสอบ 10 รอบ
    for i in range(10):
        start_time = time.perf_counter()
        first_token_time = None
        tokens_count = 0
        
        async with httpx.AsyncClient(timeout=60.0) as client:
            async with client.stream(
                "POST",
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]
                        if data == "[DONE]":
                            break
                        
                        current_time = time.perf_counter()
                        elapsed_ms = (current_time - start_time) * 1000
                        
                        if first_token_time is None:
                            first_token_time = elapsed_ms
                        
                        tokens_count += 1
        
        total_time = (time.perf_counter() - start_time) * 1000
        
        results["time_to_first_token_ms"].append(first_token_time or 0)
        results["total_time_ms"].append(total_time)
        results["tokens_received"].append(tokens_count)
        
        print(f"Run {i+1}: TTFT={first_token_time:.1f}ms, "
              f"Total={total_time:.1f}ms, Tokens={tokens_count}")
    
    # คำนวณค่าเฉลี่ย
    print("\n=== Benchmark Results ===")
    print(f"Avg Time-to-First-Token: "
          f"{statistics.mean(results['time_to_first_token_ms']):.1f}ms")
    print(f"Avg Total Time: "
          f"{statistics.mean(results['total_time_ms']):.1f}ms")
    print(f"Avg Tokens: {statistics.mean(results['tokens_received']):.0f}")
    print(f"Avg Throughput: "
          f"{statistics.mean(results['tokens_received']) / (statistics.mean(results['total_time_ms'])/1000):.1f} tokens/sec")

if __name__ == "__main__":
    asyncio.run(benchmark_stream())

ผล benchmark ที่ได้จา�

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง