Streaming Module

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

Quick Start

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)

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)

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

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
Style Differently
Collect Separately

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="")

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="")

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.

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.

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.

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

from praisonaiagents.streaming import create_text_printer_callback

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

from praisonaiagents.streaming import create_metrics_callback, StreamMetrics

metrics = StreamMetrics()
callback = create_metrics_callback(metrics)
agent.stream_emitter.add_callback(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	Notes
OpenAI	✅	✅	`o1`, `o1-mini` models
LiteLLM	✅	✅	Via `provider_specific_fields`
Gemini	✅	❌	Standard streaming only

LiteLLM automatically extracts reasoning content from provider_specific_fields.reasoning_content and emits it as StreamEvent(is_reasoning=True).

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

Don’t let callback errors break streaming. The emitter catches exceptions silently.

Guide

Reference

Quick Start

StreamEvent

Event Types

REQUEST_START

HEADERS_RECEIVED

FIRST_TOKEN

DELTA_TEXT

DELTA_TOOL_CALL

STREAM_END

Reasoning Content

Multi-Agent Context

StreamMetrics

StreamEventEmitter

Built-in Callbacks

Provider Support

Best Practices

​Quick Start

​StreamEvent

​Event Types

REQUEST_START

HEADERS_RECEIVED

FIRST_TOKEN

DELTA_TEXT

DELTA_TOOL_CALL

STREAM_END

​Reasoning Content

​Multi-Agent Context

​StreamMetrics

​StreamEventEmitter

​Built-in Callbacks

​Provider Support

​Best Practices

Quick Start

StreamEvent

Event Types

Reasoning Content

Multi-Agent Context

StreamMetrics

StreamEventEmitter

Built-in Callbacks

Provider Support

Best Practices