Streaming & SSE

LLM responses generate one token at a time over seconds to minutes. Waiting for the full response before sending it to the user creates unacceptable latency. Streaming via Server-Sent Events (SSE) delivers each token immediately, giving users sub-second time-to-first-token even for long responses.

---

1. The OpenAI-Compatible SSE Format

The de facto standard for LLM streaming:

POST /v1/chat/completions
{"model": "llama-70b", "messages": [...], "stream": true}

Response (text/event-stream):
data: {"id":"chatcmpl-1","choices":[{"delta":{"role":"assistant"},"index":0}]}

data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"Hello"},"index":0}]}

data: {"id":"chatcmpl-1","choices":[{"delta":{"content":" world"},"index":0}]}

data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"!"},"index":0}],"usage":{"prompt_tokens":10,"completion_tokens":3}}

data: [DONE]

Key details:

• Each chunk is a data: line followed by a blank line
• The delta field contains incremental content (not cumulative)
• The last data chunk includes usage (token counts)
• data: [DONE] signals stream end
• No event: field used (just data:)

---

2. Server Implementation

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio, json, time

app = FastAPI()

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest):
    if not request.stream:
        # Non-streaming: wait for full response
        response = await engine.generate(request)
        return JSONResponse(response.dict())
    
    # Streaming: yield tokens as generated
    async def token_stream():
        request_id = f"chatcmpl-{uuid4().hex[:8]}"
        input_tokens = count_tokens(request.messages)
        output_tokens = 0
        
        # First chunk: role
        yield f"data: {json.dumps(format_chunk(request_id, delta={'role': 'assistant'}))}\n\n"
        
        # Token chunks
        async for token in engine.stream(request):
            output_tokens += 1
            chunk = format_chunk(request_id, delta={"content": token.text})
            
            # Include usage in final chunk
            if token.is_last:
                chunk["usage"] = {
                    "prompt_tokens": input_tokens,
                    "completion_tokens": output_tokens,
                    "total_tokens": input_tokens + output_tokens,
                }
            
            yield f"data: {json.dumps(chunk)}\n\n"
        
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        token_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Disable Nginx buffering
        },
    )

---

3. Client Implementation

import httpx, json

async def stream_chat(messages: list[dict], model: str = "llama-70b"):
    """Consume SSE stream from LLM endpoint."""
    async with httpx.AsyncClient(timeout=120) as client:
        async with client.stream(
            "POST", "http://llm-gateway/v1/chat/completions",
            json={"model": model, "messages": messages, "stream": True},
            headers={"Authorization": "Bearer sk-xxx"},
        ) as response:
            full_text = ""
            async for line in response.aiter_lines():
                if not line.startswith("data: "):
                    continue
                
                data = line[6:]  # Strip "data: " prefix
                if data == "[DONE]":
                    break
                
                chunk = json.loads(data)
                delta = chunk["choices"][0].get("delta", {})
                content = delta.get("content", "")
                
                if content:
                    full_text += content
                    print(content, end="", flush=True)  # Print token by token
            
            return full_text

Reconnection Handling

SSE connections can drop (network issues, proxy timeouts). Implement retry:

async def stream_with_retry(messages, model, max_retries=3):
    """Stream with automatic reconnection."""
    for attempt in range(max_retries):
        try:
            return await stream_chat(messages, model)
        except (httpx.ReadTimeout, httpx.RemoteProtocolError) as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # Exponential backoff
                logger.warning(f"Stream interrupted, retrying in {wait}s: {e}")
                await asyncio.sleep(wait)
            else:
                raise

---

4. Backpressure Handling

What happens when the client can’t consume tokens as fast as the server generates them?

Server generates: 50 tokens/sec
Client processes: 10 tokens/sec (slow network, busy rendering)

Without backpressure: Server buffers tokens in memory → OOM
With backpressure:    Server slows generation to match client speed

Server-Side Backpressure

