Streaming & Structured Outputs
SSE streaming, incremental JSON parsing, and reliable structured output at production scale.
Streaming + structured outputs
Streaming sends tokens to the client as they're generated, dramatically improving perceived latency (time-to-first-token < 500ms vs multi-second wait). Structured outputs constrain the model to valid JSON. Combining them is hard: tool calls and JSON arrive as fragments across chunks, so you need an incremental parser that accumulates partial JSON and parses only when complete.
Why this combination is hard
Most production LLM apps need both streaming (for UX) and structured outputs (for reliability). But streaming structured output means parsing partial JSON across chunks — naive json.loads crashes on every chunk. Get this wrong and your agent breaks intermittently for long tool calls.
Streaming architecture
Client → POST → API opens SSE stream → LLM provider stream → for each chunk: detect tool_call vs content; if content, stream token to client; if tool_call, accumulate arguments until tool call complete, then parse + execute → feed observation back to LLM → continue. Server logs the full assembled response for observability even though only fragments were streamed.
Streaming with tool call handling
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
app = FastAPI()
client = OpenAI()
@app.post("/api/chat")
def chat(req: dict):
def generate():
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=req["messages"],
tools=req.get("tools"),
stream=True,
)
# Accumulators for tool calls (arguments arrive in fragments)
tool_call_buffers: dict[int, str] = {}
for chunk in stream:
delta = chunk.choices[0].delta
# Stream text tokens to client
if delta.content:
yield f"data: {json.dumps({'type': 'token', 'content': delta.content})}\n\n"
# Accumulate tool call arguments (fragmented across chunks)
if delta.tool_calls:
for tc in delta.tool_calls:
idx = tc.index
if idx not in tool_call_buffers:
tool_call_buffers[idx] = ""
if tc.function.arguments:
tool_call_buffers[idx] += tc.function.arguments
# On finish, parse accumulated tool calls
if chunk.choices[0].finish_reason == "tool_calls":
for idx, args_str in tool_call_buffers.items():
try:
args = json.loads(args_str) # now complete
yield f"data: {json.dumps({'type': 'tool_call', 'args': args})}\n\n"
except json.JSONDecodeError:
# Attempt repair: close braces
repaired = args_str + "}"
try:
args = json.loads(repaired)
yield f"data: {json.dumps({'type': 'tool_call', 'args': args, 'repaired': True})}\n\n"
except Exception:
yield f"data: {json.dumps({'type': 'error', 'msg': 'Invalid tool call args'})}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")Experiment: streaming vs blocking
Compare perceived latency of streaming vs blocking response.
What to observe
Streaming doesn't reduce total generation time, but it dramatically reduces perceived latency. Time-to-first-token < 500ms feels instant; > 2s feels broken. Always stream user-facing generations. Tool calls arrive fragmented — accumulate then parse.
Production streaming
Production streaming needs: server-side full-response logging (you streamed fragments but must log the whole), reconnection with resume, backpressure handling, cancellation API, heartbeat keepalive, and graceful error mid-stream. For tool calls: accumulate arguments, parse on completion, attempt repair on malformed JSON.
Challenge
Your streaming agent crashes intermittently with 'Invalid tool call: missing arguments' only for long tool calls. Diagnose and fix. (See the 'Tool Call Split Across Chunks' challenge.)
Production checklist
Production checklist
0 of 8 checked
Knowledge check
Why does json.loads crash on streaming tool call arguments?
Complete
You can now stream responses and handle fragmented structured outputs. This completes the Building LLM Applications series.
Mark this chapter as complete
Track your progress and unlock the next chapter.