Bài 8: Streaming response cho LLM API

Sau bài này bạn sẽ:

• Biết sự khác biệt giữa StreamingResponse và SSE (EventSourceResponse), khi nào dùng cái nào.
• Viết được generator async để stream token từ LLM API (OpenAI, Anthropic) qua FastAPI.
• Dùng TextIteratorStreamer để stream từ local Hugging Face model chạy trên thread riêng.
• Biết cách client (browser, curl, httpx) nhận stream.
• Xử lý lỗi giữa chừng và tránh các pitfall phổ biến.

LLM sinh text token-by-token theo kiểu autoregressive: mỗi token được tạo ra dựa trên các token trước, không thể song song hóa giữa các bước trong cùng một chuỗi. Với response dài 500 token và tốc độ 50 token/giây, nếu không streaming, user phải chờ 10 giây mới thấy chữ đầu tiên.

Với streaming, token đầu tiên xuất hiện sau khoảng 0.2–0.5 giây (time to first token — TTFT), và các token tiếp theo liên tục hiện ra trong khi model vẫn đang generate. Đây là lý do tại sao ChatGPT, Claude.ai, và hầu hết LLM UI đều dùng streaming.

3 cách stream qua HTTP

Tất cả đều dùng HTTP/1.1 chunked transfer encoding — server gửi body theo từng chunk nhỏ thay vì chờ toàn bộ response.

Với LLM API server, StreamingResponse và SSE đủ cho hầu hết trường hợp. WebSocket có overhead kết nối cao hơn và phức tạp hơn, thường chỉ dùng khi cần hai chiều (user gửi giữa chừng trong khi model đang generate).

StreamingResponse nhận một iterable (sync hoặc async generator) và gửi từng chunk ngay khi generator yield giá trị. Uvicorn flush chunk qua socket ngay lập tức — không đợi generator kết thúc.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    prompt: str

def fake_llm_stream(prompt: str):
    """Simulate LLM streaming — yield từng token."""
    words = f"Câu trả lời cho '{prompt}': token1 token2 token3 ...".split()
    for word in words:
        yield word + " "

@app.post("/chat")
async def chat(req: ChatRequest):
    def token_generator():
        for chunk in fake_llm_stream(req.prompt):
            yield chunk
    return StreamingResponse(token_generator(), media_type="text/plain")

Luồng hoạt động:

1. FastAPI nhận request POST /chat.
2. Endpoint trả về StreamingResponse với generator làm content.
3. Uvicorn gọi generator từng bước. Mỗi lần generator yield một chunk, uvicorn gửi chunk đó xuống TCP socket ngay (HTTP/1.1 chunked encoding).
4. Client nhận từng chunk riêng lẻ — không cần chờ response hoàn tất.

Async generator

Nếu nguồn data là async (gọi LLM API qua await), dùng async def generator:

@app.post("/chat-async")
async def chat_async(req: ChatRequest):
    async def token_generator():
        async for chunk in async_llm_stream(req.prompt):
            yield chunk
    return StreamingResponse(token_generator(), media_type="text/plain")

StreamingResponse hỗ trợ cả sync iterator và async generator. Với LLM API (network I/O), dùng async generator để event loop không bị block.

SSE là chuẩn HTML5 cho stream đơn chiều server → client. Format của mỗi event là plain text, mỗi field trên 1 dòng, event kết thúc bằng 2 ký tự newline:

data: Hello\n\n
data: World\n\n
event: done\ndata: \n\n

Fields hỗ trợ:

• data: <content> — nội dung event (bắt buộc).
• event: <type> — tên event tùy chỉnh (mặc định là message).
• id: <id> — dùng cho reconnection, client gửi lại Last-Event-ID header.
• retry: <ms> — thời gian client chờ trước khi reconnect (mili giây).

SSE với sse-starlette

Library sse-starlette cung cấp EventSourceResponse — wrapper tiện cho FastAPI, tự động format SSE và set header đúng (Content-Type: text/event-stream, Cache-Control: no-cache).

pip install sse-starlette

import json
from fastapi import FastAPI
from pydantic import BaseModel
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

class ChatRequest(BaseModel):
    prompt: str

