Streaming

Real-time token streaming for responsive AI interactions

Stream AI responses token-by-token as they’re generated, instead of waiting for the complete response.

from praisonaiagents import Agent

agent = Agent(instructions="You are a helpful assistant")

for chunk in agent.start("Explain streaming in one sentence", stream=True):
    print(chunk, end="", flush=True)

The user sends a prompt; tokens stream back incrementally instead of waiting for the full reply.

Quick Start

1. Install

   pip install praisonaiagents

2. Auto-detect (Default)

   from praisonaiagents import Agent
   
   agent = Agent(instructions="You are a helpful assistant")
   # No stream= argument — the SDK auto-detects what the provider supports
   agent.start("Write a short story")

   By default the SDK tries streaming first and silently falls back to non-streaming if your provider’s sync client doesn’t support it — multi-agent workflows on providers like Deepseek now Just Work. Sampling knobs you pass (max_tokens, top_p, temperature) are honoured whether the streaming attempt or the fallback runs, and if streaming raises the SDK surfaces the original cause on the error’s __cause__ — you won’t lose the real error to a fallback TypeError.

3. Force Streaming

   from praisonaiagents import Agent
   
   agent = Agent(instructions="You are a helpful assistant")
   
   for chunk in agent.start("Write a short story", stream=True):
       print(chunk, end="", flush=True)

4. Shape the output while streaming

   Pass sampling knobs to start() the same way you would to a non-streaming call.

   from praisonaiagents import Agent
   
   agent = Agent(instructions="You are a helpful assistant")
   
   for chunk in agent.start(
       "Write a haiku",
       stream=True,
       max_tokens=64,
       top_p=0.9,
       temperature=0.7,
   ):
       print(chunk, end="", flush=True)

5. Control reasoning effort while streaming

   A reasoning model honours reasoning_effort on the streaming path too.

   from praisonaiagents import Agent
   
   agent = Agent(instructions="Solve tricky problems step by step.", llm="o4-mini")
   
   for chunk in agent.start(
       "Design a load balancer for a spiky workload.",
       stream=True,
       reasoning_effort="high",
   ):
       print(chunk, end="", flush=True)

Sampling knobs on the streaming path

start(stream=True) forwards temperature, max_tokens, top_p, and reasoning_effort straight to the provider stream.

Agent-attribute fallback — set the level once at construction and every streamed turn honours it.

from praisonaiagents import Agent

# Attribute is used when the call kwarg isn't passed.
agent = Agent(instructions="Reason carefully.", llm="o4-mini", reasoning_effort="high")

for chunk in agent.start("Design a load balancer for a spiky workload.", stream=True):
    print(chunk, end="", flush=True)

What happens when streaming fails

Streaming can fail mid-flight — a context-length overflow, a provider hiccup, a sync adapter that refuses to stream — and when it does, start(stream=True) and iter_stream() fall back to a single non-streaming chat() call so your caller still gets an answer.

sequenceDiagram
    participant U as Your code
    participant S as start(stream=True)
    participant SC as streaming client
    participant C as chat() fallback
    participant P as provider

    U->>S: start("...", stream=True, max_tokens=5)
    S->>SC: streamed request
    SC-->>S: RuntimeError("context too long")
    S->>S: strip stream=True<br/>keep only kwargs chat() accepts
    S->>C: chat("...", ...filtered kwargs..., stream=False)
    C->>P: non-streamed request
    P-->>C: full answer
    C-->>S: answer
    S-->>U: yielded as one block

    Note over S,C: If chat() also raises,<br/>the raised exception's __cause__<br/>is the original streaming error.

    classDef user fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef process fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef client fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef provider fill:#10B981,stroke:#7C90A0,color:#fff

    class U user
    class S process
    class SC client
    class C agent
    class P provider

The fallback contract:

Internal helpers

One private helper on ChatMixin implements the contract above and is what the regression tests drive directly.

This is internal — do not import it from application code. It is named here so debug traces are readable and so subclasses that override chat() know that adding parameters to the override signature automatically opts those parameters into the fallback.

