Half-Duplex Mode (VAD-based Voice Conversation)

{
  "type": "prepare",
  "system_prompt": "You are a helpful assistant.",
  "config": {
    "vad": {
      "threshold": 0.8,
      "min_speech_duration_ms": 128,
      "min_silence_duration_ms": 800,
      "speech_pad_ms": 30
    },
    "generation": {
      "max_new_tokens": 256,
      "length_penalty": 1.1,
      "temperature": 0.7
    },
    "tts": {
      "enabled": true
    },
    "session": {
      "timeout_s": 180
    }
  },
  "ref_audio_base64": "<base64 reference audio>"
}

{
  "type": "audio_chunk",
  "audio_base64": "<base64 PCM float32, 16kHz, ~0.5s>"
}

{
  "type": "turn_done",
  "turn_index": 2,
  "text": "Sure, I can help you with that."
}

{"session_id": "stream_abc123"}

const sessionId = 'hdx_' + Math.random().toString(36).slice(2, 10);
const ws = new WebSocket(`wss://${location.host}/ws/half_duplex/${sessionId}`);
let aiSpeaking = false;
let audioContext, captureNode;

// -- Reference audio for voice cloning (base64 PCM float32, 16kHz) --
const refAudioBase64 = getRefAudioBase64();

ws.onopen = () => console.log('Connected, waiting for queue...');

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  switch (msg.type) {
    case 'queued':
      console.log(`Queue position: #${msg.position}, ETA: ${msg.estimated_wait_s}s`);
      break;

    case 'queue_done':
      // GPU assigned — send prepare with system_content containing ref audio.
      // system_content follows the model's best practice: [text, audio, text].
      // The audio item embeds the reference voice used for both LLM context and TTS cloning.
      ws.send(JSON.stringify({
        type: 'prepare',
        system_content: [
          { type: 'text', text: 'Mimic the voice from the audio sample.' },
          { type: 'audio', data: refAudioBase64 },         // reference voice
          { type: 'text', text: 'You are a helpful assistant. Reply naturally.' },
        ],
        config: {
          vad: { threshold: 0.8, min_silence_duration_ms: 800 },
          generation: { max_new_tokens: 256, temperature: 0.7 },
          tts: { enabled: true },
          session: { timeout_s: 180 },
        },
      }));
      break;

    case 'prepared':
      console.log(`Session ready (timeout: ${msg.timeout_s}s)`);
      startMicCapture();  // begin sending audio_chunk
      break;

    case 'vad_state':
      // Server-side VAD detected speech start / end
      console.log(msg.speaking ? 'User speaking...' : 'User stopped');
      break;

    case 'generating':
      // Server is processing user speech and starting generation.
      // Stop sending audio to prevent echo feedback.
      aiSpeaking = true;
      console.log(`Generating (speech: ${msg.speech_duration_ms}ms)`);
      break;

    case 'chunk':
      // Streaming token: incremental text and/or audio segment
      if (msg.text_delta) process.stdout.write(msg.text_delta);
      if (msg.audio_data) playAudio(msg.audio_data);  // PCM float32, 24kHz
      break;

    case 'turn_done':
      // Turn complete — resume mic after playback finishes + 800ms buffer
      // to avoid capturing the AI's own audio output.
      console.log(`\nTurn ${msg.turn_index} done: ${msg.text}`);
      setTimeout(() => { aiSpeaking = false; }, getPlaybackRemaining() + 800);
      break;

    case 'timeout':
      console.log(`Session timed out (${msg.elapsed_s}s)`);
      break;
    case 'error':
      console.error('Error:', msg.error);
      break;
  }
};

async function startMicCapture() {
  const stream = await navigator.mediaDevices.getUserMedia({ audio: { sampleRate: 16000 } });
  audioContext = new AudioContext({ sampleRate: 16000 });
  await audioContext.audioWorklet.addModule('capture-processor.js');
  const source = audioContext.createMediaStreamSource(stream);
  captureNode = new AudioWorkletNode(audioContext, 'capture-processor');
  source.connect(captureNode);

  // AudioWorklet is event-driven, NOT timer-based:
  // The audio rendering thread accumulates mic samples and fires 'chunk'
  // when the buffer reaches ~0.5s. No sleep or polling needed.
  captureNode.port.onmessage = (e) => {
    if (e.data.type === 'chunk' && !aiSpeaking && ws.readyState === WebSocket.OPEN) {
      ws.send(JSON.stringify({
        type: 'audio_chunk',
        audio_base64: float32ToBase64(e.data.audio),
      }));
    }
  };
}

