How to Handle Streaming Transcription Responses in aisuite

aisuite unifies real-time speech-to-text across OpenAI, Google, and Deepgram by exposing a single async generator that yields StreamingTranscriptionChunk objects, allowing you to process partial transcripts as they arrive while switching providers with only a configuration change.

The andrewyng/aisuite library abstracts away provider-specific streaming complexities when handling streaming transcription responses in aisuite. By implementing a consistent Audio → Transcription interface across OpenAI Whisper, Google Vertex AI, and Deepgram, the framework lets you consume real-time transcription chunks through a unified async iterator without managing WebSocket connections or vendor-specific event types.

Architecture of Streaming Transcription in aisuite

The Abstract Provider Contract

In aisuite/provider.py (lines 68-98), the base Audio.Transcription class defines the abstract method create_stream_output. This method serves as the standard entry point that all ASR providers must implement, ensuring that every backend returns an async generator of standardized chunks regardless of their underlying protocol.

The Unified Data Model

The StreamingTranscriptionChunk class in aisuite/framework/message.py normalizes output across all services:

class StreamingTranscriptionChunk(BaseModel):
    text: str                       # The transcribed snippet

    is_final: bool                  # True when the chunk marks the end of a segment

    confidence: Optional[float]     # Confidence score (if supplied)

Provider-Specific Implementations

Each provider translates its native streaming protocol into the common chunk format:

• OpenAI: OpenAIAudio.Transcriptions.create_stream_output in aisuite/providers/openai_provider.py (lines 111-194)
• Google: GoogleAudio.Transcriptions.create_stream_output in aisuite/providers/google_provider.py (lines 83-134)
• Deepgram: DeepgramAudio.Transcriptions.create_stream_output in aisuite/providers/deepgram_provider.py (lines 118-227)

How the Streaming Flow Works

1. Client Invocation: When you call client.provider.audio.transcriptions.create_stream_output(...), the client façade in aisuite/client.py (lines 560-562) forwards the request directly to the active provider's concrete implementation.

2. Provider Request Preparation:

   • OpenAI & Google: Inject stream=True (or provider-specific equivalents) and optionally enforce response_format='verbose_json' when requesting timestamp granularities.
   • Deepgram: Constructs a chunked audio pipeline and opens an async WebSocket connection, registering callbacks for transcript, error, and close events.

3. Event Translation: Each provider iterates over raw streaming events and maps them to StreamingTranscriptionChunk:

   • OpenAI: Inspects event.type for transcript.text.delta (yielding is_final=False) and transcript.text.done (yielding is_final=True).
   • Google: Yields a chunk for every alternative in the streamed response, using result.is_final to populate the is_final field.
   • Deepgram: WebSocket callbacks push chunks into a queue.Queue; the async generator drains this queue until the connection signals closure.

4. Unified Consumption: Regardless of backend, you receive an async iterator of StreamingTranscriptionChunk objects, enabling real-time processing of partial results.

Code Example: Real-Time Transcription

import asyncio
from aisuite.client import AISuiteClient

async def stream_transcription():
    # Initialise a client with the desired provider (e.g. OpenAI)

    client = AISuiteClient(provider="openai", config={"api_key": "YOUR_KEY"})
    
    # Pick a model that supports streaming (e.g. "whisper-1")

    model = "whisper-1"

    # Path to a local audio file or any file‑like object

    audio_path = "speech.wav"

    # Call the streaming API – this returns an async generator

    async for chunk in client.provider.audio.transcriptions.create_stream_output(
        model=model,
        file=audio_path,
    ):
        # Each chunk arrives as soon as the service has a partial result

        print(f"{'FINAL' if chunk.is_final else 'PARTIAL'}: {chunk.text}")

# Run the coroutine

asyncio.run(stream_transcription())

This identical pattern works with Google or Deepgram providers by changing only the provider argument and model name.

Provider-Specific Streaming Mechanisms

OpenAI Whisper Streaming

The OpenAI provider in aisuite/providers/openai_provider.py (lines 111-194) handles the Whisper API's server-sent events. It distinguishes between interim deltas and final segments by checking event.type, mapping these to is_final=False and is_final=True respectively.

Google Vertex AI Streaming

Implemented in aisuite/providers/google_provider.py (lines 83-134), the Google provider streams from Vertex AI's speech-to-text service, extracting alternatives from each response packet and preserving the service's native is_final flag in the unified chunk.

Deepgram WebSocket Streaming

The Deepgram implementation in aisuite/providers/deepgram_provider.py (lines 118-227) manages a full WebSocket lifecycle. It handles audio chunking, connection management, and callback registration, abstracting the complexity of Deepgram's streaming protocol behind the simple async generator interface.

Summary

• Unified Interface: All providers implement create_stream_output in aisuite/provider.py, returning provider-agnostic async generators.
• Chunk Semantics: StreamingTranscriptionChunk from aisuite/framework/message.py provides text, is_final, and optional confidence fields, essential for distinguishing interim from final results.
• Zero Code Changes: Switching between OpenAI, Google, and Deepgram requires only updating the provider configuration, not your streaming consumption logic.
• Real-Time Processing: The architecture supports immediate processing of partial transcripts for live captioning or voice-controlled applications.

Frequently Asked Questions

What return type does create_stream_output provide when handling streaming transcription responses in aisuite?

The method returns an async generator that yields StreamingTranscriptionChunk objects. These chunks contain text (the transcribed snippet), is_final (boolean indicating segment completion), and optional confidence scores, providing a consistent interface across OpenAI, Google, and Deepgram providers.

How does aisuite distinguish between partial and final transcription segments?

The is_final boolean field in StreamingTranscriptionChunk indicates whether a chunk completes a transcript segment. OpenAI sets this based on event.type (deltas vs. done events), Google maps it directly from result.is_final, and Deepgram determines it through WebSocket message callbacks, all normalized in the unified data model.

Can I switch from OpenAI to Deepgram without rewriting my streaming code?

Yes. Because aisuite/client.py (lines 560-562) forwards calls to the abstract provider interface, you only need to change the provider parameter when initializing AISuiteClient. The create_stream_output method signature and return type remain identical, ensuring your async iteration logic works across all supported ASR services.

Why does the Deepgram provider use a queue-based approach while others use direct iteration?

Deepgram's implementation in aisuite/providers/deepgram_provider.py (lines 118-227) uses WebSocket callbacks that push data into a queue.Queue because WebSocket event handling is callback-driven rather than iterator-based. The provider wraps this in an async generator to maintain the uniform streaming interface required by the abstract base class in aisuite/provider.py.

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