# How Streaming Works in aisuite Across Different LLM Providers: A Code-Level Breakdown

> Explore the code-level breakdown of LLM streaming in aisuite. Learn how it unifies provider streams into a standard ChatCompletionChunk object for seamless integration.

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

---

**aisuite unifies LLM streaming behind a single `client.chat.completions.create(..., stream=True)` call by converting every provider's native stream into a standard `ChatCompletionChunk` object, automatically supplying an async bridge for providers that only implement synchronous streaming.**

Streaming in aisuite is engineered to hide vendor-specific differences behind one consistent Python API. According to the `andrewyng/aisuite` source code, the library normalizes disparate formats—whether they come from OpenAI, Anthropic, Gemini, or another service—into a single OpenAI-style chunk structure. Every returned object is an instance of `aisuite.framework.chat_completion_chunk.ChatCompletionChunk`, giving you the same fields regardless of which model is running in the background.

## Three-Layer Streaming Architecture

The streaming implementation is split into three layers. The **provider base class** in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) defines the interface contract and a generic async fallback. **Provider-specific overrides** in files such as [`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), and [`aisuite/providers/gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/gemini_provider.py) translate native SDK streams into the common chunk format. The **client façade** in [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) inspects the `stream=True` flag and delegates to the appropriate provider method.

This design means that adding streaming support for a new provider usually requires implementing only a synchronous generator; the base class handles the async counterpart automatically.

## Provider Base Class and the Default Async Bridge

The abstract `Provider` class in [`aisuite/provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/provider.py) establishes the streaming contract. By default, `chat_completions_create_stream` raises an `LLMError`:

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

```

If a provider overrides this sync method but does not define an async equivalent, the base class supplies `achat_completions_create_stream`. This method runs the synchronous stream in a worker thread through `run_in_executor` and forwards each chunk through an `asyncio.Queue`:

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

```

Any provider that implements the sync generator instantly gains a non-blocking async interface without extra code.

## How Each LLM Provider Implements Streaming in aisuite

Providers that support streaming natively override `chat_completions_create_stream` to return their own generators, then normalize the output to aisuite's `ChatCompletionChunk`.

### OpenAI: Direct Pass-Through

[`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py) returns the OpenAI SDK's native iterator directly because it already emits objects matching aisuite's expected shape. The synchronous and asynchronous implementations are thin wrappers:

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

```

No additional conversion is required, so OpenAI streams move straight from the SDK to your loop.

### Anthropic: Event-to-Chunk Conversion

In [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py)—with the streaming logic starting around line 387—the Anthropic SDK returns a stream of mixed-type events. The provider consumes these events and feeds each one through `AnthropicMessageConverter.convert_stream_event`, defined at 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 tool-use start events, text deltas, and final message metadata, mapping them into `ChatCompletionChunk` objects with correct `finish_reason` and usage fields.

### Gemini: Stateful Folding

[`aisuite/providers/gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/gemini_provider.py) (streaming logic begins around line 409) uses a small state machine called `_StreamState`, found at lines 31–92, to fold Gemini's `generate_content_stream` chunks into OpenAI-shaped chunks. It also synthesizes tool calls and attaches usage metadata:

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

```

Because Gemini's streaming API delivers whole tool-call parts, `_StreamState` can emit each call as a single chunk without accumulating incremental deltas across multiple events.

### Other Providers: Thread-Based Fallback

Providers such as **Mistral**, **Cohere**, and **Fireworks** do not implement their own `chat_completions_create_stream`. When you request `stream=True` with these backends, aisuite falls back to the base class bridge. The synchronous stream runs inside a background thread, while your async code receives chunks via the `asyncio.Queue` mechanism shown previously. This guarantees a uniform non-blocking interface for every supported vendor.

## Practical Streaming Examples with aisuite

### Synchronous Streaming with Any Provider

The following snippet works with OpenAI, Anthropic, Gemini, or any other provider that either natively supports streaming or relies on the default bridge:

```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, flush=True)

```

Each `chunk` is a `ChatCompletionChunk`, so you can consume it the same way regardless of the backend.

### Async Streaming with Tool Calls (Anthropic)

Anthropic's async stream is converted automatically. Tool-call chunks appear as `delta.tool_calls` just as they would in OpenAI's format:

```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("Tool call:", tool_calls[0].function.name, tool_calls[0].function.arguments)

asyncio.run(main())

```

`AnthropicMessageConverter` normalizes the native event stream so the caller never handles Anthropic-specific event types directly.

### Async Streaming via the Base Bridge (Mistral)

If a provider lacks a native async stream, aisuite uses the base bridge. Here is how you stream from Mistral without blocking the event loop:

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

```

Because `achat_completions_create_stream` is inherited from the base class, the sync stream executes in a worker thread and yields chunks asynchronously.

## Summary

- **aisuite/provider.py** defines the streaming contract and a generic async bridge using `asyncio.Queue` and `run_in_executor`.
- **OpenAI** streaming in [`aisuite/providers/openai_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/openai_provider.py) is a direct pass-through because the SDK already returns the correct chunk shape.
- **Anthropic** streaming in [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py) relies on `AnthropicMessageConverter.convert_stream_event` (lines 62–154) to map mixed SDK events into `ChatCompletionChunk` objects.
- **Gemini** streaming in [`aisuite/providers/gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/gemini_provider.py) uses `_StreamState` (lines 31–92) to fold native chunks and synthesize tool calls and usage metadata.
- **All other providers** inherit the default bridge, ensuring every backend supports `stream=True` with a non-blocking async interface.
- The client façade in [`aisuite/client.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/client.py) routes `stream=True` to the right provider method, always returning uniform `ChatCompletionChunk` instances.

## Frequently Asked Questions

### What object does aisuite return during streaming?

aisuite returns instances of `aisuite.framework.chat_completion_chunk.ChatCompletionChunk` for every provider. This object mirrors the OpenAI streaming delta format, exposing fields such as `choices[0].delta.content` and `choices[0].delta.tool_calls`.

### Does every provider in aisuite support streaming?

Yes. Native providers such as OpenAI, Anthropic, and Gemini implement their own `chat_completions_create_stream` generators. Providers without a native implementation inherit the base class async bridge, which runs a synchronous stream in a background worker thread so that `stream=True` never blocks the event loop.

### How does aisuite convert Anthropic streaming events into OpenAI chunks?

In [`aisuite/providers/anthropic_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/anthropic_provider.py), each event from the Anthropic SDK is passed to `AnthropicMessageConverter.convert_stream_event`. This method—located at lines 62–154—normalizes tool-use start signals, text deltas, and final metadata into the standard `ChatCompletionChunk` shape, preserving usage counts and finish reasons.

### Can I receive tool calls while streaming through aisuite?

Yes. Both Anthropic and Gemini handle tool-call streaming. Anthropic's native events are mapped to `ChatCompletionChunk` deltas containing `tool_calls`, while Gemini's `_StreamState` machine in [`aisuite/providers/gemini_provider.py`](https://github.com/andrewyng/aisuite/blob/main/aisuite/providers/gemini_provider.py) emits complete tool-call chunks as soon as the full call is received from the underlying stream.