# How Streaming Works Across Different LLM Providers in aisuite

> Discover how aisuite unifies LLM streaming across providers using a three-layer architecture. Understand the base Provider class, normalized streams, and client facade for seamless integration.

- Repository: [Andrew Ng/aisuite](https://github.com/andrewyng/aisuite)
- Tags: internals
- Published: 2026-08-03

---

**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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) |

### Layer 1: The Provider Base Class and Async Bridge

In [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/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:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py), OpenAI's SDK already returns objects matching aisuite's expected shape, so the implementation is minimal:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/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:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/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:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py), the `Completions.create` method inspects the `stream` parameter and delegates to the appropriate provider method:

```python

# 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)

```python
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)

```python
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)

```python
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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) | Base contract and async bridge | [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) |
| [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py) | Direct pass-through streaming | [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py) |
| [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py) | Event stream normalization | [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py) |
| [`aisuite/providers/gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/gemini_provider.py) | Stateful chunk conversion | [`aisuite/providers/gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/gemini_provider.py) |
| [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) | Public API routing | [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) |
| [`tests/providers/test_streaming_provider.py`](https://github.com/andrewyng/aisuite/blob/main/tests/providers/test_streaming_provider.py) | Contract compliance tests | [`tests/providers/test_streaming_provider.py`](https://github.com/andrewyng/aisuite/blob/main/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.