How Streaming Works Across Different LLM Providers in aisuite

aisuite unifies streaming across all LLM providers through a three-layer architecture: a base Provider class that defines the contract and provides an async bridge, provider-specific implementations that normalize native SDK streams into OpenAI-shaped ChatCompletionChunk objects, and a client façade that routes stream=True calls to the appropriate method.

Streaming in aisuite (the open-source Python library from Andrew Ng's team) lets you receive partial responses from any supported LLM provider using identical code. The library handles the messy differences between OpenAI's direct iterators, Anthropic's event streams, and Gemini's stateful chunks—so you write stream=True once and it works everywhere.

The Three-Layer Streaming Architecture

The streaming implementation spans three layers with clear separation of responsibilities:

Layer	Responsibility	Key Source File
Provider base class	Defines contract and supplies generic async bridge	aisuite/provider.py
Provider-specific overrides	Convert native SDK streams to unified format	Provider files in aisuite/providers/
Client façade	Exposes public API and routes to provider methods	aisuite/client.py

Layer 1: The Provider Base Class and Async Bridge

In aisuite/provider.py, the abstract Provider class establishes the streaming contract through two methods:

• chat_completions_create_stream(self, model, messages, **kwargs) — synchronous streaming, raises LLMError by default
• achat_completions_create_stream(self, model, messages, **kwargs) — asynchronous streaming with automatic fallback

The base async implementation is the key innovation. Any provider that implements only the sync method automatically gains async support without additional code:

async def achat_completions_create_stream(self, model, messages, **kwargs):
    # Default async bridge – runs the sync method in a thread and

    # yields chunks via an asyncio.Queue.

    loop = asyncio.get_running_loop()
    queue: asyncio.Queue = asyncio.Queue()

    def produce():
        try:
            for chunk in self.chat_completions_create_stream(model, messages, **kwargs):
                loop.call_soon_threadsafe(queue.put_nowait, ("chunk", chunk))
        except BaseException as exc:
            loop.call_soon_threadsafe(queue.put_nowait, ("error", exc))
        finally:
            loop.call_soon_threadsafe(queue.put_nowait, ("done", None))

    loop.run_in_executor(None, produce)
    while True:
        kind, payload = await queue.get()
        if kind == "chunk":
            yield payload
        elif kind == "error":
            raise payload
        else:
            return

This bridge pushes sync streaming into a background thread and forwards chunks through an asyncio.Queue, ensuring non-blocking behavior for providers without native async support.

Layer 2: Provider-Specific Streaming Implementations

Each concrete provider either passes through native streams directly or performs conversion to the unified ChatCompletionChunk format.

OpenAI Provider: Direct Pass-Through

In aisuite/providers/openai_provider.py, OpenAI's SDK already returns objects matching aisuite's expected shape, so the implementation is minimal:

def chat_completions_create_stream(self, model, messages, **kwargs):
    stream = self.client.chat.completions.create(
        model=model,
        messages=self.transformer.convert_request(messages),
        stream=True,
        **kwargs,
    )
    yield from stream

async def achat_completions_create_stream(self, model, messages, **kwargs):
    stream = await self.aclient.chat.completions.create(
        model=model,
        messages=self.transformer.convert_request(messages),
        stream=True,
        **kwargs,
    )
    async for chunk in stream:
        yield chunk

Anthropic Provider: Event-to-Chunk Conversion

In aisuite/providers/anthropic_provider.py (streaming logic starts at line 387), Anthropic's event-based stream requires normalization. The AnthropicMessageConverter.convert_stream_event method (lines 62-154) transforms mixed event types into standard chunks:

def chat_completions_create_stream(self, model, messages, **kwargs):
    events = self.client.messages.create(
        model=model,
        system=system_message,
        messages=converted_messages,
        stream=True,
        **kwargs,
    )
    state = {}
    for event in events:
        chunk = self.converter.convert_stream_event(event, state)
        if chunk is not None:
            yield chunk

async def achat_completions_create_stream(self, model, messages, **kwargs):
    events = await self.async_client.messages.create(
        model=model,
        system=system_message,
        messages=converted_messages,
        stream=True,
        **kwargs,
    )
    state = {}
    async for event in events:
        chunk = self.converter.convert_stream_event(event, state)
        if chunk is not None:
            yield chunk

The converter handles tool-use start events, text deltas, and final message events, mapping Anthropic's stop_reason to OpenAI's finish_reason and preserving usage metadata.

Gemini Provider: Stateful Chunk Folding

Gemini's streaming API in aisuite/providers/gemini_provider.py (starting line 409) requires more complex handling. The _StreamState class (lines 31-92) manages tool-call indexing and finish reason tracking:

def chat_completions_create_stream(self, model, messages, **kwargs):
    request = self._request_kwargs(model, messages, kwargs)
    state = _StreamState()
    yield state.role_chunk()
    for chunk in self.client.models.generate_content_stream(**request):
        for out in state.convert(chunk, self.converter):
            yield out
    yield state.final_chunk(self.converter)

Unlike other providers, Gemini delivers complete tool-call parts rather than incremental deltas, so the state machine emits each call as a single chunk.

Other Providers: Automatic Async Bridge Fallback

Providers like Mistral, Cohere, and Fireworks that don't implement chat_completions_create_stream automatically use the base class async bridge. When you call acreate(..., stream=True) with these providers, aisuite executes their sync stream method in a background thread—guaranteeing a consistent async interface.

Layer 3: Client Façade API

In aisuite/client.py, the Completions.create method inspects the stream parameter and delegates to the appropriate provider method:

# Synchronous streaming

stream = client.chat.completions.create(
    model="openai:gpt-4o-mini",
    messages=[{"role": "user", "content": "Tell me a joke"}],
    stream=True,
)
for chunk in stream:
    print(chunk.choices[0].delta.content, end="")

# Asynchronous streaming

async for chunk in client.chat.completions.acreate(
    model="anthropic:claude-3-5-sonnet-20240620",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    stream=True,
):
    print(chunk.choices[0].delta.content, end="")

Both calls return objects of type aisuite.framework.chat_completion_chunk.ChatCompletionChunk, regardless of the underlying provider.

Code Examples: Streaming with Different LLM Providers

Example 1: Basic Sync Streaming (Any Provider)

from aisuite import Client

client = Client(provider_configs={"openai": {"api_key": "sk-…"}})

stream = client.chat.completions.create(
    model="openai:gpt-4o-mini",
    messages=[{"role": "user", "content": "Summarise the plot of Inception"}],
    stream=True,
)

for chunk in stream:
    # chunk is always a ChatCompletionChunk

    if delta := chunk.choices[0].delta.content:
        print(delta, flush=True)

This pattern works identically for OpenAI, Anthropic, and Gemini. For providers without native streaming, the base async bridge handles execution.

Example 2: Async Streaming with Tool Calls (Anthropic)

import asyncio
from aisuite import Client

client = Client(provider_configs={"anthropic": {"api_key": "…"}})

async def main():
    async for chunk in client.chat.completions.acreate(
        model="anthropic:claude-3-5-sonnet-20240620",
        messages=[{"role": "user", "content": "What's the weather in Paris?"}],
        tools=[{
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Fetch weather data",
                "parameters": {
                    "type": "object",
                    "properties": {"city": {"type": "string"}},
                    "required": ["city"]
                },
            }
        }],
        stream=True,
    ):
        # Tool-call chunks appear as delta.tool_calls

        if tool_calls := chunk.choices[0].delta.tool_calls:
            print("Tool call:", tool_calls[0].function.name)

