Streaming transcription

For applications requiring the lowest latency, use the real-time WebSocket API. This provides streaming transcription with incremental results. You have two ways to connect:

The Python SDK (client.beta.realtime.transcription()), which handles the WebSocket, reconnection, and audio replay for you. Use this for most applications.
The raw WebSocket protocol, for languages other than Python or when you need full control over the wire.

Switch between them with the tabs below.

The server uses Voice Activity Detection (VAD) to automatically segment speech. You can tune VAD parameters for your audio characteristics. See the Voice activity detection guide for configuration details and common presets.

Python SDK
Raw WebSocket protocol

The Python SDK for real-time transcription is in beta, and the API surface may change before it stabilizes. Share feedback with [email protected].

The SDK opens the WebSocket, streams your audio, and returns transcription events. When the connection drops mid-conversation, the session holds on to the speech the server hadn’t transcribed yet, reconnects automatically, and picks up where the transcript left off, so words spoken during the outage still come back as text.Install the SDK with the realtime extra:

pip install "together[realtime]"

Basic usage

Call client.beta.realtime.transcription() to open a session, feed audio with session.append(), and consume events by iterating the session. Each TranscriptDelta is an interim result that updates while a phrase is being spoken; each TranscriptCompleted is the finalized transcript for one utterance.

import asyncio

from together import AsyncTogether
from together.realtime import TranscriptCompleted, TranscriptDelta


async def main():
    client = AsyncTogether()

    async with client.beta.realtime.transcription(
        model="openai/whisper-large-v3",
        sample_rate=16_000,
    ) as session:
        # Feed audio from your capture source as it arrives (any chunk size).
        await session.append(pcm_chunk)  # 16 kHz mono 16-bit PCM

        async for event in session:
            if isinstance(event, TranscriptDelta):
                print("interim:", event.text)
            elif isinstance(event, TranscriptCompleted):
                print("final:", event.text)

        # Finalize whatever was said last.
        transcript = await session.flush()
        print("full transcript:", transcript)


asyncio.run(main())

session.append() never blocks on network state, so it is safe to call from a capture loop. Instead of iterating the session, you can pass an event_callback= function to handle events as they arrive.

Audio format

Audio in is 16 kHz mono 16-bit PCM (pcm_s16le_16000). Resample your source before appending. Passing sample_rate= lets the SDK reject a mismatch loudly instead of silently transcribing the wrong sample rate.

Utterance boundaries

Utterance boundaries are detected server-side by default. Final transcripts arrive on their own as the speaker pauses. To control segmentation yourself, pass turn_detection={"type": "none"} and call await session.commit() when each segment ends.

Session events

Iterate the session (or pass event_callback=) to receive normalized events. Import them from together.realtime.

Event	Meaning
`SessionStarted`	The session opened. `session_id` is useful for correlating with server-side logs.
`TranscriptDelta`	Interim text for the current utterance. Updates as the phrase is spoken.
`TranscriptCompleted`	Finalized transcript for one utterance.
`TranscriptFailed`	One utterance failed server-side. The session continues.
`Reconnecting`	A transient failure occurred and the SDK is reconnecting with backoff. No action needed.
`Reconnected`	The connection recovered. `replayed_seconds` reports how much speech was replayed.
`BufferGap`	Audio was dropped beyond recovery (an outage longer than the retention window). `dropped_seconds` lost.

TranscriptDelta and TranscriptCompleted may also include optional quality fields when the server sends them: logprobs (avg_logprob, token_logprobs, token_texts) and tokens (per-token token_id, text, and confidence).

Reconnection and replay

When the connection drops, the session emits Reconnecting and Reconnected events and retries on its own. By default the SDK makes up to two same-endpoint reconnect attempts (reconnect={"max_attempts": 2}) before raising. Transcripts recomputed from speech carried across the reconnect are marked replayed=True and may overlap text you already received. Voice agents that act on each final result can set buffer={"max_replay_seconds": 0} to resume live with no re-emission instead.

Failover across endpoints

If an endpoint fails for good, calls raise RealtimeConnectionError. When the server reports it cannot currently serve (exc.code == "no_healthy_workers", including WebSocket close code 4503), the SDK raises immediately with no same-endpoint retry so you can rotate. To keep a conversation alive across endpoint outages, run a failover ring: on failure, session.pending_audio() hands you the un-transcribed speech to seed a new session on another endpoint.

import asyncio