Debugging a failed fallback:

from praisonaiagents import Agent

agent = Agent(instructions="You are a helpful assistant")

try:
    for chunk in agent.start("Very long prompt...", stream=True, max_tokens=5):
        print(chunk, end="", flush=True)
except Exception as e:
    # The fallback error itself
    print(f"Fallback failed: {e!r}")
    # The real streaming cause — pre-#4731 this was None
    print(f"Original streaming error: {e.__cause__!r}")

Choosing the Right Method

Since PR #5739, chat(stream=True) mirrors start(stream=True)‘s sync-adapter behaviour: on adapters that cannot stream, it falls back to a non-streaming call and returns the completed text — it no longer returns None.

Common Patterns

Terminal Streaming

from praisonaiagents import Agent

agent = Agent(instructions="You are a helpful assistant")

# Tokens appear as they arrive
for chunk in agent.start("Explain quantum computing", stream=True):
    print(chunk, end="", flush=True)

chat(stream=True) Sync-Adapter Fallback

from praisonaiagents import Agent

agent = Agent(instructions="You are a helpful assistant")

# chat(stream=True) on the default sync OpenAI adapter used to return None.
# Since PR #5739 it transparently falls back to non-streaming — you get text.
reply = agent.chat("Summarise streaming in one sentence.", stream=True)
print(reply)   # -> a complete answer, never None

App Integration with iter_stream()

Best for integrating into your own application — yields raw chunks with no display overhead.

from praisonaiagents import Agent

agent = Agent(instructions="You are a helpful assistant")

full_response = ""
for chunk in agent.iter_stream("Write a haiku"):
    full_response += chunk
    # Send to your UI, WebSocket, or processing pipeline

print(full_response)

The interactive CLI (praisonai chat / praisonai code) consumes iter_stream() directly since PR #2906 — every token you see in the terminal is a real model delta, not a post-hoc word replay. If the provider does not stream, the CLI falls back to a single non-streamed chat() call and prints the completed answer as one block. See Interactive TUI.

Streaming with Callbacks

Hook into every streaming event for fine-grained control.

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

def on_event(event: StreamEvent):
    if event.type == StreamEventType.DELTA_TEXT:
        print(event.content, end="", flush=True)
    elif event.type == StreamEventType.FIRST_TOKEN:
        print("⚡ First token received!")
    elif event.type == StreamEventType.STREAM_END:
        print("\n✅ Done!")

agent = Agent(instructions="You are a helpful assistant")
agent.stream_emitter.add_callback(on_event)
agent.start("Tell me a joke", stream=True)

FastAPI SSE Integration

Pipe streaming tokens directly to a web client using Server-Sent Events.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from praisonaiagents import Agent

app = FastAPI()

@app.get("/stream")
async def stream_response(prompt: str):
    agent = Agent(instructions="You are a helpful assistant")

    def generate():
        for chunk in agent.iter_stream(prompt):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Async Streaming

import asyncio
from praisonaiagents import Agent

async def main():
    agent = Agent(instructions="You are a helpful assistant")
    result = await agent.astart("Write a poem", stream=True)
    print(result)

asyncio.run(main())

Resuming after an interrupt

An interrupted streamed run keeps its partial answer in the session store, so praisonai run --continue / --session <id> picks up whatever was produced before the interrupt.

from praisonaiagents import Agent

agent = Agent(
    instructions="You are a helpful assistant",
    memory={"session_id": "write-doc"},
)

# Stream and interrupt (Ctrl-C)
for chunk in agent.start("Write a long doc on streaming", stream=True):
    print(chunk, end="", flush=True)

# Later — pick up right where it stopped
praisonai run --continue
# or
praisonai run --session write-doc

