Next.js App Router SSE สำหรับ Streaming AI Response: Server Actions และ Performance Optimization

ในยุคที่ AI-powered applications กำลังเติบโตอย่างรวดเร็ว การส่ง response แบบ real-time กลายเป็นความต้องการหลักของผู้ใช้งาน ในบทความนี้ผมจะพาทุกท่านเจาะลึกการใช้งาน Server-Sent Events (SSE) ร่วมกับ Next.js App Router สำหรับการ stream AI response โดยใช้ Server Actions เป็นสถาปัตยกรรมหลัก พร้อมกับแนะนำ การเชื่อมต่อกับ HolySheep AI ที่ให้บริการ AI API ด้วยความหน่วงต่ำกว่า 50ms และอัตราค่าบริการที่ประหยัดกว่า 85% เมื่อเทียบกับผู้ให้บริการอื่น

SSE vs WebSocket: ทำไมต้องเลือก SSE

ก่อนจะเข้าสู่การ implement ผมอยากให้ทำความเข้าใจความแตกต่างระหว่าง SSE และ WebSocket กันก่อน

• Server-Sent Events (SSE): เป็น one-way communication จาก server ไปยัง client เหมาะสำหรับ streaming AI response ที่ server ต้องส่งข้อมูลมาเรื่อยๆ
• WebSocket: เป็น full-duplex communication ที่ต้องการการส่งข้อมูลสองทิศทาง ซับซ้อนกว่าและใช้ overhead มากกว่า
• สำหรับ AI Streaming: SSE เพียงพอและ implement ง่ายกว่ามาก

สถาปัตยกรรมโดยรวม

สถาปัตยกรรมที่เราจะสร้างประกอบด้วย 3 ส่วนหลัก:

1. Server Action: รับ request จาก client และส่งต่อไปยัง AI API
2. Streaming Response Handler: จัดการ stream จาก AI และส่งไปยัง client
3. Client Component: แสดงผลแบบ real-time พร้อม UI states ต่างๆ

การตั้งค่า Environment และ API Client

เริ่มต้นด้วยการสร้าง API client ที่รองรับ streaming จาก HolySheep AI:

// lib/holysheep-stream.ts
const BASE_URL = 'https://api.holysheep.ai/v1';

interface StreamOptions {
  model: string;
  messages: Array<{ role: string; content: string }>;
  temperature?: number;
  max_tokens?: number;
}

export async function createAIStream(
  options: StreamOptions,
  apiKey: string
): Promise> {
  const response = await fetch(${BASE_URL}/chat/completions, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': Bearer ${apiKey},
    },
    body: JSON.stringify({
      model: options.model,
      messages: options.messages,
      temperature: options.temperature ?? 0.7,
      max_tokens: options.max_tokens ?? 2048,
      stream: true, // เปิด streaming mode
    }),
  });

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

  if (!response.body) {
    throw new Error('Response body is null');
  }

  // Transform SSE format เป็น readable stream
  return new ReadableStream({
    async start(controller) {
      const reader = response.body!.getReader();
      const decoder = new TextDecoder();
      let buffer = '';

      try {
        while (true) {
          const { done, value } = await reader.read();
          
          if (done) {
            // ส่ง remaining buffer ก่อนปิด
            if (buffer.trim()) {
              controller.enqueue(new TextEncoder().encode(buffer));
            }
            controller.close();
            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]') {
                controller.close();
                return;
              }

              try {
                const parsed = JSON.parse(data);
                const content = parsed.choices?.[0]?.delta?.content;
                
                if (content) {
                  controller.enqueue(
                    new TextEncoder().encode(content)
                  );
                }
              } catch {
                // Skip malformed JSON
              }
            }
          }
        }
      } catch (error) {
        controller.error(error);
      }
    },
  });
}

// ฟังก์ชันสำหรับ parse chunks ใน client side
export function parseStreamChunk(chunk: string): string {
  try {
    const data = JSON.parse(chunk);
    return data.choices?.[0]?.delta?.content ?? '';
  } catch {
    return '';
  }
}

Server Action Implementation

ต่อไปคือการสร้าง Server Action ที่จัดการ streaming โดยใช้ React's experimental streaming:

'use server';

import { createAIStream } from '@/lib/holysheep-stream';

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

interface StreamChatOptions {
  messages: ChatMessage[];
  model?: 'gpt-4.1' | 'claude-sonnet-4.5' | 'gemini-2.5-flash' | 'deepseek-v3.2';
  temperature?: number;
}

