asr_agent/conversation_worker.py

@@ -182,10 +182,17 @@ async def _asr_transcribe(audio_data: np.ndarray, asr_engine=None) -> str:
         async for chunk, is_done in _asr_mimo_stream(audio_data):
             if chunk:
                 text += chunk
-        return text.strip()
-    loop = asyncio.get_running_loop()
-    result = await loop.run_in_executor(None, asr_engine.transcribe_array, audio_data)
-    return result.get("text", "").strip()
+        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):
@@ -240,11 +247,10 @@ async def _asr_mimo_stream(audio_data: np.ndarray):
 #  TTS
 # ═══════════════════════════════════════════════════════════════
 
-async def _tts_stream(text: str, voice: str = "Chloe"):
-    """Mimo v2.5 streaming TTS. Yields (pcm_chunk: np.ndarray, sr: int, is_final: bool)."""
+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:
-        yield np.array([], dtype=np.float32), 24000, True
-        return
+        return None
     if voice.startswith("data:"):
         model = "mimo-v2.5-tts-voiceclone"
     else:
@@ -258,7 +264,6 @@ async def _tts_stream(text: str, voice: str = "Chloe"):
             {"role": "assistant", "content": text},
         ],
         "audio": {"format": "wav", "voice": voice},
-        "stream": True,
     }
 
     async with aiohttp.ClientSession() as s:
@@ -268,28 +273,13 @@ async def _tts_stream(text: str, voice: str = "Chloe"):
                 json=body, headers=h,
                 timeout=aiohttp.ClientTimeout(total=30),
             ) 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", {})
-                        audio = delta.get("audio")
-                        if audio and isinstance(audio, dict) and "data" in audio:
-                            pcm = np.frombuffer(
-                                base64.b64decode(audio["data"]), dtype=np.int16
-                            ).astype(np.float32) / 32768.0
-                            if len(pcm) > 0:
-                                yield pcm, 24000, False
-                    except Exception:
-                        continue
+                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 stream failed")
-    yield np.array([], dtype=np.float32), 24000, True
+            logger.exception("TTS failed")
+    return None
 
 
 def _clean_tts_text(text: str) -> str:
@@ -345,7 +335,7 @@ class _TTSSegmenter:
 
 # ── VAD ──
 VAD_MODEL_PATH = os.environ.get("VAD_MODEL_PATH") or os.path.join(
-    os.path.dirname(os.path.abspath(__file__)), "silero_vad.onnx")
+    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"))
@@ -512,10 +502,37 @@ class Worker:
         sr = 16000
         vad = _SileroVAD()
         busy = False
-        play_tasks: list[asyncio.Task] = []
+
+        # ── 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, play_tasks
+            nonlocal busy, gen
             stream = rtc.AudioStream(track, sample_rate=sr, num_channels=1)
             fc = 0
             async for ev in stream:
@@ -535,13 +552,12 @@ class Worker:
 
                 if was_speech and vad.should_end() and vad.min_speech_met():
                     logger.info("[%s] VAD END blen=%d", sid, len(buf.buffer))
-                    for pt in play_tasks:
-                        if not pt.done():
-                            pt.cancel()
-                    play_tasks.clear()
+                    gen += 1
+                    cur_gen["v"] = -1  # stop current playback immediately
+                    _drain_queue(play_q)
                     busy = True
                     try:
-                        await self._transcribe_and_respond(buf, play_tasks)
+                        await self._transcribe_and_respond(buf, play_q, gen)
                     except Exception:
                         logger.exception("[%s] transcribe failed", sid)
                     busy = False
@@ -550,13 +566,12 @@ class Worker:
 
                 if vad.in_speech and vad.total_samples >= int(MAX_SPEECH_S * sr):
                     logger.info("[%s] VAD MAX blen=%d", sid, len(buf.buffer))
-                    for pt in play_tasks:
-                        if not pt.done():
-                            pt.cancel()
-                    play_tasks.clear()
+                    gen += 1
+                    cur_gen["v"] = -1
+                    _drain_queue(play_q)
                     busy = True
                     try:
-                        await self._transcribe_and_respond(buf, play_tasks)
+                        await self._transcribe_and_respond(buf, play_q, gen)
                     except Exception:
                         logger.exception("[%s] max failed", sid)
                     busy = False
