Build a realtime voice agent: streaming speech in, reply, speech out

"""A voice-agent loop on Fish Audio realtime speech-to-text.

    pip install "websockets>=14" msgpack
    export FISH_API_KEY=...
    python voice_agent.py speech.wav

The "microphone" is a 16-bit mono WAV file streamed in real time, and think() / speak() are
placeholders: replace them with your LLM and your text-to-speech playback.
"""
import asyncio
import contextlib
import os
import re
import sys
import time
import wave

import msgpack
import websockets

URL = "wss://api.fish.audio/v1/asr/realtime"
CHUNK_MS = 100
TAGS = re.compile(r"<\|[^|]*\|>")  # speaker tags such as <|speaker:0|>
T0 = time.monotonic()


def log(*args):
    print(f"{time.monotonic() - T0:6.2f}s", *args)


def words(text):
    """The transcript without speaker tags."""
    return TAGS.sub("", text or "").strip()


async def think(history):
    """Placeholder LLM: replace with a call to your model."""
    await asyncio.sleep(0.5)
    return f"You asked: {history[-1]['content']} Let me check that for you, it will only take a moment."


async def speak(text):
    """Placeholder TTS: returns when the last sample has played."""
    await asyncio.sleep(len(text) * 0.06)


def duck(on):
    """Placeholder: lower (or restore) the agent's playback volume."""
    log("  duck" if on else "  unduck")


class Agent:
    def __init__(self, ws):
        self.ws = ws
        self.history = []     # the conversation, for your LLM
        self.prepared = None  # (turn_index, task): a reply started on eager_end_of_turn
        self.reply = None     # task: the reply being prepared or played
        self.speaking = False
        self.hung_up = False  # the caller's audio has ended: answer nothing more

    async def send(self, message):
        if message["type"] != "audio":
            log("->", message)
        await self.ws.send(msgpack.packb(message))

    def start_thinking(self, text):
        return asyncio.create_task(think(self.history + [{"role": "user", "content": text}]))

    def drop_prepared(self):
        if self.prepared:
            self.prepared[1].cancel()
            self.prepared = None

    async def stop(self):
        """Barge-in: stop the reply (its finally sends agent_speaking false)."""
        if self.reply:
            self.reply.cancel()
            with contextlib.suppress(asyncio.CancelledError, websockets.ConnectionClosed):
                await self.reply
            self.reply = None

    async def play(self, thinking):
        """Speak a reply. Runs from committed until the last sample has played."""
        try:
            self.speaking = True
            await self.send({"type": "agent_speaking", "speaking": True})  # committed to replying
            text = await thinking  # the LLM may still be working: that is part of the reply
            await self.send({"type": "context_add", "items": [{"role": "assistant", "text": text}]})
            self.history.append({"role": "assistant", "content": text})
            await speak(text)
        finally:
            self.speaking = False
            await self.send({"type": "agent_speaking", "speaking": False})  # played out, or stopped

    async def on_event(self, m):
        t = m["type"]
        if t == "eager_end_of_turn":
            self.drop_prepared()
            if words(m["text"]):  # start the reply early; committed will have the same text
                self.prepared = (m["turn_index"], self.start_thinking(words(m["text"])))
        elif t == "turn_resumed":
            self.drop_prepared()  # the user went on: that text is not final
        elif t == "speech_start":
            if self.speaking:
                duck(True)  # maybe an interruption, maybe "mm-hm"
        elif t == "user_backchannel":
            duck(False)  # the user only acknowledged: keep talking
        elif t == "user_turn_taken":
            await self.stop()  # the user is taking the turn: stop talking
        elif t == "committed":
            text = words(m["text"])
            if m.get("user_turn", True) and text and not self.hung_up:
                if self.prepared and self.prepared[0] == m["turn_index"]:
                    thinking = self.prepared[1]  # reuse the reply started on eager_end_of_turn
                    self.prepared = None
                else:
                    self.drop_prepared()
                    thinking = self.start_thinking(text)
                self.history.append({"role": "user", "content": text})
                await self.stop()
                self.reply = asyncio.create_task(self.play(thinking))
            else:  # not the user's turn (noise, another voice, a backchannel), or no words
                if self.prepared and self.prepared[0] == m["turn_index"]:
                    self.drop_prepared()
                if self.speaking:
                    duck(False)


async def microphone(agent, wav, rate):
    """Stream the file in real time, then end the session."""
    loop = asyncio.get_running_loop()
    start, sent_ms = loop.time(), 0
    while chunk := wav.readframes(rate * CHUNK_MS // 1000):
        await agent.send({"type": "audio", "data": chunk})
        sent_ms += CHUNK_MS
        await asyncio.sleep(max(0.0, start + sent_ms / 1000 - loop.time()))
    agent.hung_up = True
    await agent.stop()  # agent_speaking false goes before audio_done
    await agent.send({"type": "audio_done"})


async def main(path):
    wav = wave.open(path, "rb")
    rate = wav.getframerate()
    headers = {"Authorization": f"Bearer {os.environ['FISH_API_KEY']}"}
    async with websockets.connect(URL, additional_headers=headers) as ws:
        agent = Agent(ws)
        await agent.send({
            "type": "config",
            "audio": {"format": "pcm_s16le", "sample_rate": rate, "channels": 1},
            "turn_detection": {"mode": "auto", "eager": True},
            "context": {"terms": ["Main Street"]},
        })
        mic = None
        try:
            async for frame in ws:
                m = msgpack.unpackb(frame)
                shown = {k: v for k, v in m.items() if k in ("turn_index", "text", "user_turn", "turn_kind", "reason")}
                log("<-", m["type"], shown)
                if m["type"] == "session_started":
                    mic = asyncio.create_task(microphone(agent, wav, rate))
                await agent.on_event(m)
        finally:
            if mic:
                mic.cancel()
            await agent.stop()
    log("closed", ws.close_code)


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1]))