export async function streamChat(
  options: StreamChatOptions,
  apiKey: string
): Promise> {
  const model = options.model ?? 'deepseek-v3.2';
  
  // Map model names ตาม HolySheep API
  const modelMap: Record = {
    'gpt-4.1': 'gpt-4.1',
    'claude-sonnet-4.5': 'claude-sonnet-4.5',
    'gemini-2.5-flash': 'gemini-2.5-flash',
    'deepseek-v3.2': 'deepseek-v3.2',
  };

  return createAIStream(
    {
      model: modelMap[model],
      messages: options.messages.map((m) => ({
        role: m.role,
        content: m.content,
      })),
      temperature: options.temperature,
      max_tokens: 4096,
    },
    apiKey
  );
}

// Server Action สำหรับ non-streaming (ถ้าต้องการ)
export async function chatCompletion(
  options: StreamChatOptions,
  apiKey: string
): Promise {
  const stream = await streamChat(options, apiKey);
  const reader = stream.getReader();
  let result = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    result += new TextDecoder().decode(value);
  }

  return result;
}

Client Component สำหรับ Streaming UI

ตอนนี้มาสร้าง Client Component ที่จะ consume streaming response:

'use client';

import { useState, useCallback, useRef } from 'react';
import { streamChat } from './actions';

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

export default function AIChatComponent() {
  const [messages, setMessages] = useState([]);
  const [input, setInput] = useState('');
  const [isStreaming, setIsStreaming] = useState(false);
  const [error, setError] = useState(null);
  const abortControllerRef = useRef(null);

  const handleSubmit = useCallback(
    async (e: React.FormEvent) => {
      e.preventDefault();
      
      if (!input.trim() || isStreaming) return;

      const userMessage: Message = {
        id: crypto.randomUUID(),
        role: 'user',
        content: input.trim(),
        timestamp: new Date(),
      };

      setMessages((prev) => [...prev, userMessage]);
      setInput('');
      setIsStreaming(true);
      setError(null);

      // สร้าง Assistant message placeholder
      const assistantMessageId = crypto.randomUUID();
      setMessages((prev) => [
        ...prev,
        {
          id: assistantMessageId,
          role: 'assistant',
          content: '',
          timestamp: new Date(),
        },
      ]);

      try {
        // เรียก Server Action
        const stream = await streamChat(
          {
            messages: [
              ...messages.map((m) => ({
                role: m.role,
                content: m.content,
              })),
              { role: 'user', content: input.trim() },
            ],
            model: 'deepseek-v3.2', // เลือก model ที่ประหยัดที่สุด
          },
          process.env.NEXT_PUBLIC_HOLYSHEEP_API_KEY!
        );

        const reader = stream.getReader();
        const decoder = new TextDecoder();
        let accumulatedContent = '';

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

          accumulatedContent += decoder.decode(value, { stream: true });
          
          // Update message content แบบ real-time
          setMessages((prev) =>
            prev.map((msg) =>
              msg.id === assistantMessageId
                ? { ...msg, content: accumulatedContent }
                : msg
            )
          );
        }
      } catch (err) {
        console.error('Stream error:', err);
        setError(err instanceof Error ? err.message : 'เกิดข้อผิดพลาด');
        setIsStreaming(false);
        
        // ลบ placeholder message
        setMessages((prev) => prev.filter((m) => m.id !== assistantMessageId));
      }
    },
    [input, isStreaming, messages]
  );

  const cancelStream = useCallback(() => {
    if (abortControllerRef.current) {
      abortControllerRef.current.abort();
      setIsStreaming(false);
    }
  }, []);

  return (
    <div className="flex flex-col h-[600px] max-w-2xl mx-auto p-4">
      {/* Messages Display */}
      <div className="flex-1 overflow-y-auto space-y-4 mb-4">
        {messages.map((msg) => (
          <div
            key={msg.id}
            className={`flex ${
              msg.role === 'user' ? 'justify-end' : 'justify-start'
            }`}
          >
            <div
              className={`max-w-[80%] p-4 rounded-lg ${
                msg.role === 'user'
                  ? 'bg-blue-600 text-white'
                  : 'bg-gray-100 text-gray-900'
              }`}
            >
              <div className="whitespace-pre-wrap">{msg.content}</div>
              {msg.role === 'assistant' && isStreaming && (
                <span className="inline-block animate-pulse ml-1">▍</span>
              )}
            </div>
          </div>
        ))}
        
        {error && (
          <div className="text-red-500 p-4 bg-red-50 rounded-lg">
            {error}
          </div>
        )}
      </div>

      {/* Input Form */}
      <form onSubmit={handleSubmit} className="flex gap-2">
        <input
          type="text"
          value={input}
          onChange={(e) => setInput(e.target.value)}
          disabled={isStreaming}
          placeholder="พิมพ์ข้อความของคุณ..."
          className="flex-1 p-3 border rounded-lg focus:outline-none focus:ring-2 focus:ring-blue-500"
        />
        {isStreaming ? (
          <button
            type="button"
            onClick={cancelStream}
            className="px-6 py-3 bg-red-500 text-white rounded-lg hover:bg-red-600"
          >
            หยุด
          </button>
        ) : (
          <button
            type="submit"
            disabled={!input.trim()}
            className="px-6 py-3 bg-blue-600 text-white rounded-lg hover:bg-blue-700 disabled:opacity-50"
          >
            ส่ง
          </button>
        )}
      </form>
    </div>
  );
}