sequenceDiagram
    participant User
    participant Agent
    participant Store as Session Store

    User->>Agent: start(prompt, stream=True)
    Agent->>Store: persist user turn
    loop tokens
        Agent-->>User: chunk
        Agent->>Store: upsert_partial_assistant_message (debounced)
    end
    Note over User,Agent: Ctrl-C mid-stream
    Agent->>Store: force-flush partial (GeneratorExit)
    User->>Agent: praisonai run --continue
    Store-->>Agent: history incl. partial turn
    Agent-->>User: resume with partial answer visible

    classDef user fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef store fill:#10B981,stroke:#7C90A0,color:#fff

    class User user
    class Agent agent
    class Store store

The streaming path persists the user turn as soon as the stream starts, then flushes the growing partial assistant turn as tokens arrive. Partial writes are debounced (~5 writes/sec) and the retention/compaction window is enforced only on finalize, so streaming does not thrash the disk. Three internal helpers drive this — _persist_assistant_delta, _finalize_assistant_turn, and _discard_partial_assistant_turn — you’ll see their names in debug logs but never call them directly.

Streaming with Knowledge

Streaming responses retrieve and inject knowledge context using the same normalization as the non-streaming sync path.

from praisonaiagents import Agent

agent = Agent(
    name="Research Assistant",
    instructions="Answer using the knowledge base.",
    knowledge=["docs/handbook.pdf"],
)

for chunk in agent.start_stream("Summarise the onboarding policy"):
    print(chunk, end="", flush=True)

Streaming with Tools

When your agent uses tools, streaming happens in two phases: the initial response that decides to call tools, and a follow-up response that synthesizes the tool results.

Tools can also emit incremental progress while they run using emit_tool_progress() — these arrive as TOOL_PROGRESS events in your stream callback before the tool returns its result. See Tool Progress Streaming.

Chained follow-up tool calls

Streaming agents can now chain tool calls through the follow-up turn — a second tool call returned by the follow-up is executed, its result is fed back to the model, and the loop continues until the model answers with plain content.

from praisonaiagents import Agent

agent = Agent(
    name="Researcher",
    instructions="Call step1 first, then step2 with step1's result.",
    tools=[step1, step2],
)

for chunk in agent.start("Run the pipeline", stream=True):
    print(chunk, end="", flush=True)

• Bounds. The total tools-executed count is capped by max_tool_calls_per_turn (seeded with the first streamed batch, so the cap spans every follow-up batch); the follow-up iteration count is capped by max_iterations (fallback self.max_iter, default 20).
• Ollama. In the single-shot (non-parallel) mode, follow-up batches resolve chained-arg placeholders and run one call at a time, mirroring the first streamed batch so an argument that references an earlier call’s result in the same batch is resolved before dispatch.
• Cancellation. A Stop between follow-up iterations short-circuits with the existing stream-cancel sentinel.

Parallel tool calls under streaming

The streaming accumulator now handles the three cases providers hit in practice — a tool-call index greater than the current list, parallel calls that reuse index=0 with different ids, and id/name fields that arrive in a later chunk.

Stopping a streamed tool batch

A Stop / interrupt_controller.request() during a streamed tool phase now short-circuits pending tool calls with typed cancelled results and ends the stream — no leftover completion request is issued for the cancelled turn.

from praisonaiagents import Agent
from praisonaiagents.agent import InterruptController

controller = InterruptController()

agent = Agent(
    instructions="Run the requested tools.",
    tools=["shell", "search"],
    interrupt_controller=controller,
)

# From a UI Stop button / signal handler on another thread:
# controller.request("user_stop")

for chunk in agent.start("Search the web and download the result", stream=True):
    print(chunk, end="", flush=True)

The non-streaming fallback loop

Providers that cannot stream with tools run a bounded, non-streamed tool loop under the hood.

Any provider whose _supports_streaming_tools() returns False takes this path — currently Anthropic (anthropic/claude-sonnet-4-20250514), Gemini (gemini/gemini-2.0-flash), and Ollama (ollama/llama3.2).