@app.post("/chat-sse")
async def chat_sse(req: ChatRequest):
    async def event_generator():
        # Giả lập stream từ LLM
        tokens = ["Hello", " world", "!", " How", " are", " you?"]
        for token in tokens:
            # yield dict → sse-starlette format tự động
            yield {
                "event": "token",
                "data": json.dumps({"type": "token", "content": token})
            }
        # Event kết thúc
        yield {
            "event": "done",
            "data": json.dumps({"type": "done"})
        }

    return EventSourceResponse(event_generator())

Output wire format mà client nhận được:

event: token
data: {"type": "token", "content": "Hello"}

event: token
data: {"type": "token", "content": " world"}

...

event: done
data: {"type": "done"}

SSE vs StreamingResponse — khi nào chọn cái nào

Dùng StreamingResponse khi:

• Client là non-browser (curl, httpx, Python script).
• Chỉ cần stream text thuần, không cần event type.
• Muốn giảm overhead format tối thiểu.

Dùng SSE khi:

• Frontend dùng React/JavaScript và cần phân biệt loại event (token, error, done).
• Cần browser EventSource API với auto-reconnect.
• Muốn gửi metadata kèm mỗi chunk (ví dụ: finish reason, usage stats).

OpenAI SDK (openai>=1.0) hỗ trợ streaming qua stream=True. Mỗi chunk trong stream có choices[0].delta.content — có thể là None (chunk đầu) hoặc string (token).

import json
import os
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
openai_client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4o-mini"

