Skip to content

Streaming Responses

Mojentic supports streaming responses from LLMs, providing a better user experience by showing content as it arrives rather than waiting for the complete response. Streaming works seamlessly with tool calling, allowing you to stream content before, during, and after tool execution.

Overview

The LLMBroker.generate_stream() method mirrors the generate() method but yields content chunks as they arrive from the LLM. This is particularly useful for:

  • Long-form content generation (stories, articles, explanations)
  • Interactive chat applications where users want immediate feedback
  • Agentic workflows where tools are called and you want to show progress
  • Any scenario where reducing perceived latency improves UX

Supported Gateways

Gateway Streaming Support Tool Support Notes
OllamaGateway ✅ Full ✅ Full Tool calls arrive complete
OMLXGateway ✅ Full ✅ Full OpenAI-compatible SSE, keepalives ignored
OpenAIGateway ✅ Full ✅ Full Handles incremental tool arguments
AnthropicGateway ❌ Not yet ❌ N/A Coming soon

Basic Usage

Simple Text Streaming

from mojentic.llm import LLMBroker
from mojentic.llm.gateways import OllamaGateway
from mojentic.llm.gateways.models import LLMMessage

broker = LLMBroker(
    model="qwen2.5:7b",
    gateway=OllamaGateway()
)

# Stream the response
stream = broker.generate_stream(
    messages=[LLMMessage(content="Tell me a short story about a dragon")],
    temperature=0.7
)

# Print chunks as they arrive
for chunk in stream:
    print(chunk, end='', flush=True)

print("\nDone!")

Streaming with OpenAI

import os
from mojentic.llm import LLMBroker
from mojentic.llm.gateways import OpenAIGateway
from mojentic.llm.gateways.models import LLMMessage

broker = LLMBroker(
    model="gpt-4o-mini",
    gateway=OpenAIGateway(api_key=os.getenv("OPENAI_API_KEY"))
)

stream = broker.generate_stream(
    messages=[LLMMessage(content="Count to 10, one number per line")],
    temperature=0.7
)

for chunk in stream:
    print(chunk, end='', flush=True)

Streaming with Tool Calling

One of the most powerful features is that streaming works seamlessly with tool calling. The broker automatically:

  1. Streams content as it arrives
  2. Detects when tool calls are made
  3. Executes the tools
  4. Recursively streams the response after tool execution

Example with Date Resolution

from mojentic.llm import LLMBroker
from mojentic.llm.gateways import OpenAIGateway
from mojentic.llm.gateways.models import LLMMessage
from mojentic.llm.tools.date_resolver import ResolveDateTool
import os

broker = LLMBroker(
    model="gpt-4o-mini",
    gateway=OpenAIGateway(api_key=os.getenv("OPENAI_API_KEY"))
)

date_tool = ResolveDateTool()

# This will:
# 1. Stream any initial content
# 2. Call the date tool when needed
# 3. Stream the response after tool execution
stream = broker.generate_stream(
    messages=[
        LLMMessage(content="Tell me a story about tomorrow and next week")
    ],
    tools=[date_tool],
    temperature=0.7
)

for chunk in stream:
    print(chunk, end='', flush=True)

Content Before Tool Calls

Modern LLMs are optimized to call tools directly without extra explanation. If you want to see the LLM's reasoning before it calls tools, explicitly ask for it:

stream = broker.generate_stream(
    messages=[
        LLMMessage(content="What is the date tomorrow? "
                          "First, explain what you're going to do, "
                          "then call the tool.")
    ],
    tools=[date_tool]
)

# Output will be like:
# "I will use the date resolution tool to find tomorrow's date..."
# [tool call happens]
# "Tomorrow is November 13, 2025."

How It Works

For Ollama

  1. Content chunks arrive and are yielded immediately
  2. Tool calls arrive complete in the stream
  3. When a tool call is detected, it's executed
  4. The broker recursively calls generate_stream() with the tool result
  5. The response after tool execution is also streamed

