Streaming AI Responses: Xử Lý Theo Thời Gian Thực

Streaming AI responses là kỹ thuật trả về kết quả từ mô hình ngôn ngữ lớn (LLM) theo từng đoạn (chunk) thay vì đợi toàn bộ văn bản hoàn thành. User nhận token đầu tiên trong < 500ms, cải thiện trải nghiệm tương tác đáng kể so với batch processing (phải đợi 5-20 giây). Cốt lõi là Server-Sent Events (SSE) hoặc WebSocket để gửi dữ liệu incremental từ server về client real-time.

Streaming AI Là Gì?

Khi bạn gọi một LLM API (OpenAI GPT, Claude, Gemini), có hai cách nhận kết quả:

1. Batch (non-streaming): Gửi request → đợi → nhận toàn bộ response một lúc.
2. Streaming: Gửi request → nhận token đầu tiên ngay lập tức → token tiếp theo liên tục → cho đến khi kết thúc.

Ví dụ trực quan:

• Batch: User gửi câu hỏi → loading spinner 8 giây → toàn bộ câu trả lời hiện ra.
• Streaming: User gửi câu hỏi → từ đầu tiên xuất hiện sau 0.3 giây → câu trả lời “chảy” ra từng từ như đánh máy.

Lợi ích cốt lõi:

• Time To First Token (TTFT) thấp: User thấy progress ngay, giảm tỷ lệ bỏ cuộc.
• Perceived performance cao hơn thực tế: 10 giây streaming cảm nhận nhanh hơn 8 giây batch.
• Trải nghiệm tự nhiên: Giống chat với con người hơn là query database.

Trade-off:

• Phức tạp hơn về mặt kỹ thuật (xử lý stream, error mid-stream).
• Khó cache toàn bộ response (phải cache từng chunk hoặc chỉ cache sau khi hoàn thành).
• Không phù hợp với workflow cần toàn bộ output trước khi xử lý tiếp (ví dụ phân tích sentiment, lưu database).

Cách Streaming Hoạt Động Ở Tầng Protocol

Server-Sent Events (SSE) — Chuẩn Phổ Biến

Đa số API LLM (OpenAI, Anthropic, Google) dùng SSE cho streaming:

Request:

POST /v1/chat/completions
Content-Type: application/json
Accept: text/event-stream

{
  "model": "gpt-4o",
  "messages": [...],
  "stream": true
}

Response stream (mỗi dòng là một event):

data: {"choices":[{"delta":{"content":"Xin"}}]}

data: {"choices":[{"delta":{"content":" chào"}}]}

data: {"choices":[{"delta":{"content":"!"}}]}

data: [DONE]

Client đọc từng dòng data:, parse JSON, ghép delta.content lại thành câu hoàn chỉnh.

Đặc điểm SSE:

• One-way (server → client).
• Tự động reconnect nếu mất kết nối.
• Dễ implement hơn WebSocket cho use case này.
• Chạy trên HTTP/2 hoặc HTTP/1.1 (long-polling).

WebSocket — Khi Cần Two-way

Một số framework (LangServe, custom backend) dùng WebSocket để:

• Client gửi message mới mid-stream (ví dụ “dừng lại”, “bổ sung câu hỏi”).
• Streaming cả input lẫn output (ví dụ voice chat).

Trade-off: Phức tạp hơn SSE, nhưng linh hoạt hơn cho tương tác hai chiều.

Triển Khai Streaming Với Các API Phổ Biến

OpenAI (GPT-4, GPT-4o)

Python SDK:

from openai import OpenAI

client = OpenAI(api_key="sk-...")

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Giải thích streaming AI"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

JavaScript/TypeScript:

import OpenAI from 'openai';

const openai = new OpenAI({ apiKey: 'sk-...' });

const stream = await openai.chat.completions.create({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Giải thích streaming AI' }],
  stream: true,
});

for await (const chunk of stream) {
  process.stdout.write(chunk.choices[0]?.delta?.content || '');
}

Lưu ý:

• stream=True trong Python, stream: true trong JS.
• Mỗi chunk có thể chứa metadata khác (ví dụ finish_reason, usage).

Anthropic (Claude)

Python:

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Giải thích streaming AI"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

TypeScript:

import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic({ apiKey: 'sk-ant-...' });

const stream = await client.messages.stream({
  model: 'claude-3-5-sonnet-20241022',
  max_tokens: 1024,
  messages: [{ role: 'user', content: 'Giải thích streaming AI' }],
});

for await (const chunk of stream) {
  if (chunk.type === 'content_block_delta' && chunk.delta.type === 'text_delta') {
    process.stdout.write(chunk.delta.text);
  }
}

Khác biệt so với OpenAI:

• Anthropic có nhiều event type hơn (message_start, content_block_start, content_block_delta, message_stop).
• Python SDK có wrapper .text_stream tiện hơn.

Google Gemini

Python:

import google.generativeai as genai

genai.configure(api_key="...")
model = genai.GenerativeModel('gemini-1.5-pro')