@app.post("/chat/openai/stream")
async def chat_openai_stream(req: ChatRequest):
    async def token_generator():
        stream = await openai_client.chat.completions.create(
            model=req.model,
            messages=req.messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content is not None:
                yield delta.content

    return StreamingResponse(token_generator(), media_type="text/plain")

Nếu muốn trả về SSE JSON (để frontend có thể parse finish_reason, usage, v.v.):

from sse_starlette.sse import EventSourceResponse

@app.post("/chat/openai/sse")
async def chat_openai_sse(req: ChatRequest):
    async def event_generator():
        try:
            stream = await openai_client.chat.completions.create(
                model=req.model,
                messages=req.messages,
                stream=True,
            )
            async for chunk in stream:
                delta = chunk.choices[0].delta
                finish_reason = chunk.choices[0].finish_reason
                if delta.content is not None:
                    yield {
                        "event": "token",
                        "data": json.dumps({
                            "type": "token",
                            "content": delta.content
                        })
                    }
                if finish_reason is not None:
                    yield {
                        "event": "done",
                        "data": json.dumps({
                            "type": "done",
                            "finish_reason": finish_reason
                        })
                    }
        except Exception as e:
            yield {
                "event": "error",
                "data": json.dumps({"type": "error", "message": str(e)})
            }

    return EventSourceResponse(event_generator())

Lưu ý: AsyncOpenAI trả về async stream — iterate bằng async for. SDK cũng có OpenAI (sync client) với stream=True trả về sync iterator, nhưng dùng sync client trong async def endpoint là blocking — nên dùng AsyncOpenAI.

Anthropic SDK (anthropic>=0.30) cung cấp AsyncAnthropic với context manager client.messages.stream(). Iterate stream.text_stream để lấy text chunks.

import json
import os
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic

app = FastAPI()
anthropic_client = anthropic.AsyncAnthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

class ChatRequest(BaseModel):
    messages: list[dict]
    system: str = "You are a helpful assistant."
    model: str = "claude-3-5-haiku-20241022"
    max_tokens: int = 1024

@app.post("/chat/anthropic/stream")
async def chat_anthropic_stream(req: ChatRequest):
    async def token_generator():
        async with anthropic_client.messages.stream(
            model=req.model,
            max_tokens=req.max_tokens,
            system=req.system,
            messages=req.messages,
        ) as stream:
            async for text in stream.text_stream:
                yield text

    return StreamingResponse(token_generator(), media_type="text/plain")

Nếu cần truy cập stop reason và usage sau khi stream xong, dùng await stream.get_final_message() bên trong context manager:

from sse_starlette.sse import EventSourceResponse

@app.post("/chat/anthropic/sse")
async def chat_anthropic_sse(req: ChatRequest):
    async def event_generator():
        try:
            async with anthropic_client.messages.stream(
                model=req.model,
                max_tokens=req.max_tokens,
                system=req.system,
                messages=req.messages,
            ) as stream:
                async for text in stream.text_stream:
                    yield {
                        "event": "token",
                        "data": json.dumps({"type": "token", "content": text})
                    }
                # Lấy thông tin kết thúc sau khi stream hoàn tất
                final = await stream.get_final_message()
                yield {
                    "event": "done",
                    "data": json.dumps({
                        "type": "done",
                        "stop_reason": final.stop_reason,
                        "input_tokens": final.usage.input_tokens,
                        "output_tokens": final.usage.output_tokens
                    })
                }
        except Exception as e:
            yield {
                "event": "error",
                "data": json.dumps({"type": "error", "message": str(e)})
            }

    return EventSourceResponse(event_generator())

Anthropic cũng hỗ trợ cách thứ hai — raw event streaming qua client.messages.create(stream=True) — nhưng client.messages.stream() với text_stream thường đủ dùng và code gọn hơn.

Hugging Face Transformers cung cấp TextIteratorStreamer (từ transformers>=4.36) — một streamer object có thể iterate để lấy token từng bước trong khi model đang generate.

Cơ chế hoạt động: model.generate() là blocking — nó không trả về cho đến khi sinh xong toàn bộ sequence. Để stream được, chạy model.generate() trong một thread riêng và đặt TextIteratorStreamer làm streamer argument. Main thread iterate streamer trong khi thread kia generate.

import asyncio
import threading
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from transformers import AutoTokenizer, AutoModelForCausalLM, TextIteratorStreamer

app = FastAPI()

# Dùng model nhẹ cho demo — SmolLM2-135M hoặc Llama-3.2-1B
MODEL_ID = "HuggingFaceTB/SmolLM2-135M-Instruct"

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForCausalLM.from_pretrained(MODEL_ID)
model.eval()

class ChatRequest(BaseModel):
    prompt: str
    max_new_tokens: int = 200

@app.post("/chat/local/stream")
async def chat_local_stream(req: ChatRequest):
    async def token_generator():
        # Tokenize input
        inputs = tokenizer(req.prompt, return_tensors="pt")
        input_ids = inputs["input_ids"]

        # TextIteratorStreamer: skip_prompt=True bỏ qua phần echo input
        streamer = TextIteratorStreamer(
            tokenizer,
            skip_prompt=True,
            skip_special_tokens=True
        )

        generate_kwargs = {
            "input_ids": input_ids,
            "streamer": streamer,
            "max_new_tokens": req.max_new_tokens,
            "do_sample": False,
        }

        # Chạy generate trong thread riêng — không block event loop
        thread = threading.Thread(target=model.generate, kwargs=generate_kwargs)
        thread.start()

        # Main thread (trong async generator) iterate streamer
        # streamer là iterable — nó block cho đến khi có token mới hoặc generation xong
        for text_chunk in streamer:
            yield text_chunk
            # Nhường event loop để FastAPI xử lý request khác
            await asyncio.sleep(0)

        thread.join()

    return StreamingResponse(token_generator(), media_type="text/plain")

Điểm quan trọng:

• await asyncio.sleep(0) trong generator — cần thiết để yield control về event loop giữa các token, cho phép uvicorn flush chunk và xử lý request khác. Nếu bỏ qua, generator chạy xong hết rồi mới trả về (buffer toàn bộ).
• skip_prompt=True — tránh echo lại phần input trong output.
• skip_special_tokens=True — bỏ các token đặc biệt như <eos>, <pad>.
• Với model nặng trên GPU, thêm .to("cuda") cho cả model lẫn input_ids.

Lưu ý về thread safety

Nếu nhiều request gọi cùng lúc vào cùng 1 model instance, model.generate() sẽ bị serialized bởi GIL (và GPU context nếu dùng GPU). Với mục đích demo hoặc load thấp thì chấp nhận được. Production cần model pool hoặc batching (Bài 51).

curl

Flag -N (hoặc --no-buffer) tắt buffering của curl, in chunk ngay khi nhận:

# StreamingResponse (text/plain)
curl -N -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Giải thích gradient descent"}'

# SSE endpoint
curl -N -X POST http://localhost:8000/chat-sse \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Giải thích gradient descent"}'

httpx (Python client)

Dùng httpx.stream() với context manager, iterate response.iter_text() hoặc iter_lines() để xử lý từng chunk:

import httpx

def consume_stream(prompt: str):
    with httpx.stream(
        "POST",
        "http://localhost:8000/chat",
        json={"prompt": prompt},
        timeout=60.0
    ) as response:
        for chunk in response.iter_text():
            print(chunk, end="", flush=True)
    print()  # newline cuối

consume_stream("Giải thích gradient descent")

Với SSE endpoint, parse từng dòng để tách data: field:

import json
import httpx

def consume_sse(prompt: str):
    with httpx.stream(
        "POST",
        "http://localhost:8000/chat-sse",
        json={"prompt": prompt},
        timeout=60.0
    ) as response:
        for line in response.iter_lines():
            if line.startswith("data: "):
                raw = line[len("data: "):]
                try:
                    event = json.loads(raw)
                    if event["type"] == "token":
                        print(event["content"], end="", flush=True)
                    elif event["type"] == "done":
                        print()
                        break
                    elif event["type"] == "error":
                        print(f"\nError: {event['message']}")
                        break
                except json.JSONDecodeError:
                    pass

consume_sse("Giải thích gradient descent")

Browser — fetch + ReadableStream

Browser EventSource API chỉ hỗ trợ GET. Với POST (phần lớn chat endpoint), dùng fetch + ReadableStream.getReader():

async function streamChat(prompt) {
  const response = await fetch("/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ prompt }),
  });

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

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const text = decoder.decode(value, { stream: true });
    // Append text vào UI
    document.getElementById("output").textContent += text;
  }
}