Performance Optimization และ Concurrency Control

สำหรับ production environment การจัดการ concurrency และ performance เป็นสิ่งสำคัญมาก:

Rate Limiting และ Queue Management

// lib/stream-queue.ts
import { EventEmitter } from 'events';

interface QueuedRequest {
  id: string;
  userId: string;
  stream: ReadableStream;
  resolve: (controller: ReadableStreamDefaultController) => void;
  reject: (error: Error) => void;
  timestamp: number;
}

class StreamQueueManager extends EventEmitter {
  private queue: Map = new Map();
  private processing: Set = new Set();
  private userConnections: Map = new Map();
  
  // จำกัดจำนวน connection ต่อ user
  private readonly MAX_CONCURRENT_PER_USER = 3;
  // จำกัด queue size ทั้งหมด
  private readonly MAX_QUEUE_SIZE = 100;
  // Timeout สำหรับ queued request (ms)
  private readonly QUEUE_TIMEOUT = 30000;

  async enqueue(
    userId: string,
    generator: () => Promise>
  ): Promise> {
    // Check user limit
    const userConnections = this.userConnections.get(userId) ?? 0;
    if (userConnections >= this.MAX_CONCURRENT_PER_USER) {
      throw new Error(
        User ${userId} exceeded concurrent limit (${this.MAX_CONCURRENT_PER_USER})
      );
    }

    // Check queue size
    if (this.queue.size >= this.MAX_QUEUE_SIZE) {
      throw new Error('Server queue is full, please try again later');
    }

    return new Promise((resolve, reject) => {
      const requestId = crypto.randomUUID();
      
      const request: QueuedRequest = {
        id: requestId,
        userId,
        stream: null as any,
        resolve,
        reject,
        timestamp: Date.now(),
      };

      this.queue.set(requestId, request);
      this.processQueue();
    });
  }

  private async processQueue() {
    if (this.processing.size >= this.MAX_CONCURRENT_PER_USER * 2) {
      return; // Server busy
    }

    const request = this.queue.values().next().value;
    if (!request) return;

    this.queue.delete(request.id);
    this.processing.add(request.id);
    this.userConnections.set(
      request.userId,
      (this.userConnections.get(request.userId) ?? 0) + 1
    );

    // Set timeout
    const timeout = setTimeout(() => {
      this.cleanup(request.id, request.userId);
      request.reject(new Error('Request timeout'));
    }, this.QUEUE_TIMEOUT);

    try {
      const stream = await this.createStream(request);
      clearTimeout(timeout);
      request.resolve(this.createControlledStream(stream, request));
    } catch (error) {
      clearTimeout(timeout);
      request.reject(error as Error);
    } finally {
      this.cleanup(request.id, request.userId);
    }
  }

  private async createStream(request: QueuedRequest): Promise> {
    // เรียก Server Action จริงๆ
    const { streamChat } = await import('./actions');
    return streamChat(
      { messages: [], model: 'deepseek-v3.2' },
      process.env.HOLYSHEEP_API_KEY!
    );
  }

  private createControlledStream(
    source: ReadableStream,
    request: QueuedRequest
  ): ReadableStream {
    const reader = source
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
多模型 Cost Optimization：ค้นหา Pareto Optimal ระหว่างคุณภาพ Res
向量数据库监控：Embedding 质量评估与异常检测
ป้องกัน Prompt Injection Attack ในระบบ AI Dialogue: คู่มือ I