asyncio.run(main())

Anthropic's native events are transparently converted, so tool_calls appear in the same structure as OpenAI's format.

Example 3: Using the Async Bridge Fallback (Mistral)

client = Client(provider_configs={"mistral": {"api_key": "…"}})

# Mistral lacks native stream implementation; aisuite uses the thread-based bridge

async for chunk in client.chat.completions.acreate(
    model="mistral:mistral-large-latest",
    messages=[{"role": "user", "content": "Write a haiku"}],
    stream=True,
):
    print(chunk.choices[0].delta.content, end="")

The bridge ensures all providers support async streaming, even those with only synchronous SDKs.

Key Implementation Files

File	Streaming Role	Location
aisuite/provider.py	Base contract and async bridge	aisuite/provider.py
aisuite/providers/openai_provider.py	Direct pass-through streaming	aisuite/providers/openai_provider.py
aisuite/providers/anthropic_provider.py	Event stream normalization	aisuite/providers/anthropic_provider.py
aisuite/providers/gemini_provider.py	Stateful chunk conversion	aisuite/providers/gemini_provider.py
aisuite/client.py	Public API routing	aisuite/client.py
tests/providers/test_streaming_provider.py	Contract compliance tests	tests/providers/test_streaming_provider.py

Summary

• Unified interface: All providers expose identical streaming through create(..., stream=True) and acreate(..., stream=True)
• Automatic async bridging: Providers without native async streaming inherit thread-based execution via achat_completions_create_stream
• Normalized output: Every provider yields ChatCompletionChunk objects with OpenAI-compatible structure, including tool calls and usage metadata
• Provider-specific optimizations: OpenAI uses direct pass-through, Anthropic converts event streams, Gemini employs stateful folding
• Zero-effort extensibility: New providers only need to implement chat_completions_create_stream to gain full streaming support

Frequently Asked Questions

Does aisuite streaming work with all LLM providers?

Yes. Core providers (OpenAI, Anthropic, Gemini) implement native streaming with optimized conversions. All other providers automatically use the base async bridge, which runs synchronous streams in a background thread. According to the aisuite source code, this guarantees that stream=True never fails due to missing provider implementation.

Why do Anthropic and Gemini need conversion while OpenAI doesn't?

OpenAI's SDK already returns ChatCompletionChunk-compatible objects. Anthropic exposes a different event-based API requiring AnthropicMessageConverter.convert_stream_event to map event types. Gemini's generate_content_stream delivers chunks in yet another format handled by _StreamState. The normalization layer ensures your code receives identical structures regardless of provider.

How does the async bridge handle backpressure and errors?

The base achat_completions_create_stream pushes chunks through an asyncio.Queue with call_soon_threadsafe. Producer exceptions are captured and re-raised in the async consumer loop. The finally block guarantees queue termination, preventing hangs when the underlying stream encounters errors.

Can I use streaming with custom or local providers?

Yes. Implement chat_completions_create_stream(self, model, messages, **kwargs) in your custom provider class, yielding ChatCompletionChunk objects. The base class automatically provides achat_completions_create_stream via the thread bridge, or you can override it for native async performance.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}