Cách triển khai Streaming

Người dùng ghét việc phải chờ đợi. Streaming hiển thị phản hồi của AI ngay khi chúng được tạo ra, khiến trải nghiệm trở nên liền mạch. Hướng dẫn này trình bày các nguyên tắc cơ bản của việc Streaming hiệu quả.

Hiển thị progress ngay lập tức. Không bao giờ để người dùng phải nhìn chằm chằm vào một màn hình trống trơn.

Tại sao Streaming quan trọng#

Tác động đến tâm lý của người dùng là rất đáng kể: cảm giác các phản hồi có áp dụng Streaming nhanh hơn ngay cả khi tổng thời gian so với trường hợp không áp dụng Streaming là như nhau.

Các kỹ thuật Streaming#

Server-Sent Events (SSE)#

Dòng dữ liệu một chiều từ server đến client. Đây là lựa chọn thường xuyên được chúng tôi áp dụng đối với các phản hồi của LLM.

Ưu điểm: Đơn giản, được hỗ trợ sẵn bởi trình duyệt, có cơ chế tự động reconnect, hoạt động qua hầu hết các proxy.

Nhược điểm: Chỉ có một chiều từ server đến client và chỉ truyền được dữ liệu ở dạng text.

WebSockets#

Kết nối truyền thông hai chiều thời gian thực (real-time).

Ưu điểm: Hai chiều, hỗ trợ định dạng dữ liệu nhị phân (binary), độ trễ thấp cho việc truyền dữ liệu qua lại.

Nhược điểm: Triển khai phức tạp hơn, yêu cầu cấu hình reconnect một cách thủ công (hoặc sử dụng các thư viện có hỗ trợ sẵn).

Lựa chọn cái nào#

Quy tắc chung: Nếu chỉ có server cần gửi dữ liệu, hãy dùng SSE. Nếu cả hai phía server và client cần gửi dữ liệu real-time, hãy dùng WebSocket.

Nguyên tắc thiết kế SSE#

Định dạng dữ liệu#

SSE sử dụng định dạng text đơn giản. Mỗi message (hay còn gọi là event) là một khối text kết thúc bằng một dòng trống (\n\n), có thể chứa một số trường chính sau:

• data: — payload (bắt buộc), luôn là plain text. Nếu muốn gửi dữ liệu có cấu trúc (ví dụ JSON), thì phải chuyển thành string rồi client phải parse lại để đọc
• event: — tên kiểu event tùy chỉnh (mặc định là "message")
• id: — ID của event, trình duyệt dùng để reconnect từ đúng vị trí bị ngắt
• retry: — thời gian chờ reconnect tính bằng ms

Một message có thể có nhiều dòng data:, chúng sẽ được nối lại bằng \n:

data: dòng thứ nhất
data: dòng thứ hai

Kết quả event.data = "dòng thứ nhất\ndòng thứ hai"

Để đánh dấu kết thúc stream, chúng tôi dùng tín hiệu [DONE].

Ví dụ một stream hoàn chỉnh của agent:

event: tool_start
data: {"tool":"search","query":"keyword clustering"}

event: tool_end
data: {"tool":"search","results":3}

event: text
data: {"content":"Keyword clustering là kỹ thuật"}

event: text
data: {"content":" nhóm các từ khóa."}
data: {"content":"Đối với SEO..."}

event: done
data: [DONE]

Kiểu Event dành cho Agents#

Với agent có sử dụng tool, chúng tôi sử dụng các kiểu event sau:

Headers#

Các headers cần thiết cho SSE:

1. Content-Type: text/event-stream — báo cho trình duyệt biết đây là SSE
2. Cache-Control: no-cache — ngăn trình duyệt buffer luồng stream và sau đó truyền tải một lúc

Xử lý các trường hợp đặc biệt#

Kết nối bị ngắt#

Kết nối có thể bị ngắt, hãy luôn chuẩn bị sẵn cho trường hợp đó:

Khi dùng EventSource API: Reconnection được trình duyệt hỗ trợ và thực hiện một cách tự động.

