Stream คำตอบ AI ไปหน้าเว็บด้วย Server-Sent Events

คำตอบจาก LLM ที่ยาวสักหน่อยอาจใช้เวลาสร้างหลายวินาทีจนถึงหลายสิบวินาที ถ้าเรารอจนเสร็จแล้วค่อยส่งทั้งก้อน ผู้ใช้จะเห็นหน้าจอนิ่งพร้อม spinner และหลายคนจะกดรีเฟรชหรือคิดว่าระบบค้าง การ stream คำตอบทีละส่วนแบบที่เห็นในแอปแชต AI ทั่วไปช่วยแก้ปัญหานี้ได้ บทความนี้จะอธิบายว่าทำไมควร stream เลือก Server-Sent Events (SSE) หรือ WebSocket ดี แล้วเขียนตัวอย่างฝั่ง PHP/Yii2 ส่ง event ไปยัง JavaScript EventSource พร้อมเรื่องที่ทำให้หลายคนติดอยู่นาน คือ output buffering และ reverse proxy

ทำไมต้อง stream

• Time to first token ผู้ใช้เห็นตัวอักษรแรกเร็ว แม้เวลารวมจะเท่าเดิม ความรู้สึกว่าระบบตอบสนองดีขึ้นมาก
• เลี่ยง timeout request ที่เงียบนาน ๆ เสี่ยงโดน timeout ของ proxy หรือ load balancer ระหว่างทาง การมีข้อมูลไหลอย่างต่อเนื่องช่วยได้
• หยุดได้กลางทาง ถ้าผู้ใช้เห็นว่าคำตอบผิดทางตั้งแต่ต้น ก็กดหยุดได้ ประหยัดทั้งเวลาและ token

ผู้ให้บริการ LLM ส่วนใหญ่มีโหมด streaming ซึ่งส่งคำตอบกลับมาเป็นชิ้นเล็ก ๆ (มักใช้รูปแบบ SSE เช่นกัน) งานของเราคือรับชิ้นเหล่านั้นแล้วส่งต่อไปยังเบราว์เซอร์

SSE vs WebSocket

การ stream คำตอบ AI เป็นงานทางเดียวโดยธรรมชาติ ผู้ใช้ส่งคำถามครั้งเดียว แล้ว server ส่งคำตอบไหลกลับมา SSE จึงเหมาะกว่าและเข้ากับ stack PHP + nginx ที่มีอยู่ได้ทันที WebSocket จะคุ้มเมื่อต้องการสื่อสารสองทางตลอดเวลา เช่น แชตหลายคนแบบ real-time

อีกทางเลือกหนึ่งที่บางทีมใช้คือ long polling หรือให้หน้าเว็บถาม server ซ้ำทุกวินาทีว่ามีข้อความใหม่หรือยัง วิธีนี้ทำได้แต่สิ้นเปลือง request และตอบสนองช้ากว่า เมื่อเทียบกับ SSE ที่เปิดการเชื่อมต่อเดียวแล้วรับข้อมูลได้ทันทีที่มี ข้อจำกัดของ SSE ที่ควรรู้คือ ถ้าเว็บยังใช้ HTTP/1.1 เบราว์เซอร์จำกัดจำนวนการเชื่อมต่อพร้อมกันต่อโดเมน การเปิดหลายแท็บที่ stream พร้อมกันอาจทำให้ request อื่นต้องรอ ถ้าใช้ HTTP/2 ข้อจำกัดนี้จะลดลงมาก

รูปแบบข้อความของ SSE

SSE คือ HTTP response ที่มี Content-Type: text/event-stream และไม่ปิดการเชื่อมต่อทันที ข้อมูลแต่ละ event เป็นบรรทัดข้อความ จบ event ด้วยบรรทัดว่าง

event: token
data: {"text":"สวัส"}

event: token
data: {"text":"ดีครับ"}

: บรรทัดที่ขึ้นต้นด้วย colon คือ comment ใช้เป็น heartbeat ได้