from together import AsyncTogether
from together.realtime import RealtimeConnectionError

# Independent deployments of the same model. On failure, move to the next.
endpoints = [
    ("https://api.together.ai/v1", "openai/whisper-large-v3-endpoint1"),
    ("https://api.together.ai/v1", "openai/whisper-large-v3-endpoint2"),
]


async def transcribe_with_failover():
    carry_over = b""  # audio a failed endpoint received but never transcribed

    for base_url, model in endpoints:
        client = AsyncTogether(base_url=base_url)
        session = client.beta.realtime.transcription(
            model=model,
            sample_rate=16_000,
            reconnect={"max_attempts": 0},  # switch endpoints immediately
        )
        try:
            async with session:
                if carry_over:
                    await session.append(carry_over)
                # ... stream the rest of your audio ...
            break
        except RealtimeConnectionError as exc:
            # Includes exc.code == "no_healthy_workers". Take back
            # untranscribed audio.
            carry_over = session.pending_audio()


asyncio.run(transcribe_with_failover())

Synchronous usage

Together().beta.realtime.transcription(...) mirrors the async API on a background thread. Use it for a handful of concurrent sessions. For high concurrency, use the async client.

Key parameters

Parameter	Description
`model`	Model to use, for example `openai/whisper-large-v3`.
`sample_rate`	Sample rate of the audio you append. Lets the SDK validate the format.
`input_audio_format`	Wire audio format. Defaults to `pcm_s16le_16000`.
`language`	Source language hint passed through to the service.
`prompt`	Text prompt to bias transcription, passed through to the service.
`turn_detection`	Turn-detection config (for example `{"type": "none"}`, `min_silence_duration_ms`, `max_speech_duration_s`).
`reconnect`	Reconnection policy. Defaults to `max_attempts=2`. Use `{"max_attempts": 0}` to fail over immediately.
`buffer`	Replay buffer config, for example `{"max_replay_seconds": 0}` to resume live without re-emitting speech.
`keepalive_silence`	When `True`, send silence so idle sessions stay alive past the server idle timeout.
`event_callback`	A function called with each event, as an alternative to iterating the session.

For full manual control over the raw wire events with no automatic recovery, use client.beta.realtime.connect().

Use the raw WebSocket protocol from languages other than Python, or when you need full control over the wire. The Python SDK wraps this same protocol.

Establish a connection

Connect to: wss://api.together.ai/v1/realtime?model={model}&input_audio_format=pcm_s16le_16000Headers:

{
  'Authorization': 'Bearer $TOGETHER_API_KEY',
  'OpenAI-Beta': 'realtime=v1'
}

Query parameters

Parameter	Type	Required	Description
model	string	Yes	Model to use (e.g., `openai/whisper-large-v3`)
input_audio_format	string	Yes	Audio format: `pcm_s16le_16000`

Client-to-server messages

Append audio to buffer

{
  "type": "input_audio_buffer.append",
  "audio": "base64-encoded-audio-chunk"
}

Send audio data in base64-encoded PCM format.

Commit audio buffer

{
  "type": "input_audio_buffer.commit"
}

Forces transcription of any remaining audio in the server-side buffer.

Server-to-client messages

Delta events (intermediate results)

{
  "type": "conversation.item.input_audio_transcription.delta",
  "delta": "The quick brown fox jumps"
}

Delta events are intermediate transcriptions. The model is still processing and may revise the output. Each delta message overrides the previous delta.

Completed events (final results)

{
  "type": "conversation.item.input_audio_transcription.completed",
  "transcript": "The quick brown fox jumps over the lazy dog"
}

Completed events are final transcriptions. The model is confident about this text. The next delta event continues from where this completed.

Real-time example

import asyncio
import base64
import json
import os
import sys

import numpy as np
import sounddevice as sd
import websockets

# Configuration
API_KEY = os.getenv("TOGETHER_API_KEY")
MODEL = "openai/whisper-large-v3"
SAMPLE_RATE = 16000
BATCH_SIZE = 4096  # 256ms batches for optimal performance

if not API_KEY:
    print("Error: Set TOGETHER_API_KEY environment variable")
    sys.exit(1)


