realtime_client.py

#!/usr/bin/env python3
"""Example client for the llama-server /v1/realtime websocket, with the openai SDK.

Start a server with an audio model, e.g. LFM3-Audio:

    llama-server -hf LiquidAI/LFM3-Audio-2B-GGUF

Spoken conversation, the audio of each wav file is one user turn. It is sent in chunks as it would come from a
microphone, the server prefills it as it arrives so the response starts right after the last chunk:

    python examples/realtime_client.py --instructions "Respond with interleaved text and audio." question.wav

Output modalities: left out, the model picks; "text", "audio" or "text,audio" (interleaved) force the response.
Without --instructions, a forced mode comes with the system prompt of the LFM3-Audio or LFM3-Omni model card:

    python examples/realtime_client.py --output-modalities text,audio --out answer.wav question.wav

Tool calls: --tools gives the conversation a demo tool, the client runs its calls and sends the results back:

    python examples/realtime_client.py --tools --text "Set the volume to 30."

Text turns go in with --text, transcription sessions transcribe each file:

    python examples/realtime_client.py --text "What is the capital of France?"
    python examples/realtime_client.py --type transcription --language en speech.wav

Live transcription models (Voxtral Realtime, LFM3-Audio live ASR) transcribe while the audio is sent, a transcription
session opens with ?intent=transcription:

    python examples/realtime_client.py --type transcription speech.wav

Requirements: pip install "openai[realtime]" numpy
"""

import argparse
import asyncio
import base64
import json
import time
import wave

import numpy as np
from openai import AsyncOpenAI


# system prompts of the LiquidAI/LFM3-Audio-2B and LiquidAI/LFM3-Omni-2B model cards for a forced output mode
MODE_INSTRUCTIONS = {
    "audio": {"text,audio": "Respond with interleaved text and audio."},
    "omni":  {"text": "Respond in text.", "text,audio": "Respond in interleaved text and audio."},
}
# and with tools: calls, or an answer in interleaved text and speech
TOOLS_INSTRUCTIONS = "Respond with function calls, and interleaved text and audio.\n\nList of tools:"

# the demo tool of --tools
DEMO_TOOLS = [{
    "type": "function",
    "name": "set_volume",
    "description": "Set the device speaker volume.",
    "parameters": {
        "type": "object",
        "properties": {"volume": {"type": "integer", "description": "Volume from 0 to 100."}},
        "required": ["volume"],
    },
}]


def run_tool(name, arguments):
    if name == "set_volume":
        return {"volume": json.loads(arguments)["volume"]}
    return {"error": f"unknown tool {name}"}


def read_wav(path, rate):
    # PCM16 wav -> mono int16 at the given rate
    with wave.open(path, "rb") as f:
        assert f.getsampwidth() == 2, "the example reads PCM16 wav files"
        x = np.frombuffer(f.readframes(f.getnframes()), dtype=np.int16).astype(np.float32)
        x = x.reshape(-1, f.getnchannels()).mean(axis=1)
        sr = f.getframerate()
    if sr != rate:
        n = int(round(len(x) * rate / sr))
        x = np.interp(np.arange(n) * sr / rate, np.arange(len(x)), x)
    return np.clip(x, -32768, 32767).astype(np.int16)


def write_wav(path, pcm, rate):
    with wave.open(path, "wb") as f:
        f.setnchannels(1)
        f.setsampwidth(2)
        f.setframerate(rate)
        f.writeframes(pcm.tobytes())