async def token_stream_with_backpressure():
    """Detect slow clients and apply backpressure."""
    buffer = asyncio.Queue(maxsize=100)  # Buffer up to 100 tokens
    
    # Producer: generate tokens
    async def produce():
        async for token in engine.stream(request):
            try:
                buffer.put_nowait(token)
            except asyncio.QueueFull:
                # Client is slow — wait for buffer to drain
                logger.warning("Client backpressure — buffer full")
                await buffer.put(token)  # Blocks until space available
    
    producer = asyncio.create_task(produce())
    
    # Consumer: yield to client at client's pace
    while True:
        token = await asyncio.wait_for(buffer.get(), timeout=30)
        if token is DONE_SENTINEL:
            break
        yield f"data: {json.dumps(format_token(token))}\n\n"
    
    yield "data: [DONE]\n\n"

Connection Timeout

Kill streams that stall:

# Server: kill idle streams after 30s with no token sent
STREAM_IDLE_TIMEOUT = 30  # seconds between tokens

# Client: kill streams that don't produce tokens for 60s
async with httpx.AsyncClient(timeout=httpx.Timeout(
    connect=5.0,
    read=60.0,      # 60s between any data received
    write=5.0,
    pool=5.0,
)) as client:
    ...

---

5. Proxy Configuration

Every proxy between server and client must be configured for streaming:

Nginx

location /v1/chat/completions {
    proxy_pass http://llm_backend;
    
    # CRITICAL for streaming:
    proxy_buffering off;           # Don't buffer response body
    proxy_cache off;               # Don't cache streamed responses
    proxy_http_version 1.1;        # HTTP/1.1 for chunked encoding
    proxy_set_header Connection "";
    
    # Timeouts for long generations:
    proxy_read_timeout 300s;       # 5 min between chunks
    proxy_send_timeout 300s;
    
    # Disable response buffering header:
    add_header X-Accel-Buffering no;
    
    # Buffer size (small — just for headers):
    proxy_buffer_size 4k;
    proxy_buffers 4 4k;
}

AWS ALB/CloudFront

ALB: Streaming works by default (HTTP/1.1 chunked)
     - Set idle timeout to 300s (default 60s is too short for LLMs)
     
CloudFront: DOES NOT support SSE well
     - Buffers response until complete
     - Use ALB directly, or use WebSocket alternative

Common Proxy Pitfalls

Issue	Symptom	Fix
Response buffering ON	Client gets all tokens at once	proxy_buffering off
Read timeout too short	Stream drops mid-generation	Increase to 300s
CDN buffering	Tokens arrive in bursts	Bypass CDN for streaming
HTTP/2 multiplexing	Works but complex	HTTP/1.1 is simpler for SSE
gzip compression	Delays token delivery	Disable compression for SSE

---

6. Aggregation: Logging Complete Responses

The gateway needs the full response for logging, but tokens stream through:

class StreamAggregator:
    """Aggregate streaming tokens while forwarding to client."""
    
    async def proxy_and_log(self, request, backend):
        full_response = []
        input_tokens = 0
        output_tokens = 0
        start_time = time.monotonic()
        ttft = None
        
        async def aggregating_stream():
            nonlocal ttft, output_tokens
            async for chunk_bytes in backend.stream(request):
                chunk = json.loads(chunk_bytes)
                delta = chunk.get("choices", [{}])[0].get("delta", {})
                content = delta.get("content", "")
                
                if content:
                    if ttft is None:
                        ttft = time.monotonic() - start_time
                    full_response.append(content)
                    output_tokens += 1
                
                yield f"data: {chunk_bytes.decode()}\n\n"
            
            yield "data: [DONE]\n\n"
            
            # Log after stream completes
            await log_request(
                prompt=request.messages,
                response="".join(full_response),
                ttft_ms=ttft * 1000,
                total_ms=(time.monotonic() - start_time) * 1000,
                output_tokens=output_tokens,
            )
        
        return StreamingResponse(aggregating_stream(), media_type="text/event-stream")

---

Self-Check

• Implement an OpenAI-compatible SSE streaming endpoint
• Handle client reconnection with exponential backoff
• Configure Nginx to properly proxy SSE streams without buffering
• Implement backpressure detection for slow clients
• Aggregate streaming tokens for logging while forwarding to the client

---

References

• Server-Sent Events (SSE) Specification
• OpenAI API Streaming Documentation
• For gateway integration, see API Gateway