> ## Documentation Index > Fetch the complete documentation index at: https://praison.ai/docs/llms.txt > Use this file to discover all available pages before exploring further. # 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. ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} graph LR A[Agent]:::agent --> B[LLM]:::agent B --> C[StreamEvent]:::tool C --> D[Your Callback]:::agent classDef agent fill:#8B0000,color:#fff classDef tool fill:#189AB4,color:#fff ``` ## Quick Start ```python Basic Streaming theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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) ``` ```python With Reasoning theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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) ``` ```python Async Reasoning theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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. > [!NOTE] > > `start(..., stream=True)` and `astart(...)` accept `max_tokens` and `top_p` alongside `temperature` — each is forwarded to the provider only when present. See [Streaming → Sampling knobs on the streaming path](https://praison.ai/docs/features/streaming.md#sampling-knobs-on-the-streaming-path). ## StreamEvent Every streaming chunk is wrapped in a `StreamEvent` with full context. > [accordion] 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 ```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} sequenceDiagram participant A as Agent participant L as LLM participant C as Callback A->>L: Request L-->>C: REQUEST_START L-->>C: HEADERS_RECEIVED L-->>C: FIRST_TOKEN loop Streaming L-->>C: DELTA_TEXT L-->>C: DELTA_TEXT (is_reasoning=True) end L-->>C: LAST_TOKEN L-->>C: STREAM_END ``` > [card] REQUEST\_START > > Before API call is made > [card] HEADERS\_RECEIVED > > HTTP 200 headers arrive > [card] FIRST\_TOKEN > > First content delta (TTFT marker) > [card] DELTA\_TEXT > > Text content chunk > [card] DELTA\_TOOL\_CALL > > Tool call chunk > [card] STREAM\_END > > Stream completed ## Reasoning Content Models like `o1`, `o1-mini`, and `deepseek-reasoner` emit reasoning/thinking content before the final answer. > [tab] Detect Reasoning > > ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} > 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="") > ``` > [tab] Style Differently > > ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} > 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="") > ``` > [tab] Collect Separately > > ```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} > 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} def callback(event: StreamEvent): if event.agent_id: print(f"[{event.agent_id}] ", end="") if event.type == StreamEventType.DELTA_TEXT: print(event.content, end="") ``` > [accordion] 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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) ``` > [accordion] 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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 ```python Text Printer theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.streaming import create_text_printer_callback # Simple text printer callback = create_text_printer_callback(flush=True) agent.stream_emitter.add_callback(callback) ``` ```python Metrics Collector theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} from praisonaiagents.streaming import create_metrics_callback, StreamMetrics metrics = StreamMetrics() callback = create_metrics_callback(metrics) agent.stream_emitter.add_callback(callback) ``` ```python Logging Callback theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}} 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 | > [!TIP] > > `is_reasoning=True` is emitted on both sync and async streaming paths. On async Responses-API runs, reasoning-summary text arrives as `DELTA_TEXT` deltas with `is_reasoning=True`, identical to the sync path. > [!TIP] > > LiteLLM automatically extracts reasoning content from `provider_specific_fields.reasoning_content` and emits it as `StreamEvent(is_reasoning=True)`. ## Best Practices **1. Use is\_reasoning for UI** Style reasoning content differently (collapsed, dimmed, or hidden) **2. Track Metrics** Monitor TTFT and tokens/second for performance optimization **3. Handle Errors** Check for `StreamEventType.ERROR` events **4. Multi-Agent Safety** Each agent has its own emitter - no shared state > [!WARNING] > > Don't let callback errors break streaming. The emitter catches exceptions silently. This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.