Streaming LLM outputs break traditional logging. A structured logger for streaming responses lets you capture token deltas, latency, and errors as discrete events instead of dumping raw text into stdout. This tutorial builds a minimal, production-minded version in Python that works with any OpenAI-compatible chat completions stream.
Prerequisites
- Python 3.11 or newer
openaiPython package v1.0+ (pip install openai)- An API key and base URL for an OpenAI-compatible endpoint (OpenAI, a self-hosted vLLM server, or a gateway)
- Familiarity with context managers and generator iteration
No external logging framework is required; we write JSON Lines to a file or stderr.
Why naive logging fails for streams
Printing each token as it arrives produces interleaved, unparseable output when multiple requests run concurrently. You lose request identity, cannot compute time-to-first-token (TTFT), and error handling becomes guesswork. A structured logger for streaming responses emits one JSON object per event, tagged with a request ID and event type, so you can grep, ingest to ClickHouse, or replay later without custom parsers.
The core problem is that streaming inverts the normal request/response cycle. Instead of one log line per call, you get N lines for N tokens plus lifecycle events. If you treat them as plain strings, you cannot answer basic questions: which provider served the request? How many tokens were cached? Did the stream fail after 200 tokens?
Designing the event schema
We use four event types: start, token, stop, error. Every event shares a base set of fields:
{
"ts": "2024-05-12T18:22:01.123Z",
"request_id": "req_01H9...",
"model": "gpt-4o-mini",
"event": "token",
"data": { "text": "Hello", "tokens": 1 }
}ts is UTC ISO-8601 with milliseconds. request_id is a client-generated UUID. data carries event-specific payload. Keeping the schema append-only (never rename event to type later) avoids breaking downstream consumers.
Building the logger class
Create stream_logger.py. The class wraps the OpenAI client and yields tokens while logging each event.
import json
import sys
import time
import uuid
from datetime import datetime, timezone
class StreamLogger:
def __init__(self, client, out=None, model="gpt-4o-mini"):
self.client = client
self.out = out or sys.stdout
self.model = model
self.request_id = None
def _emit(self, event, data):
rec = {
"ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
"request_id": self.request_id,
"model": self.model,
"event": event,
"data": data,
}
self.out.write(json.dumps(rec) + "\n")
self.out.flush()
def chat(self, messages, **kwargs):
self.request_id = uuid.uuid4().hex
start = time.monotonic()
self._emit("start", {"messages": messages})
first_token_ts = None
try:
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
stream=True,
stream_options={"include_usage": True},
**kwargs,
)
full_text = ""
for chunk in stream:
if not chunk.choices:
continue
delta = chunk.choices[0].delta
text = delta.content or ""
if text:
if first_token_ts is None:
first_token_ts = time.monotonic()
full_text += text
self._emit("token", {"text": text})
usage = getattr(chunk, "usage", None)
self._emit("stop", {
"full_text": full_text,
"ttft_ms": int((first_token_ts - start) * 1000) if first_token_ts else None,
"total_ms": int((time.monotonic() - start) * 1000),
"usage": usage.model_dump() if usage else None,
})
return full_text
except Exception as e:
self._emit("error", {"error": str(e), "type": type(e).__name__})
raiseThe stream_options={"include_usage": True} flag asks the provider to send a final chunk with token counts. We compute TTFT from the first emitted token, not the stop event.
Wiring the client
import sys
from openai import OpenAI
from stream_logger import StreamLogger
client = OpenAI(api_key="sk-...", base_url="https://api.openai.com/v1")
logger = StreamLogger(client, out=sys.stderr, model="gpt-4o-mini")
text = logger.chat([
{"role": "user", "content": "Explain streaming logging in one sentence."}
])Run it. The function returns concatenated text; stderr receives JSON Lines.
Expected output at checkpoint
For a short response, stderr shows:
{"ts":"2024-05-12T18:22:01.123Z","request_id":"req_01h9...","model":"gpt-4o-mini","event":"start","data":{"messages":[{"role":"user","content":"Explain streaming logging in one sentence."}]}}
{"ts":"2024-05-12T18:22:01.450Z","request_id":"req_01h9...","model":"gpt-4o-mini","event":"token","data":{"text":"Streaming"}}
{"ts":"2024-05-12T18:22:01.502Z","request_id":"req_01h9...","model":"gpt-4o-mini","event":"token","data":{"text":" logging"}}
{"ts":"2024-05-12T18:22:01.980Z","request_id":"req_01h9...","model":"gpt-4o-mini","event":"stop","data":{"full_text":"Streaming logging captures token events.","ttft_ms":327,"total_ms":857,"usage":{"prompt_tokens":12,"completion_tokens":5,"total_tokens":17}}}The usage object appears only if the provider supports include_usage. A structured logger for streaming responses should treat usage as optional and never assume its shape.
Testing without an API key
Use unittest.mock to simulate a stream and validate the logger offline:
class FakeChunk:
def __init__(self, text, usage=None):
self.choices = [type("C", (), {"delta": type("D", (), {"content": text})()})]
self.usage = usage
class FakeStream:
def __init__(self, chunks):
self._c = chunks
def __iter__(self):
return iter(self._c)
class FakeCompletions:
def create(self, **kwargs):
return FakeStream([
FakeChunk("Hello "),
FakeChunk("world"),
FakeChunk(None, usage=type("U", (), {"model_dump": lambda self: {"total_tokens": 2}})()),
])
class FakeClient:
chat = type("Chat", (), {"completions": FakeCompletions()})()
logger = StreamLogger(FakeClient(), out=sys.stdout)
logger.chat([{"role": "user", "content": "test"}])This prints start, two token events, and stop with mocked usage. Drop it into a pytest suite to catch schema regressions.
Handling provider fallback and routing
If you point the client at a gateway that fronts multiple providers, the model string may not tell you who served the token. Some gateways add response headers like x-provider or x-routed-to. Capture them from the underlying HTTP response if your SDK exposes it. For example, n4n.ai offers automatic fallback when a provider is rate-limited or degraded, and honors client routing directives; log the resolved provider by reading stream.response.headers (OpenAI v1 exposes response on the stream object in some versions, or use client.with_streaming_response). When fallback occurs before the stream starts, the header tells you the winner.
# After creating stream, if available:
resp = getattr(stream, "response", None)
if resp and "x-provider" in resp.headers:
self._emit("start", {"provider": resp.headers["x-provider"]})If the header is absent, fall back to logging the requested model only. Never block the stream waiting for headers that may not exist.
Error and retry visibility
Streaming errors often surface as APIConnectionError or RateLimitError mid-iteration. The except block in StreamLogger.chat emits an error event with the exception type. Pipe these to alerting. Because the logger does not swallow exceptions, callers can implement retry logic; the request ID stays constant across retries if you move ID generation outside, letting you correlate attempts.
def chat_with_retry(logger, messages, tries=3):
for i in range(tries):
try:
return logger.chat(messages, request_id=logger.request_id)
except Exception as e:
if i == tries - 1:
raiseRefactor chat to accept an optional request_id parameter so retries share correlation.
Async variant
The same pattern works with AsyncOpenAI. Replace time.monotonic with asyncio.get_event_loop().time() and use async for. Emit via synchronous write to stdout; under uvloop this is safe for moderate volumes.
async def chat_async(self, messages):
self.request_id = uuid.uuid4().hex
self._emit("start", {"messages": messages})
stream = await self.client.chat.completions.create(model=self.model, messages=messages, stream=True)
async for chunk in stream:
# same delta handling as sync versionShipping the logs
JSON Lines is natively ingested by Vector, Fluent Bit, or jq for debugging:
python app.py 2>stream.log
jq 'select(.event=="error")' stream.logA structured logger for streaming responses turns opaque token firehoses into queryable timelines. You now have TTFT, per-request text, and error context without custom parsers.
Extending the schema
Add fields as needed: temperature, finish_reason, cached_tokens (if the provider returns cache-control hints). If your gateway forwards provider cache-control hints, log prompt_tokens_details.cached_tokens from usage to track cache hit rates. Keep the schema append-only; never rename fields in place. With this foundation, you can add sampling, redaction of PII in messages, or per-tenant routing metadata.