sequenceDiagram
    participant U as User
    participant S as get_response_stream
    participant M as Model (non-streamed)
    participant T as Tool

    U->>S: start(stream=True) with tools
    Note over S: provider does not stream with tools<br/>(Anthropic / Gemini / Ollama)

    loop bounded by max_iter and max_tool_calls_per_turn
        S->>M: request (tools=..., stream=False)
        M-->>S: tool_calls (+ optional prose)
        S-->>U: yield prose (if any)
        S->>T: execute each tool_call
        T-->>S: result (or error → reported to model)
        S->>S: append tool results to messages
    end

    alt provider stalls (repeated call / empty answer / iteration limit)
        S->>M: request (tools=[], stream=False, wrap_up prompt)
        M-->>S: final answer
        S-->>U: yield final answer
    else
        M-->>S: final answer (no tool_calls)
        S-->>U: yield final answer
    end

    classDef user fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef stream fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef model fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef tool fill:#189AB4,stroke:#7C90A0,color:#fff

    class U user
    class S stream
    class M model
    class T tool

Stall detection and tools-disabled finalisation

Since PraisonAI PR #4870, the fallback stops when a provider makes no progress and asks the model to answer with tools disabled.

Four ways the loop can stop early:

After any of these, the loop asks the model once more with tools= removed and the wrap-up prompt "The tools have already been called and their results are above. Do not call any tools. Answer the original question now using those results." — the same bounded finalisation get_response and get_response_async already use.

If the finalisation returns empty or the fallback message, the stream raises LLMResponseError("Provider produced no answer after N tool call(s) on the streaming path (<reason>).") rather than yielding nothing.

Symptom this fixes. On a small local model the stream path used to run the same tool up to ten times and yield an empty string. Since #4870 the same scenario emits one tool call and a real model-written sentence.

What the fallback does now:

1. Issues a non-streamed request with tools= still attached.
2. If the response has no tool_calls, yields whatever prose is there and stops.
3. If it has tool_calls, records the assistant turn (with tool_calls for OpenAI-shaped providers, plain content for Ollama), yields any accompanying prose, then executes each tool.
4. A failing tool is reported to the model as {"error": "..."} rather than aborting the run.
5. Loops back and asks the model again with the tool-result messages appended.

Prose emitted alongside tool calls is still yielded to the caller. _create_tool_message — now a real method — mirrors the non-streaming loop’s formatting, including Ollama’s natural-language variant.

Tool activity telemetry

Streaming UIs now see every tool call through the tool_call display callback.

Register a tool_call handler and it fires on the streaming path with tool_name, tool_input, tool_output, elapsed_time, and success — the same kwargs as the non-streaming path.

from praisonaiagents import Agent, register_display_callback

def on_tool_call(tool_name=None, tool_input=None, success=None, **kwargs):
    print(f"🔧 {tool_name}({tool_input}) — {'ok' if success else 'failed'}")

register_display_callback("tool_call", on_tool_call)

agent = Agent(instructions="You are a weather assistant", tools=[get_weather])

for chunk in agent.start("What's the weather in Paris?", stream=True):
    print(chunk, end="", flush=True)

The Phase 2 follow-up is fetched with stream=False and yielded as one block — it is not a token stream. It carries the accumulated messages (system prompt, user turn, the assistant tool-call message, and the tool-result messages) and keeps tools, so the model may call another tool rather than being forced to answer.

from praisonaiagents import Agent, tool

@tool
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: 72°F, sunny"

agent = Agent(
    instructions="You are a weather assistant",
    tools=[get_weather]
)

for chunk in agent.start("What's the weather in Paris?", stream=True):
    print(chunk, end="", flush=True)

On the custom-LLM path both phases go through the same _completion_with_retry wrapper, so transient rate-limit or network errors are retried automatically without any caller intervention. If every retry is exhausted on the follow-up, the stream ends with the error sentinel documented below rather than dropping silently.

Streaming with long histories

Context management runs on the streaming path exactly like it does on the non-streaming path — on both streaming branches.

If your agent has a ContextManager configured (auto-enabled when the agent has tools; see Context Management Overview), long chat histories are compacted before the streamed request is sent — same budget, same strategy, same behaviour as chat().

