Real-time voice agent in 80 lines (STT → LLM → TTS)

Build it in four steps

1. STEP 01

   Open the mic, stream to Deepgram

   Send raw 16-bit PCM from the mic to AIgateway's Deepgram-compatible streaming endpoint. Partial transcripts come back every ~100ms; treat each finalized utterance as a turn.

   import asyncio, websockets, pyaudio
   
   STT_URL = "wss://api.aigateway.sh/v1/audio/transcriptions/stream?model=deepgram/nova-3"
   
   async def listen(on_utterance):
       mic = pyaudio.PyAudio().open(rate=16000, channels=1, format=pyaudio.paInt16,
                                    input=True, frames_per_buffer=1024)
       async with websockets.connect(STT_URL, extra_headers={"Authorization": "Bearer sk-aig-..."}) as ws:
           async def mic_loop():
               while True:
                   await ws.send(mic.read(1024, exception_on_overflow=False))
           asyncio.create_task(mic_loop())
           async for msg in ws:
               evt = __import__("json").loads(msg)
               if evt.get("is_final"):
                   await on_utterance(evt["transcript"])

2. STEP 02

   Stream the transcript to Kimi K2.6

   Each utterance becomes a user turn. Stream the LLM response token-by-token so the TTS can start speaking before the LLM is done thinking.

   from openai import AsyncOpenAI
   
   client = AsyncOpenAI(base_url="https://api.aigateway.sh/v1", api_key="sk-aig-...")
   SYSTEM = "You are a helpful voice assistant. Keep replies under 3 sentences, conversational, no markdown."
   
   async def think(history: list[dict], on_token):
       stream = await client.chat.completions.create(
           model="moonshot/kimi-k2.6",
           messages=[{"role": "system", "content": SYSTEM}, *history],
           stream=True,
           extra_headers={"x-aig-tag": "voice.llm"},
       )
       async for chunk in stream:
           tok = chunk.choices[0].delta.content
           if tok:
               await on_token(tok)

3. STEP 03

   Speak tokens with ElevenLabs Turbo

   Open a TTS websocket, push sentences as they form (flush on ". " / "? " / "! "). ElevenLabs streams audio back in real time; write it to the speaker.

   TTS_URL = "wss://api.aigateway.sh/v1/audio/speech/stream?model=elevenlabs/eleven-turbo-v3&voice=rachel"
   
   async def speak():
       buf = ""
       async with websockets.connect(TTS_URL, extra_headers={"Authorization": "Bearer sk-aig-..."}) as ws:
           speaker = pyaudio.PyAudio().open(rate=24000, channels=1, format=pyaudio.paInt16, output=True)
   
           async def on_token(tok):
               nonlocal buf
               buf += tok
               if buf and buf[-1] in ".?!":
                   await ws.send(__import__("json").dumps({"text": buf}))
                   buf = ""
   
           async def drain():
               async for audio in ws:
                   speaker.write(audio)
   
           asyncio.create_task(drain())
           return on_token

4. STEP 04

   Wire it together

   Hook the three stages up and you've got a full voice loop. Keep history in memory, add a hang-up keyword, ship.

   async def main():
       history = []
       on_token = await speak()
   
       async def on_utterance(text: str):
           print("user:", text)
           if text.strip().lower() in {"bye", "goodbye", "hang up"}:
               raise SystemExit
           history.append({"role": "user", "content": text})
           await think(history, on_token)
   
       await listen(on_utterance)
   
   asyncio.run(main())

FAQ

More examples