Add realtime ASR preview updates

README.md

@@ -96,6 +96,8 @@ TEXT_MODEL_API_KEY=none
 
 VOICEPRINT_STORE=/data/asr_voiceprints.json
 VOICEPRINT_MATCH_THRESHOLD=0.45
+REALTIME_PARTIAL_MIN_SECONDS=1.5
+REALTIME_PARTIAL_INTERVAL_SECONDS=2.0
 ```
 
 说明：

asr/main.py

@@ -10,6 +10,7 @@ PUNC 模型: punc_ct-transformer_zh-cn-common-vocab272727-pytorch
 import os
 import re
 import json
+import asyncio
 import tempfile
 import logging
 import subprocess
@@ -100,6 +101,8 @@ model = None
 VOICE_DETECTED_URL = os.environ.get("VOICE_DETECTED_URL", "http://127.0.0.1:18000")
 VOICEPRINT_STORE = os.environ.get("VOICEPRINT_STORE", "/tmp/asr_voiceprints.json")
 VOICEPRINT_MATCH_THRESHOLD = float(os.environ.get("VOICEPRINT_MATCH_THRESHOLD", "0.45"))
+REALTIME_PARTIAL_INTERVAL_SECONDS = float(os.environ.get("REALTIME_PARTIAL_INTERVAL_SECONDS", "2.0"))
+REALTIME_PARTIAL_MIN_SECONDS = float(os.environ.get("REALTIME_PARTIAL_MIN_SECONDS", "1.5"))
 
 
 def _voiceprint_key(database_name: str, collection_name: str) -> str:
@@ -245,9 +248,39 @@ async def health():
     return {"status": "healthy"}
 
 
+def _write_pcm_wav(audio_bytes: bytes) -> str:
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
+        tmp_path = tmp.name
+
+    with wave.open(tmp_path, "wb") as wav_file:
+        wav_file.setnchannels(1)
+        wav_file.setsampwidth(2)
+        wav_file.setframerate(16000)
+        wav_file.writeframes(audio_bytes)
+
+    return tmp_path
+
+
+def _text_delta(previous: str, current: str) -> str:
+    previous = previous or ""
+    current = current or ""
+    if not current or current == previous:
+        return ""
+    if current.startswith(previous):
+        return current[len(previous):]
+
+    max_overlap = min(len(previous), len(current), 80)
+    for size in range(max_overlap, 0, -1):
+        if previous[-size:] == current[:size]:
+            return current[size:]
+    return current
+
+
 async def run_asr_websocket(websocket: WebSocket):
     await websocket.accept(subprotocol="binary")
     audio_chunks = []
+    last_partial_at = 0.0
+    last_partial_text = ""
 
     try:
         while True:
@@ -255,6 +288,33 @@ async def run_asr_websocket(websocket: WebSocket):
 
             if "bytes" in message and message["bytes"]:
                 audio_chunks.append(message["bytes"])
+                audio_seconds = sum(len(chunk) for chunk in audio_chunks) / 32000.0
+                now = asyncio.get_running_loop().time()
+
+                if (
+                    audio_seconds >= REALTIME_PARTIAL_MIN_SECONDS
+                    and now - last_partial_at >= REALTIME_PARTIAL_INTERVAL_SECONDS
+                ):
+                    last_partial_at = now
+                    partial_path = None
+                    try:
+                        partial_path = _write_pcm_wav(b"".join(audio_chunks))
+                        result = await run_asr(partial_path, language="zh", response_format="json")
+                        partial_text = result.get("text", "") if isinstance(result, dict) else ""
+                        if partial_text and partial_text != last_partial_text:
+                            await websocket.send_json({
+                                "mode": "2pass-online",
+                                "text": partial_text,
+                                "timestamp": str(int(audio_seconds * 1000)),
+                                "is_full_text": True,
+                                "is_final": False,
+                            })
+                            last_partial_text = partial_text
+                    except Exception as e:
+                        logger.warning(f"实时部分识别失败: {e}")
+                    finally:
+                        if partial_path and os.path.exists(partial_path):
+                            os.unlink(partial_path)
                 continue
 
             if "text" in message and message["text"]:
@@ -276,15 +336,7 @@ async def run_asr_websocket(websocket: WebSocket):
     tmp_path = None
     try:
         if audio_chunks:
-            with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
-                tmp_path = tmp.name
-
-            with wave.open(tmp_path, "wb") as wav_file:
-                wav_file.setnchannels(1)
-                wav_file.setsampwidth(2)
-                wav_file.setframerate(16000)
-                wav_file.writeframes(b"".join(audio_chunks))
-
+            tmp_path = _write_pcm_wav(b"".join(audio_chunks))
             result = await run_asr(tmp_path, language="zh", response_format="json")
             if isinstance(result, dict):
                 text = result.get("text", "")

config.env

@@ -29,6 +29,8 @@ LLM_ROLE_HISTORY_MAX_CHARS=1200
 VOICE_DETECTED_URL=http://voice-dected:8000
 VOICEPRINT_STORE=/data/asr_voiceprints.json
 VOICEPRINT_MATCH_THRESHOLD=0.45
+REALTIME_PARTIAL_MIN_SECONDS=1.5
+REALTIME_PARTIAL_INTERVAL_SECONDS=2.0
 
 # fastapi_wss legacy/local ASR-with-speaker helper
 ASR_LOCAL_URL=http://asr-test:59805/asr

speakr/src/api/asr_ws.py

@@ -157,13 +157,16 @@ def live_asr_proxy(client_ws):
                 upstream.send(message)
                 continue
 
-            if payload.get('type') == 'stop':
-                upstream.send(json.dumps({'is_speaking': False}, ensure_ascii=False))
-                stop_requested.set()
-                deadline = time.time() + 3.0
-                while not upstream_closed.is_set() and time.time() < deadline:
-                    time.sleep(0.05)
-                break
+            if payload.get('type') == 'stop':
+                logger.info('Realtime ASR stop requested, waiting for upstream final result')
+                upstream.send(json.dumps({'is_speaking': False}, ensure_ascii=False))
+                stop_requested.set()
+                deadline = time.time() + 25.0
+                while not upstream_closed.is_set() and time.time() < deadline:
+                    time.sleep(0.05)
+                if not upstream_closed.is_set():
+                    logger.warning('Realtime ASR upstream did not close before final wait timeout')
+                break
             if 'is_speaking' in payload or payload.get('mode'):
                 upstream.send(json.dumps(payload, ensure_ascii=False))
     except Exception as exc:
@@ -176,4 +179,4 @@ def live_asr_proxy(client_ws):
             pass
         upstream_closed.set()
         _send_json(client_ws, {'type': 'state', 'state': 'closed'})
-        time.sleep(0.05)
+        time.sleep(0.05)

speakr/templates/index/src/asr/realtime-asr-pipeline.js

@@ -1218,7 +1218,11 @@ export class RealtimeAsrPipeline {
         }
       }
 
-      this.recTextOnline += shouldBreak ? text + "\n" : text;
+      if (parsedMessage.is_full_text) {
+        this.recTextOnline = shouldBreak ? text + "\n" : text;
+      } else {
+        this.recTextOnline += shouldBreak ? text + "\n" : text;
+      }
       this.lastOnlineTimestamp = currentTimestamp;
     } else if (mode === "2pass-offline") {
       // online 通道也可能回显 offline 句子，这里只作为交接预览保存。