$ cat ai-engineering/streaming.md

Streaming and Backpressure

How SSE streaming handles partial output, cancellation, tool-call JSON, and backpressure.

Jatin Bansal@blog:~/ai-engineering$ open streaming

A support chatbot ships at 09:00. By 10:30 the on-call has a flood of complaints: “the bot froze on me.” The dashboards say latency p50 is 4.8s, p99 is 12s; well within the budget on paper. The bug isn’t latency. The bot is not streaming. Users see a spinner for five seconds, then a wall of text. The same five seconds with token-by-token streaming would have read as fast even though the total wall time is identical. Streaming is rarely about throughput; it is about time-to-first-token (TTFT) becoming visible to the user, and about giving the user a cancel button that actually works.

Streaming semantics

Streaming pushes the decode loop’s output incrementally over one open HTTP response instead of buffering the complete message. Anthropic, OpenAI, Google, Mistral, and open-source serving stacks expose this primitive through stream=true or a dedicated stream method. The client receives typed server-sent events until a terminal event closes the response.

Streaming changes several observable properties:

TTFT becomes the latency that matters. Total wall time is unchanged; perceived latency drops to the time-to-first-token. On a 1,500-token answer at ~80 tokens/sec, that’s 12.5s down to ~200ms; a 60× reduction in perceived latency for the same compute.
Cancellation is cheap. Closing the HTTP connection mid-stream interrupts provider-side decoding, so billing stops near the cancellation point.
Progressive rendering. The interface can paint text and tool-call arguments as they arrive, showing activity before a tool call is complete.
Bounded memory. The server never holds the full response in memory; each chunk is forwarded to the client and discarded. For a 100k-token response this is the difference between a 500MB buffer per request and a fixed ~64KB window.

Streaming leaves throughput unchanged. The model still decodes one token per step; total tokens-per-second is unchanged. Reducing total generation time requires changes such as a smaller model, prompt caching, or latency optimization.

Producer and consumer

The GPU produces tokens during decoding, and the client consumes them in its rendering loop. The provider API, reverse proxy, CDN, and application backend sit between those endpoints. Any layer can buffer, batch, or drop chunks.

The implicit contract: every actor in the chain forwards each chunk immediately and applies no buffering beyond what TCP requires. The implicit failure mode: somebody in the chain buffers. The most common culprits are gzip compression layers that wait for a window of bytes before flushing, Nginx’s default proxy_buffering on which holds the upstream response in memory until close, and CDN edges that don’t recognize the text/event-stream MIME type. If an API streams while the deployed application waits for the complete message, the proxy chain is the likely cause.

SSE event types

Both Anthropic and OpenAI use SSE frames of the form event: <name>\ndata: <json>\n\n, but their event grammars differ.

Anthropic Messages API streaming uses a block-oriented grammar. A response is bracketed by message_start and message_stop. Inside, each content block (a text block, a tool_use block, a thinking block) is bracketed by content_block_start / content_block_stop, with a sequence of content_block_delta events carrying the incremental payload. The delta type identifies the payload:

text_delta; append .text to the text buffer.
input_json_delta; append .partial_json to a buffer indexed by the block index; this is the streaming form of a tool call’s input argument.
thinking_delta; append .thinking for extended thinking blocks; treat as a separate buffer.
signature_delta; the cryptographic signature for the thinking block, only emitted in extended-thinking mode.

message_delta carries top-level changes such as the final stop_reason and usage. ping is a keep-alive. An error event is fatal and closes the client stream.

The mental model: text and tool-call arguments are streamed in parallel. A single assistant turn that contains a text preamble and two tool calls will interleave four delta streams (one text, two input_json_delta, plus any thinking) under the same message_start. The accumulator must be indexed by (content_block_index, delta_type).

The OpenAI Responses API emits semantic events with the prefix response.*. The shape is similar: response.created opens the stream, response.output_text.delta carries text chunks by concatenating .delta strings, response.function_call_arguments.delta carries streaming tool-call arguments, and response.completed closes. Tool calls have their own progress events (response.function_call.in_progress, .completed) which the older Chat Completions streaming chunks did not. New integrations generally use Responses API; Chat Completions retains its older streaming format.

SSE fits most text-generation streams because it is unidirectional and uses a plain HTTP response. It provides:

Standard load balancers, CDNs, and reverse proxies handle SSE without special configuration. WebSockets need explicit Upgrade: websocket handling and sticky sessions at every hop.
Automatic reconnect via Last-Event-ID is part of the SSE spec; reconnect logic on WebSockets is the caller’s problem.
No protocol upgrade; the connection is HTTP throughout, observable in the same logs and traces as the rest of the stack.

WebSockets win when both sides need to send data on the same connection at high frequency; voice-mode interrupts mid-generation, collaborative agents, anything that needs full-duplex. For 95% of chat products and tool-using agents, SSE is the right transport and adding WebSockets is undirected complexity. The 2026 industry shift toward WebSockets for interactive AI is real but narrow: it’s bidirectional voice and real-time interrupt protocols, not chat. See Hivenet’s overview of SSE vs WebSockets for LLM apps for a careful breakdown.

Anthropic streaming in Python

Stream a message, accumulate text and a tool call in parallel, handle abort. Install: pip install anthropic. The Anthropic SDK ships a high-level messages.stream context manager that handles event accumulation, while the lower-level events remain useful for debugging framework behavior.

python

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
import json
import signal
from anthropic import Anthropic

client = Anthropic()

TOOLS = [{
    "name": "get_weather",
    "description": "Get current weather for a location.",
    "input_schema": {
        "type": "object",
        "properties": {"location": {"type": "string"}},
        "required": ["location"],
    },
}]