event: done
data: {}

เราส่ง data เป็น JSON เพื่อให้ขึ้นบรรทัดใหม่ในข้อความได้อย่างปลอดภัย (ถ้าข้อมูลมี newline ดิบ จะทำให้รูปแบบ event เพี้ยน) และกำหนดชื่อ event เพื่อแยกประเภท เช่น token, failure, done

ฝั่ง PHP: ส่ง event จาก Yii2

ข้อจำกัดหนึ่งของ EventSource คือเรียกได้แค่ GET จึงนิยมแบ่งเป็นสองขั้น ขั้นแรก POST คำถามมาเก็บไว้ก่อน (ผ่าน CSRF และตรวจสิทธิ์ตามปกติ) แล้วได้ id กลับไป ขั้นที่สองเปิด EventSource ด้วย id นั้น ตัวอย่างนี้เก็บคำถามไว้ใน cache แบบใช้ครั้งเดียว

<?php
namespace frontend\controllers;

use Yii;
use yii\web\Controller;
use yii\web\NotFoundHttpException;

class AiController extends Controller
{
    // ขั้นที่ 1: รับคำถาม (POST + CSRF) แล้วคืน id
    public function actionPrepare()
    {
        Yii::$app->response->format = \yii\web\Response::FORMAT_JSON;
        $q = trim((string)Yii::$app->request->post('question', ''));
        if ($q === '' || mb_strlen($q) > 2000) {
            throw new \yii\web\BadRequestHttpException('คำถามไม่ถูกต้อง');
        }
        $id = Yii::$app->security->generateRandomString(32);
        Yii::$app->cache->set("ai-q:{$id}", [
            'user' => Yii::$app->user->id,
            'q'    => $q,
        ], 120);
        return ['id' => $id];
    }

    // ขั้นที่ 2: stream คำตอบ
    public function actionStream(string $id)
    {
        $job = Yii::$app->cache->get("ai-q:{$id}");
        if (!$job || $job['user'] !== Yii::$app->user->id) {
            throw new NotFoundHttpException();
        }
        Yii::$app->cache->delete("ai-q:{$id}");

        // ปล่อย session lock ไม่อย่างนั้น request อื่นของผู้ใช้คนนี้จะรอจน stream จบ
        Yii::$app->session->close();

        // บอก Yii ว่าเราจัดการ output เอง
        Yii::$app->response->isSent = true;

        header('Content-Type: text/event-stream; charset=utf-8');
        header('Cache-Control: no-cache');
        header('X-Accel-Buffering: no'); // ปิด buffer ของ nginx สำหรับ response นี้

        // ปิด output buffer ของ PHP ทุกชั้น
        while (ob_get_level() > 0) {
            ob_end_flush();
        }
        set_time_limit(0);

        $send = function (string $event, array $data): void {
            echo "event: {$event}\n";
            echo 'data: ' . json_encode($data, JSON_UNESCAPED_UNICODE) . "\n\n";
            flush();
        };

        try {
            foreach ((new \common\components\LlmStream())->stream($job['q']) as $chunk) {
                if (connection_aborted()) {
                    break; // ผู้ใช้ปิดหน้าหรือกดหยุด
                }
                $send('token', ['text' => $chunk]);
            }
            $send('done', []);
        } catch (\Throwable $e) {
            Yii::error($e->getMessage(), __METHOD__);
            $send('failure', ['message' => 'เกิดข้อผิดพลาด กรุณาลองใหม่']);
        }
    }
}

สังเกตว่า connection_aborted() จะรู้ว่าผู้ใช้ตัดการเชื่อมต่อก็ต่อเมื่อ PHP พยายามส่งข้อมูลออกไปแล้ว จึงต้องเช็กหลัง flush รอบก่อน ๆ ซึ่งในลูปนี้เกิดขึ้นทุกชิ้นอยู่แล้ว

รับ stream จาก LLM ด้วย Guzzle