@@ -571,16 +586,19 @@ class Worker:
             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_tasks: list[asyncio.Task] | None = None):
-        if play_tasks is None:
-            play_tasks = []
+    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()
         txt = await _asr_transcribe(all_audio, self.asr)
         if not txt:
             return
@@ -588,28 +606,41 @@ class Worker:
         logger.info("ASR: %s", txt[:80])
         await self._send({"type": "utterance", "text": txt, "seq": 0})
 
-        # ── Streaming LLM + concurrent TTS playback ──
+        # ── 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})
-                    for s in seg.feed(delta):
-                        logger.info("TTS seg(%d): %s", len(s), s[:40])
-                        play_tasks.append(asyncio.create_task(self._tts_and_play(s)))
+                    tts_segments.extend(seg.feed(delta))
 
                 if is_final:
                     remaining = seg.flush()
                     if remaining:
-                        play_tasks.append(asyncio.create_task(self._tts_and_play(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([
@@ -618,37 +649,6 @@ class Worker:
             ])
             self.hist[:] = self.hist[-20:]
 
-    async def _tts_and_play(self, text: str):
-        """Stream TTS audio and play chunks as they arrive."""
-        try:
-            chunks: list[np.ndarray] = []
-            async for pcm, sr, is_final in _tts_stream(text):
-                if len(pcm) > 0:
-                    chunks.append(pcm)
-                if is_final:
-                    break
-            if chunks:
-                audio = np.concatenate(chunks)
-                if len(audio) > 0:
-                    await self._play(audio, 24000)
-        except (asyncio.CancelledError, asyncio.TimeoutError):
-            pass
-        except Exception:
-            logger.exception("TTS+play failed")
-
-    async def _play(self, audio: np.ndarray, sr: int):
-        src = rtc.AudioSource(sr, 1)
-        tk = rtc.LocalAudioTrack.create_audio_track("tts", src)
-        await self.room.local_participant.publish_track(tk)
-        for i in range(0, len(audio), sr // 50):
-            c = audio[i: i + sr // 50]
-            await src.capture_frame(rtc.AudioFrame(
-                data=c.tobytes(), sample_rate=sr,
-                num_channels=1, samples_per_channel=len(c),
-            ))
-            await asyncio.sleep(0.018)
-        return tk
-
     async def _send(self, msg: dict):
         try:
             await self.room.local_participant.publish_data(
@@ -668,5 +668,15 @@ async def main():
     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/dispatcher.py

@@ -0,0 +1,103 @@
+"""
+LiveKit Worker Dispatcher — per-room ASR worker management.
+App calls POST /connect → gets tokens for a unique room → connects.
+Worker is spawned for that room automatically.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import sys
+import time
+
+import jwt
+from aiohttp import web
+
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "whisper_asr"))
+
+logger = logging.getLogger("dispatcher")
+
+LIVEKIT_URL = os.environ.get("LIVEKIT_URL", "ws://localhost:7880")
+API_KEY = "devkey"
+API_SECRET = "secretsecretsecretsecretsecret12"
+
+
+def _room_token(room_name: str, identity: str, *, can_create: bool = False) -> str:
+    n = int(time.time())
+    return jwt.encode({
+        "iss": API_KEY, "sub": identity,
+        "nbf": n - 60, "exp": n + 24 * 3600,
+        "video": {
+            "roomJoin": True, "roomCreate": can_create,
+            "room": room_name,
+            "canPublish": True, "canSubscribe": True, "canPublishData": True,
+        },
+    }, API_SECRET, algorithm="HS256")
+
+
+_workers: dict[str, asyncio.Task] = {}
+
+
+async def _spawn_worker(room_name: str):
+    """Run a worker for one room. Exits when room is empty or cancelled."""
+    from conversation_worker import Worker
+    try:
+        logger.info("Worker starting room=%s", room_name)
+        w = Worker(room_name, identity="asr-bot")
+        await w.run()
+    except asyncio.CancelledError:
+        pass
+    except Exception:
+        logger.exception("Worker %s failed", room_name)
+    finally:
+        logger.info("Worker stopped room=%s", room_name)
+
+
+async def handle_connect(request: web.Request) -> web.Response:
+    """POST /connect — return tokens for a new unique room and spawn worker."""
+    try:
+        body = await request.json()
+        app_identity = body.get("identity", f"user-{int(time.time()*1000)%100000}")
+    except Exception:
+        app_identity = f"user-{int(time.time()*1000)%100000}"
+
+    room_name = f"room-{int(time.time_ns())}"
+
+    # Spawn worker (runs in background, connects to room)
+    task = asyncio.create_task(_spawn_worker(room_name))
+    _workers[room_name] = task
+
+    logger.info("Room %s created for %s (total workers: %d)", room_name, app_identity, len(_workers))
+
+    return web.json_response({
+        "room": room_name,
+        "url": LIVEKIT_URL,
+        "token": _room_token(room_name, app_identity, can_create=True),
+    })
+
+
+async def handle_health(request: web.Request) -> web.Response:
+    # Clean up finished workers
+    for name, task in list(_workers.items()):
+        if task.done():
+            _workers.pop(name, None)
+    return web.json_response({"status": "ok", "active_workers": len(_workers)})
+
+
+async def main():
+    logging.basicConfig(level=logging.INFO)
+    app = web.Application()
+    app.router.add_post("/connect", handle_connect)
+    app.router.add_get("/health", handle_health)
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(runner, "0.0.0.0", 9000)
+    await site.start()
+    logger.info("Dispatcher listening on :9000")
+    await asyncio.Future()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

flutter_asr_client/android/app/build.gradle.kts

@@ -5,6 +5,9 @@ plugins {
     id("dev.flutter.flutter-gradle-plugin")
 }
 
+import java.util.Properties
+import java.io.FileInputStream
+
 android {
     namespace = "com.dinnoni.asr_client"
     compileSdk = flutter.compileSdkVersion
@@ -19,6 +22,27 @@ android {
         jvmTarget = JavaVersion.VERSION_17.toString()
     }
 
+    val keystorePropertiesFile = rootProject.file("key.properties")
+    val keystoreProperties = Properties()
+    if (keystorePropertiesFile.exists()) {
+        keystoreProperties.load(FileInputStream(keystorePropertiesFile))
+    }
+
+    signingConfigs {
+        create("release") {
+            keyAlias = keystoreProperties["keyAlias"] as String
+            keyPassword = keystoreProperties["keyPassword"] as String
+            storeFile = file(keystoreProperties["storeFile"] as String)
+            storePassword = keystoreProperties["storePassword"] as String
+        }
+        getByName("debug") {
+            keyAlias = keystoreProperties["keyAlias"] as String
+            keyPassword = keystoreProperties["keyPassword"] as String
+            storeFile = file(keystoreProperties["storeFile"] as String)
+            storePassword = keystoreProperties["storePassword"] as String
+        }
+    }
+
     defaultConfig {
         // TODO: Specify your own unique Application ID (https://developer.android.com/studio/build/application-id.html).
         applicationId = "com.dinnoni.asr_client"
@@ -32,8 +56,15 @@ android {
 
     buildTypes {
         release {
-            // TODO: Add your own signing config for the release build.
-            // Signing with the debug keys for now, so `flutter run --release` works.
+            signingConfig = signingConfigs.getByName("release")
+            isMinifyEnabled = false
+            isShrinkResources = false
+            proguardFiles(
+                getDefaultProguardFile("proguard-android-optimize.txt"),
+                "proguard-rules.pro"
+            )
+        }
+        debug {
             signingConfig = signingConfigs.getByName("debug")
         }
     }

flutter_asr_client/android/app/proguard-rules.pro

@@ -0,0 +1,8 @@
+# Flutter / Dart
+-keep class io.flutter.app.** { *; }
+-keep class io.flutter.plugin.** { *; }
+-keep class io.flutter.util.** { *; }
+-keep class io.flutter.view.** { *; }
+-keep class io.flutter.** { *; }
+-keep class io.flutter.plugins.** { *; }
+-dontwarn io.flutter.**

flutter_asr_client/lib/common/constants.dart

@@ -13,5 +13,5 @@ abstract final class AppConstants {
   static const reconnectBaseDelay = Duration(seconds: 1);
   static const reconnectMaxDelay = Duration(seconds: 30);
   static const appName = '检测数据采集';
-  static const defaultLiveKitUrl = 'ws://localhost:7888';
+  static const defaultLiveKitUrl = 'ws://36.152.142.37:10003';
 }