response = model.generate_content("Giải thích streaming AI", stream=True)

for chunk in response:
    print(chunk.text, end="", flush=True)

Gemini streaming đơn giản hơn (mỗi chunk là text thuần), nhưng ít metadata hơn OpenAI/Anthropic.

Streaming Trong Framework (LangChain, LlamaIndex)

LangChain

Method 1: .stream()

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", streaming=True)
prompt = ChatPromptTemplate.from_template("Giải thích {topic}")
chain = prompt | llm

for chunk in chain.stream({"topic": "streaming AI"}):
    print(chunk.content, end="", flush=True)

Method 2: Callback handlers

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = ChatOpenAI(
    model="gpt-4o",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)
llm.invoke("Giải thích streaming AI")
# Output tự động stream ra stdout

LangChain streaming hoạt động với:

• Chains phức tạp (multi-step).
• Agents (streaming cả thought process và final answer).
• Retrieval (streaming retrieved docs + LLM response).

LlamaIndex

from llama_index.llms.openai import OpenAI
from llama_index.core.llms import ChatMessage

llm = OpenAI(model="gpt-4o")

messages = [ChatMessage(role="user", content="Giải thích streaming AI")]
response = llm.stream_chat(messages)

for token in response:
    print(token.delta, end="", flush=True)

Query engine streaming:

from llama_index.core import VectorStoreIndex

index = VectorStoreIndex.from_documents(docs)
query_engine = index.as_query_engine(streaming=True)

response = query_engine.query("Streaming AI là gì?")
response.print_response_stream()

Frontend: Hiển Thị Stream Trong UI

React + OpenAI Stream

import { useState } from 'react';

function ChatBox() {
  const [response, setResponse] = useState('');

  async function handleSubmit(prompt: string) {
    setResponse('');
    const res = await fetch('/api/chat', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ prompt }),
    });

    const reader = res.body?.getReader();
    const decoder = new TextDecoder();

    while (true) {
      const { done, value } = await reader!.read();
      if (done) break;
      
      const chunk = decoder.decode(value);
      const lines = chunk.split('\n').filter(line => line.startsWith('data:'));
      
      for (const line of lines) {
        if (line === 'data: [DONE]') break;
        const json = JSON.parse(line.slice(5));
        const content = json.choices[0]?.delta?.content || '';
        setResponse(prev => prev + content);
      }
    }
  }

  return <div>{response}</div>;
}

Backend API route (Next.js):