ฝั่งที่เราเป็น client ของ LLM ก็อ่านแบบ stream ได้ โดยเปิด option stream ของ Guzzle แล้วอ่าน body ทีละบรรทัด รูปแบบของแต่ละชิ้น (ชื่อ field ที่เก็บข้อความ วิธีส่งสัญญาณจบ) ขึ้นกับเอกสารของผู้ให้บริการ ตัวอย่างนี้แยกส่วนนั้นไว้ใน extractDelta()

<?php
namespace common\components;

use GuzzleHttp\Client;

final class LlmStream
{
    public function stream(string $question): \Generator
    {
        $http = new Client(['connect_timeout' => 5, 'read_timeout' => 60]);
        $res = $http->post(getenv('LLM_API_URL'), [
            'stream'  => true,
            'headers' => ['Authorization' => 'Bearer ' . getenv('LLM_API_KEY')],
            'json'    => [
                'model'    => getenv('LLM_MODEL'),
                'stream'   => true,
                'messages' => [['role' => 'user', 'content' => $question]],
            ],
        ]);

        $body = $res->getBody();
        $buffer = '';
        while (!$body->eof()) {
            $buffer .= $body->read(1024);
            while (($pos = strpos($buffer, "\n")) !== false) {
                $line = rtrim(substr($buffer, 0, $pos), "\r");
                $buffer = substr($buffer, $pos + 1);
                if (!str_starts_with($line, 'data:')) {
                    continue;
                }
                $payload = trim(substr($line, 5));
                $text = $this->extractDelta($payload);
                if ($text !== null && $text !== '') {
                    yield $text;
                }
            }
        }
    }

    // ปรับตามรูปแบบ streaming ของผู้ให้บริการที่ใช้
    private function extractDelta(string $payload): ?string
    {
        $json = json_decode($payload, true);
        return is_array($json) ? ($json['choices'][0]['delta']['content'] ?? null) : null;
    }
}

การอ่านแบบนี้เก็บ buffer ไว้ต่อบรรทัดที่ถูกตัดครึ่ง และ substr กับ strpos ทำงานระดับ byte โดยตัดที่ \n เท่านั้น จึงไม่ทำให้ตัวอักษรไทยแบบ UTF-8 ถูกตัดกลางตัว

ฝั่ง JavaScript: EventSource

const csrf = document.querySelector('meta[name="csrf-token"]').content;
const output = document.getElementById('answer');

async function ask(question) {
  output.textContent = '';
  const res = await fetch('/ai/prepare', {
    method: 'POST',
    headers: { 'Content-Type': 'application/x-www-form-urlencoded', 'X-CSRF-Token': csrf },
    body: new URLSearchParams({ question }),
  });
  const { id } = await res.json();

  const es = new EventSource('/ai/stream?id=' + encodeURIComponent(id));

  es.addEventListener('token', (e) => {
    // ใช้ textContent ไม่ใช่ innerHTML เพื่อกัน XSS
    output.textContent += JSON.parse(e.data).text;
  });
  es.addEventListener('done', () => es.close());
  es.addEventListener('failure', (e) => {
    output.textContent += '\n' + JSON.parse(e.data).message;
    es.close();
  });
  es.onerror = () => {
    // การเชื่อมต่อหลุด: ปิดเสมอ ไม่อย่างนั้น EventSource จะ reconnect เองและเรียกซ้ำ
    es.close();
  };
}

เราตั้งชื่อ event ข้อผิดพลาดฝั่ง server ว่า failure แทน error เพราะ error เป็นชื่อ event ที่ EventSource ใช้แจ้งปัญหาการเชื่อมต่ออยู่แล้ว ถ้าใช้ชื่อซ้ำจะแยกสองกรณีออกจากกันยาก จุดที่ต้องระวังอีกข้อคือ EventSource จะ reconnect อัตโนมัติ เมื่อการเชื่อมต่อหลุด ซึ่งดีสำหรับฟีดข่าว แต่สำหรับคำตอบ AI แปลว่าอาจยิง LLM ซ้ำ เราจึงปิดเมื่อได้ done, failure หรือเมื่อเกิด error และฝั่ง server ลบ id ออกจาก cache หลังใช้ครั้งแรก ถ้าต้องการ POST โดยตรง สามารถใช้ fetch() แล้วอ่าน response.body เป็น ReadableStream แทนได้ แต่ต้อง parse รูปแบบ SSE เอง

