Streaming

Trace streaming LLM responses.

Risicare provides utilities for tracing streaming LLM responses with proper chunk tracking and span attribution.

Why Streaming Needs Special Handling

Standard contextvars do not propagate across yield boundaries in async generators (a Python limitation, PEP 568). The streaming utilities use the span registry for ID-based lookup instead, ensuring span context is available throughout the stream.

traced_stream (Async)

Wrap async streaming responses with span-aware chunk tracking:

from risicare import traced_stream, get_tracer
 
async def generate():
    with get_tracer().start_span("llm_stream") as span:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Write a story"}],
            stream=True,
        )
        async for chunk in traced_stream(span.span_id, response, event_name="chunk"):
            yield chunk.choices[0].delta.content or ""
 
async def main():
    async for text in generate():
        print(text, end="")

Read a generator like this to its end, or close it with contextlib.aclosing() where you stop early. A generator that is dropped part-way and closed later, in another task or thread, can leave its ended span as the current span, so the next span becomes its child (a known issue that the release does not fix).

Parameters

Parameter	Type	Default	Description
`span_id`	`str`	required	The span ID to track events under (from `span.span_id`)
`stream`	`AsyncIterator`	required	The async iterator to wrap
`event_name`	`str`	`"chunk"`	Name for chunk events recorded on the span

traced_stream_sync (Sync)

Wrap synchronous streaming responses:

from risicare import traced_stream_sync, get_tracer
 
with get_tracer().start_span("llm_stream") as span:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Write a story"}],
        stream=True,
    )
    for chunk in traced_stream_sync(span.span_id, response, event_name="chunk"):
        print(chunk.choices[0].delta.content or "", end="")

Parameters

Parameter	Type	Default	Description
`span_id`	`str`	required	The span ID to track events under
`stream`	`Iterator`	required	The iterator to wrap
`event_name`	`str`	`"chunk"`	Name for chunk events recorded on the span

How It Works

The streaming utilities:

Look up the span by ID from the span registry
Track each chunk with an event on the span (chunk number and size)
Record totals when the stream completes (stream.total_chunks, stream.total_size)
End the span, outside a with block, when the stream completes, raises or is closed (status unset). Inside a with tracer.start_span() block, as in the samples, the block ends the span: status ok, or error when the code raises, also before the stream starts

# Span attributes after streaming completes:
{
    "stream.total_chunks": 42,
    "stream.total_size": 1560,
}

Span Registry

The span registry allows ID-based span retrieval in contexts where contextvars do not propagate (async generators, process pools):

from risicare import register_span, get_span_by_id, unregister_span
 
# Register a span for later retrieval
register_span(span, ttl_seconds=60)
 
# Retrieve the span by ID (from any context)
same_span = get_span_by_id(span.span_id)
 
# Clean up when done
unregister_span(span.span_id)

TTL-Based Cleanup

Registered spans have a default TTL of 60 seconds and are automatically cleaned up when expired. For long-running streams, use extend_span_ttl(span_id, additional_seconds) to keep the span in the registry. traced_stream reads the registry one time, when the stream starts.

Custom Streaming

For custom streaming implementations, use the tracer span context manager:

from risicare import get_tracer
 
async def my_streaming_function():
    tracer = get_tracer()
    with tracer.start_span("streaming-operation") as span:
        total_tokens = 0
        async for chunk in some_stream():
            total_tokens += count_tokens(chunk)
            span.set_attribute("tokens_so_far", total_tokens)
            yield chunk
        span.set_attribute("total_tokens", total_tokens)

Generator Functions

Decorators and Generators

Standard decorators do not work well with generator functions. Use traced_stream instead:

from risicare import traced_stream, get_tracer
 
# Do this
async def stream_response():
    with get_tracer().start_span("llm_stream") as span:
        async for chunk in traced_stream(span.span_id, client.stream()):
            yield chunk

Streaming Span Attributes

Attribute	Description
`stream.total_chunks`	Total number of chunks received
`stream.total_size`	Total accumulated size across all chunks

Configuration

Content Capture

Streamed content, like all prompt and completion content, is off by default. To capture it:

risicare.init(
    trace_content=True  # Capture streaming content
)

Advanced Span Registry Utilities

extend_span_ttl is exported from risicare (since 0.5.1) and from risicare_core. get_span_registry_stats is exported from risicare_core only.

extend_span_ttl

Extend the TTL of a registered span. Use this for long-running streams that may exceed the default 60-second TTL.

from risicare import extend_span_ttl
 
# Returns True if span was found and TTL extended, False otherwise
extend_span_ttl(span_id: str, additional_seconds: float) -> bool

from risicare import extend_span_ttl, register_span, get_tracer
 
async def relay(long_running_stream):
    tracer = get_tracer()
    with tracer.start_span("long-stream") as span:
        register_span(span, ttl_seconds=60)
        async for chunk in long_running_stream:
            # Keep the span alive during a multi-minute stream
            extend_span_ttl(span.span_id, 60)
            yield chunk

get_span_registry_stats

Get registry statistics for debugging span lifecycle issues.

from risicare_core import get_span_registry_stats
 
stats = get_span_registry_stats() -> dict

Returns:

Key	Type	Description
`total_entries`	int	Total spans in registry (active + expired)
`active_entries`	int	Spans that have not expired
`expired_entries`	int	Spans past their TTL (awaiting cleanup)
`operation_count`	int	Number of `register()` calls that the registry has counted

JavaScript / TypeScript

The JS SDK provides tracedStream() for tracing async iterables. Unlike Python, Node.js AsyncLocalStorage propagates through async generators natively, so no span registry workaround is needed.

import { tracedStream } from 'risicare';
 
const stream = tracedStream(asyncIterable, { name: 'llm-stream' });
for await (const chunk of stream) {
    process.stdout.write(chunk);
}
// Span automatically records stream.chunk_count and stream.completed

You can also pass a string name directly:

const stream = tracedStream(asyncIterable, 'my-stream');

Next Steps

Decorators

Add context with decorators

Learn more

Context

Context propagation

Learn more

Edit this page on GitHub

PreviousContext NextJavaScript