实时语音 Agent：全双工对话与低延迟链路

麦克风 → VAD → ASR → LLM → TTS → 扬声器

麦克风 → 音频 Tokenizer → 多模态 LLM → 音频解码 → 扬声器

import torch
import numpy as np

class SileroVAD:
    def __init__(self, threshold: float = 0.5, sr: int = 16000):
        self.model, _ = torch.hub.load(
            repo_or_dir="snakers4/silero-vad", model="silero_vad"
        )
        self.threshold = threshold
        self.sr = sr
        self.reset()

    def reset(self):
        self.model.reset_states()
        self.speech_ms = 0
        self.silence_ms = 0

    def process(self, chunk: np.ndarray, chunk_ms: int = 32) -> str:
        """输入 32ms 音频块，返回状态：silence / speech / end。"""
        prob = self.model(torch.from_numpy(chunk), self.sr).item()
        if prob > self.threshold:
            self.speech_ms += chunk_ms
            self.silence_ms = 0
            return "speech" if self.speech_ms > 96 else "silence"
        self.silence_ms += chunk_ms
        # 静音超过 500ms 且此前有语音，判定说完
        if self.silence_ms > 500 and self.speech_ms > 200:
            self.reset()
            return "end"
        return "silence"

async def is_turn_complete(partial_text: str) -> bool:
    """用极小的分类模型判断是否说完。"""
    prompt = f"判断下面这句话是否语义完整，只回答 yes 或 no：\n{partial_text}"
    ans = await small_llm.complete(prompt, max_tokens=2, temperature=0)
    return ans.strip().lower().startswith("y")

class StreamingASR:
    def __init__(self, model):
        self.model = model
        self.buffer = []

    async def feed(self, audio_chunk: np.ndarray):
        self.buffer.append(audio_chunk)
        # 每 200ms 出一次部分结果
        partial = self.model.transcribe(np.concatenate(self.buffer), partial=True)
        yield {"type": "partial", "text": partial}

    async def finalize(self):
        text = self.model.transcribe(np.concatenate(self.buffer), partial=False)
        self.buffer.clear()
        return {"type": "final", "text": text}

result = asr.transcribe(
    audio,
    initial_prompt="对话涉及的技术名词：Qwen、vLLM、KV Cache、LoRA。",
    hotwords=["Qwen", "vLLM", "LoRA"],
)

VOICE_SYSTEM_PROMPT = """你是语音助手。规则：
1. 回答必须口语化、简短，控制在 2~3 句以内。
2. 不要使用 Markdown、列表、代码块——它们无法被朗读。
3. 数字用中文读法表达（"三点五"而非"3.5"）。
4. 需要长内容时，先给结论并询问是否展开。
"""

import re

SENT_END = re.compile(r"[。！？!?；;]")

async def llm_to_tts(llm_stream, tts):
    buf = ""
    async for token in llm_stream:
        buf += token
        # 遇到句末标点且长度够，立刻合成
        if SENT_END.search(token) and len(buf) >= 8:
            await tts.speak(buf)
            buf = ""
    if buf.strip():
        await tts.speak(buf)

import httpx

async def stream_tts(text: str, out_queue):
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream(
            "POST", "http://tts:8000/synthesize",
            json={"text": text, "format": "pcm_s16le", "sample_rate": 24000},
        ) as resp:
            async for chunk in resp.aiter_bytes(4096):
                await out_queue.put(chunk)   # 立即推给播放器

class VoiceSession:
    def __init__(self):
        self.state = "idle"          # idle / listening / thinking / speaking
        self.playback_queue = asyncio.Queue()
        self.llm_task: asyncio.Task | None = None

    async def on_speech_start(self):
        if self.state == "speaking":
            await self.barge_in()

    async def barge_in(self):
        # 1. 清空播放缓冲
        while not self.playback_queue.empty():
            self.playback_queue.get_nowait()
        # 2. 取消 LLM / TTS 任务
        if self.llm_task and not self.llm_task.done():
            self.llm_task.cancel()
        # 3. 记录已播出的文本到历史
        self.history.append({"role": "assistant",
                             "content": self.spoken_text, "truncated": True})
        self.state = "listening"

const stream = await navigator.mediaDevices.getUserMedia({
  audio: {
    echoCancellation: true,
    noiseSuppression: true,
    autoGainControl: true,
    sampleRate: 16000,
  },
});

单会话资源占用（服务端）：
  VAD:      ~1% 单核
  ASR:      流式模型，约 0.3~0.5 GPU 等效（并发批处理可摊薄）
  LLM:      与其他会话共享批处理，按 token 计
  TTS:      流式合成，约 0.2 GPU 等效

单张 A100 (80G) 参考容量：
  级联式，7B 模型，平均回答 80 token：
    ~40~80 并发会话（受 LLM 批处理与 TTS 显存共同限制）

IDLE_TIMEOUT_S = 90

async def reap_idle_sessions(sessions: dict):
    now = time.monotonic()
    for sid, sess in list(sessions.items()):
        if now - sess.last_activity > IDLE_TIMEOUT_S and sess.state == "idle":
            await sess.close()          # 释放 ASR/TTS 资源
            del sessions[sid]

async def safe_tts(text: str, primary, fallback):
    try:
        return await asyncio.wait_for(primary.synthesize(text), timeout=1.5)
    except asyncio.TimeoutError:
        return await fallback.synthesize(text)