How Streaming Works Across Different LLM Providers in aisuite

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

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.

Client Facade – The Public API

The Chat.Completions.create method in 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, the implementation simply yields the SDK's generator directly.

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

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

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.

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.

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.

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 provides a thread-based bridge, enabling async streaming for providers without native async implementations.
• Source Files: Core logic resides in aisuite/provider.py, aisuite/providers/openai_provider.py, aisuite/providers/anthropic_provider.py, aisuite/providers/gemini_provider.py, and 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 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.

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