Use single FunASR 2pass stream for realtime preview

fastapi_wss/src/a2a_wss.py

@@ -835,7 +835,12 @@ async def forward_audio(websocket):
             logger.info(f"已连接到 ASR 服务: {TARGET_WS_URL}")
 
             # 发送 ASR 配置（双通道模式，PCM 音频格式）
-            config = {"mode": "2pass", "chunk_size": [10, 10, 10], "wav_format": "pcm"}
+            config = {
+                "mode": "2pass",
+                "chunk_size": [5, 10, 5],
+                "chunk_interval": 5,
+                "wav_format": "pcm",
+            }
             await target_ws.send(json.dumps(config))
             logger.debug(f"发送 ASR 配置: {config}")
 

speakr/src/api/asr_ws.py

@@ -56,10 +56,10 @@ def _hotwords_payload(logger, scene=SCENE_CG):
 
 def _asr_init_payload(mode, hotwords):
     return {
-        'chunk_size': [10, 10, 10],
+        'chunk_size': [5, 10, 5],
         'wav_name': 'h5',
         'is_speaking': True,
-        'chunk_interval': 10,
+        'chunk_interval': 5,
         'itn': True,
         'mode': mode,
         'hotwords': hotwords,
@@ -85,6 +85,7 @@ def live_asr_proxy(client_ws):
     upstream_closed = threading.Event()
     stop_requested = threading.Event()
     upstream_holder = {'ws': None}
+    upstream_message_count = {'value': 0}
 
     def on_open(upstream):
         upstream_holder['ws'] = upstream
@@ -93,8 +94,42 @@ def live_asr_proxy(client_ws):
         _send_json(client_ws, {'type': 'state', 'state': 'connected', 'mode': mode, 'scene': scene})
 
     def on_message(_upstream, message):
+        upstream_message_count['value'] += 1
+        outbound_message = message
+        if upstream_message_count['value'] <= 20:
+            try:
+                payload = json.loads(message)
+                if payload.get('mode') == 'online':
+                    payload['mode'] = '2pass-online'
+                    outbound_message = json.dumps(payload, ensure_ascii=False)
+                elif payload.get('mode') == 'offline':
+                    payload['mode'] = '2pass-offline'
+                    outbound_message = json.dumps(payload, ensure_ascii=False)
+                logger.info(
+                    'Realtime ASR upstream message #%s: mode=%s text=%r',
+                    upstream_message_count['value'],
+                    payload.get('mode'),
+                    (payload.get('text') or '')[:80],
+                )
+            except Exception:
+                logger.info(
+                    'Realtime ASR upstream message #%s: %r',
+                    upstream_message_count['value'],
+                    str(message)[:160],
+                )
+        else:
+            try:
+                payload = json.loads(message)
+                if payload.get('mode') == 'online':
+                    payload['mode'] = '2pass-online'
+                    outbound_message = json.dumps(payload, ensure_ascii=False)
+                elif payload.get('mode') == 'offline':
+                    payload['mode'] = '2pass-offline'
+                    outbound_message = json.dumps(payload, ensure_ascii=False)
+            except Exception:
+                outbound_message = message
         try:
-            client_ws.send(message)
+            client_ws.send(outbound_message)
         except Exception:
             upstream_closed.set()
 

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

@@ -187,10 +187,9 @@ export class RealtimeAsrPipeline {
   start() {
     this.disposed = false;
 
-    const onlineStarted = this.wsOnline?.wsStart() === 1;
+    const onlineStarted = this.wsOnline?.wsStart("2pass") === 1;
-    const offlineStarted = this.wsOffline?.wsStart("offline") === 1;
+    this.setActive(onlineStarted);
-    this.setActive(onlineStarted || offlineStarted);
+    return onlineStarted ? 1 : 0;
-    return onlineStarted || offlineStarted ? 1 : 0;
   }
 
   // 立即停止：用于暂停或离开录音视图。关闭 socket 和 Recorder 回调，
@@ -752,11 +751,7 @@ export class RealtimeAsrPipeline {
   // 预览文本 = 待矫正 offline 文本 + online/offline 交接文本 + 当前 online 临时文本。
   syncRealtimePreview() {
     // 【兜底去重】无论前面逻辑是否产生重复标点，最终渲染前统一清理
-    const previewText = this._deduplicateConsecutivePunctuation(
+    const previewText = this._deduplicateConsecutivePunctuation(this.recTextOnline);
-      this.getPendingSentencePreviewText() +
-        this.offlineSentenceHandoffText +
-        this.recTextOnline
-    );
     this.onPreviewTextChange(previewText);
   }
 
@@ -1202,7 +1197,7 @@ export class RealtimeAsrPipeline {
     const timestamp = parsedMessage.timestamp;
     const messageEpoch = this.liveTranscriptionEpoch;
 
-    if (mode === "2pass-online") {
+    if (mode === "2pass-online" || mode === "online") {
       // 低延迟预览文本可能不完整，因此不进入正式 segmentList。
       const currentTimestamp = this.extractTimestampMs(timestamp);
       if (this.isStaleRealtimeMessage(currentTimestamp, messageEpoch)) {
@@ -1224,13 +1219,25 @@ export class RealtimeAsrPipeline {
         this.recTextOnline += shouldBreak ? text + "\n" : text;
       }
       this.lastOnlineTimestamp = currentTimestamp;
-    } else if (mode === "2pass-offline") {
+    } else if (mode === "2pass-offline" || mode === "offline") {
       // online 通道也可能回显 offline 句子，这里只作为交接预览保存。
       // 正式句子只通过 offline 通道进入矫正流程。
-      this.recTextOnline = "";
+      if (text.trim()) {
-      this.offlineSentenceHandoffText = text || this.recTextOnline || "";
+        const pendingSentence = this.ensurePendingSentence(
-      this.recTextOnline = "";
+          text,
-      this.lastOnlineTimestamp = null;
+          timestamp,
+          messageEpoch
+        );
+        if (
+          pendingSentence &&
+          !pendingSentence.processed &&
+          !pendingSentence.correctionRequested
+        ) {
+          pendingSentence.correctionRequested = true;
+          this.setPendingCorrectionCount(this.pendingCorrectionCount + 1);
+          this._requestCorrection(pendingSentence);
+        }
+      }
     }
 
     this.syncRealtimePreview();
@@ -1249,7 +1256,7 @@ export class RealtimeAsrPipeline {
     const messageEpoch = this.liveTranscriptionEpoch;
     const messageTimestampMs = this.extractTimestampMs(timestamp);
 
-    if (mode === "2pass-offline") {
+    if (mode === "2pass-offline" || mode === "offline") {
       // 文本先到：先放入 pending 队列刷新预览，再调用 LLM 矫正接口，
       // 矫正完成后才进入正式 segmentList。
       if (this.isStaleRealtimeMessage(messageTimestampMs, messageEpoch)) {
@@ -1257,8 +1264,10 @@ export class RealtimeAsrPipeline {
       }
 
       const text = "" + parsedMessage.text;
-      this.recTextOnline = "";
+      if (!text.trim()) {
-      this.lastOnlineTimestamp = null;
+        this.syncRealtimePreview();
+        return;
+      }
       const pendingSentence = this.ensurePendingSentence(
         text,
         timestamp,

speakr/templates/index/src/asr/wsconnecter.js

@@ -54,7 +54,9 @@ export function WebSocketConnectMethod(config) {
     if (speechSokt) {
       speechSokt.close();
     }
-    if (e === "offline" || (e && e !== "online")) {
+    if (e === "2pass") {
+      modeType = "2pass";
+    } else if (e === "offline" || (e && e !== "online")) {
       modeType = "offline";
     } else {
       modeType = "online";