sequenceDiagram
    participant U as User
    participant A as Agent
    participant CM as ContextManager
    participant L as LLM

    U->>A: start(prompt, stream=True)
    A->>A: _build_messages
    alt context_manager present
        A->>CM: _apply_context_management(messages, system_prompt, tools)
        CM-->>A: optimized messages
    end
    A->>L: streamed request (bounded)
    L-->>A: token
    A-->>U: token
    L-->>A: token
    A-->>U: token

    classDef user fill:#10B981,stroke:#7C90A0,color:#fff
    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef cm fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef llm fill:#F59E0B,stroke:#7C90A0,color:#fff

    class U user
    class A agent
    class CM cm
    class L llm

The same flow now runs on both streaming branches — the OpenAI-client branch and the custom-LLM branch. Which one your agent takes is decided by the routing flag _using_custom_llm, not the vendor.

Which streaming branch does my agent use?

Two branches feed the streaming path, and they are chosen by _using_custom_llm:

The trigger is the routing flag, not the vendor — llm="openai/gpt-4o-mini" takes the custom-LLM branch too. Before PR #4753 the custom-LLM rows were ❌: compaction was silently skipped, and long conversations were sent to the provider in full.

from praisonaiagents import Agent

def search(query: str) -> str:
    """A sample tool."""
    return f"Results for {query}"

# context_manager auto-enabled because tools are present
agent = Agent(instructions="You are a helpful assistant", tools=[search])

# Chat history can grow indefinitely across calls — compaction now
# runs on both entry points.
for _ in range(500):
    agent.chat("Ask a long-running question...")

# Streaming path now compacts before sending, same as non-streaming.
for chunk in agent.start("What did we discuss?", stream=True):
    print(chunk, end="", flush=True)

When you don’t have context management (Agent(context=False), or an agent with no tools and no explicit context=True), streaming and non-streaming both send the raw history — as before. The self.context_manager guard skips the whole step, so there is zero overhead when context management is disabled.

Error Handling in the Stream

If the follow-up LLM call fails after retries, the stream ends with a visible error sentence instead of silently dropping. This sentinel is emitted from the custom-LLM streaming path (praisonaiagents/llm/llm.py::get_response_stream) — the native sync-OpenAI path does not currently emit it.

You may receive this exact sentinel string:

[Error: Failed to generate final response after tool execution (ref: followup-1713957912345). Please retry. If it continues, try reducing prompt size.]

Detect the error sentinel in your stream consumer:

from praisonaiagents import Agent

agent = Agent(instructions="You are a helpful assistant", tools=[...])

full = ""
for chunk in agent.iter_stream("Research and summarize quantum computing"):
    full += chunk
    print(chunk, end="", flush=True)

if "[Error:" in full and "ref:" in full:
    # Surface ref to your logs / retry externally
    print(f"\n⚠️ Error detected, check logs for correlation ID")

StreamEvent Protocol

Every streaming chunk emits a StreamEvent with full context.

TODO_UPDATED fires on every todo_add / todo_update — subscribe to render a live checklist. See Live Todo Streaming.

MODEL_FALLBACK fires the moment the runtime swaps models. Subscribe to render a “switched to backup” notice, or pair it with the HookEvent.MODEL_FALLBACK hook for programmatic reactions. See Model Fallback → Observing the Switch. The custom-LLM path (provider/model strings, model objects, base_url) emits it too, so one subscription covers every agent shape.

Reacting to Retries

When the agent hits a rate limit or transient error, it emits a RETRY event before it waits, then retries. The signal fires on both the sync and async retry loops, and reaches both add_callback(sync_fn) and add_async_callback(async_fn) consumers. Emission is guarded so there is zero overhead when nothing is listening.