def stream_with_tools(user_msg: str) -> dict:
    text_parts: list[str] = []
    # tool call inputs arrive as a stream of JSON fragments per content block
    tool_inputs: dict[int, str] = {}
    tool_meta: dict[int, dict] = {}

    cancelled = {"flag": False}
    signal.signal(signal.SIGINT, lambda *_: cancelled.__setitem__("flag", True))

    with client.messages.stream(
        model="claude-opus-4-7",
        max_tokens=1024,
        tools=TOOLS,
        messages=[{"role": "user", "content": user_msg}],
    ) as stream:
        for event in stream:
            if cancelled["flag"]:
                stream.close()  # closes the underlying HTTP response
                break

            if event.type == "content_block_start":
                if event.content_block.type == "tool_use":
                    tool_meta[event.index] = {
                        "id": event.content_block.id,
                        "name": event.content_block.name,
                    }
                    tool_inputs[event.index] = ""
            elif event.type == "content_block_delta":
                d = event.delta
                if d.type == "text_delta":
                    text_parts.append(d.text)
                    print(d.text, end="", flush=True)
                elif d.type == "input_json_delta":
                    tool_inputs[event.index] += d.partial_json
            elif event.type == "message_stop":
                break

    # final, complete tool inputs — only safe to json.loads once the stream is done
    tools_finalized = [
        {"id": tool_meta[i]["id"], "name": tool_meta[i]["name"],
         "input": json.loads(tool_inputs[i])}
        for i in sorted(tool_inputs)
        if tool_inputs[i]
    ]
    return {"text": "".join(text_parts), "tools": tools_finalized}

Text and tool-call JSON arrive interleaved under different event.index values, so the accumulator keeps separate buffers by index. input_json_delta chunks are fragments of one JSON string and become safe for json.loads only after the matching content_block_stop. Incremental rendering requires a partial JSON parser. Calling stream.close() closes the HTTP socket, allowing the provider to stop decoding within a few tokens.

Partial JSON

input_json_delta carries raw JSON fragments that may need partial rendering. Naive incremental parsing; JSON.parse(buffer) after every delta; fails on every fragment until the very last one. The right tool is a permissive parser that closes the open brackets and quotes in flight: partial-json-parser (TypeScript) and partial-json-parser (Python) are the canonical libraries; LangChain’s JsonOutputParser ships an equivalent for its streaming chain APIs. Parser state should persist across calls; re-parsing from byte zero on every chunk turns the rendering loop O(n²) on response length, which gets brutal past 5–10k tokens. The same pattern shows up in structured output when rendering a generateObject-style payload as the model decodes it; the same parser applies.

Failure modes

Proxy buffering is the silent killer. Nginx defaults to proxy_buffering on, which holds the upstream response until the buffer fills or the upstream closes. The symptom is “the SDK streams, the curl streams, my deployed app doesn’t.” The proxy configuration requires proxy_buffering off, proxy_cache off, and X-Accel-Buffering: no as a response header from the application (Nginx honors it per-response). The same flag exists in different forms for Cloudflare (Cache-Control: no-transform plus disabling Auto Minify on the response), Vercel/Netlify Edge, and most other CDNs. Deployment testing is necessary because localhost bypasses the proxy.

Cancellation must propagate. A “Stop” button that just hides the streaming <div> while the upstream call keeps running is the worst kind of bug: the interface reports cancellation while generation and billing continue. The cancellation chain is: client AbortController → route handler req.signal → SDK abortSignal → fetch to provider → provider closes the GPU stream. Break any link and the cancel is decorative.

Tool-call deltas are interleaved, not sequential. The model can emit a text preamble and two parallel tool calls in the same assistant turn (Anthropic’s parallel tool use), and the content_block_delta events arrive interleaved under different indices. The accumulator must be a map keyed by index, not a single flat string. Treating the stream as a single linear text buffer collapses the two tool calls into one corrupted JSON blob.

Backpressure exists, but inverted. TCP receive-window pressure on a slow client will cause the upstream to pause decoding; the GPU sits idle, the billing meter still ticks for the reservation. The signal to watch is server-side throughput per stream dropping below 50–60 tokens/sec on a known-fast model. When proxying through Node.js, the response.write() calls return false when the client buffer is full; the handler must await drain rather than buffer in memory. The Node.js stream backpressure guide is the right reference. Skipping this is how a slow mobile client OOMs a Node worker handling a 100k-token response.

Accumulating the complete response can waste memory: an arr.push(token); return arr.join("") pattern that re-allocates on every chunk turns a streaming endpoint into a memory-heavy one. For a 50k-token response with ~2-character tokens, that’s ~2.5 million string allocations and 100KB of doubled-up buffers. Chunks should flow downstream immediately. When later processing needs the full text, the SDK can accumulate it once through finalText or accumulate().

Refusal mid-stream is a thing. The model can decide to refuse partway through a response; typical pattern is a partial text block followed by a message_delta carrying stop_reason: "refusal" (Anthropic) or a response.refusal.delta (OpenAI’s Responses API). The UI must handle the case where the stream ends with a refusal after the user has already seen some tokens; those orphaned tokens should not render as a complete answer.

Streaming composes with strict structured output. OpenAI’s response_format: json_schema and Anthropic’s strict tool use emit JSON fragments that may be invalid until the block is complete. Schema validation therefore waits until content_block_stop.

Retries on a streaming connection don’t compose naively. The standard SDK retry policy assumes a single buffered response. On a streaming call that drops 8k tokens in, retrying replays the whole prompt from scratch; billed again, with no automatic continuation. Long-form generation can use either a transparent retry only on initial-connection failures (most SDKs do this; check the docs), or an application-level checkpoint that re-prompts with <continue from here> semantics, which is awkward and prone to drift. Most production systems just surface stream errors to the user and let them retry by hand.