Streaming Module

Real-time streaming events for LLM responses with reasoning support

Stream LLM responses in real-time with full control over text, tool calls, and reasoning content.

Diagram

Quick Start

Basic Streaming

from praisonaiagents import Agent
from praisonaiagents.streaming import StreamEvent, StreamEventType

def my_callback(event: StreamEvent):
    if event.type == StreamEventType.DELTA_TEXT:
        print(event.content, end="", flush=True)

agent = Agent(name="assistant")
agent.stream_emitter.add_callback(my_callback)
response = agent.start("Hello!", stream=True)

With Reasoning

from praisonaiagents import Agent
from praisonaiagents.streaming import StreamEvent, StreamEventType

def my_callback(event: StreamEvent):
    if event.type == StreamEventType.DELTA_TEXT:
        if event.is_reasoning:
            # Style reasoning content differently (dim gray)
            print(f"\033[90m{event.content}\033[0m", end="")
        else:
            print(event.content, end="")

agent = Agent(name="thinker", llm="o1-mini")
agent.stream_emitter.add_callback(my_callback)
response = agent.start("Think step by step...", stream=True)

Async Reasoning

import asyncio
from praisonaiagents import Agent
from praisonaiagents.streaming import StreamEvent, StreamEventType

def my_callback(event: StreamEvent):
    if event.type == StreamEventType.DELTA_TEXT:
        tag = "[thinking] " if event.is_reasoning else ""
        print(f"{tag}{event.content}", end="", flush=True)

async def main():
    agent = Agent(name="thinker", llm="gpt-5")
    agent.stream_emitter.add_callback(my_callback)
    await agent.astart("Solve: how many r's in strawberry?")

asyncio.run(main())

Reasoning-summary deltas arrive on both sync (start/chat) and async (astart/achat) paths — the same callback works for both.

StreamEvent

Every streaming chunk is wrapped in a StreamEvent with full context.

StreamEvent Fields

Field	Type	Default	Description
`type`	`StreamEventType`	Required	Event type (see below)
`timestamp`	`float`	Auto	High-precision timestamp
`content`	`str`	`None`	Text content for DELTA_TEXT
`tool_call`	`dict`	`None`	Tool call data
`metadata`	`dict`	`None`	Additional metadata
`error`	`str`	`None`	Error message
`is_reasoning`	`bool`	`False`	Whether content is reasoning/thinking
`agent_id`	`str`	`None`	Agent identifier (multi-agent)
`session_id`	`str`	`None`	Session identifier
`run_id`	`str`	`None`	Run identifier

Event Types

Diagram

REQUEST_START

Before API call is made

HEADERS_RECEIVED

HTTP 200 headers arrive

FIRST_TOKEN

First content delta (TTFT marker)

DELTA_TEXT

Text content chunk

DELTA_TOOL_CALL

Tool call chunk

STREAM_END

Stream completed

Reasoning Content

Models like o1, o1-mini, and deepseek-reasoner emit reasoning/thinking content before the final answer.

Detect Reasoning

python

def callback(event: StreamEvent):
    if event.type == StreamEventType.DELTA_TEXT:
        if event.is_reasoning:
            # This is thinking/reasoning content
            print(f"[Thinking] {event.content}")
        else:
            # This is the actual response
            print(event.content, end="")

Style Differently

python

def callback(event: StreamEvent):
    if event.type == StreamEventType.DELTA_TEXT:
        if event.is_reasoning:
            # Dim gray for reasoning
            print(f"\033[90m{event.content}\033[0m", end="")
        else:
            # Normal for response
            print(event.content, end="")

Collect Separately

python

reasoning_parts = []
response_parts = []

def callback(event: StreamEvent):
    if event.type == StreamEventType.DELTA_TEXT:
        if event.is_reasoning:
            reasoning_parts.append(event.content)
        else:
            response_parts.append(event.content)

# After streaming:
full_reasoning = "".join(reasoning_parts)
full_response = "".join(response_parts)

Multi-Agent Context

In multi-agent workflows, identify which agent emitted each event.

python

def callback(event: StreamEvent):
    if event.agent_id:
        print(f"[{event.agent_id}] ", end="")

    if event.type == StreamEventType.DELTA_TEXT:
        print(event.content, end="")

Context Fields

Field	Use Case
`agent_id`	Identify agent in multi-agent workflows
`session_id`	Track conversation sessions
`run_id`	Correlate events within a single run

StreamMetrics

Track streaming performance with built-in metrics.

python

from praisonaiagents.streaming import StreamMetrics

metrics = StreamMetrics()

def callback(event: StreamEvent):
    metrics.update_from_event(event)
    if event.type == StreamEventType.DELTA_TEXT:
        print(event.content, end="")

# After streaming:
print(metrics.format_summary())
# Output: TTFT: 245ms | Stream: 1200ms | Total: 1445ms | Tokens: 150 (125.0/s)

Metrics Properties

Property	Description
`ttft`	Time To First Token (seconds)
`stream_duration`	Token streaming duration
`total_time`	End-to-end request time
`tokens_per_second`	Token generation rate

StreamEventEmitter

Each agent has its own emitter for thread-safe streaming.

python

from praisonaiagents import Agent

agent = Agent(name="assistant")

# Add sync callback
agent.stream_emitter.add_callback(my_sync_callback)

# Add async callback
agent.stream_emitter.add_async_callback(my_async_callback)

# Enable metrics
agent.stream_emitter.enable_metrics()

# After streaming
metrics = agent.stream_emitter.get_metrics()

Built-in Callbacks

Text Printer

from praisonaiagents.streaming import create_text_printer_callback

# Simple text printer
callback = create_text_printer_callback(flush=True)
agent.stream_emitter.add_callback(callback)

Metrics Collector

from praisonaiagents.streaming import create_metrics_callback, StreamMetrics

metrics = StreamMetrics()
callback = create_metrics_callback(metrics)
agent.stream_emitter.add_callback(callback)

Logging Callback

from praisonaiagents.streaming import create_logging_callback

logger, callback = create_logging_callback(verbose=True, metrics=True)
agent.stream_emitter.add_callback(callback)

# After streaming
print(logger.get_metrics_summary())

Provider Support

Streaming with reasoning content is supported across providers:

Provider	Streaming	Reasoning	Async parity	Notes
OpenAI (Responses API)	✅	✅	✅	`gpt-5`, `o3`, `o4-mini`, reasoning-capable `gpt-4.1` tiers
OpenAI (Chat Completions)	✅	✅	✅	`o1`, `o1-mini`
LiteLLM	✅	✅	✅	Via `provider_specific_fields.reasoning_content`
Gemini	✅	❌	—	Standard streaming only

Best Practices

Use is_reasoning for UI
Style reasoning content differently (collapsed, dimmed, or hidden)
Track Metrics
Monitor TTFT and tokens/second for performance optimization
Handle Errors
Check for StreamEventType.ERROR events
Multi-Agent Safety
Each agent has its own emitter – no shared state