Buffering และ timeout หลัง reverse proxy

อาการคลาสสิกคือ ทดสอบในเครื่องแล้วไหลดี พอขึ้น server จริง คำตอบโผล่มาทีเดียวทั้งก้อนตอนจบ สาเหตุคือมี buffer อยู่ระหว่างทางหลายชั้น ไล่ตรวจตามนี้

• PHP output buffering ค่า output_buffering ใน php.ini และ zlib.output_compression ทำให้ข้อมูลถูกกักไว้ เราปิดด้วยลูป ob_end_flush() แล้วตามด้วย flush() ทุก event
• nginx โดยปกติ buffer response จาก upstream header X-Accel-Buffering: no สั่งปิดเฉพาะ response นี้ ใช้ได้ทั้งกับ fastcgi_pass และ proxy_pass
• gzip การบีบอัดอาจรอข้อมูลสะสม ควรปิด gzip สำหรับ location ของ stream
• Timeout proxy มักตัดการเชื่อมต่อที่ไม่มีข้อมูลเกินเวลาที่กำหนด ให้ปรับ fastcgi_read_timeout หรือ proxy_read_timeout ให้ยาวพอ และส่ง comment : ping เป็นระยะถ้ามีช่วงที่ต้องรอนาน
• CDN และ load balancer บางตัวมี buffer และ idle timeout ของตัวเอง ต้องตรวจการตั้งค่าของบริการที่ใช้

# nginx: ตัวอย่าง location สำหรับ endpoint stream
location /ai/stream {
    gzip off;
    fastcgi_buffering off;
    fastcgi_read_timeout 300s;
    include fastcgi_params;
    fastcgi_param SCRIPT_FILENAME $document_root/index.php;
    fastcgi_pass php-fpm:9000;
}

สุดท้าย อย่าลืมว่าทุก stream ที่เปิดอยู่ยึด PHP-FPM worker หนึ่งตัวไว้ตลอด ถ้ามีผู้ใช้พร้อมกันมาก ต้องคำนวณ pm.max_children ให้พอ และจำกัดจำนวน stream ต่อผู้ใช้

สรุป

• Stream ช่วยให้ผู้ใช้เห็นคำตอบเร็ว ลดความเสี่ยง timeout และหยุดกลางทางได้
• SSE เหมาะกับการส่งคำตอบ AI เพราะเป็นงานทางเดียวและใช้กับ PHP-FPM ได้เลย
• ฝั่ง PHP ต้องปิด output buffer, flush ทุก event, ปิด session lock และตั้ง X-Accel-Buffering: no
• ฝั่ง JS ให้ปิด EventSource เมื่อจบ เมื่อ failure หรือเมื่อการเชื่อมต่อ error เพื่อกัน reconnect ที่ทำให้เรียก LLM ซ้ำ และแสดงผลด้วย textContent

ลองทำ

ก่อนต่อกับ LLM จริง ให้สร้าง action ทดสอบที่ส่ง event token เป็นคำว่า "หนึ่ง" ถึง "สิบ" ห่างกันคำละ 0.5 วินาที (ใช้ usleep(500000)) แล้วเปิดผ่าน EventSource ในเบราว์เซอร์ ทดลองทั้งแบบเรียกตรงและผ่าน nginx ถ้าคำทั้งหมดโผล่มาพร้อมกันตอนจบ ให้ไล่ตรวจ buffer ทีละชั้นตามรายการในบทความ จนกว่าจะเห็นคำไหลออกมาทีละคำ