Voice Assistant Pipeline

import asyncio
import time
from openai import AsyncOpenAI

client = AsyncOpenAI()

SENTENCE_ENDINGS = {'.', '!', '?', '\n'}

async def transcribe(audio_path: str) -> str:
    """ASR: audio file → text via Whisper."""
    with open(audio_path, "rb") as f:
        transcript = await client.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            response_format="text"
        )
    return transcript

async def stream_llm_to_tts(
    user_text: str,
    conversation: list[dict],
    on_audio_chunk: callable
):
    """Stream LLM response sentence-by-sentence into TTS.

    This is the core optimization: we don't wait for the full
    LLM response before starting TTS. Each complete sentence
    is sent to TTS immediately, and audio chunks play as they
    arrive.
    """
    conversation.append({"role": "user", "content": user_text})

    # Stream LLM response
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",         # Fast TTFT (~150ms)
        messages=conversation,
        max_tokens=200,              # Keep voice responses concise
        stream=True
    )

    sentence_buffer = ""
    full_response = ""
    tts_tasks = []

    async for chunk in stream:
        token = chunk.choices[0].delta.content or ""
        sentence_buffer += token
        full_response += token

        # Flush on sentence boundary
        if token and token[-1] in SENTENCE_ENDINGS and len(sentence_buffer.strip()) > 10:
            sentence = sentence_buffer.strip()
            sentence_buffer = ""

            # Fire TTS for this sentence concurrently
            task = asyncio.create_task(
                speak_sentence(sentence, on_audio_chunk)
            )
            tts_tasks.append(task)

    # Flush remaining buffer
    if sentence_buffer.strip():
        task = asyncio.create_task(
            speak_sentence(sentence_buffer.strip(), on_audio_chunk)
        )
        tts_tasks.append(task)

    # Wait for all TTS tasks to complete
    await asyncio.gather(*tts_tasks)

    conversation.append({"role": "assistant", "content": full_response})
    return full_response

async def speak_sentence(text: str, on_audio_chunk: callable):
    """TTS: text → streaming audio chunks."""
    response = await client.audio.speech.create(
        model="tts-1",              # Low-latency model
        voice="nova",
        input=text,
        response_format="opus"      # Efficient for streaming
    )

    async for chunk in response.iter_bytes(chunk_size=4096):
        on_audio_chunk(chunk)        # Play immediately

async def voice_assistant_loop():
    """Main loop with timing instrumentation."""
    conversation = [{
        "role": "system",
        "content": (
            "You are a voice assistant. Respond in 1-3 short sentences. "
            "Be conversational and concise — the user is listening, not reading."
        )
    }]

    print("Voice assistant ready. Ctrl+C to exit.")

    while True:
        # In production: record with VAD, save to temp file
        audio_path = await record_with_vad()  # Your recording function

        t0 = time.perf_counter()
        user_text = await transcribe(audio_path)
        t_asr = time.perf_counter() - t0

        if not user_text.strip():
            continue

        print(f"User: {user_text}  [ASR: {t_asr*1000:.0f}ms]")

        t1 = time.perf_counter()
        first_audio = False

        def on_chunk(chunk):
            nonlocal first_audio, t1
            if not first_audio:
                print(f"  [First audio: {(time.perf_counter()-t1)*1000:.0f}ms]")
                first_audio = True
            play_audio_chunk(chunk)    # Your playback function

        response = await stream_llm_to_tts(
            user_text, conversation, on_chunk
        )
        t_total = time.perf_counter() - t0
        print(f"Assistant: {response}  [Total: {t_total*1000:.0f}ms]")