The observable story

When a streaming endpoint fails in production, you need to know where it broke. Was the prompt malformed? Did the provider time out? Did a chunk arrive corrupted? Structured logging and clean error handling turn a mystery into a fixable problem.

service.py
python
import logging
import time

logger = logging.getLogger(__name__)

async def generate_story_stream(request: StoryRequest):
    start = time.time()
    logger.info(
        "story.stream.start",
        extra={"character": request.character_name, "theme": request.story_theme},
    )

    try:
        chunk_count = 0
        async for chunk in llm_provider.generate_stream(build_story_prompt(request)):
            chunk_count += 1
            yield chunk

        logger.info(
            "story.stream.complete",
            extra={"chunks": chunk_count, "duration_ms": int((time.time() - start) * 1000)},
        )
    except Exception as exc:
        logger.exception("story.stream.failed", extra={"error": str(exc)})
        raise

Structured logs carry context (character, theme, chunk count, duration) so you can filter and aggregate them in any log platform. Exceptions are logged with stack traces before being re-raised.

Be careful here. Prompts often contain user data, and responses can be long. Log metadata (request ID, character name, duration, chunk count, token usage) by default, and only log full prompts or responses behind a debug flag. This keeps log volume manageable and avoids leaking sensitive content into your observability pipeline.

Quiz: Quiz

Loading practice…