How aisuite Handles Streaming Across LLM Providers: Architecture and Implementation

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, 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:

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:

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, the OpenAI implementation forwards the SDK's native stream directly without conversion, as OpenAI's format already matches aisuite's ChatCompletionChunk structure:

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, 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):

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, 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:

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

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

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)

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, 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. 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 defines the base contract and async bridge; aisuite/providers/openai_provider.py, anthropic_provider.py (starting line 387), and gemini_provider.py (starting line 409) contain provider-specific implementations; and aisuite/client.py routes user requests to these provider methods based on the stream=True flag.

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