sequenceDiagram
    participant Agent
    participant LLM
    participant Emitter as StreamEmitter
    participant Sync as sync callback<br/>(add_callback)
    participant Async as async callback<br/>(add_async_callback)

    Agent->>LLM: async request
    LLM-->>Agent: 429 rate-limit
    Agent->>Agent: honour retry-after
    Agent->>Emitter: emit_async RETRY (attempt, max_attempts, delay, reason)
    Emitter-->>Sync: RETRY (via emit)
    Emitter-->>Async: RETRY (via emit_async)
    Agent->>LLM: retry

    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef llm fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef emit fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef consumer fill:#10B981,stroke:#7C90A0,color:#fff

    class Agent agent
    class LLM llm
    class Emitter emit
    class Sync,Async consumer

The RETRY event carries its details in event.metadata:

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

def on_event(event: StreamEvent):
    if event.type == StreamEventType.RETRY:
        m = event.metadata or {}
        print(
            f"Retrying in {m['delay']:.1f}s "
            f"(attempt {m['attempt']}/{m['max_attempts']}) — {m['reason']}"
        )

agent = Agent(instructions="Answer concisely.")
agent.stream_emitter.add_callback(on_event)
agent.start("Summarise the news")

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

async def on_event(event: StreamEvent):
    if event.type == StreamEventType.RETRY:
        m = event.metadata or {}
        await broadcast_status(
            f"Retrying in {m['delay']:.1f}s "
            f"(attempt {m['attempt']}/{m['max_attempts']}) — {m['reason']}"
        )

async def main():
    agent = Agent(instructions="Answer concisely.")
    agent.stream_emitter.add_async_callback(on_event)
    await agent.astart("Summarise the news")

asyncio.run(main())

Token usage on the streaming path

Streamed calls now record real prompt and completion tokens, not zeros.

Before PraisonAI PR #4819, the Chat Completions streaming path’s choices guard swallowed the trailing usage-only chunk, and the Responses API streaming path never inspected the response.completed event — so CompletionUsage(0, 0, 0) reached your metrics collector on every streamed turn. Since #4819:

• process_stream_chunks reads the usage chunk before the choices guard, and process_stream_response requests stream_options={"include_usage": True} on every streamed call (a caller-supplied stream_options is respected).
• _stream_responses_api / _stream_responses_api_async capture the final response from response.completed and hand it to _track_token_usage; the sync path also fires the llm_end display callback (tokens_in, tokens_out, latency_ms).

