Half-Duplex 模式（VAD 半双工语音对话）

{
  "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 参考音频>"
}

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

{
  "type": "turn_done",
  "turn_index": 2,
  "text": "好的，我可以帮你解答这个问题。"
}

{"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;

// -- 声音克隆参考音频 (base64 PCM float32, 16kHz) --
const refAudioBase64 = getRefAudioBase64();

ws.onopen = () => console.log('已连接，等待排队...');

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  switch (msg.type) {
    case 'queued':
      console.log(`排队位置: #${msg.position}，预计等待: ${msg.estimated_wait_s}s`);
      break;

    case 'queue_done':
      // GPU 已分配——发送 prepare，通过 system_content 嵌入参考音频。
      // system_content 遵循模型最佳实践格式：[text, audio, text]。
      // 其中的 audio 项用于 LLM 上下文嵌入和 TTS 声音克隆。
      ws.send(JSON.stringify({
        type: 'prepare',
        system_content: [
          { type: 'text', text: '模仿音频样本的音色并生成新的内容。' },
          { type: 'audio', data: refAudioBase64 },         // 参考音色
          { type: 'text', text: '你是一个有帮助的助手，请自然地回复。' },
        ],
        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(`会话就绪（超时: ${msg.timeout_s}s）`);
      startMicCapture();  // 开始发送 audio_chunk
      break;

    case 'vad_state':
      // 服务端 VAD 检测到语音开始/结束
      console.log(msg.speaking ? '用户正在说话...' : '用户停止说话');
      break;

    case 'generating':
      // 服务端正在处理用户语音并开始生成回复，
      // 此时暂停发送麦克风音频以防止回声反馈。
      aiSpeaking = true;
      console.log(`开始生成（语音时长: ${msg.speech_duration_ms}ms）`);
      break;

    case 'chunk':
      // 流式 token：增量文本和/或音频片段
      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':
      // 本轮完成 — 等播放完毕 + 800ms 缓冲后恢复麦克风，
      // 避免采集到 AI 自身的音频输出。
      console.log(`\n轮次 ${msg.turn_index} 完成: ${msg.text}`);
      setTimeout(() => { aiSpeaking = false; }, getPlaybackRemaining() + 800);
      break;

    case 'timeout':
      console.log(`会话超时（${msg.elapsed_s}s）`);
      break;
    case 'error':
      console.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 是事件驱动的，不是定时器：
  // 音频渲染线程实时累积麦克风采样，缓冲区满约 0.5 秒时自动触发 'chunk'。
  // 无需 sleep 或轮询。
  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:
    """加载 WAV 文件，返回 base64 编码的 PCM float32 (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):
    """读取音频文件，切分为 0.5 秒的 float32 块并返回 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. 等待排队分配
        while True:
            msg = json.loads(await ws.recv())
            if msg["type"] == "queue_done":
                break
            print(f"排队中 #{msg.get('position')}")

        # 2. 发送 prepare — system_content 嵌入参考音频用于声音克隆。
        #    格式遵循模型最佳实践：[text, audio, text]。
        ref_b64 = load_ref_audio(ref_audio_path) if ref_audio_path else None
        system_content = [
            {"type": "text", "text": "模仿音频样本的音色并生成新的内容。"},
            {"type": "audio", "data": ref_b64},         # 参考音色
            {"type": "text", "text": "你是一个有帮助的助手，请自然地回复。"},
        ] 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"会话就绪: {msg['session_id']}")

        # 3. 并发发送音频和接收消息
        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,
                }))
                # 模拟实时麦克风节奏：浏览器端 AudioWorklet 由音频渲染线程
                # 实时驱动，缓冲区满时自动触发——无需 sleep。
                # 这里从文件读取，需要 sleep 来匹配实时节奏。
                await asyncio.sleep(0.5)
            # 等待服务端完成最后一轮生成
            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("正在说话..." if msg["speaking"] else "停止说话")
                elif msg["type"] == "generating":
                    print(f"开始生成（语音时长 {msg['speech_duration_ms']}ms）")
                elif msg["type"] == "chunk":
                    if msg.get("text_delta"):
                        print(msg["text_delta"], end="", flush=True)
                elif msg["type"] == "turn_done":
                    print(f"\n--- 轮次 {msg['turn_index']} 完成 ---")
                elif msg["type"] in ("stopped", "timeout"):
                    break

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

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