Cartesia

SAMBANOVA_API_KEY="your-sambanova-api-key"
CARTESIA_API_KEY="your-cartesia-api-key"

import asyncio
import os

import sounddevice as sd
from cartesia import AsyncCartesia
from dotenv import load_dotenv
from sambanova import AsyncSambaNova

load_dotenv()

QUESTION = 'In two sentences, why is the sky blue?'
SAMPLE_RATE = 44100


async def main() -> None:
    llm = AsyncSambaNova(api_key=os.environ['SAMBANOVA_API_KEY'])
    tts = AsyncCartesia(api_key=os.environ['CARTESIA_API_KEY'])
    speaker = sd.RawOutputStream(samplerate=SAMPLE_RATE, channels=1, dtype='int16')
    speaker.start()

    async with tts.tts.websocket_connect() as conn:
        # One context = one utterance. Text pushed into it is spoken as it arrives.
        ctx = conn.context(
            model_id='sonic-3.6',
            voice={'mode': 'id', 'id': '6ccbfb76-1fc6-48f7-b71d-91ac6298247b'},
            output_format={'container': 'raw', 'encoding': 'pcm_s16le', 'sample_rate': SAMPLE_RATE},
            language='en',
        )

        async def write() -> None:
            stream = await llm.chat.completions.create(
                model='Meta-Llama-3.3-70B-Instruct',
                messages=[{'role': 'user', 'content': QUESTION}],
                stream=True,
            )
            async for chunk in stream:
                text = chunk.choices[0].delta.content if chunk.choices else None
                if text:  # only the answer is spoken, never the model's reasoning
                    print(text, end='', flush=True)
                    await ctx.push(text)
            await ctx.no_more_inputs()

        async def speak() -> None:
            async for event in ctx.receive():
                if event.type == 'chunk' and event.audio:
                    await asyncio.to_thread(speaker.write, event.audio)
                elif event.type == 'error':
                    raise RuntimeError(event.message or event.title or 'unknown error')

        await asyncio.gather(write(), speak())

    speaker.stop()  # waits for the last audio to finish playing
    speaker.close()
    await tts.close()
    print()


asyncio.run(main())