cleanup: merge worker files, remove simplified version

asr_agent/conversation_worker.py

@@ -1,682 +0,0 @@
-"""
-Full Pipeline Worker: VAD -> ASR -> LLM -> TTS
-  VAD: Silero VAD (ONNX) with dual-threshold + sliding window
-  VP:  AEC/ANS/AGC handled by LiveKit WebRTC
-  ASR: Qwen3-ASR (local model, batch on speech-end)
-  LLM: streaming via OpenAI-compatible API (vLLM), <think> tag filtering
-  TTS: Mimo API (remote), xiaozhi-style two-tier sentence segmentation
-"""
-
-from __future__ import annotations
-
-import asyncio
-import base64
-import json
-import logging
-import os
-import re
-import sys
-import time
-from collections import deque
-from dataclasses import dataclass, field
-
-import numpy as np
-try:
-    import onnxruntime
-    _HAS_ONNX = True
-except ImportError:
-    _HAS_ONNX = False
-
-import aiohttp
-import jwt
-from livekit import rtc
-
-_root = os.path.join(os.path.dirname(os.path.abspath(__file__)), "whisper_asr")
-sys.path.insert(0, _root)
-from audio_processor import AudioBuffer
-from qwen_engine import QwenASREngine
-
-logger = logging.getLogger("worker")
-
-# ── Env ──
-LIVEKIT_URL = os.environ.get("LIVEKIT_URL", "ws://localhost:7880")
-ASR_MODEL = os.environ.get("ASR_MODEL_PATH", "Qwen/Qwen3-ASR-0.6B")
-VLLM_URL = os.environ.get("VLLM_URL", "http://127.0.0.1:8000/v1")
-LLM_MODEL = os.environ.get("LLM_MODEL", "qwen3.6-35b-awq")
-MIMO_KEY = os.environ.get("MIMO_KEY", "")
-MIMO_API_BASE = os.environ.get("MIMO_API_BASE", "https://token-plan-cn.xiaomimimo.com/v1")
-
-# Provider switches: "vllm" | "mimo" for LLM, "qwen" | "mimo" for ASR
-LLM_PROVIDER = os.environ.get("LLM_PROVIDER", "vllm")
-ASR_PROVIDER = os.environ.get("ASR_PROVIDER", "qwen")
-
-
-# ── LLM ──
-LLM_MAX_TOKENS  = 128
-LLM_TEMPERATURE = 0.7
-LLM_TIMEOUT     = 25
-
-SYSTEM_PROMPT = (
-    "你是友好的中文语音助手，名字叫狄诺尼试验员。"
-    "回答简洁自然，2-3句即可，用口语化中文。"
-    "不要使用 Markdown、代码块、表格或特殊符号。"
-    "不要输出括号注释、不要使用英文缩写。"
-)
-
-# ── TTS segmentation (xiaozhi-style two-tier) ──
-FIRST_SENTENCE_PUNCT = "，、,。！？；:：\n"
-SENTENCE_END_PUNCT   = "。！？!?\n"
-MIN_TTS_CHARS        = 2
-
-
-# ═══════════════════════════════════════════════════════════════
-#  JWT
-# ═══════════════════════════════════════════════════════════════
-
-def _token(room, ident):
-    n = int(time.time())
-    return jwt.encode({
-        "iss": "devkey", "sub": ident, "name": ident,
-        "nbf": n - 60, "exp": n + 6 * 3600,
-        "video": {"roomJoin": True, "room": room,
-                   "canPublish": True, "canSubscribe": True, "canPublishData": True},
-    }, "secretsecretsecretsecretsecret12", algorithm="HS256")
-
-
-# ═══════════════════════════════════════════════════════════════
-#  LLM (streaming, <think> filtering)
-# ═══════════════════════════════════════════════════════════════
-
-async def _llm_stream(prompt: str, hist: list[dict]):
-    """Stream LLM tokens, dispatching based on LLM_PROVIDER env."""
-    if LLM_PROVIDER == "mimo":
-        async for x in _llm_stream_mimo(prompt, hist):
-            yield x
-        return
-    async for x in _llm_stream_vllm(prompt, hist):
-        yield x
-
-
-async def _llm_stream_vllm(prompt: str, hist: list[dict]):
-    """vLLM streaming via OpenAI-compatible SSE."""
-    msgs = [{"role": "system", "content": SYSTEM_PROMPT}]
-    msgs.extend(hist)
-    msgs.append({"role": "user", "content": prompt})
-
-    async with aiohttp.ClientSession() as s:
-        async with s.post(
-            f"{VLLM_URL}/chat/completions",
-            json={"model": LLM_MODEL, "messages": msgs,
-                  "max_tokens": LLM_MAX_TOKENS, "temperature": LLM_TEMPERATURE,
-                  "stream": True,
-                  "chat_template_kwargs": {"enable_thinking": False}},
-            timeout=aiohttp.ClientTimeout(total=LLM_TIMEOUT),
-        ) as r:
-            async for line in r.content:
-                line = line.decode().strip()
-                if not line.startswith("data: "):
-                    continue
-                data = line[6:]
-                if data == "[DONE]":
-                    break
-                try:
-                    chunk = json.loads(data)
-                    delta = chunk.get("choices", [{}])[0].get("delta", {})
-                    text = delta.get("content", "")
-                    if text:
-                        if "</think>" in text:
-                            text = text.split("</think>")[-1]
-                        if "<think>" in text:
-                            text = text.split("<think>")[0]
-                        if text.strip():
-                            yield text, False
-                except Exception:
-                    continue
-    yield "", True
-
-
-async def _llm_stream_mimo(prompt: str, hist: list[dict]):
-    """Mimo v2.5 LLM streaming via OpenAI-compatible SSE."""
-    if not MIMO_KEY:
-        yield "", True
-        return
-    msgs = [{"role": "system", "content": SYSTEM_PROMPT}]
-    msgs.extend(hist)
-    msgs.append({"role": "user", "content": prompt})
-
-    h = {"api-key": MIMO_KEY, "Content-Type": "application/json"}
-    async with aiohttp.ClientSession() as s:
-        async with s.post(
-            f"{MIMO_API_BASE}/chat/completions",
-            json={"model": "mimo-v2.5", "messages": msgs,
-                  "max_tokens": LLM_MAX_TOKENS, "temperature": LLM_TEMPERATURE,
-                  "stream": True},
-            headers=h, timeout=aiohttp.ClientTimeout(total=LLM_TIMEOUT),
-        ) as r:
-            async for line in r.content:
-                line = line.decode().strip()
-                if not line.startswith("data: "):
-                    continue
-                data = line[6:]
-                if data == "[DONE]":
-                    break
-                try:
-                    chunk = json.loads(data)
-                    delta = chunk.get("choices", [{}])[0].get("delta", {})
-                    text = delta.get("content", "")
-                    if text and text.strip():
-                        yield text, False
-                except Exception:
-                    continue
-    yield "", True
-
-
-# ═══════════════════════════════════════════════════════════════
-#  ASR (dispatch: qwen-local | mimo-api)
-# ═══════════════════════════════════════════════════════════════
-
-async def _asr_transcribe(audio_data: np.ndarray, asr_engine=None) -> str:
-    """Dispatch ASR based on ASR_PROVIDER env. Return final text."""
-    if ASR_PROVIDER == "mimo":
-        text = ""
-        async for chunk, is_done in _asr_mimo_stream(audio_data):
-            if chunk:
-                text += chunk
-        text = text.strip()
-    else:
-        loop = asyncio.get_running_loop()
-        result = await loop.run_in_executor(None, asr_engine.transcribe_array, audio_data)
-        text = result.get("text", "").strip()
-    # Filter single-char filler results (common ASR noise)
-    fillers = {"嗯", "啊", "哦", "呃", "咦", "噢", "唔", "哎"}
-    if text and len(text) <= 2 and all(c in "嗯啊哦呃咦噢唔哎？！。，、~·~" for c in text):
-        logger.info("ASR filtered filler: %s", text)
-        return ""
-    return text
-
-
-async def _asr_mimo_stream(audio_data: np.ndarray):
-    """Mimo v2.5 streaming ASR. Yields (text_snapshot, is_final)."""
-    if not MIMO_KEY:
-        yield "", True
-        return
-    import io, wave
-    buf = io.BytesIO()
-    with wave.open(buf, "wb") as wf:
-        wf.setnchannels(1)
-        wf.setsampwidth(2)
-        wf.setframerate(16000)
-        wf.writeframes((audio_data * 32767).astype(np.int16).tobytes())
-    audio_b64 = base64.b64encode(buf.getvalue()).decode()
-
-    h = {"api-key": MIMO_KEY, "Content-Type": "application/json"}
-    async with aiohttp.ClientSession() as s:
-        async with s.post(
-            f"{MIMO_API_BASE}/chat/completions",
-            json={
-                "model": "mimo-v2.5-asr",
-                "messages": [{"role": "user", "content": [
-                    {"type": "input_audio", "input_audio": {
-                        "data": f"data:audio/wav;base64,{audio_b64}"}}
-                ]}],
-                "stream": True,
-                "asr_options": {"language": "auto"},
-            },
-            headers=h, timeout=aiohttp.ClientTimeout(total=20),
-        ) as r:
-            async for line in r.content:
-                line = line.decode().strip()
-                if not line.startswith("data: "):
-                    continue
-                data = line[6:]
-                if data == "[DONE]":
-                    yield "", True
-                    return
-                try:
-                    chunk = json.loads(data)
-                    delta = chunk.get("choices", [{}])[0].get("delta", {})
-                    text = delta.get("content", "")
-                    if text:
-                        yield text, False
-                except Exception:
-                    continue
-    yield "", True
-
-
-# ═══════════════════════════════════════════════════════════════
-#  TTS
-# ═══════════════════════════════════════════════════════════════
-
-async def _tts(text: str, voice: str = "Chloe") -> np.ndarray | None:
-    """Mimo TTS. Returns int16 PCM at 24kHz mono, or None on failure."""
-    if not MIMO_KEY:
-        return None
-    if voice.startswith("data:"):
-        model = "mimo-v2.5-tts-voiceclone"
-    else:
-        model = "mimo-v2.5-tts"
-
-    h = {"api-key": MIMO_KEY, "Content-Type": "application/json"}
-    body: dict = {
-        "model": model,
-        "messages": [
-            {"role": "user", "content": ""},
-            {"role": "assistant", "content": text},
-        ],
-        "audio": {"format": "wav", "voice": voice},
-    }
-
-    async with aiohttp.ClientSession() as s:
-        try:
-            async with s.post(
-                f"{MIMO_API_BASE}/chat/completions",
-                json=body, headers=h,
-                timeout=aiohttp.ClientTimeout(total=30),
-            ) as r:
-                result = await r.json()
-                b64 = result.get("choices", [{}])[0].get("message", {}).get("audio", {}).get("data", "")
-                if b64:
-                    return np.frombuffer(base64.b64decode(b64), dtype=np.int16)
-        except Exception:
-            logger.exception("TTS failed")
-    return None
-
-
-def _clean_tts_text(text: str) -> str:
-    text = re.sub(r"\*{1,3}(.*?)\*{1,3}", r"\1", text)
-    text = re.sub(r"`{1,3}.*?`{1,3}", "", text)
-    text = re.sub(r"\[([^\]]+)]\([^)]+\)", r"\1", text)
-    text = re.sub(r"#{1,6}\s*", "", text)
-    text = re.sub(r"[>\-]\s", "", text)
-    text = re.sub(r"[\U0001F300-\U0001F9FF]", "", text)
-    return text.strip()
-
-
-# ═══════════════════════════════════════════════════════════════
-#  TTS Segmenter (xiaozhi-style: two-tier, processed_chars)
-# ═══════════════════════════════════════════════════════════════
-
-@dataclass
-class _TTSSegmenter:
-    buffer: str = ""
-    processed: int = 0
-    is_first: bool = True
-
-    def feed(self, text: str) -> list[str]:
-        self.buffer += text
-        segments = []
-        puncts = FIRST_SENTENCE_PUNCT if self.is_first else SENTENCE_END_PUNCT
-        unprocessed = self.buffer[self.processed:]
-
-        last_pos = -1
-        for ch in puncts:
-            pos = unprocessed.rfind(ch)
-            if pos > last_pos:
-                last_pos = pos
-
-        if last_pos >= 0:
-            seg_raw = unprocessed[:last_pos + 1]
-            seg_clean = _clean_tts_text(seg_raw)
-            if len(seg_clean) >= MIN_TTS_CHARS:
-                segments.append(seg_clean)
-                self.processed += len(seg_raw)
-                if self.is_first:
-                    self.is_first = False
-        return segments
-
-    def flush(self) -> str:
-        remaining = self.buffer[self.processed:].strip()
-        if remaining:
-            remaining = _clean_tts_text(remaining)
-        self.processed = len(self.buffer)
-        return remaining
-
-
-
-# ── VAD ──
-VAD_MODEL_PATH = os.environ.get("VAD_MODEL_PATH") or os.path.join(
-    os.path.dirname(os.path.abspath(__file__)), "whisper_asr", "silero_vad.onnx")
-VAD_THRESHOLD_HIGH = float(os.environ.get("VAD_THRESHOLD_HIGH", "0.5"))
-VAD_THRESHOLD_LOW  = float(os.environ.get("VAD_THRESHOLD_LOW", "0.2"))
-VAD_WINDOW_SIZE    = int(os.environ.get("VAD_WINDOW_SIZE", "5"))
-VAD_VOICE_RATIO    = float(os.environ.get("VAD_VOICE_RATIO", "0.5"))
-MIN_SPEECH_S       = float(os.environ.get("MIN_SPEECH_S", "0.3"))
-MIN_SILENCE_S      = float(os.environ.get("MIN_SILENCE_S", "0.4"))
-MAX_SPEECH_S       = float(os.environ.get("MAX_SPEECH_S", "8.0"))
-
-# Global ONNX session (shared across participants)
-_vad_session = None
-
-
-def _get_vad_session():
-    global _vad_session
-    if _vad_session is not None:
-        return _vad_session
-    if not _HAS_ONNX or not os.path.exists(VAD_MODEL_PATH):
-        return None
-    opts = onnxruntime.SessionOptions()
-    opts.inter_op_num_threads = 1
-    opts.intra_op_num_threads = 1
-    _vad_session = onnxruntime.InferenceSession(
-        VAD_MODEL_PATH, providers=["CPUExecutionProvider"], sess_options=opts,
-    )
-    logger.info("Silero VAD loaded (%s)", VAD_MODEL_PATH)
-    return _vad_session
-
-
-# ═══════════════════════════════════════════════════════════════
-#  Silero VAD (ONNX, per-participant state)
-# ═══════════════════════════════════════════════════════════════
-
-class _SileroVAD:
-    """Xiaozhi-style Silero VAD with dual-threshold + sliding window."""
-
-    def __init__(self):
-        self._sess = _get_vad_session()
-        self._state = np.zeros((2, 1, 128), dtype=np.float32)
-        self._context = np.zeros((1, 64), dtype=np.float32)
-        self._window: deque = deque(maxlen=max(VAD_WINDOW_SIZE, 1))
-        self.in_speech = False
-        self.speech_start_sample = 0
-        self.silence_counter = 0
-        self.total_samples = 0
-        self._sr = np.array(16000, dtype=np.int64)
-
-    def process(self, frame: np.ndarray) -> bool:
-        """Process 160-sample (10ms) frame. Call every frame. Returns in_speech."""
-        n = len(frame)
-
-        if self._sess is not None:
-            # ── Silero ONNX inference ──
-            # Silero processes 512-sample windows. Accumulate frames.
-            if not hasattr(self, '_buf'):
-                self._buf = np.array([], dtype=np.float32)
-            self._buf = np.concatenate([self._buf, frame]).astype(np.float32)
-
-            is_voice = False
-            while len(self._buf) >= 512:
-                chunk = self._buf[:512]
-                self._buf = self._buf[512:]
-                audio_in = chunk.reshape(1, -1)
-                inp = np.concatenate([self._context, audio_in], axis=1).astype(np.float32)
-                out, self._state = self._sess.run(
-                    None, {"input": inp, "state": self._state, "sr": self._sr},
-                )
-                self._context = inp[:, -64:]
-                prob = out.item()
-                is_voice = prob >= VAD_THRESHOLD_HIGH or (
-                    prob > VAD_THRESHOLD_LOW and (
-                        self._window[-1] if self._window else False
-                    )
-                )
-                self._window.append(is_voice)
-        else:
-            # ── Fallback: energy-based ──
-            energy = float(np.sqrt(np.mean(frame ** 2)))
-            if energy > 0.015:
-                is_voice = True
-            elif energy < 0.005:
-                is_voice = False
-            else:
-                is_voice = self._window[-1] if self._window else False
-            self._window.append(is_voice)
-
-        have_voice = sum(self._window) / max(len(self._window), 1) >= VAD_VOICE_RATIO
-
-        if have_voice and not self.in_speech:
-            logger.info("VAD START")
-            self.in_speech = True
-            self.speech_start_sample = self.total_samples
-            self.silence_counter = 0
-        elif not have_voice and self.in_speech:
-            self.silence_counter += n
-        elif have_voice:
-            self.silence_counter = 0
-        self.total_samples += n
-        return self.in_speech
-
-    def should_end(self) -> bool:
-        return self.in_speech and self.silence_counter >= int(MIN_SILENCE_S * 16000)
-
-    def min_speech_met(self) -> bool:
-        return (self.total_samples - self.speech_start_sample) / 16000 >= MIN_SPEECH_S
-
-    def reset(self):
-        self._window.clear()
-        self.in_speech = False
-        self.speech_start_sample = 0
-        self.silence_counter = 0
-        self._state = np.zeros((2, 1, 128), dtype=np.float32)
-        self._context = np.zeros((1, 64), dtype=np.float32)
-        if hasattr(self, '_buf'):
-            self._buf = np.array([], dtype=np.float32)
-
-
-# ═══════════════════════════════════════════════════════════════
-#  Worker
-# ═══════════════════════════════════════════════════════════════
-
-class Worker:
-    def __init__(self, room, identity="asr-bot"):
-        self.rn, self.id = room, identity
-        self.room = rtc.Room()
-        self.tasks: dict = {}
-        self.hist: list[dict] = []
-
-    async def run(self):
-        logger.info("Loading ASR model...")
-        self.asr = QwenASREngine(model_id=ASR_MODEL, language=None)
-        logger.info("ASR ready")
-
-        self.room.on("track_subscribed", self._on_track)
-        self.room.on("participant_disconnected", self._on_off)
-        self.room.on("participant_connected", lambda p: None)
-        self.room.on("track_published", lambda pub, p: None)
-        await self.room.connect(LIVEKIT_URL, _token(self.rn, self.id))
-        logger.info("Worker ready (VAD+ASR+LLM+TTS)")
-
-        for p in self.room.remote_participants.values():
-            for pub in p.track_publications.values():
-                if pub.track and pub.kind == rtc.TrackKind.KIND_AUDIO:
-                    self._on_track(pub.track, pub, p)
-        try:
-            await asyncio.Future()
-        finally:
-            await self.room.disconnect()
-
-    def _on_track(self, t, _, p):
-        s = p.sid
-        if s in self.tasks:
-            return
-        logger.info("TRACK %s", p.identity)
-        self.tasks[s] = asyncio.create_task(self._run(s, t))
-
-    def _on_off(self, p):
-        x = self.tasks.pop(p.sid, None)
-        if x:
-            x.cancel()
-
-    async def _run(self, sid, track):
-        logger.info("_run %s", sid)
-        buf = AudioBuffer(max_duration=MAX_SPEECH_S + 4, sample_rate=16000)
-        sr = 16000
-        vad = _SileroVAD()
-        busy = False
-
-        # ── Single persistent TTS output track per session ──
-        # One track is reused for all turns so audio never cross-streams
-        # between turns/segments (old code published a new track each call).
-        tts_src = rtc.AudioSource(24000, 1)
-        tts_track = rtc.LocalAudioTrack.create_audio_track("tts", tts_src)
-        await self.room.local_participant.publish_track(tts_track)
-        play_q: "asyncio.Queue[tuple[int, np.ndarray]]" = asyncio.Queue()
-        gen = 0                 # turn generation; bumped to flush in-flight audio
-        cur_gen = {"v": -1}    # generation currently being played by _player
-
-        async def _player():
-            """Drain TTS audio sequentially on the single track."""
-            while True:
-                g, audio = await play_q.get()
-                cur_gen["v"] = g
-                for i in range(0, len(audio), 24000 // 50):
-                    if cur_gen["v"] != g:
-                        break  # a newer turn started; drop this stale audio
-                    c = audio[i: i + 24000 // 50]
-                    await tts_src.capture_frame(rtc.AudioFrame(
-                        data=c.tobytes(), sample_rate=24000,
-                        num_channels=1, samples_per_channel=len(c),
-                    ))
-                    await asyncio.sleep(0.018)
-                cur_gen["v"] = -1
-
-        player_task = asyncio.create_task(_player())
-
-        async def _acc():
-            nonlocal busy, gen
-            stream = rtc.AudioStream(track, sample_rate=sr, num_channels=1)
-            fc = 0
-            async for ev in stream:
-                fc += 1
-                if busy:
-                    continue
-                arr = np.frombuffer(ev.frame.data, dtype=np.int16).astype(np.float32) / 32768.0
-                buf.append(arr)
-
-                was_speech = vad.in_speech
-                vad.process(arr)
-
-                if fc % 100 == 0:
-                    logger.info("[%s] af#%d e=%.4f vad=%s blen=%d",
-                                sid, fc, float(np.sqrt(np.mean(arr ** 2))),
-                                vad.in_speech, len(buf.buffer))
-
-                if was_speech and vad.should_end() and vad.min_speech_met():
-                    logger.info("[%s] VAD END blen=%d", sid, len(buf.buffer))
-                    gen += 1
-                    cur_gen["v"] = -1  # stop current playback immediately
-                    _drain_queue(play_q)
-                    busy = True
-                    try:
-                        await self._transcribe_and_respond(buf, play_q, gen)
-                    except Exception:
-                        logger.exception("[%s] transcribe failed", sid)
-                    busy = False
-                    vad.reset()
-                    vad.total_samples = len(buf.buffer)
-
-                if vad.in_speech and vad.total_samples >= int(MAX_SPEECH_S * sr):
-                    logger.info("[%s] VAD MAX blen=%d", sid, len(buf.buffer))
-                    gen += 1
-                    cur_gen["v"] = -1
-                    _drain_queue(play_q)
-                    busy = True
-                    try:
-                        await self._transcribe_and_respond(buf, play_q, gen)
-                    except Exception:
-                        logger.exception("[%s] max failed", sid)
-                    busy = False
-                    vad.reset()
-                    vad.total_samples = len(buf.buffer)
-
-            logger.info("[%s] stream end fc=%d", sid, fc)
-
-        try:
-            await _acc()
-        except asyncio.CancelledError:
-            pass
-        except Exception:
-            logger.exception("[%s] _acc failed", sid)
-        finally:
-            player_task.cancel()
-            try:
-                await self.room.local_participant.unpublish_track(tts_track)
-            except Exception:
-                pass
-
-    async def _transcribe_and_respond(self, buf: AudioBuffer, play_q, gen: int):
-        all_audio = buf.get_all()
-        buf.clear()
-        if len(all_audio) <= 1600:
-            return
-
-        txt = await _asr_transcribe(all_audio, self.asr)
-        if not txt:
-            return
-
-        logger.info("ASR: %s", txt[:80])
-        await self._send({"type": "utterance", "text": txt, "seq": 0})
-
-        # ── Streaming LLM, collect segments for ordered TTS ──
-        reply_full = ""
-        seg = _TTSSegmenter()
-        tts_segments: list[str] = []
-
-        try:
-            async for delta, is_final in _llm_stream(txt, self.hist):
-                if delta:
-                    reply_full += delta
-                    await self._send({"type": "reply_partial", "text": reply_full, "seq": 0})
-                    tts_segments.extend(seg.feed(delta))
-
-                if is_final:
-                    remaining = seg.flush()
-                    if remaining:
-                        tts_segments.append(remaining)
-                    break
-        except Exception:
-            logger.exception("LLM stream failed")
-            reply_full = "抱歉，我暂时无法回答。"
-
-        # Ordered TTS: generate concurrently but enqueue in order
-        if tts_segments:
-            tasks = [asyncio.create_task(_tts(s)) for s in tts_segments]
-            for i, task in enumerate(tasks):
-                try:
-                    pcm = await asyncio.wait_for(task, timeout=15)
-                    if pcm is not None and len(pcm) > 0:
-                        logger.info("TTS seg(%d/%d): %s", i+1, len(tasks), tts_segments[i][:40])
-                        await play_q.put((gen, pcm))
-                except (asyncio.TimeoutError, asyncio.CancelledError):
-                    break
-                except Exception:
-                    logger.exception("TTS seg %d failed", i+1)
-
-        if reply_full:
-            await self._send({"type": "reply", "text": reply_full})
-            self.hist.extend([
-                {"role": "user", "content": txt},
-                {"role": "assistant", "content": reply_full},
-            ])
-            self.hist[:] = self.hist[-20:]
-
-    async def _send(self, msg: dict):
-        try:
-            await self.room.local_participant.publish_data(
-                json.dumps(msg, ensure_ascii=False).encode(),
-                reliable=True, topic="transcription",
-            )
-        except Exception:
-            pass
-
-
-async def main():
-    import argparse
-    p = argparse.ArgumentParser()
-    p.add_argument("--room", required=True)
-    a = p.parse_args()
-    logging.basicConfig(level=logging.INFO)
-    await Worker(a.room).run()
-
-
-def _drain_queue(q: "asyncio.Queue") -> None:
-    """Discard any queued-but-not-yet-played TTS audio."""
-    while not q.empty():
-        try:
-            q.get_nowait()
-        except Exception:
-            break
-
-
-if __name__ == "__main__":
-    asyncio.run(main())
-

asr_agent/worker.py

@@ -1,227 +1,682 @@
 """
-LiveKit ASR Worker — subscribes to participant audio, runs Qwen ASR,
-and sends transcriptions back as room data messages.
+Full Pipeline Worker: VAD -> ASR -> LLM -> TTS
+  VAD: Silero VAD (ONNX) with dual-threshold + sliding window
+  VP:  AEC/ANS/AGC handled by LiveKit WebRTC
+  ASR: Qwen3-ASR (local model, batch on speech-end)
+  LLM: streaming via OpenAI-compatible API (vLLM), <think> tag filtering
+  TTS: Mimo API (remote), xiaozhi-style two-tier sentence segmentation
 """
 
 from __future__ import annotations
 
 import asyncio
+import base64
 import json
 import logging
 import os
+import re
 import sys
 import time
-import traceback
+from collections import deque
+from dataclasses import dataclass, field
 
-import jwt
 import numpy as np
+try:
+    import onnxruntime
+    _HAS_ONNX = True
+except ImportError:
+    _HAS_ONNX = False
+
+import aiohttp
+import jwt
 from livekit import rtc
 
-# Import whisper_asr modules (same directory or adjacent).
-_asr_root = os.path.join(os.path.dirname(__file__), "whisper_asr")
-sys.path.insert(0, _asr_root)
-from audio_processor import AudioBuffer      # noqa: E402
-from qwen_engine import QwenASREngine        # noqa: E402
-from transcript_processor import TranscriptProcessor  # noqa: E402
-
-logger = logging.getLogger("asr-worker")
-
-LIVEKIT_URL = os.environ.get("LIVEKIT_URL", "ws://localhost:10003")
-API_KEY = "devkey"
-API_SECRET = "secretsecretsecretsecretsecret12"
-ASR_MODEL_PATH = os.environ.get("ASR_MODEL_PATH", "/data/models/Qwen3-ASR")
-TTS_MODEL_PATH = os.environ.get("TTS_MODEL_PATH", "/data/models/Qwen3-TTS")
-
-TRANSCRIBE_INTERVAL = 2.0
-BUFFER_DURATION = 5.0
-SILENCE_REPEATS = 3
-SILENCE_SECONDS = 3.0
-
-
-def create_token(room_name: str, identity: str) -> str:
-    now = int(time.time())
-    return jwt.encode(
-        {
-            "iss": API_KEY,
-            "sub": identity,
-            "name": identity,
-            "nbf": now - 60,
-            "exp": now + 6 * 3600,
-            "video": {
-                "roomJoin": True,
-                "room": room_name,
-                "canPublish": True,
-                "canSubscribe": True,
-                "canPublishData": True,
+_root = os.path.join(os.path.dirname(os.path.abspath(__file__)), "whisper_asr")
+sys.path.insert(0, _root)
+from audio_processor import AudioBuffer
+from qwen_engine import QwenASREngine
+
+logger = logging.getLogger("worker")
+
+# ── Env ──
+LIVEKIT_URL = os.environ.get("LIVEKIT_URL", "ws://localhost:7880")
+ASR_MODEL = os.environ.get("ASR_MODEL_PATH", "Qwen/Qwen3-ASR-0.6B")
+VLLM_URL = os.environ.get("VLLM_URL", "http://127.0.0.1:8000/v1")
+LLM_MODEL = os.environ.get("LLM_MODEL", "qwen3.6-35b-awq")
+MIMO_KEY = os.environ.get("MIMO_KEY", "")
+MIMO_API_BASE = os.environ.get("MIMO_API_BASE", "https://token-plan-cn.xiaomimimo.com/v1")
+
+# Provider switches: "vllm" | "mimo" for LLM, "qwen" | "mimo" for ASR
+LLM_PROVIDER = os.environ.get("LLM_PROVIDER", "vllm")
+ASR_PROVIDER = os.environ.get("ASR_PROVIDER", "qwen")
+
+
+# ── LLM ──
+LLM_MAX_TOKENS  = 128
+LLM_TEMPERATURE = 0.7
+LLM_TIMEOUT     = 25
+
+SYSTEM_PROMPT = (
+    "你是友好的中文语音助手，名字叫狄诺尼试验员。"
+    "回答简洁自然，2-3句即可，用口语化中文。"
+    "不要使用 Markdown、代码块、表格或特殊符号。"
+    "不要输出括号注释、不要使用英文缩写。"
+)
+
+# ── TTS segmentation (xiaozhi-style two-tier) ──
+FIRST_SENTENCE_PUNCT = "，、,。！？；:：\n"
+SENTENCE_END_PUNCT   = "。！？!?\n"
+MIN_TTS_CHARS        = 2
+
+
+# ═══════════════════════════════════════════════════════════════
+#  JWT
+# ═══════════════════════════════════════════════════════════════
+
+def _token(room, ident):
+    n = int(time.time())
+    return jwt.encode({
+        "iss": "devkey", "sub": ident, "name": ident,
+        "nbf": n - 60, "exp": n + 6 * 3600,
+        "video": {"roomJoin": True, "room": room,
+                   "canPublish": True, "canSubscribe": True, "canPublishData": True},
+    }, "secretsecretsecretsecretsecret12", algorithm="HS256")
+
+
+# ═══════════════════════════════════════════════════════════════
+#  LLM (streaming, <think> filtering)
+# ═══════════════════════════════════════════════════════════════
+
+async def _llm_stream(prompt: str, hist: list[dict]):
+    """Stream LLM tokens, dispatching based on LLM_PROVIDER env."""
+    if LLM_PROVIDER == "mimo":
+        async for x in _llm_stream_mimo(prompt, hist):
+            yield x
+        return
+    async for x in _llm_stream_vllm(prompt, hist):
+        yield x
+
+
+async def _llm_stream_vllm(prompt: str, hist: list[dict]):
+    """vLLM streaming via OpenAI-compatible SSE."""
+    msgs = [{"role": "system", "content": SYSTEM_PROMPT}]
+    msgs.extend(hist)
+    msgs.append({"role": "user", "content": prompt})
+
+    async with aiohttp.ClientSession() as s:
+        async with s.post(
+            f"{VLLM_URL}/chat/completions",
+            json={"model": LLM_MODEL, "messages": msgs,
+                  "max_tokens": LLM_MAX_TOKENS, "temperature": LLM_TEMPERATURE,
+                  "stream": True,
+                  "chat_template_kwargs": {"enable_thinking": False}},
+            timeout=aiohttp.ClientTimeout(total=LLM_TIMEOUT),
+        ) as r:
+            async for line in r.content:
+                line = line.decode().strip()
+                if not line.startswith("data: "):
+                    continue
+                data = line[6:]
+                if data == "[DONE]":
+                    break
+                try:
+                    chunk = json.loads(data)
+                    delta = chunk.get("choices", [{}])[0].get("delta", {})
+                    text = delta.get("content", "")
+                    if text:
+                        if "</think>" in text:
+                            text = text.split("</think>")[-1]
+                        if "<think>" in text:
+                            text = text.split("<think>")[0]
+                        if text.strip():
+                            yield text, False
+                except Exception:
+                    continue
+    yield "", True
+
+
+async def _llm_stream_mimo(prompt: str, hist: list[dict]):
+    """Mimo v2.5 LLM streaming via OpenAI-compatible SSE."""
+    if not MIMO_KEY:
+        yield "", True
+        return
+    msgs = [{"role": "system", "content": SYSTEM_PROMPT}]
+    msgs.extend(hist)
+    msgs.append({"role": "user", "content": prompt})
+
+    h = {"api-key": MIMO_KEY, "Content-Type": "application/json"}
+    async with aiohttp.ClientSession() as s:
+        async with s.post(
+            f"{MIMO_API_BASE}/chat/completions",
+            json={"model": "mimo-v2.5", "messages": msgs,
+                  "max_tokens": LLM_MAX_TOKENS, "temperature": LLM_TEMPERATURE,
+                  "stream": True},
+            headers=h, timeout=aiohttp.ClientTimeout(total=LLM_TIMEOUT),
+        ) as r:
+            async for line in r.content:
+                line = line.decode().strip()
+                if not line.startswith("data: "):
+                    continue
+                data = line[6:]
+                if data == "[DONE]":
+                    break
+                try:
+                    chunk = json.loads(data)
+                    delta = chunk.get("choices", [{}])[0].get("delta", {})
+                    text = delta.get("content", "")
+                    if text and text.strip():
+                        yield text, False
+                except Exception:
+                    continue
+    yield "", True
+
+
+# ═══════════════════════════════════════════════════════════════
+#  ASR (dispatch: qwen-local | mimo-api)
+# ═══════════════════════════════════════════════════════════════
+
+async def _asr_transcribe(audio_data: np.ndarray, asr_engine=None) -> str:
+    """Dispatch ASR based on ASR_PROVIDER env. Return final text."""
+    if ASR_PROVIDER == "mimo":
+        text = ""
+        async for chunk, is_done in _asr_mimo_stream(audio_data):
+            if chunk:
+                text += chunk
+        text = text.strip()
+    else:
+        loop = asyncio.get_running_loop()
+        result = await loop.run_in_executor(None, asr_engine.transcribe_array, audio_data)
+        text = result.get("text", "").strip()
+    # Filter single-char filler results (common ASR noise)
+    fillers = {"嗯", "啊", "哦", "呃", "咦", "噢", "唔", "哎"}
+    if text and len(text) <= 2 and all(c in "嗯啊哦呃咦噢唔哎？！。，、~·~" for c in text):
+        logger.info("ASR filtered filler: %s", text)
+        return ""
+    return text
+
+
+async def _asr_mimo_stream(audio_data: np.ndarray):
+    """Mimo v2.5 streaming ASR. Yields (text_snapshot, is_final)."""
+    if not MIMO_KEY:
+        yield "", True
+        return
+    import io, wave
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wf:
+        wf.setnchannels(1)
+        wf.setsampwidth(2)
+        wf.setframerate(16000)
+        wf.writeframes((audio_data * 32767).astype(np.int16).tobytes())
+    audio_b64 = base64.b64encode(buf.getvalue()).decode()
+
+    h = {"api-key": MIMO_KEY, "Content-Type": "application/json"}
+    async with aiohttp.ClientSession() as s:
+        async with s.post(
+            f"{MIMO_API_BASE}/chat/completions",
+            json={
+                "model": "mimo-v2.5-asr",
+                "messages": [{"role": "user", "content": [
+                    {"type": "input_audio", "input_audio": {
+                        "data": f"data:audio/wav;base64,{audio_b64}"}}
+                ]}],
+                "stream": True,
+                "asr_options": {"language": "auto"},
             },
-        },
-        API_SECRET,
-        algorithm="HS256",
+            headers=h, timeout=aiohttp.ClientTimeout(total=20),
+        ) as r:
+            async for line in r.content:
+                line = line.decode().strip()
+                if not line.startswith("data: "):
+                    continue
+                data = line[6:]
+                if data == "[DONE]":
+                    yield "", True
+                    return
+                try:
+                    chunk = json.loads(data)
+                    delta = chunk.get("choices", [{}])[0].get("delta", {})
+                    text = delta.get("content", "")
+                    if text:
+                        yield text, False
+                except Exception:
+                    continue
+    yield "", True
+
+
+# ═══════════════════════════════════════════════════════════════
+#  TTS
+# ═══════════════════════════════════════════════════════════════
+
+async def _tts(text: str, voice: str = "Chloe") -> np.ndarray | None:
+    """Mimo TTS. Returns int16 PCM at 24kHz mono, or None on failure."""
+    if not MIMO_KEY:
+        return None
+    if voice.startswith("data:"):
+        model = "mimo-v2.5-tts-voiceclone"
+    else:
+        model = "mimo-v2.5-tts"
+
+    h = {"api-key": MIMO_KEY, "Content-Type": "application/json"}
+    body: dict = {
+        "model": model,
+        "messages": [
+            {"role": "user", "content": ""},
+            {"role": "assistant", "content": text},
+        ],
+        "audio": {"format": "wav", "voice": voice},
+    }
+
+    async with aiohttp.ClientSession() as s:
+        try:
+            async with s.post(
+                f"{MIMO_API_BASE}/chat/completions",
+                json=body, headers=h,
+                timeout=aiohttp.ClientTimeout(total=30),
+            ) as r:
+                result = await r.json()
+                b64 = result.get("choices", [{}])[0].get("message", {}).get("audio", {}).get("data", "")
+                if b64:
+                    return np.frombuffer(base64.b64decode(b64), dtype=np.int16)
+        except Exception:
+            logger.exception("TTS failed")
+    return None
+
+
+def _clean_tts_text(text: str) -> str:
+    text = re.sub(r"\*{1,3}(.*?)\*{1,3}", r"\1", text)
+    text = re.sub(r"`{1,3}.*?`{1,3}", "", text)
+    text = re.sub(r"\[([^\]]+)]\([^)]+\)", r"\1", text)
+    text = re.sub(r"#{1,6}\s*", "", text)
+    text = re.sub(r"[>\-]\s", "", text)
+    text = re.sub(r"[\U0001F300-\U0001F9FF]", "", text)
+    return text.strip()
+
+
+# ═══════════════════════════════════════════════════════════════
+#  TTS Segmenter (xiaozhi-style: two-tier, processed_chars)
+# ═══════════════════════════════════════════════════════════════
+
+@dataclass
+class _TTSSegmenter:
+    buffer: str = ""
+    processed: int = 0
+    is_first: bool = True
+
+    def feed(self, text: str) -> list[str]:
+        self.buffer += text
+        segments = []
+        puncts = FIRST_SENTENCE_PUNCT if self.is_first else SENTENCE_END_PUNCT
+        unprocessed = self.buffer[self.processed:]
+
+        last_pos = -1
+        for ch in puncts:
+            pos = unprocessed.rfind(ch)
+            if pos > last_pos:
+                last_pos = pos
+
+        if last_pos >= 0:
+            seg_raw = unprocessed[:last_pos + 1]
+            seg_clean = _clean_tts_text(seg_raw)
+            if len(seg_clean) >= MIN_TTS_CHARS:
+                segments.append(seg_clean)
+                self.processed += len(seg_raw)
+                if self.is_first:
+                    self.is_first = False
+        return segments
+
+    def flush(self) -> str:
+        remaining = self.buffer[self.processed:].strip()
+        if remaining:
+            remaining = _clean_tts_text(remaining)
+        self.processed = len(self.buffer)
+        return remaining
+
+
+
+# ── VAD ──
+VAD_MODEL_PATH = os.environ.get("VAD_MODEL_PATH") or os.path.join(
+    os.path.dirname(os.path.abspath(__file__)), "whisper_asr", "silero_vad.onnx")
+VAD_THRESHOLD_HIGH = float(os.environ.get("VAD_THRESHOLD_HIGH", "0.5"))
+VAD_THRESHOLD_LOW  = float(os.environ.get("VAD_THRESHOLD_LOW", "0.2"))
+VAD_WINDOW_SIZE    = int(os.environ.get("VAD_WINDOW_SIZE", "5"))
+VAD_VOICE_RATIO    = float(os.environ.get("VAD_VOICE_RATIO", "0.5"))
+MIN_SPEECH_S       = float(os.environ.get("MIN_SPEECH_S", "0.3"))
+MIN_SILENCE_S      = float(os.environ.get("MIN_SILENCE_S", "0.4"))
+MAX_SPEECH_S       = float(os.environ.get("MAX_SPEECH_S", "8.0"))
+
+# Global ONNX session (shared across participants)
+_vad_session = None
+
+
+def _get_vad_session():
+    global _vad_session
+    if _vad_session is not None:
+        return _vad_session
+    if not _HAS_ONNX or not os.path.exists(VAD_MODEL_PATH):
+        return None
+    opts = onnxruntime.SessionOptions()
+    opts.inter_op_num_threads = 1
+    opts.intra_op_num_threads = 1
+    _vad_session = onnxruntime.InferenceSession(
+        VAD_MODEL_PATH, providers=["CPUExecutionProvider"], sess_options=opts,
     )
-
-
-class ASRWorker:
-    def __init__(self, room_name: str, identity: str = "asr-bot"):
-        self._room_name = room_name
-        self._identity = identity
-        self._room = rtc.Room()
-        self._engine: QwenASREngine | None = None
-        self._processors: dict[str, TranscriptProcessor] = {}
-        self._buffers: dict[str, AudioBuffer] = {}
-        self._tasks: dict[str, asyncio.Task] = {}
+    logger.info("Silero VAD loaded (%s)", VAD_MODEL_PATH)
+    return _vad_session
+
+
+# ═══════════════════════════════════════════════════════════════
+#  Silero VAD (ONNX, per-participant state)
+# ═══════════════════════════════════════════════════════════════
+
+class _SileroVAD:
+    """Xiaozhi-style Silero VAD with dual-threshold + sliding window."""
+
+    def __init__(self):
+        self._sess = _get_vad_session()
+        self._state = np.zeros((2, 1, 128), dtype=np.float32)
+        self._context = np.zeros((1, 64), dtype=np.float32)
+        self._window: deque = deque(maxlen=max(VAD_WINDOW_SIZE, 1))
+        self.in_speech = False
+        self.speech_start_sample = 0
+        self.silence_counter = 0
+        self.total_samples = 0
+        self._sr = np.array(16000, dtype=np.int64)
+
+    def process(self, frame: np.ndarray) -> bool:
+        """Process 160-sample (10ms) frame. Call every frame. Returns in_speech."""
+        n = len(frame)
+
+        if self._sess is not None:
+            # ── Silero ONNX inference ──
+            # Silero processes 512-sample windows. Accumulate frames.
+            if not hasattr(self, '_buf'):
+                self._buf = np.array([], dtype=np.float32)
+            self._buf = np.concatenate([self._buf, frame]).astype(np.float32)
+
+            is_voice = False
+            while len(self._buf) >= 512:
+                chunk = self._buf[:512]
+                self._buf = self._buf[512:]
+                audio_in = chunk.reshape(1, -1)
+                inp = np.concatenate([self._context, audio_in], axis=1).astype(np.float32)
+                out, self._state = self._sess.run(
+                    None, {"input": inp, "state": self._state, "sr": self._sr},
+                )
+                self._context = inp[:, -64:]
+                prob = out.item()
+                is_voice = prob >= VAD_THRESHOLD_HIGH or (
+                    prob > VAD_THRESHOLD_LOW and (
+                        self._window[-1] if self._window else False
+                    )
+                )
+                self._window.append(is_voice)
+        else:
+            # ── Fallback: energy-based ──
+            energy = float(np.sqrt(np.mean(frame ** 2)))
+            if energy > 0.015:
+                is_voice = True
+            elif energy < 0.005:
+                is_voice = False
+            else:
+                is_voice = self._window[-1] if self._window else False
+            self._window.append(is_voice)
+
+        have_voice = sum(self._window) / max(len(self._window), 1) >= VAD_VOICE_RATIO
+
+        if have_voice and not self.in_speech:
+            logger.info("VAD START")
+            self.in_speech = True
+            self.speech_start_sample = self.total_samples
+            self.silence_counter = 0
+        elif not have_voice and self.in_speech:
+            self.silence_counter += n
+        elif have_voice:
+            self.silence_counter = 0
+        self.total_samples += n
+        return self.in_speech
+
+    def should_end(self) -> bool:
+        return self.in_speech and self.silence_counter >= int(MIN_SILENCE_S * 16000)
+
+    def min_speech_met(self) -> bool:
+        return (self.total_samples - self.speech_start_sample) / 16000 >= MIN_SPEECH_S
+
+    def reset(self):
+        self._window.clear()
+        self.in_speech = False
+        self.speech_start_sample = 0
+        self.silence_counter = 0
+        self._state = np.zeros((2, 1, 128), dtype=np.float32)
+        self._context = np.zeros((1, 64), dtype=np.float32)
+        if hasattr(self, '_buf'):
+            self._buf = np.array([], dtype=np.float32)
+
+
+# ═══════════════════════════════════════════════════════════════
+#  Worker
+# ═══════════════════════════════════════════════════════════════
+
+class Worker:
+    def __init__(self, room, identity="asr-bot"):
+        self.rn, self.id = room, identity
+        self.room = rtc.Room()
+        self.tasks: dict = {}
+        self.hist: list[dict] = []
 
     async def run(self):
-        logger.info("Loading Qwen ASR model...")
-        self._engine = QwenASREngine(
-            model_id=ASR_MODEL_PATH, language=None
-        )
-
-        self._room.on("track_subscribed", self._on_track_subscribed)
-        self._room.on(
-            "participant_disconnected", self._on_participant_disconnected
-        )
-        self._room.on("participant_connected", lambda p: None)
-        self._room.on("track_published", lambda pub, p: None)
-
-        token = create_token(self._room_name, self._identity)
-        logger.info(f"Connecting to room: {self._room_name}")
-        await self._room.connect(LIVEKIT_URL, token)
-        logger.info(f"ASR worker ready. Participants: {len(self._room.remote_participants)}")
-
-        # Subscribe to any tracks that already exist in the room.
-        for p in self._room.remote_participants.values():
-            logger.info(f"Existing participant: {p.identity}, tracks: {len(p.track_publications)}")
+        logger.info("Loading ASR model...")
+        self.asr = QwenASREngine(model_id=ASR_MODEL, language=None)
+        logger.info("ASR ready")
+
+        self.room.on("track_subscribed", self._on_track)
+        self.room.on("participant_disconnected", self._on_off)
+        self.room.on("participant_connected", lambda p: None)
+        self.room.on("track_published", lambda pub, p: None)
+        await self.room.connect(LIVEKIT_URL, _token(self.rn, self.id))
+        logger.info("Worker ready (VAD+ASR+LLM+TTS)")
+
+        for p in self.room.remote_participants.values():
             for pub in p.track_publications.values():
-                logger.info(f"  pub kind={pub.kind}, source={pub.source.name if pub.source else 'N/A'}, track={type(pub.track).__name__ if pub.track else 'None'}, subscribed={pub.subscribed}")
-                if pub.track is not None and pub.kind == rtc.TrackKind.KIND_AUDIO:
-                    self._on_track_subscribed(pub.track, pub, p)
-
-        logger.info(f"Subscribed tracks: {list(self._buffers.keys())}")
-
+                if pub.track and pub.kind == rtc.TrackKind.KIND_AUDIO:
+                    self._on_track(pub.track, pub, p)
         try:
             await asyncio.Future()
         finally:
-            await self._room.disconnect()
-
-    def _on_track_subscribed(
-        self,
-        track: rtc.RemoteAudioTrack,
-        publication: rtc.RemoteTrackPublication,
-        participant: rtc.RemoteParticipant,
-    ):
-        sid = participant.sid
-        if sid in self._buffers:
+            await self.room.disconnect()
+
+    def _on_track(self, t, _, p):
+        s = p.sid
+        if s in self.tasks:
             return
+        logger.info("TRACK %s", p.identity)
+        self.tasks[s] = asyncio.create_task(self._run(s, t))
+
+    def _on_off(self, p):
+        x = self.tasks.pop(p.sid, None)
+        if x:
+            x.cancel()
+
+    async def _run(self, sid, track):
+        logger.info("_run %s", sid)
+        buf = AudioBuffer(max_duration=MAX_SPEECH_S + 4, sample_rate=16000)
+        sr = 16000
+        vad = _SileroVAD()
+        busy = False
+
+        # ── Single persistent TTS output track per session ──
+        # One track is reused for all turns so audio never cross-streams
+        # between turns/segments (old code published a new track each call).
+        tts_src = rtc.AudioSource(24000, 1)
+        tts_track = rtc.LocalAudioTrack.create_audio_track("tts", tts_src)
+        await self.room.local_participant.publish_track(tts_track)
+        play_q: "asyncio.Queue[tuple[int, np.ndarray]]" = asyncio.Queue()
+        gen = 0                 # turn generation; bumped to flush in-flight audio
+        cur_gen = {"v": -1}    # generation currently being played by _player
+
+        async def _player():
+            """Drain TTS audio sequentially on the single track."""
+            while True:
+                g, audio = await play_q.get()
+                cur_gen["v"] = g
+                for i in range(0, len(audio), 24000 // 50):
+                    if cur_gen["v"] != g:
+                        break  # a newer turn started; drop this stale audio
+                    c = audio[i: i + 24000 // 50]
+                    await tts_src.capture_frame(rtc.AudioFrame(
+                        data=c.tobytes(), sample_rate=24000,
+                        num_channels=1, samples_per_channel=len(c),
+                    ))
+                    await asyncio.sleep(0.018)
+                cur_gen["v"] = -1
+
+        player_task = asyncio.create_task(_player())
+
+        async def _acc():
+            nonlocal busy, gen
+            stream = rtc.AudioStream(track, sample_rate=sr, num_channels=1)
+            fc = 0
+            async for ev in stream:
+                fc += 1
+                if busy:
+                    continue
+                arr = np.frombuffer(ev.frame.data, dtype=np.int16).astype(np.float32) / 32768.0
+                buf.append(arr)
+
+                was_speech = vad.in_speech
+                vad.process(arr)
+
+                if fc % 100 == 0:
+                    logger.info("[%s] af#%d e=%.4f vad=%s blen=%d",
+                                sid, fc, float(np.sqrt(np.mean(arr ** 2))),
+                                vad.in_speech, len(buf.buffer))
+
+                if was_speech and vad.should_end() and vad.min_speech_met():
+                    logger.info("[%s] VAD END blen=%d", sid, len(buf.buffer))
+                    gen += 1
+                    cur_gen["v"] = -1  # stop current playback immediately
+                    _drain_queue(play_q)
+                    busy = True
+                    try:
+                        await self._transcribe_and_respond(buf, play_q, gen)
+                    except Exception:
+                        logger.exception("[%s] transcribe failed", sid)
+                    busy = False
+                    vad.reset()
+                    vad.total_samples = len(buf.buffer)
+
+                if vad.in_speech and vad.total_samples >= int(MAX_SPEECH_S * sr):
+                    logger.info("[%s] VAD MAX blen=%d", sid, len(buf.buffer))
+                    gen += 1
+                    cur_gen["v"] = -1
+                    _drain_queue(play_q)
+                    busy = True
+                    try:
+                        await self._transcribe_and_respond(buf, play_q, gen)
+                    except Exception:
+                        logger.exception("[%s] max failed", sid)
+                    busy = False
+                    vad.reset()
+                    vad.total_samples = len(buf.buffer)
+
+            logger.info("[%s] stream end fc=%d", sid, fc)
 
-        logger.info(f"Audio track subscribed: {participant.identity} ({sid})")
-        logger.info(f"  track kind={track.kind}, muted={track.muted}, stream_state={track.stream_state}")
-
-        self._buffers[sid] = AudioBuffer(
-            max_duration=BUFFER_DURATION * 2,
-            sample_rate=16000,
-        )
-        self._processors[sid] = TranscriptProcessor(
-            silence_repeats=SILENCE_REPEATS,
-            silence_seconds=SILENCE_SECONDS,
-        )
-        self._tasks[sid] = asyncio.create_task(self._process(sid, track))
-
-    def _on_participant_disconnected(self, participant: rtc.RemoteParticipant):
-        sid = participant.sid
-        t = self._tasks.pop(sid, None)
-        if t:
-            t.cancel()
-        self._buffers.pop(sid, None)
-        self._processors.pop(sid, None)
-
-    async def _process(self, sid: str, track: rtc.RemoteAudioTrack):
-        buf = self._buffers[sid]
-        proc = self._processors[sid]
-
-        async def _accumulate():
+        try:
+            await _acc()
+        except asyncio.CancelledError:
+            pass
+        except Exception:
+            logger.exception("[%s] _acc failed", sid)
+        finally:
+            player_task.cancel()
             try:
-                stream = rtc.AudioStream(track, sample_rate=16000, num_channels=1)
-                count = 0
-                async for ev in stream:
-                    frame = ev.frame
-                    data = frame.data
-                    count += 1
-                    if count == 1:
-                        logger.info(f"[{sid}] first audio frame: {len(data)} bytes, sr={frame.sample_rate}, ch={frame.num_channels}")
-                    arr = (
-                        np.frombuffer(data, dtype=np.int16).astype(np.float32)
-                        / 32768.0
-                    )
-                    buf.append(arr)
-                logger.info(f"[{sid}] stream ended after {count} frames")
+                await self.room.local_participant.unpublish_track(tts_track)
             except Exception:
-                logger.exception(f"[{sid}] accumulate failed")
+                pass
 
-        async def _transcribe():
-            tick = 0
-            last_pos = 0
-            while True:
-                await asyncio.sleep(TRANSCRIBE_INTERVAL)
-                tick += 1
-                total = len(buf.buffer)
-                if total - last_pos < 3200:
-                    continue
-                context = int(0.5 * 16000)
-                audio = buf.buffer[max(0, last_pos - context):]
-                if len(audio) <= 1600:
-                    continue
+    async def _transcribe_and_respond(self, buf: AudioBuffer, play_q, gen: int):
+        all_audio = buf.get_all()
+        buf.clear()
+        if len(all_audio) <= 1600:
+            return
 
-                loop = asyncio.get_running_loop()
-                try:
-                    result = await loop.run_in_executor(
-                        None, self._engine.transcribe_array, audio
-                    )
-                except Exception:
-                    logger.exception("transcription failed")
-                    continue
+        txt = await _asr_transcribe(all_audio, self.asr)
+        if not txt:
+            return
 
-                new_samples = total - last_pos
-                last_pos = total
-                raw = result.get("text", "").strip()
-                if tick % 5 == 0:
-                    logger.info(f"[{sid}] transcribe tick={tick} raw='{raw}' new={new_samples}")
-                if not raw:
-                    continue
+        logger.info("ASR: %s", txt[:80])
+        await self._send({"type": "utterance", "text": txt, "seq": 0})
 
-                for msg in proc.feed(raw):
-                    logger.info(f"[{sid}] msg: {msg['type']} '{msg['text'][:50]}'")
-                    await self._send(msg)
+        # ── Streaming LLM, collect segments for ordered TTS ──
+        reply_full = ""
+        seg = _TTSSegmenter()
+        tts_segments: list[str] = []
 
-        t = asyncio.create_task(_transcribe())
         try:
-            await _accumulate()
-        finally:
-            t.cancel()
+            async for delta, is_final in _llm_stream(txt, self.hist):
+                if delta:
+                    reply_full += delta
+                    await self._send({"type": "reply_partial", "text": reply_full, "seq": 0})
+                    tts_segments.extend(seg.feed(delta))
+
+                if is_final:
+                    remaining = seg.flush()
+                    if remaining:
+                        tts_segments.append(remaining)
+                    break
+        except Exception:
+            logger.exception("LLM stream failed")
+            reply_full = "抱歉，我暂时无法回答。"
+
+        # Ordered TTS: generate concurrently but enqueue in order
+        if tts_segments:
+            tasks = [asyncio.create_task(_tts(s)) for s in tts_segments]
+            for i, task in enumerate(tasks):
+                try:
+                    pcm = await asyncio.wait_for(task, timeout=15)
+                    if pcm is not None and len(pcm) > 0:
+                        logger.info("TTS seg(%d/%d): %s", i+1, len(tasks), tts_segments[i][:40])
+                        await play_q.put((gen, pcm))
+                except (asyncio.TimeoutError, asyncio.CancelledError):
+                    break
+                except Exception:
+                    logger.exception("TTS seg %d failed", i+1)
+
+        if reply_full:
+            await self._send({"type": "reply", "text": reply_full})
+            self.hist.extend([
+                {"role": "user", "content": txt},
+                {"role": "assistant", "content": reply_full},
+            ])
+            self.hist[:] = self.hist[-20:]
 
-    async def _send(self, message: dict):
+    async def _send(self, msg: dict):
         try:
-            await self._room.local_participant.publish_data(
-                json.dumps(message, ensure_ascii=False).encode(),
-                reliable=True,
-                topic="transcription",
+            await self.room.local_participant.publish_data(
+                json.dumps(msg, ensure_ascii=False).encode(),
+                reliable=True, topic="transcription",
             )
         except Exception:
-            logger.exception("publish_data failed")
+            pass
 
 
 async def main():
     import argparse
-
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--room", required=True)
-    parser.add_argument("--identity", default="asr-bot")
-    args = parser.parse_args()
-
+    p = argparse.ArgumentParser()
+    p.add_argument("--room", required=True)
+    a = p.parse_args()
     logging.basicConfig(level=logging.INFO)
+    await Worker(a.room).run()
+
 
-    worker = ASRWorker(room_name=args.room, identity=args.identity)
-    await worker.run()
+def _drain_queue(q: "asyncio.Queue") -> None:
+    """Discard any queued-but-not-yet-played TTS audio."""
+    while not q.empty():
+        try:
+            q.get_nowait()
+        except Exception:
+            break
 
 
 if __name__ == "__main__":
     asyncio.run(main())
+