feat(tts-asr): streaming Mimo TTS (voiceclone) + streaming ASR

- ASR: mio-v2.5-asr streaming SSE, correct audio data URI format
- TTS: mio-v2.5-tts / mio-v2.5-tts-voiceclone streaming
- TTS voice clone: auto-detected when voice starts with 'data:'
- TTS: 24kHz PCM16LE mono, chunked playback
- MIMO_API_BASE unified across ASR/LLM/TTS
- _play supports any sample rate

asr_agent/conversation_worker.py

@@ -176,19 +176,23 @@ async def _llm_stream_mimo(prompt: str, hist: list[dict]):
 # ═══════════════════════════════════════════════════════════════
 
 async def _asr_transcribe(audio_data: np.ndarray, asr_engine=None) -> str:
-    """Dispatch ASR based on ASR_PROVIDER env."""
+    """Dispatch ASR based on ASR_PROVIDER env. Return final text."""
     if ASR_PROVIDER == "mimo":
-        return await _asr_mimo(audio_data)
-    # Default: local Qwen3-ASR
+        text = ""
+        async for chunk, is_done in _asr_mimo_stream(audio_data):
+            if chunk:
+                text = chunk  # last partial becomes final
+        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()
 
 
-async def _asr_mimo(audio_data: np.ndarray) -> str:
-    """Mimo v2.5 ASR via API."""
+async def _asr_mimo_stream(audio_data: np.ndarray):
+    """Mimo v2.5 streaming ASR. Yields (text_snapshot, is_final)."""
     if not MIMO_KEY:
-        return ""
+        yield "", True
+        return
     import io, wave
     buf = io.BytesIO()
     with wave.open(buf, "wb") as wf:
@@ -202,44 +206,90 @@ async def _asr_mimo(audio_data: np.ndarray) -> str:
     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": audio_b64, "format": "wav"}}
-                ]}
-            ]},
+            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,
+                "extra_body": {"asr_options": {"language": "auto"}},
+            },
             headers=h, timeout=aiohttp.ClientTimeout(total=20),
         ) as r:
-            try:
-                result = await r.json()
-                return result.get("choices", [{}])[0].get("message", {}).get("content", "")
-            except Exception:
-                logger.exception("Mimo ASR failed")
-                return ""
+            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) -> np.ndarray | None:
-    key = os.environ.get("MIMO_KEY", "")
-    if not key:
-        return None
-    h = {"api-key": key, "Content-Type": "application/json"}
+async def _tts_stream(text: str, voice: str = "Chloe"):
+    """Mimo v2.5 streaming TTS. Yields (pcm_chunk: np.ndarray, sr: int, is_final: bool)."""
+    if not MIMO_KEY:
+        yield np.array([], dtype=np.float32), 24000, True
+        return
+    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},
+        "stream": True,
+    }
+
     async with aiohttp.ClientSession() as s:
-        async with s.post(
-            "https://token-plan-cn.xiaomimimo.com/v1/chat/completions",
-            json={"model": "mimo-v2.5-tts", "messages": [
-                {"role": "user", "content": "请用自然语速朗读。"},
-                {"role": "assistant", "content": text},
-            ], "audio": {"format": "wav", "voice": "Chloe"}},
-            headers=h, timeout=aiohttp.ClientTimeout(total=20),
-        ) as r:
-            try:
-                b64 = (await r.json())["choices"][0]["message"]["audio"]["data"]
-                return np.frombuffer(base64.b64decode(b64), dtype=np.int16)
-            except Exception:
-                return None
+        try:
+            async with s.post(
+                f"{MIMO_API_BASE}/chat/completions",
+                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
+        except Exception:
+            logger.exception("TTS stream failed")
+    yield np.array([], dtype=np.float32), 24000, True
 
 
 def _clean_tts_text(text: str) -> str:
@@ -569,12 +619,19 @@ class Worker:
             self.hist[:] = self.hist[-20:]
 
     async def _tts_and_play(self, text: str):
-        """Fetch TTS audio and play it immediately (background task)."""
+        """Stream TTS audio and play chunks as they arrive."""
         try:
-            wav = await asyncio.wait_for(_tts(text), timeout=15)
-            if wav is not None and len(wav) > 0:
-                await self._play(wav, 24000)
-        except (asyncio.TimeoutError, asyncio.CancelledError):
+            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")