For OpenAI

The OpenAI implementation is more complex because tool arguments arrive incrementally:

  1. Content chunks arrive and are yielded immediately
  2. Tool call chunks arrive piece-by-piece:
  3. First chunk: id, name, empty arguments
  4. Subsequent chunks: fragments of arguments JSON
  5. Tool arguments are accumulated until finish_reason == 'tool_calls'
  6. Complete tool calls are parsed and executed
  7. Recursive streaming continues with tool results

OpenAI and oMLX follow-up requests include every parallel tool call in the assistant message. Each tool result keeps the matching call id, including when the id and arguments arrive in separate stream chunks. This also applies to ordinary tool turns.

Example of OpenAI tool call streaming: 

Chunk 1: id='call_123', name='get_weather', arguments=''
Chunk 2: arguments='{"'
Chunk 3: arguments='location'
Chunk 4: arguments='":"'
Chunk 5: arguments='San'
Chunk 6: arguments=' Francisco'
Chunk 7: arguments='"}'
Finish: tool_calls

Method Signature

def generate_stream(
    self,
    messages: List[LLMMessage],
    tools=None,
    config: Optional[CompletionConfig] = None,
    temperature=None,
    num_ctx=None,
    max_tokens=None,
    correlation_id: str = None
) -> Iterator[str]:
    """
    Generate a streaming text response from the LLM.

    Yields content chunks as they arrive. When tool calls are detected,
    executes them and recursively streams the response.

    Parameters
    ----------
    messages : List[LLMMessage]
        A list of messages to send to the LLM.
    tools : List[Tool]
        A list of tools to use with the LLM.
    config : CompletionConfig, optional
        Configuration object controlling generation behavior (preferred).
    temperature : float, optional
        (Legacy) The temperature to use for the response.
    num_ctx : int, optional
        (Legacy) The number of context tokens to use.
    max_tokens : int, optional
        (Legacy) The maximum number of tokens to generate.
    correlation_id : str
        UUID string for tracing events.

    Note
    ----
    Prefer using `config=CompletionConfig(...)` over individual kwargs.
    If both are provided, config takes precedence.

    Yields
    ------
    str
        Content chunks as they arrive from the LLM.
    """

Best Practices

1. Always Flush Output

When printing streaming content, use flush=True to ensure immediate display:

for chunk in stream:
    print(chunk, end='', flush=True)

2. Handle Exceptions

Wrap streaming in try/except to handle network errors gracefully:

try:
    for chunk in broker.generate_stream(messages=messages, tools=tools):
        print(chunk, end='', flush=True)
except Exception as e:
    print(f"\nError during streaming: {e}")

3. Use with Tracer for Debugging

The tracer system records all streaming activity:

from mojentic.tracer import TracerSystem

tracer = TracerSystem()
broker = LLMBroker(
    model="gpt-4o-mini",
    gateway=OpenAIGateway(),
    tracer=tracer
)

# Stream with full observability
stream = broker.generate_stream(messages=messages, tools=tools)
for chunk in stream:
    print(chunk, end='', flush=True)

# Review what happened
events = tracer.get_events()

4. Check Gateway Support

Not all gateways support streaming. Check before use:

if hasattr(broker.adapter, 'complete_stream'):
    stream = broker.generate_stream(messages=messages)
else:
    response = broker.generate(messages=messages)

Single-turn streaming with completion evidence

generate_stream() yields text and never tells you whether the provider finished. When incomplete output must never be used as a result (for example, before you act on generated JSON), use generate_stream_events(). It yields typed events and ends every stream with exactly one terminal event:

Event Meaning
StreamContent(text) Visible assistant content, in order
StreamCompleted(metadata) Terminal success. metadata is a CompletionMetadata
StreamError(reason, detail, metadata) Terminal failure. reason is a StreamErrorReason