Khi dùng fetch() với ReadableStream: Bạn phải tự cấu hình reconnection với cơ chế exponential backoff, bởi vì fetch() được sinh ra để dành cho HTTP API với mục đích chung.

Huỷ bỏ#

Người dùng nên có quyền huỷ bỏ các yêu cầu Streaming, tại vì:

• Người dùng nhận ra họ đã gửi đi sai yêu cầu
• Phản hồi đang đi sai hướng
• Người dùng thay đổi quyết định

Phía client dùng AbortController để huỷ request:

const controller = new AbortController();

// Bắt đầu stream
fetch("/api/chat/sse", { signal: controller.signal });

// Khi người dùng bấm "Dừng"
controller.abort();

Phía server cần phát hiện khi client ngắt kết nối và xử lý logic ngắt:

@app.get("/api/chat/sse")
async def chat_sse(request: Request):
    async def generate():
        with client.messages.stream(
            model="...",
            max_tokens=1024,
            messages=[{"role": "user", "content": "..."}],
        ) as stream:
            for text in stream.text_stream:
                # Kiểm tra client đã ngắt kết nối chưa
                if await request.is_disconnected():
                    stream.close()
                    break
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

Trong ví dụ minh hoạ trên, khi controller.abort() được gọi phía client, trình duyệt đóng kết nối TCP. Server sử dụng FastAPI phát hiện qua request.is_disconnected(), từ đó gọi stream.close() để huỷ request đến LLM API, tránh lãng phí token.

Rate Limiting#

Nhiều luồng Streaming đồng thời có thể làm quá tải server hay giới hạn API.

Giải pháp:

• Giới hạn số luồng đồng thời cho mỗi người dùng (2-3 là hợp lý)
• Trả về lỗi với mã 429 (Too Many Requests) khi vượt quá giới hạn
• Client xử lý lỗi được trả về này

Chunk bị phân mảnh#

Khi dùng fetch() với ReadableStream để nhận SSE, dữ liệu có thể đến dưới dạng chunk không hoàn chỉnh — một message có thể bị cắt giữa chừng qua nhiều network packet.

Giải pháp: Buffer dữ liệu nhận được, chỉ parse khi gặp \n\n (dấu hiệu kết thúc một message hoàn chỉnh).

Lưu ý: Nếu dùng EventSource API thì không cần lo vấn đề này — trình duyệt đã tự động thực hiện.

Tối ưu frontend#

Re-render#

Mỗi khối dữ liệu Streaming sẽ kích hoạt re-render trong React. Ở tốc độ truyền cao, điều này gây ra quá nhiều lần re-render, ảnh hưởng đến trải nghiệm của người dùng.

Giải pháp: Buffer các khối dữ liệu trong một ref, re-render sau một khoảng thời gian:

const bufferRef = useRef("");

// Xử lý SSE — ghi vào ref, không re-render
function onChunk(chunk) {
  bufferRef.current += chunk;
}

// Cập nhật dữ liệu vào state của React sau một khoảng thời gian
useEffect(() => {
  const id = setInterval(() => {
    if (bufferRef.current) {
      setText((prev) => prev + bufferRef.current);
      bufferRef.current = "";
    }
  }, 50);
  return () => clearInterval(id);
}, []);

Với nội dung Markdown có Streaming, có thể sử dụng llm-ui — một thư viện React được thiết kế để hiển thị đầu ra của LLM.

Nội dung dài#

Khi các message tích lũy trong giao diện, kích thước của DOM cũng tăng lên tương ứng — dẫn đến việc hiển thị chậm hơn và sử dụng nhiều bộ nhớ.

Giải pháp: Ảo hóa danh sách message để chỉ các message có thể nhìn thấy mới được đưa vào DOM:

import { Virtuoso } from "react-virtuoso";

<Virtuoso
  data={messages}
  followOutput="smooth"
  itemContent={(index, message) => <ChatMessage {...message} />}
/>;

react-virtuoso chỉ render các message trong vùng hiển thị, bất kể độ dài tổng thể của cuộc trò chuyện.