Both get_response and get_response_async now accept return_token_usage=True (parity restored in PraisonAI PR #5305) and package the returned string as (text, TokenUsage) extracted from the final raw response, whether the response came from Chat Completions (prompt_tokens/completion_tokens) or the Responses API (input_tokens/output_tokens).

# Async — same parameter as sync since PraisonAI PR #5305
text, usage = await agent.llm.get_response_async(prompt, return_token_usage=True)
print(usage.prompt_tokens, usage.completion_tokens, usage.total_tokens)

Token counters populate even for a purely streamed call — no callbacks, no _emit:

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

metrics = StreamMetrics()

def on_event(event: StreamEvent):
    metrics.update_from_event(event)

agent = Agent(instructions="You are a helpful assistant")
agent.stream_emitter.add_callback(on_event)

for chunk in agent.start("Explain AI briefly", stream=True):
    print(chunk, end="", flush=True)

print(metrics.format_summary())
# tokens_in / tokens_out are now non-zero on a streamed turn

The llm_end display callback now fires on the Responses API streaming path (sync) too, carrying the same tokens_in / tokens_out / latency_ms fields as the non-streaming path.

Metrics

Track Time To First Token (TTFT) and throughput. Token counters tokens_in / tokens_out are now accurate on the streaming path (see Token usage on the streaming path).

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

metrics = StreamMetrics()

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

agent = Agent(instructions="You are a helpful assistant")
agent.stream_emitter.add_callback(on_event)
agent.start("Explain AI briefly", stream=True)

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

Key Concepts

Time To First Token (TTFT)

Request → [TTFT] → First Token → [Streaming] → Last Token → Done

TTFT is the time before the first token arrives. This is provider latency — the model must process your prompt before generating. Streaming does NOT reduce TTFT, but it shows progress immediately.

Streaming vs Non-Streaming

Streaming runs the same pre-turn checks

Streamed turns pass the same pre-turn gate as chat() / achat() before the first chunk is yielded (PR #5863).

• Rate limiter: iter_stream() / start(stream=True) acquires the agent’s configured rate_limiter, so a streamed turn cannot bypass the request budget.
• BEFORE_AGENT hooks: a blocking result ends the generator immediately — no model call, no chunks. A hook returning modified_input["prompt"] rewrites the prompt the model sees, exactly like on chat().

from praisonaiagents import Agent

agent = Agent(name="Writer", instructions="Write product copy")

for chunk in agent.iter_stream("Announce our new espresso machine"):
    print(chunk, end="", flush=True)

Streaming and guardrails

Streaming splits guardrail handling: input guardrails run, output guardrails do not.

Input guardrails — enforced on streaming (since PR #4462)

Since PraisonAI PR #4462, iter_stream() and start(stream=True) validate the prompt before the first token is yielded — the full prompt is known up front, so input validation gates the stream exactly like chat(). A blocked prompt yields the single chunk [Input blocked by guardrail: <reason>] and stops; the durable-run record is never opened.

Output guardrails — bypassed on streaming

Output guardrails validate the full response before it is returned to the caller. Token-level streaming yields tokens as the model produces them — there is no full response to validate until streaming completes, and buffering tokens until then would break the streaming contract.

Since PraisonAI PR #3632, the SDK emits a clear warning as soon as streaming begins on an agent with an output guardrail attached:

from praisonaiagents import Agent

agent = Agent(
    name="Writer",
    instructions="Write product copy",
    guardrails="Must be professional and under 200 words",
)

# Emits: WARNING Agent Writer: output guardrail is not applied to streamed
# responses (iter_stream / stream=True). Use chat() for guardrail-validated output.
for chunk in agent.iter_stream("Announce our new espresso machine"):
    print(chunk, end="", flush=True)

# Same warning is emitted here — both entry points share one streaming generator.
for chunk in agent.start("Announce our new espresso machine", stream=True):
    print(chunk, end="", flush=True)

See Guardrails → Streaming and guardrails for the guardrail-side view.

Managed-backend streaming (backend=)

When `Agent(backend=

) is used, start(stream=True) delegates to backend.stream(prompt)` and yields each chunk as the backend emits it — even from ordinary sync code with no running event loop.

Prior to PraisonAI PR #3908, the sync (“no running event loop”) branch buffered internally, so the generator yielded only after the whole response was produced; since #3908 it is truly incremental, mirroring the running-loop branch.

from praisonaiagents import Agent

agent = Agent(name="assistant", backend=my_managed_backend)

for chunk in agent.start("Write a 2000-word report.", stream=True):
    print(chunk, end="", flush=True)   # text appears progressively — no full-response wait

Stopping a managed stream early

Explicitly closing the stream tears down the upstream run — on Anthropic that stops the server-side session and stops token billing.

import asyncio
import contextlib
from praisonaiagents import Agent

agent = Agent(name="assistant", backend=my_managed_backend)

async def main():
    async with contextlib.aclosing(agent.astart("Write a long report.", stream=True)) as stream:
        async for chunk in stream:
            print(chunk, end="", flush=True)
            if user_disconnected():
                break
    # Leaving the `async with` closes the generator deterministically, which
    # sets the stop event and calls backend.interrupt() on a daemon thread.

asyncio.run(main())

Closing a sync managed stream early

Since PR #6073 (merged 2026-10-27), the sync generator from agent.start(prompt, stream=True) uses one hardened bridge whether or not your code already has a running event loop. Closing it early stops the backend cleanly.

from praisonaiagents import Agent

agent = Agent(name="assistant", backend=my_managed_backend)

stream = agent.start("Write a long report.", stream=True)
print(next(stream))   # read one chunk
stream.close()        # producer stops; the backend's own cleanup runs

• Bounded buffer — at most 64 chunks are buffered; a slow consumer pauses the producer instead of growing memory.
• Early close stops the producer — close() (or garbage collection) signals the producer thread, which stops within ~100 ms.
• Backend cleanup runs — the bridge calls aclose() on the backend’s async generator, so its finally blocks execute.
• No shutdown hangs — the producer runs on a daemon thread.
• Errors surface once — a RuntimeError from the backend is raised to you exactly once; backend.stream() is never re-run.

This path streams like any other start(stream=True) call — see Choosing the Right Method and Managed Runtime Protocol for the backend contract.

CLI Usage

# Stream responses in terminal
praisonai chat --stream "Tell me a joke"

# With verbose output
praisonai chat --stream --verbose "Explain quantum computing"

Best Practices

Troubleshooting

“Streaming seems to buffer before showing anything”

This is TTFT, not buffering. The model is generating the first token. Check:

• Model complexity (larger models have higher TTFT)
• Prompt length (longer prompts take longer to process)
• Network latency to the API

“Tokens appear in chunks, not one at a time”

Normal. Providers may batch tokens for efficiency.

“Stream ends with [Error: Failed to generate final response after tool execution (ref: followup-...)]“

The follow-up LLM call (the one that synthesizes tool results into a final answer) failed after the built-in retries. Common causes:

• Persistent rate limit — pair streaming with a Rate Limiter at higher RPM, or back off the caller.
• Context-length overflow — reduce conversation history or tool-result size.
• Provider outage — include the ref: ID when reporting. The internal log line (ref=..., model=..., error=...) makes it searchable.

“Streaming with tools returns nothing on Anthropic / Gemini / Ollama”

Fixed in PraisonAI PR #4757. If you are on an older version, upgrade — earlier versions called an undefined method inside a swallowed exception and dropped every tool call. The failure symptom was a Streaming failed with unexpected error log line followed by an empty response. See The non-streaming fallback loop.

“Streaming is not supported in sync OpenAIAdapter” / Deepseek multi-agent crash

Fixed on start(stream=True) with single-agent smart-fallback in PR #1734. Since PR #4731 the fallback filters kwargs (max_tokens, top_p, and anything else chat() doesn’t declare) and chains the original streaming exception via raise ... from streaming_error; PR #4734 factored the whole fallback into ChatMixin._stream_fallback_chat (signature-driven, so it is robust to future chat() changes) and added the _praisonai_stream_fallback_exhausted sentinel so the outer handler does not run the fallback twice. If you were previously catching TypeError: chat() got an unexpected keyword argument 'max_tokens' as a proxy for “streaming failed”, switch to inspecting exc.__cause__ instead — the real cause is now preserved. For multi-agent teams that use sync-only providers, explicitly disable streaming with output=["verbose", {"stream": False}] or similar. See Multi-Agent Output for configuration options.

chat(stream=True) was not covered by that fallback until PR #5739 (fixes #5737). Before #5739, an explicit chat(prompt, stream=True) against the sync OpenAI adapter raised ValueError("Streaming is not supported in sync OpenAIAdapter…") inside _chat_completion, which the outer except handler swallowed — so the call returned None. Since #5739, _chat_completion catches that specific ValueError and retries once with stream=False, so the caller receives the completed text. Any other ValueError (and any unrelated error) is still re-raised and surfaces as the normal typed LLMError — the fix is scoped to the “Streaming is not supported” message, so error signals are never swallowed into None.

If you were previously branching on result is None from chat(stream=True) as a proxy for “the sync adapter can’t stream”, switch to one of:

• rely on the new fallback — the returned string is the real response;
• call achat() for native async streaming when you need streaming semantics;
• call chat(stream=False) explicitly if you want to skip the streaming attempt entirely.

Routed Claude / Gemini crash on the async path

Routed Claude and Gemini models (Bedrock / Vertex AI / OpenRouter prefixes) now stream correctly on the async path. Older versions misdetected them as OpenAI and crashed. Provider detection now resolves bedrock/anthropic., vertex_ai/claude-, openrouter/anthropic/ to Anthropic and vertex_ai/gemini- to Gemini, handing each to the correct streaming adapter. See Routed model provider detection.

Related