Nothing follows the terminal event.

from mojentic.llm import (
    CompletionConfig, LLMBroker, ResponseFormat, StreamCompleted, StreamContent, StreamError,
)
from mojentic.llm.gateways import OpenAIGateway

broker = LLMBroker(model="gpt-4o-mini", gateway=OpenAIGateway())
config = CompletionConfig(response_format=ResponseFormat(type="json_object"))

received = []
for event in broker.generate_stream_events(messages, config):
    match event:
        case StreamContent(text=text):
            received.append(text)
        case StreamCompleted(metadata=metadata):
            result = "".join(received)
            print(metadata.finish_reason, metadata.usage)
        case StreamError(reason=reason, metadata=metadata):
            # "received" is evidence of what arrived, not a result. Do not use it.
            print(f"Turn failed: {reason.value}", metadata)

CompletionMetadata holds finish_reason, usage, provider_model and a provider metadata map (for example Ollama's durations). Each field is None when the provider did not report it.

Client-measured timing

Every dispatched completion on Ollama, OpenAI and oMLX reports independent monotonic client timing, in milliseconds:

Field Meaning
time_to_first_token_ms First dispatch to the first nonempty token-bearing frame; None without a token
first_token_kind thinking, content or tool_call; None without a token
request_duration_ms First dispatch to terminal success/failure; None if never dispatched

For generate_stream_events, read these fields directly on terminal event.metadata (CompletionMetadata). For recovery streams, use terminal event.metrics (StreamMetrics); completed response metadata also carries the measured fields. Existing provider usage and durations retain their original names, units and values, including oMLX usage.time_to_first_token in seconds.

Legacy complete_stream, LLMBroker.generate_stream and ChatSession.send_stream keep their existing chunk/text ordering. Read gateway.last_stream_metrics, broker.last_stream_metrics or session.last_stream_metrics after consumption, failure or supported local close. Legacy exceptions retain their original types and expose error.stream_metrics when timing is available. These properties describe the latest operation on the shared gateway; use per-event terminal evidence when consuming concurrent streams. Never treat a measured duration as proof of successful completion.

for text in session.send_stream("Hello"):
    print(text, end="", flush=True)
metrics = session.last_stream_metrics
if metrics is not None:
    print(metrics.first_token_kind, metrics.time_to_first_token_ms,
          metrics.request_duration_ms)

Stream construction, request building and potentially blocking initial request capture/attempt callbacks are excluded. The first actual HTTP dispatch starts the clock. It continues through admitted retries, their delays and callbacks; each request returning tool results starts a fresh clock. Tool execution time belongs to neither completion. Keepalives, empty frames, tool names and empty argument strings do not count. The first argument fragment counts before a tool call is assembled; Ollama's complete argument objects count, including {}. Within the first token-bearing frame the precedence is thinking, content, then tool arguments, regardless of delivery order.

Success and failure after dispatch carry duration even without tokens. Closing a synchronous iterator yields no synthetic terminal event. Direct event-stream local close leaves timing on the gateway. Recovery-backed complete_stream, generate_stream and send_stream retain timing and available provider metadata immediately after close(), including after content but before terminal evidence. Closing an unsent iterator records no timing. Recovery RecoveryCall.cancel() produces the existing typed cancelled terminal with timing if dispatched; aclose() releases resources without inventing a terminal. Remote termination remains unknown. Anthropic streaming and unsupported model/routes remain unsupported and produce no dispatch timing.

Completion rules

  • OpenAI: success requires finish_reason: "stop" and then data: [DONE]. [DONE] after any other finish reason (for example length) is an INCOMPLETE_COMPLETION error that carries the finish reason, usage and model. The request sets stream_options: {"include_usage": true} so usage is reported.
  • Ollama: success requires a frame with done: true and done_reason: "stop". Any other done_reason is an INCOMPLETE_COMPLETION error with the same evidence. A final frame without done_reason is also INCOMPLETE_COMPLETION, so Ollama servers too old to send done_reason cannot use this API.

An INCOMPLETE_STREAM error carries whatever evidence arrived before the stream ended (for example the model or usage), and None when nothing arrived.

StreamErrorReason Cause
INCOMPLETE_COMPLETION The provider finished for a reason other than stop
INCOMPLETE_STREAM The stream ended without the provider's terminal marker
PROVIDER_ERROR The provider sent an error frame or an error response
UNEXPECTED_TOOL_CALLS The provider asked for a tool call
INVALID_STREAM_EVENT A frame could not be decoded or had an unexpected shape
REQUEST_FAILED The request could not be sent, or the connection failed mid-stream
STREAM_EVENTS_UNSUPPORTED The gateway or model does not support this API. No request was sent

Content that arrived before a StreamError is evidence of what the provider sent. It is not a result, even when it happens to be valid JSON.

Behaviour

  • The broker sends exactly one HTTP request. It supplies no tools, forces max_tool_iterations to zero and never retries.
  • Call close() on the generator to cancel the request, or use contextlib.closing around a loop that may break early. A retained generator stays open after break. Closing it closes the HTTP response. An early stop is not an error and yields no further event. The tracer keeps the LLM call event but records no response event for that turn.
  • OpenAIGateway, OllamaGateway and OMLXGateway support this API. Other gateways, such as AnthropicGateway, yield a single STREAM_EVENTS_UNSUPPORTED error before any request.
  • The tracer records the LLM call when the request starts, and the response (content so far plus the provider evidence) when the terminal event arrives. See Provider evidence in response traces.
  • generate_stream() keeps its existing behaviour.

Limitations

Structured Output

Streaming does not work with structured output (generate_object()). This is an API limitation from OpenAI:

# This will raise NotImplementedError
stream = broker.generate_stream(
    messages=messages,
    object_model=MyPydanticModel  # ❌ Not supported
)

# Use non-streaming for structured output
obj = broker.generate_object(
    messages=messages,
    object_model=MyPydanticModel  # ✅ Works
)

Requested response formats

Streaming cannot return a parsed object, but it can ask the provider for JSON. Set CompletionConfig(response_format=...) and the OpenAI and Ollama gateways forward it in the streaming request, the same way they do without streaming:

from mojentic.llm import CompletionConfig, ResponseFormat

config = CompletionConfig(response_format=ResponseFormat(type="json_object"))
for chunk in broker.generate_stream(messages=messages, config=config):
    print(chunk, end="", flush=True)

Add json_schema=... to request JSON schema mode. This records the request. It is not proof that the provider enforced the format, so validate the assembled content before you use it. See Requesting a response format.

Model Compatibility

OpenAI's streaming respects model capabilities. Reasoning models and some older models may not support streaming:

# The gateway will raise NotImplementedError if model doesn't support streaming
try:
    stream = broker.generate_stream(
        messages=messages,
        model="o1-preview"  # May not support streaming
    )
except NotImplementedError as e:
    print(f"Model doesn't support streaming: {e}")

Performance Considerations

Latency vs. Throughput

  • Streaming: Lower perceived latency, better UX, same total time
  • Non-streaming: Higher perceived latency, worse UX, same total time

Streaming doesn't make responses faster—it shows them sooner.

Tool Call Overhead

Each tool call in streaming mode: 1. Waits for complete tool arguments (OpenAI) 2. Executes the tool synchronously 3. Makes a new streaming request

For workflows with many tool calls, this adds network round-trips.

Memory Usage

Streaming uses less memory than buffering full responses, especially for long-form content.

Examples

See the complete working example at src/_examples/streaming.py in the repository.

API Reference

For detailed API documentation, see:

Streaming uses the same request fidelity rules as ordinary completion: unspecified temperature, context and output limits are omitted; use max_tokens for an explicit output limit. Session max_context remains a local history budget.