# How Streaming Works Across Different LLM Providers in aisuite

> Learn how aisuite unifies LLM streaming across OpenAI, Anthropic, and Gemini providers with a single API. Explore seamless integration and automatic async bridging for consistent results.

- Repository: [Andrew Ng/aisuite](https://github.com/andrewyng/aisuite)
- Tags: deep-dive
- Published: 2026-07-28

---

**aisuite unifies streaming across OpenAI, Anthropic, Gemini, and other providers behind a single `client.chat.completions.create(..., stream=True)` API by converting native SDK streams into a standard OpenAI-shaped `ChatCompletionChunk` format, with automatic async bridging for providers that only support synchronous streaming.**

The aisuite library by Andrew Ng eliminates provider-specific fragmentation by routing all streaming requests through a three-layer architecture. When you enable streaming across different LLM providers in aisuite, every provider—whether native or adapted—returns identical chunk objects, allowing you to switch models without rewriting streaming logic.

## The Three-Layer Streaming Architecture

The streaming implementation spans three architectural layers that abstract provider differences behind a consistent Python interface.

### Provider Base Class – The Contract

The abstract `Provider` class in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) defines the streaming contract through two methods: `chat_completions_create_stream` for synchronous streaming and `achat_completions_create_stream` for async. By default, the sync method raises `LLMError`, forcing concrete providers to implement their own logic. The async method provides a generic bridge that runs the synchronous stream in a background worker thread and forwards chunks through an `asyncio.Queue`.

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

    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

```

### Provider-Specific Overrides

Each concrete provider in the `aisuite/providers/` directory either forwards the native SDK stream (OpenAI) or adapts it (Anthropic, Gemini). This layer handles the normalization of disparate event formats into the standard `ChatCompletionChunk` structure defined in [`aisuite/framework/chat_completion_chunk.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/framework/chat_completion_chunk.py).

### Client Facade – The Public API

The `Chat.Completions.create` method in [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) inspects the `stream=True` parameter and delegates to the appropriate provider streaming method. This ensures that whether you call OpenAI, Anthropic, or a provider using the async bridge, you receive a uniform iterator yielding identical chunk objects.

## OpenAI Streaming: Direct Pass-Through

OpenAI requires no format conversion because its native streaming chunks already match the aisuite standard. In [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py), the implementation simply yields the SDK's generator directly.

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

```

Because OpenAI's response objects align perfectly with aisuite's framework, the provider acts as a transparent proxy without additional conversion overhead.

## Anthropic Streaming: Event-to-Chunk Conversion

Anthropic's streaming API delivers mixed-type events that must be normalized. In [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py) (starting at line 387), the `chat_completions_create_stream` method consumes the SDK's event stream and uses `AnthropicMessageConverter.convert_stream_event` (lines 62-154) to transform each event into a `ChatCompletionChunk`.

```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 maintains state across events to handle tool-use start markers, content deltas, and final message boundaries, mapping Anthropic's `stop_reason` to OpenAI's `finish_reason` while aggregating usage metadata.

## Gemini Streaming: Stateful Folding

Gemini's streaming implementation in [`aisuite/providers/gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/gemini_provider.py) (starting at line 409) uses a `_StreamState` class (lines 31-92) to manage the conversion of `generate_content_stream` chunks into OpenAI-shaped objects. Gemini delivers whole tool-call parts rather than incremental tokens, requiring a state machine to track indices and synthesize usage reports.

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

```

The `_StreamState` captures the assistant role, accumulates tool-call indices, and emits a final chunk containing the finish reason and usage statistics, ensuring parity with OpenAI's streaming behavior.

## Async Streaming for All Providers via the Bridge

Providers like **Mistral**, **Cohere**, and **Fireworks** that do not implement `chat_completions_create_stream` automatically inherit the base class's async bridge behavior. When you call `client.chat.completions.acreate(..., stream=True)` with these providers, aisuite executes the provider's synchronous stream method in a background thread using `loop.run_in_executor`, yielding chunks through an `asyncio.Queue` without blocking the event loop.

This guarantees that **every** provider in aisuite supports async streaming, either natively (OpenAI, Anthropic) or through the thread-based fallback, delivering a non-blocking interface regardless of the underlying SDK's capabilities.

## Practical Implementation Examples

### Basic Synchronous Streaming

Use the standard `create` method with any supported provider. The returned iterator yields `ChatCompletionChunk` objects containing incremental content in `chunk.choices[0].delta.content`.

```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:
    if delta := chunk.choices[0].delta.content:
        print(delta, end="", flush=True)

```

### Async Streaming with Tool Calls (Anthropic)

The unified interface handles tool calling consistently across providers. Anthropic's native tool-use events are converted automatically, appearing as `delta.tool_calls` in the chunk objects.

```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,
    ):
        if tool_calls := chunk.choices[0].delta.tool_calls:
            print(f"Tool: {tool_calls[0].function.name}")

asyncio.run(main())

```

### Using Providers Without Native Async Support

For providers like Mistral that lack native async streaming, the base bridge automatically handles the thread management. You write the same async code regardless of the provider's native capabilities.

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

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="")

```

## Summary

- **Unified API**: `client.chat.completions.create(..., stream=True)` works identically across OpenAI, Anthropic, Gemini, and other providers in aisuite.
- **Three-Layer Architecture**: The base `Provider` class defines contracts, concrete providers handle normalization, and the `Client` facade exposes the public interface.
- **Automatic Conversion**: Anthropic uses `convert_stream_event` (lines 62-154) and Gemini uses `_StreamState` (lines 31-92) to transform native events into standard `ChatCompletionChunk` objects.
- **Universal Async Support**: The `achat_completions_create_stream` method in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) provides a thread-based bridge, enabling async streaming for providers without native async implementations.
- **Source Files**: Core logic resides in [`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), [`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), and [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py).

## Frequently Asked Questions

### Does aisuite support streaming for every LLM provider?

Yes. Providers with native streaming APIs (OpenAI, Anthropic, Gemini) implement `chat_completions_create_stream` directly. All other providers inherit the base class's async bridge, which runs synchronous streams in a background thread via `asyncio.Queue`, ensuring every provider supports `stream=True`.

### Why do Anthropic and Gemini require conversion while OpenAI does not?

OpenAI's streaming response format matches aisuite's `ChatCompletionChunk` structure exactly, allowing direct pass-through. Anthropic and Gemini use different event schemas and chunk structures, so aisuite normalizes their responses through provider-specific converters to maintain interface consistency.

### How does the async bridge handle backpressure and errors?

The bridge in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) uses an `asyncio.Queue` to communicate between the worker thread and the async generator. Exceptions in the synchronous stream are captured and re-raised in the async context, while the queue buffers chunks until consumed, preventing memory exhaustion for fast streams.

### Can tool calling be used with streaming across all providers?

Yes. When streaming with tool calls, providers like Anthropic emit tool-use start events and partial JSON arguments. The `AnthropicMessageConverter` normalizes these into `ChatCompletionChunk` objects with `delta.tool_calls` populated, matching OpenAI's tool-calling format. The unified interface means your tool-handling code works without modification when switching providers.