class RealtimeTranscriber:
    """Realtime transcription client for Together AI."""

    def __init__(self):
        self.ws = None
        self.stream = None
        self.is_ready = False
        self.audio_buffer = np.array([], dtype=np.float32)
        self.audio_queue = asyncio.Queue()

    async def connect(self):
        """Connect to Together AI API."""
        url = (
            f"wss://api.together.ai/v1/realtime"
            f"?intent=transcription"
            f"&model={MODEL}"
            f"&input_audio_format=pcm_s16le_16000"
            f"&authorization=Bearer {API_KEY}"
        )

        self.ws = await websockets.connect(
            url,
            subprotocols=[
                "realtime",
                f"openai-insecure-api-key.{API_KEY}",
                "openai-beta.realtime-v1",
            ],
        )

    async def send_audio(self):
        """Capture and send audio to API."""

        def audio_callback(indata, frames, time, status):
            self.audio_queue.put_nowait(indata.copy().flatten())

        # Start microphone stream
        self.stream = sd.InputStream(
            samplerate=SAMPLE_RATE,
            channels=1,
            dtype="float32",
            blocksize=1024,
            callback=audio_callback,
        )
        self.stream.start()

        # Process and send audio
        while True:
            try:
                audio = await asyncio.wait_for(
                    self.audio_queue.get(), timeout=0.1
                )

                if self.ws and self.is_ready:
                    # Add to buffer
                    self.audio_buffer = np.concatenate(
                        [self.audio_buffer, audio]
                    )

                    # Send when buffer is full
                    while len(self.audio_buffer) >= BATCH_SIZE:
                        batch = self.audio_buffer[:BATCH_SIZE]
                        self.audio_buffer = self.audio_buffer[BATCH_SIZE:]

                        # Convert float32 to int16 PCM
                        audio_int16 = (
                            np.clip(batch, -1.0, 1.0) * 32767
                        ).astype(np.int16)
                        audio_base64 = base64.b64encode(
                            audio_int16.tobytes()
                        ).decode()

                        # Send to API
                        await self.ws.send(
                            json.dumps(
                                {
                                    "type": "input_audio_buffer.append",
                                    "audio": audio_base64,
                                }
                            )
                        )

            except asyncio.TimeoutError:
                continue
            except Exception as e:
                print(f"Error: {e}", file=sys.stderr)
                break

    async def receive_transcriptions(self):
        """Receive and display transcription results."""
        current_interim = ""

        try:
            async for message in self.ws:
                data = json.loads(message)

                if data["type"] == "session.created":
                    self.is_ready = True

                elif (
                    data["type"]
                    == "conversation.item.input_audio_transcription.delta"
                ):
                    # Interim result
                    print(
                        f"\r\033[90m{data['delta']}\033[0m", end="", flush=True
                    )
                    current_interim = data["delta"]

                elif (
                    data["type"]
                    == "conversation.item.input_audio_transcription.completed"
                ):
                    # Final result
                    if current_interim:
                        print("\r\033[K", end="")
                    print(f"\033[92m{data['transcript']}\033[0m")
                    current_interim = ""

                elif data["type"] == "error":
                    print(f"\nError: {data.get('message', 'Unknown error')}")

        except websockets.exceptions.ConnectionClosed:
            pass

    async def close(self):
        """Close connections and cleanup."""
        if self.stream:
            self.stream.stop()
            self.stream.close()

        # Flush remaining audio
        if len(self.audio_buffer) > 0 and self.ws and self.is_ready:
            try:
                audio_int16 = (
                    np.clip(self.audio_buffer, -1.0, 1.0) * 32767
                ).astype(np.int16)
                audio_base64 = base64.b64encode(audio_int16.tobytes()).decode()
                await self.ws.send(
                    json.dumps(
                        {
                            "type": "input_audio_buffer.append",
                            "audio": audio_base64,
                        }
                    )
                )
            except Exception:
                pass

        if self.ws:
            await self.ws.close()

    async def run(self):
        """Main execution loop."""
        try:
            print("🎤 Together AI Realtime Transcription")
            print("=" * 40)
            print("Connecting...")

            await self.connect()

            print("✓ Connected")
            print("✓ Recording started - speak now\n")

            # Run audio capture and transcription concurrently
            await asyncio.gather(
                self.send_audio(), self.receive_transcriptions()
            )

        except KeyboardInterrupt:
            print("\n\nStopped")
        except Exception as e:
            print(f"Error: {e}", file=sys.stderr)
        finally:
            await self.close()


async def main():
    transcriber = RealtimeTranscriber()
    await transcriber.run()


if __name__ == "__main__":
    asyncio.run(main())

import WebSocket from 'ws';
import recorder from 'node-record-lpcm16';