async def main(args):
    # the SDK derives a wss:// URL from base_url, a plain http server needs ws://
    client = AsyncOpenAI(base_url=args.url, api_key=args.api_key, websocket_base_url=args.url.replace("http", "ws", 1))
    query = {"intent": "transcription"} if args.type == "transcription" else {}
    async with client.realtime.connect(model=args.model, extra_query=query) as conn:
        ev = await conn.recv()
        if ev.type == "error":
            print("error:", ev.error.message)
            return
        # the audio goes in at the rate of the model, PCM16 mono
        session_info = ev.model_dump(warnings=False)["session"]
        audio_fmt = session_info["audio"]
        card = "omni" if "omni" in str(session_info.get("model", "")).lower() else "audio"
        in_rate = audio_fmt["input"]["format"]["rate"]
        out_rate = (audio_fmt.get("output") or {}).get("format", {}).get("rate", 24000)

        session = {"type": args.type}
        if args.type == "realtime":
            instructions = args.instructions
            if instructions is None:
                instructions = TOOLS_INSTRUCTIONS if args.tools else MODE_INSTRUCTIONS[card].get(args.output_modalities)
            if args.tools:
                session["tools"] = DEMO_TOOLS
            if instructions is not None:
                session["instructions"] = instructions
            if args.output_modalities:
                session["output_modalities"] = args.output_modalities.split(",")
        elif args.language:
            session["audio"] = {"input": {"transcription": {"language": args.language}}}
        await conn.session.update(session=session)  # ty: ignore[invalid-argument-type]
        ev = await conn.recv()
        if ev.type == "error":
            print("error:", ev.error.message)
            return

        turns = [("audio", path) for path in args.wav] + [("text", text) for text in args.text]
        for i, (kind, src) in enumerate(turns):
            t_last = 0.0

            # the input is sent while the events are read: live transcription arrives while the audio is sent
            async def send_turn():
                nonlocal t_last
                if kind == "audio":
                    pcm = read_wav(src, in_rate)
                    step = in_rate * args.chunk_ms // 1000
                    for j in range(0, len(pcm), step):
                        chunk = pcm[j:j + step]
                        await conn.input_audio_buffer.append(audio=base64.b64encode(chunk.tobytes()).decode())
                        if args.realtime:
                            await asyncio.sleep(len(chunk) / in_rate)
                    if args.type == "transcription":
                        # "final" ends the input, models that transcribe with a delay then complete the last item
                        await conn.send({"type": "input_audio_buffer.commit", "final": i == len(args.wav) - 1})  # ty: ignore[invalid-argument-type]
                else:
                    await conn.conversation.item.create(item={
                        "type": "message", "role": "user", "content": [{"type": "input_text", "text": src}]})
                if args.type == "realtime":
                    # the audio not committed yet ends the user turn
                    await conn.response.create()
                t_last = time.perf_counter()

            print(f"--- turn {i} ({kind}: {src})")
            sender = asyncio.create_task(send_turn())
            t_first = None
            audio = []
            async for ev in conn:
                if ev.type == "response.output_text.delta" or ev.type == "conversation.item.input_audio_transcription.delta":
                    t_first = t_first or time.perf_counter()
                    print(ev.delta, end="", flush=True)
                elif ev.type == "response.output_audio.delta":
                    t_first = t_first or time.perf_counter()
                    audio.append(np.frombuffer(base64.b64decode(ev.delta), dtype=np.int16))
                elif ev.type == "response.done":
                    # the calls of the response are run, their results go back and the model answers them
                    calls = [o for o in ev.model_dump()["response"]["output"] if o["type"] == "function_call"]
                    for call in calls:
                        result = json.dumps(run_tool(call["name"], call["arguments"]))
                        print(f"\n[call {call['name']}({call['arguments']}) -> {result}]", flush=True)
                        await conn.conversation.item.create(item={
                            "type": "function_call_output", "call_id": call["call_id"], "output": result})
                    if not calls:
                        break
                    await conn.response.create()
                elif ev.type == "conversation.item.input_audio_transcription.completed":
                    break
                elif ev.type == "error":
                    print("\nerror:", ev.error.message)
                    return
            await sender
            print()
            if t_first is not None and t_first > t_last:
                print(f"first output {1000 * (t_first - t_last):.0f} ms after the end of the input")
            elif t_first is not None:
                print(f"output started {1000 * (t_last - t_first):.0f} ms before the end of the input, "
                      f"done {1000 * (time.perf_counter() - t_last):.0f} ms after it")
            if audio:
                pcm_out = np.concatenate(audio)
                path = f"{args.out.removesuffix('.wav')}_{i}.wav" if len(turns) > 1 else args.out
                write_wav(path, pcm_out, out_rate)
                print(f"audio: {len(pcm_out) / out_rate:.2f} s -> {path}")


if __name__ == "__main__":
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument("wav", nargs="*", help="PCM16 wav files, one user turn each")
    p.add_argument("--text", action="append", default=[], help="a user turn as text, after the audio turns, repeatable")
    p.add_argument("--type", choices=["realtime", "transcription"], default="realtime", help="session type")
    p.add_argument("--instructions", help="system prompt of the conversation")
    p.add_argument("--tools", action="store_true", help="give the conversation a demo tool (set_volume), its calls are run")
    p.add_argument("--output-modalities", choices=["text", "audio", "text,audio"], help="default: the model picks")
    p.add_argument("--language", help="transcription language, e.g. en")
    p.add_argument("--chunk-ms", type=int, default=100, help="audio chunk size")
    p.add_argument("--realtime", action=argparse.BooleanOptionalAction, default=True, help="send the audio in real time")
    p.add_argument("--out", default="response.wav", help="where the audio of the responses goes")
    p.add_argument("--url", default="http://127.0.0.1:9931/v1")
    p.add_argument("--api-key", default="none")
    p.add_argument("--model", default="default", help="model name, used by the router")
    asyncio.run(main(p.parse_args()))