# How aisuite Handles Streaming Across LLM Providers: Architecture and Implementation

> Discover how aisuite streamlines LLM streaming. It unifies responses from providers like OpenAI, Anthropic, and Gemini into a single format for seamless integration.

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

---

**aisuite unifies streaming responses from OpenAI, Anthropic, Gemini, and other LLM providers behind a single `stream=True` parameter, automatically normalizing native streams into OpenAI-compatible `ChatCompletionChunk` objects while providing a generic async bridge for providers without native async support.**

The `andrewyng/aisuite` repository eliminates the complexity of integrating disparate streaming APIs by exposing a consistent interface through `client.chat.completions.create()`. When streaming is enabled, aisuite routes requests through provider-specific implementations that translate native token streams into standard `ChatCompletionChunk` objects, ensuring uniform behavior regardless of the underlying LLM service.

## Provider Base Class and the Async Bridge

The foundation of aisuite's streaming architecture resides in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py), where the abstract `Provider` class defines the streaming contract.

By default, `chat_completions_create_stream` raises an `LLMError`, forcing concrete providers to explicitly implement streaming support:

```python
def chat_completions_create_stream(self, model, messages, **kwargs):
    raise LLMError(f"{type(self).__name__} does not support streaming chat completions.")

```

For asynchronous operations, the base class provides `achat_completions_create_stream`, which serves as a **generic async bridge**. This method automatically wraps any synchronous streaming implementation using `asyncio.Queue` and `run_in_executor`, enabling non-blocking async streams for providers that only offer synchronous SDKs:

```python
async def achat_completions_create_stream(self, model, messages, **kwargs):
    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 ensures that **all** providers support async streaming regardless of their native SDK capabilities.

## Provider-Specific Streaming Implementations

While the base class provides fallbacks, major providers implement optimized native streaming in their respective modules.

### OpenAI: Direct Pass-Through

Located in [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py), the OpenAI implementation forwards the SDK's native stream directly without conversion, as OpenAI's format already matches aisuite's `ChatCompletionChunk` structure:

```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: Event-to-Chunk Conversion

The Anthropic provider ([`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py), starting at line 387) consumes Anthropic's event-based streaming API and normalizes events into OpenAI-shaped chunks using `AnthropicMessageConverter.convert_stream_event` (lines 62-154):

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

```

The converter handles mixed event types—including tool-use starts, content deltas, and final message events—mapping them to standard `ChatCompletionChunk` objects with proper usage metadata and finish reasons.

### Gemini: Stateful Stream Folding

Google's Gemini implementation ([`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 streaming state. This state machine folds Gemini's `generate_content_stream` chunks into OpenAI-compatible format, handling tool-call synthesis and usage reporting:

```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 incremental token streams, Gemini delivers complete tool-call parts, which the state machine emits as single chunks without incremental accumulation.

### Other Providers: Fallback Bridge

Providers such as **Mistral**, **Cohere**, and **Fireworks** do not implement `chat_completions_create_stream`. When `stream=True` is requested, aisuite falls back to the base class async bridge, executing the synchronous stream (if available) in a background thread via `run_in_executor`. This guarantees a non-blocking async interface for every supported provider.

## Practical Streaming Examples

The client façade in [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) translates the `stream=True` flag into calls to the provider's streaming methods. Both synchronous and asynchronous usage follow identical patterns across all providers.

### Synchronous Streaming

```python
from aisuite import Client

client = Client(provider_configs={"anthropic": {"api_key": "your-key"}})

stream = client.chat.completions.create(
    model="anthropic:claude-3-5-sonnet-20240620",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    stream=True,
)

for chunk in stream:
    if delta := chunk.choices[0].delta.content:
        print(delta, end="", flush=True)

```

### Async Streaming with Tool Calls

```python
import asyncio
from aisuite import Client

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

async def main():
    async for chunk in client.chat.completions.acreate(
        model="openai:gpt-4o-mini",
        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 call: {tool_calls[0].function.name}")

asyncio.run(main())

```

### Cross-Provider Async Streaming (Mistral Example)

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

# Uses base async bridge since Mistral lacks native async streaming

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

- **aisuite unifies streaming** across all LLM providers through the `stream=True` parameter in `client.chat.completions.create()`, normalizing responses to OpenAI-compatible `ChatCompletionChunk` objects.
- **Provider implementations** reside in `aisuite/providers/`, with OpenAI using direct pass-through, Anthropic converting event streams via `AnthropicMessageConverter.convert_stream_event` (lines 62-154), and Gemini employing `_StreamState` (lines 31-92) for chunk folding.
- **Async support** is guaranteed for all providers through the base class bridge in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py), which executes synchronous streams in background threads using `asyncio.Queue` and `run_in_executor`.
- **Tool calls and metadata** are handled provider-specifically but normalized to the standard format, preserving compatibility across OpenAI, Anthropic, Gemini, and other supported services.

## Frequently Asked Questions

### What happens if a provider doesn't support streaming?

If a provider lacks a native streaming implementation, aisuite falls back to the base class `achat_completions_create_stream` method in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py). This bridge runs the provider's synchronous method in a background thread using `run_in_executor`, yielding chunks through an `asyncio.Queue` to maintain a non-blocking async interface.

### How does aisuite handle different chunk formats from various providers?

aisuite converts all provider-specific stream formats into OpenAI-compatible `ChatCompletionChunk` objects. OpenAI streams pass through unchanged, while Anthropic events are transformed by `AnthropicMessageConverter.convert_stream_event` and Gemini chunks are processed through `_StreamState.convert`. This ensures every provider returns identical object structures regardless of native API differences.

### Can I use streaming with async/await for all providers?

Yes. Every provider supports async streaming via `client.chat.completions.acreate(..., stream=True)`. Providers with native async SDKs (like OpenAI and Anthropic) use their own optimized implementations, while others utilize the generic async bridge that wraps synchronous streams in threads, guaranteeing consistent async behavior across the entire provider ecosystem.

### Where is the streaming logic implemented in the aisuite source code?

The streaming architecture spans three key files: [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) defines the base contract and async bridge; [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py), [`anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/anthropic_provider.py) (starting line 387), and [`gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/gemini_provider.py) (starting line 409) contain provider-specific implementations; and [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) routes user requests to these provider methods based on the `stream=True` flag.