export async function POST(req: Request) {
  const { prompt } = await req.json();
  
  const stream = await openai.chat.completions.create({
    model: 'gpt-4o',
    messages: [{ role: 'user', content: prompt }],
    stream: true,
  });

  const encoder = new TextEncoder();
  const readable = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content || '';
        controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`));
      }
      controller.enqueue(encoder.encode('data: [DONE]\n\n'));
      controller.close();
    },
  });

  return new Response(readable, {
    headers: { 'Content-Type': 'text/event-stream' },
  });
}

Vercel AI SDK (Wrapper Tiện Hơn)

import { OpenAIStream, StreamingTextResponse } from 'ai';
import { Configuration, OpenAIApi } from 'openai-edge';

const config = new Configuration({ apiKey: process.env.OPENAI_API_KEY });
const openai = new OpenAIApi(config);

export async function POST(req: Request) {
  const { messages } = await req.json();
  
  const response = await openai.createChatCompletion({
    model: 'gpt-4o',
    stream: true,
    messages,
  });

  const stream = OpenAIStream(response);
  return new StreamingTextResponse(stream);
}

Client (với useChat hook):

import { useChat } from 'ai/react';

function Chat() {
  const { messages, input, handleInputChange, handleSubmit } = useChat();

  return (
    <div>
      {messages.map(m => <div key={m.id}>{m.content}</div>)}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
      </form>
    </div>
  );
}

Vercel AI SDK xử lý streaming, parsing, state management tự động.

Xử Lý Lỗi Mid-Stream

Thách thức: Stream có thể bị ngắt giữa chừng (network timeout, rate limit, server crash). Khác với batch (retry toàn bộ), streaming cần chiến lược phức tạp hơn.

Best practices:

1. Detect incomplete stream:

full_response = ""
try:
    for chunk in stream:
        full_response += chunk.choices[0].delta.content
except Exception as e:
    # Stream bị ngắt
    if not full_response.endswith(expected_ending):
        # Retry hoặc thông báo user
        print(f"Stream incomplete: {e}")

2. Graceful degradation:

• Hiển thị partial response + nút “Retry”.
• Lưu partial response để user không mất hết.

3. Timeout:

import signal

def timeout_handler(signum, frame):
    raise TimeoutError("Stream timeout")

signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(30)  # 30 giây timeout

try:
    for chunk in stream:
        # process
        signal.alarm(30)  # reset timer mỗi chunk
finally:
    signal.alarm(0)  # cancel timeout

4. Server-side retry logic:

• Exponential backoff cho transient errors (network glitch).
• KHÔNG retry cho permanent errors (invalid API key, quota exceeded).

Performance: Streaming vs Batch

Benchmark thực tế (GPT-4o, câu hỏi 50 từ, response 500 token):

Metric	Batch	Streaming
Time To First Token (TTFT)	2.3s	0.4s
Total Time	7.8s	8.1s
Perceived Speed (user survey)	3.2/5	4.7/5
Bandwidth	100%	103%

Nhận xét:

• Streaming chậm hơn 3-5% về total time (overhead SSE).
• Nhưng nhanh hơn 5x về TTFT.
• User cảm nhận nhanh hơn 46%.

Khi nào batch tốt hơn streaming:

• Background jobs (không có user chờ).
• Cần toàn bộ response để validate (ví dụ check JSON syntax).
• Rate limit nghiêm (mỗi request tốn quota → muốn minimize số request).

So Sánh Các Phương Pháp Streaming

Phương pháp	Protocol	Use case	Complexity
SSE	HTTP	One-way server→client (chatbot, assistant)	Thấp
WebSocket	WS	Two-way (voice chat, collaborative editing)	Trung bình
HTTP Chunked Transfer	HTTP/1.1	Legacy systems không hỗ trợ SSE	Thấp
gRPC Streaming	HTTP/2	Microservices, high-throughput	Cao

Khuyến nghị:

• Mặc định dùng SSE cho chatbot/assistant.
• WebSocket nếu cần client gửi message mid-stream.
• gRPC cho internal services (không public-facing).

Câu Hỏi Thường Gặp (FAQ)

Streaming AI responses có nhanh hơn batch processing không?

Không nhanh hơn về tổng thời gian xử lý, nhưng user nhận được token đầu tiên sớm hơn (Time To First Token < 500ms thay vì phải đợi toàn bộ response). Trải nghiệm cảm nhận nhanh hơn nhiều.

Khi nào nên dùng streaming thay vì batch?

Dùng streaming cho chatbot, assistant, UI tương tác trực tiếp với user. Dùng batch cho API backend, batch processing, khi cần toàn bộ output để xử lý tiếp (ví dụ lưu database, phân tích sentiment).

Streaming tốn bandwidth nhiều hơn không?

Hơi nhiều hơn do overhead của Server-Sent Events (SSE) hoặc WebSocket, nhưng không đáng kể (< 5%). Lợi ích UX bù đắp hoàn toàn.

LangChain và LlamaIndex có hỗ trợ streaming không?

Có. LangChain hỗ trợ .stream() và callback handlers, LlamaIndex có response_mode='stream'. Cả hai đều tương thích với OpenAI và Anthropic streaming.

Làm sao cache streaming responses?

Hai cách:

1. Cache sau khi stream xong: Ghép toàn bộ chunks lại, lưu vào cache (Redis, database), request tiếp theo trả batch.
2. Incremental cache: Lưu từng chunk vào cache (phức tạp hơn, ít dùng).

Đa số hệ thống dùng cách 1 (đơn giản hơn).

Streaming có tương thích với function calling không?

Có, nhưng phức tạp hơn:

• OpenAI streaming trả delta.tool_calls incremental (phải ghép từng phần lại).
• Anthropic trả toàn bộ tool call trong một event.

Ví dụ OpenAI:

tool_calls = {}
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.tool_calls:
        for tc in delta.tool_calls:
            if tc.index not in tool_calls:
                tool_calls[tc.index] = {"name": "", "arguments": ""}
            tool_calls[tc.index]["name"] += tc.function.name or ""
            tool_calls[tc.index]["arguments"] += tc.function.arguments or ""

Sau khi stream xong, parse arguments (JSON string) và execute function.

Kết Luận

Streaming AI responses đã trở thành chuẩn mực cho ứng dụng chat/assistant. Độ phức tạp kỹ thuật cao hơn batch, nhưng Time To First Token thấp đáng kể — user thấy kết quả ngay trong < 500ms thay vì đợi cả câu trả lời xong.

Bốn điểm cốt lõi để triển khai tốt:

1. Dùng SSE cho đa số use case (đơn giản nhất).
2. Implement error handling mid-stream cẩn thận — stream có thể bị ngắt bất cứ lúc nào.
3. Test với network throttle và timeout.
4. Monitor TTFT như metric quan trọng ngang latency tổng.

Vercel AI SDK và LangChain giải quyết phần lớn boilerplate. Bạn chỉ cần focus vào logic nghiệp vụ.

Đọc thêm:

• Function Calling Trong AI: Cách LLM Gọi Công Cụ Thực Tế — streaming kết hợp function calling cho AI agents.
• RAG Là Gì? Cách Doanh Nghiệp Dùng AI Trả Lời Từ Dữ Liệu Riêng — streaming RAG responses để giảm latency.
• Agentic AI Workflows: Orchestration Hệ Thống AI Agents — streaming trong multi-agent systems phức tạp hơn thế nào.