Với SSE endpoint qua fetch:

async function streamSSE(prompt) {
  const response = await fetch("/chat-sse", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ prompt }),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });

    // Parse SSE line by line
    const lines = buffer.split("\n");
    buffer = lines.pop(); // giữ lại dòng chưa hoàn chỉnh

    for (const line of lines) {
      if (line.startsWith("data: ")) {
        try {
          const event = JSON.parse(line.slice(6));
          if (event.type === "token") {
            document.getElementById("output").textContent += event.content;
          }
        } catch {}
      }
    }
  }
}

Browser EventSource GET-only vẫn hữu ích cho endpoint không cần body (ví dụ stream log, status updates). Chỉ với LLM chat — thường cần POST body — thì phải dùng fetch.

Khi streaming, HTTP status code đã được gửi (200 OK) trước khi generator chạy — không thể đổi lại nếu lỗi xảy ra giữa chừng. Thay vào đó, yield error chunk theo format đã định trước để client parse.

import json
import asyncio
from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel
from openai import AsyncOpenAI, APIStatusError, APIConnectionError

app = FastAPI()
openai_client = AsyncOpenAI()

class ChatRequest(BaseModel):
    messages: list[dict]

@app.post("/chat/safe-stream")
async def chat_safe_stream(req: ChatRequest):
    async def event_generator():
        try:
            stream = await openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=req.messages,
                stream=True,
            )
            async for chunk in stream:
                delta = chunk.choices[0].delta
                if delta.content is not None:
                    yield {
                        "event": "token",
                        "data": json.dumps({"type": "token", "content": delta.content})
                    }
            yield {
                "event": "done",
                "data": json.dumps({"type": "done"})
            }
        except APIStatusError as e:
            # Rate limit, auth error, v.v.
            yield {
                "event": "error",
                "data": json.dumps({
                    "type": "error",
                    "code": e.status_code,
                    "message": e.message
                })
            }
        except APIConnectionError:
            yield {
                "event": "error",
                "data": json.dumps({
                    "type": "error",
                    "message": "Connection to LLM API failed"
                })
            }
        except asyncio.CancelledError:
            # Client đóng connection — cleanup nếu cần, sau đó re-raise
            raise

    return EventSourceResponse(event_generator())

Client cancel — asyncio.CancelledError

Khi client đóng connection giữa chừng (user bấm stop, page reload), uvicorn cancel coroutine đang chạy — generator nhận asyncio.CancelledError. Nếu cần cleanup (đóng DB connection, release lock), dùng try/finally:

async def event_generator():
    try:
        async with some_resource() as res:
            async for chunk in res.stream():
                yield {"data": chunk}
    except asyncio.CancelledError:
        # Cleanup nếu cần — resource trong finally hoặc context manager
        raise  # Re-raise để uvicorn biết request bị cancel
    finally:
        # Đảm bảo resource luôn được release
        pass

Uvicorn không buffer response — nó gửi từng chunk xuống socket ngay khi generator yield. HTTP/1.1 chunked transfer encoding không cần config thêm.