function stopSession() {
  ws.send(JSON.stringify({ type: 'stop' }));
  ws.close();
}

import asyncio, json, base64
import numpy as np
import websockets

def load_ref_audio(path: str) -> str:
    """Load a WAV file and return base64-encoded PCM float32 at 16kHz."""
    import soundfile as sf
    audio, _ = sf.read(path, dtype="float32", samplerate=16000)
    return base64.b64encode(audio.tobytes()).decode()

def audio_file_to_chunks(path, chunk_duration=0.5, sr=16000):
    """Read a WAV file and yield 0.5s float32 chunks as base64."""
    import soundfile as sf
    audio, _ = sf.read(path, dtype="float32", samplerate=sr)
    chunk_size = int(sr * chunk_duration)
    for i in range(0, len(audio), chunk_size):
        yield base64.b64encode(audio[i:i + chunk_size].tobytes()).decode()

async def half_duplex_session(
    audio_path: str,
    server="wss://localhost:8006",
    ref_audio_path: str | None = "ref.wav",
):
    session_id = f"hdx_{id(object()):x}"
    url = f"{server}/ws/half_duplex/{session_id}"

    async with websockets.connect(url) as ws:
        # 1. Wait for queue assignment
        while True:
            msg = json.loads(await ws.recv())
            if msg["type"] == "queue_done":
                break
            print(f"Queued at #{msg.get('position')}")

        # 2. Prepare — system_content embeds reference audio for voice cloning.
        #    Format follows model best practice: [text, audio, text].
        ref_b64 = load_ref_audio(ref_audio_path) if ref_audio_path else None
        system_content = [
            {"type": "text", "text": "Mimic the voice from the audio sample."},
            {"type": "audio", "data": ref_b64},         # reference voice
            {"type": "text", "text": "You are a helpful assistant. Reply naturally."},
        ] if ref_b64 else None

        prepare_msg = {
            "type": "prepare",
            "config": {
                "vad": {"threshold": 0.8, "min_silence_duration_ms": 800},
                "generation": {"max_new_tokens": 256, "temperature": 0.7},
                "tts": {"enabled": True},
                "session": {"timeout_s": 60},
            },
        }
        if system_content:
            prepare_msg["system_content"] = system_content
        await ws.send(json.dumps(prepare_msg))

        msg = json.loads(await ws.recv())
        assert msg["type"] == "prepared"
        print(f"Session ready: {msg['session_id']}")

        # 3. Concurrently send audio and receive responses
        async def send_audio():
            for chunk_b64 in audio_file_to_chunks(audio_path):
                await ws.send(json.dumps({
                    "type": "audio_chunk",
                    "audio_base64": chunk_b64,
                }))
                # Simulate real-time microphone cadence: in a browser, the
                # AudioWorklet fires chunk events driven by the audio rendering
                # thread — no sleep needed. Here we sleep because we're reading
                # from a file and need to pace the chunks to match real time.
                await asyncio.sleep(0.5)
            # Wait for the server to finish generating the final turn
            await asyncio.sleep(5)
            await ws.send(json.dumps({"type": "stop"}))

        async def recv_messages():
            async for raw in ws:
                msg = json.loads(raw)
                if msg["type"] == "vad_state":
                    print("Speaking..." if msg["speaking"] else "Stopped speaking")
                elif msg["type"] == "generating":
                    print(f"Generating ({msg['speech_duration_ms']}ms speech)")
                elif msg["type"] == "chunk":
                    if msg.get("text_delta"):
                        print(msg["text_delta"], end="", flush=True)
                elif msg["type"] == "turn_done":
                    print(f"\n--- Turn {msg['turn_index']} done ---")
                elif msg["type"] in ("stopped", "timeout"):
                    break

        await asyncio.gather(send_audio(), recv_messages())

asyncio.run(half_duplex_session("test_audio.wav"))