// Configuration
const API_KEY = process.env.TOGETHER_API_KEY;
const MODEL = 'openai/whisper-large-v3';
const SAMPLE_RATE = 16000;

if (!API_KEY) {
  console.error('Error: Set TOGETHER_API_KEY environment variable');
  process.exit(1);
}

class RealtimeTranscriber {
  /** Realtime transcription client for Together AI. */
  private ws: WebSocket | null = null;
  private isReady = false;
  private currentInterim = '';

  async connect() {
    /** Connect to Together AI API. */
    const url =
      `wss://api.together.ai/v1/realtime` +
      `?intent=transcription` +
      `&model=${MODEL}` +
      `&input_audio_format=pcm_s16le_16000` +
      `&authorization=Bearer ${API_KEY}`;

    this.ws = new WebSocket(url, [
      'realtime',
      `openai-insecure-api-key.${API_KEY}`,
      'openai-beta.realtime-v1',
    ]);

    this.ws.on('message', (data) => this.receiveTranscriptions(data));
    this.ws.on('error', (err) => console.error(`Error: ${err}`));

    return new Promise((resolve) => {
      this.ws?.on('open', () => {
        resolve(null);
      });
    });
  }

  sendAudio() {
    /** Capture and send audio to API. */
    const mic = recorder.record({
      sampleRate: SAMPLE_RATE,
      threshold: 0,
      verbose: false,
    });

    mic.stream().on('data', (chunk: Buffer) => {
      if (this.ws && this.isReady && this.ws.readyState === WebSocket.OPEN) {
        this.ws.send(
          JSON.stringify({
            type: 'input_audio_buffer.append',
            audio: chunk.toString('base64'),
          })
        );
      }
    });

    mic.stream().on('error', (err) => {
      console.error('Microphone Error:', err);
    });
  }

  receiveTranscriptions(data: WebSocket.Data) {
    /** Receive and display transcription results. */
    const message = JSON.parse(data.toString());

    if (message.type === 'session.created') {
      this.isReady = true;
    } else if (
      message.type === 'conversation.item.input_audio_transcription.delta'
    ) {
      // Interim result
      process.stdout.write(`\r\x1b[90m${message.delta}\x1b[0m`);
      this.currentInterim = message.delta;
    } else if (
      message.type === 'conversation.item.input_audio_transcription.completed'
    ) {
      // Final result
      if (this.currentInterim) {
        process.stdout.write('\r\x1b[K');
      }
      console.log(`\x1b[92m${message.transcript}\x1b[0m`);
      this.currentInterim = '';
    } else if (message.type === 'error') {
      console.error(`\nError: ${message.message || 'Unknown error'}`);
    }
  }

  async run() {
    /** Main execution loop. */
    try {
      console.log('🎤 Together AI Realtime Transcription');
      console.log('='.repeat(40));
      console.log('Connecting...');

      await this.connect();

      console.log('✓ Connected');
      console.log('✓ Recording started - speak now\n');

      this.sendAudio();
    } catch (e) {
      console.error(`Error: ${e}`);
    }
  }
}

async function main() {
  const transcriber = new RealtimeTranscriber();
  await transcriber.run();
}

main();

GET STARTED

SERVERLESS

INFERENCE APIS

DEDICATED MODEL INFERENCE

DEDICATED CONTAINER INFERENCE

GPU CLUSTERS

FINE-TUNING

CODE EXECUTION

ADMINISTRATION

Basic usage

Audio format

Utterance boundaries

Session events

Reconnection and replay

Failover across endpoints

Synchronous usage

Key parameters

Establish a connection

Query parameters

Client-to-server messages

Append audio to buffer

Commit audio buffer

Server-to-client messages

Delta events (intermediate results)

Completed events (final results)

Real-time example

​Basic usage

​Audio format

​Utterance boundaries

​Session events

​Reconnection and replay

​Failover across endpoints

​Synchronous usage

​Key parameters

​Establish a connection

​Query parameters

​Client-to-server messages

​Append audio to buffer

​Commit audio buffer

​Server-to-client messages

​Delta events (intermediate results)

​Completed events (final results)

​Real-time example

Basic usage

Audio format

Utterance boundaries

Session events

Reconnection and replay

Failover across endpoints

Synchronous usage

Key parameters

Establish a connection

Query parameters

Client-to-server messages

Append audio to buffer

Commit audio buffer

Server-to-client messages

Delta events (intermediate results)

Completed events (final results)

Real-time example