Vấn đề xảy ra khi có nginx phía trước. Nginx mặc định buffer toàn bộ response từ upstream trước khi gửi đến client — streaming bị vô hiệu hóa hoàn toàn, client nhận 1 cục dù server đã gửi chunk từng bước.

Cách 1: Header per-response

Thêm header X-Accel-Buffering: no vào response — nginx đọc header này và tắt buffering cho response đó:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(req: ChatRequest):
    async def token_generator():
        # ... yield tokens
        pass

    return StreamingResponse(
        token_generator(),
        media_type="text/plain",
        headers={"X-Accel-Buffering": "no"}
    )

Cách 2: Nginx config cho location cụ thể

Tắt buffering trong nginx config cho endpoint streaming:

location /chat {
    proxy_pass http://127.0.0.1:8000;
    proxy_buffering off;
    proxy_cache off;
    # Quan trọng với HTTP/1.1 — nginx cần giữ kết nối keep-alive
    proxy_http_version 1.1;
    proxy_set_header Connection "";
}

Cả 2 cách đều hoạt động — dùng header per-response khi không muốn sửa nginx config (ví dụ môi trường shared). Dùng nginx config khi có quyền kiểm soát server và muốn áp dụng cho toàn bộ endpoint.

Kiểm tra buffering

Cách đơn giản nhất để kiểm tra: dùng curl với -N và thêm time.sleep(0.1) vào generator. Nếu token hiện ra từng bước (cách nhau ~100ms) — không bị buffer. Nếu tất cả hiện ra cùng lúc cuối cùng — đang bị buffer ở đâu đó.

Backpressure

Nếu client nhận chậm hơn server generate, các chunk được buffer trong OS socket buffer của uvicorn. Với LLM, mỗi token là vài byte — buffer tích lũy không đáng kể kể cả khi model generate nhanh hơn user đọc. Chỉ trở thành vấn đề nếu response rất dài (hàng chục nghìn token) và client bị lag mạng nặng.

Pitfall 1 — Generator đã exhausted

Generator chỉ iterate được 1 lần. Nếu truyền iterable đã dùng xong vào StreamingResponse, client nhận response rỗng:

# SAI: tokens là list đã tạo sẵn, không phải generator
tokens = list(some_function())  # đã consume iterable
return StreamingResponse(iter(tokens), media_type="text/plain")
# → OK, iter(tokens) tạo iterator mới từ list

# SAI hơn: generator đã dùng một lần
gen = some_generator()
_ = list(gen)       # exhaust generator
return StreamingResponse(gen, media_type="text/plain")
# → client nhận empty response

Nếu cần dùng lại data, convert sang list trước khi truyền vào StreamingResponse hoặc dùng factory function trả về generator mới mỗi lần gọi.

Pitfall 2 — Quên media_type

Mặc định StreamingResponse dùng application/octet-stream nếu không chỉ định media_type. Browser sẽ download file thay vì hiển thị. Luôn chỉ rõ:

StreamingResponse(generator(), media_type="text/plain")         # Text
StreamingResponse(generator(), media_type="text/event-stream")  # SSE thủ công
StreamingResponse(generator(), media_type="application/json")   # NDJSON

Pitfall 3 — Quên flush chunk cuối

Với httpx client, khi đọc stream xong cần gọi response.aclose() (hoặc dùng context manager) để đảm bảo buffer còn lại được flush và connection được đóng đúng cách:

async with httpx.AsyncClient() as client:
    async with client.stream("POST", url, json=payload) as response:
        async for chunk in response.aiter_text():
            print(chunk, end="")
# aclose() tự động khi thoát context manager

Pitfall 4 — SSE qua nginx không có proxy_buffering off

Đã đề cập ở mục 10 — đây là nguyên nhân phổ biến nhất khiến streaming "không hoạt động" trên production dù test local với curl thì chạy đúng.

Pitfall 5 — Thiếu await asyncio.sleep(0) với TextIteratorStreamer

Với local model, nếu bỏ await asyncio.sleep(0) trong vòng lặp iterate streamer, generator sẽ chạy hết vòng lặp mà không yield control về event loop — toàn bộ token bị buffer trong memory của uvicorn, client nhận 1 cục cuối cùng. Đã trình bày chi tiết ở mục 7.

Bài 9: Gradio — demo UI trong vài dòng code — Cách dùng Gradio để tạo UI demo cho model AI mà không cần viết HTML/CSS/JS.