Migrating from ElevenLabs with Manual Commits

pip install cartesia

import os
from cartesia import AsyncCartesia

client = AsyncCartesia(api_key=os.getenv("CARTESIA_API_KEY"))

async with client.stt.manual_finalize.websocket(
    model="ink-2", encoding="pcm_s16le", sample_rate=16000
) as connection:
    ...

# ElevenLabs
await elevenlabs_connection.send({
  "audio_base_64": b64encode(raw_audio),
})

# Cartesia
# raw_audio (bytes) - Raw audio data, about 100 ms at a time
await connection.send_raw(raw_audio)

# ElevenLabs
await elevenlabs_connection.send({
  "audio_base_64": audio_base_64,
})

# Cartesia
from base64 import b64decode
await connection.send_raw(b64decode(audio_base_64))

# ElevenLabs
await elevenlabs_connection.commit()

# Cartesia
await connection.send("finalize")

# ElevenLabs: close the socket immediately
await elevenlabs_connection.close()

# Cartesia: commit remaining audio
# and let the server close the socket once done
await connection.send("close")

# Cartesia: Close the socket early (optional)
await connection.close()

import asyncio
from cartesia.types.stt import STTManualFinalizeWebsocketResponse

partial_transcript = ""

def on_message(message: STTManualFinalizeWebsocketResponse) -> None:
    global partial_transcript
    if message.type == "transcript" and message.is_final:
        # Do not strip or add whitespace!
        partial_transcript += message.text
        print(f"partial_transcript: {partial_transcript}")
    elif message.type == "flush_done" or message.type == "done":
        print(f"committed_transcript: {partial_transcript}")
        partial_transcript = ""
    elif message.type == "error":
        error_code = message.error_code or "unknown_error"
        if error_code == "quota_exceeded":
            print("You are out of credits")
        elif error_code == "concurrency_limited":
            print("You have too many open STT connections")
        else:
            print(f"{error_code}: {message.message}")

connection.on("event", on_message)

# ElevenLabs dispatches to your callbacks automatically;
# with Cartesia you run the dispatch loop yourself
recv_task = asyncio.create_task(connection.dispatch_events())