feat: add realtime postprocess hotwords (#3317)

Co-authored-by: LauraGPT <lauragpt@users.noreply.github.com>

Author: LauraGPT

examples/industrial_data_pretraining/fun_asr_nano/docs/realtime_demo.md

@@ -59,6 +59,25 @@ render(committed + preview);
 
 `HOTWORDS:Tool,客製化,季會` 或 `--hotword-file` 是模型解码阶段的热词偏置，不是确定性文本替换。热词过多、太短、中英文混杂，或音频静音/低音量/长句不停顿时，模型可能把相近声音偏向热词。排查时建议先完全关闭热词跑同一段音频；如果需求是把固定误识别改成正确专名，优先在识别完成后做后处理，而不是把大量词放进模型 hotwords。
 
+实时服务也支持最终文本级后处理，适合“固定错词 -> 正确专名”的确定性纠正，不会参与模型解码，也不会在静音时把声音偏成热词。启动时可以传文件：
+
+```bash
+cat > postprocess_hotwords.txt <<'EOF'
+哈囉=>客製化
+哈罗=>客製化
+EOF
+
+funasr-realtime-server --postprocess-hotword-file postprocess_hotwords.txt
+```
+
+也可以在同一条 WebSocket 连接中发送：
+
+```text
+POSTPROCESS_HOTWORDS:哈囉=>客製化,哈罗=>客製化
+```
+
+如果你确实希望模型在解码阶段更偏向某些专名，再使用 `HOTWORDS:` / `--hotword-file`；否则优先使用 `POSTPROCESS_HOTWORDS:` / `--postprocess-hotword-file`。
+
 ## 客户端
 
 ```bash

examples/industrial_data_pretraining/fun_asr_nano/serve_realtime_ws.py

@@ -1,6 +1,14 @@
 #!/usr/bin/env python3
 """Compatibility entry point for the packaged realtime WebSocket server."""
 
+import sys
+from pathlib import Path
+
+
+repo_root = Path(__file__).resolve().parents[3]
+if (repo_root / "funasr").is_dir() and str(repo_root) not in sys.path:
+    sys.path.insert(0, str(repo_root))
+
 from funasr.bin.realtime_ws import cli_main
 
 

funasr/bin/realtime_ws.py

@@ -23,6 +23,11 @@
 import regex
 import websockets
 
+from funasr.utils.postprocess_hotwords import (
+    apply_postprocess_hotwords_to_results,
+    parse_postprocess_hotwords,
+)
+
 warnings.filterwarnings('ignore')
 logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
 logger = logging.getLogger(__name__)
@@ -73,6 +78,36 @@ def _clean_asr_text(text):
     return text.strip()
 
 
+def _postprocess_result_text(text, asr_kwargs):
+    if not text:
+        return text
+    cfg = {
+        key: asr_kwargs[key]
+        for key in (
+            "postprocess_hotwords",
+            "postprocess_hotword_file",
+            "postprocess_hotword_threshold",
+            "postprocess_hotword_fuzzy",
+            "return_postprocess_hotword_matches",
+        )
+        if key in asr_kwargs
+    }
+    if not cfg:
+        return text
+    results = apply_postprocess_hotwords_to_results([{"text": text}], cfg)
+    return results[0].get("text", text)
+
+
+def _parse_postprocess_hotwords_command(payload):
+    explicit, fuzzy_targets = parse_postprocess_hotwords(
+        payload.replace(",", "\n")
+    )
+    if fuzzy_targets:
+        for target in fuzzy_targets:
+            explicit[target] = target
+    return explicit
+
+
 from funasr.models.fsmn_vad_streaming.dynamic_vad import DynamicStreamingVAD
 
 
@@ -587,6 +622,7 @@ def _decode_segment(self, seg):
             )
             text = results[0]["text"] if results else ""
             text = _clean_asr_text(text)
+            text = _postprocess_result_text(text, self.asr_kwargs)
             self.prev_seg_text = text
             return text
         except Exception as e:
@@ -675,6 +711,13 @@ def load_models(args):
             _asr_kwargs["language"] = args.language
             logger.info(f"Language: {args.language}")
 
+        postprocess_file = getattr(args, "postprocess_hotword_file", "")
+        if postprocess_file:
+            _asr_kwargs["postprocess_hotword_file"] = postprocess_file
+            _asr_kwargs["postprocess_hotword_fuzzy"] = False
+            _asr_kwargs["return_postprocess_hotword_matches"] = True
+            logger.info(f"Loaded postprocess hotwords from '{postprocess_file}'")
+
         if getattr(args, "endpoint_mode", "server") == "server":
             logger.info("Loading VAD: fsmn-vad (streaming)")
             _vad_model = AutoModel(
@@ -770,6 +813,18 @@ async def handle_client(websocket, args):
                     session.asr_kwargs["hotwords"] = hotwords
                     await websocket.send(json.dumps({"event": "hotwords_set", "hotwords": hotwords}))
                     logger.info(f"Hotwords set: {len(hotwords)} words")
+                elif cmd.upper().startswith("POSTPROCESS_HOTWORDS:"):
+                    payload = cmd.split(":", 1)[1]
+                    hotwords = _parse_postprocess_hotwords_command(payload)
+                    session.asr_kwargs = dict(session.asr_kwargs)
+                    session.asr_kwargs["postprocess_hotwords"] = hotwords
+                    session.asr_kwargs["postprocess_hotword_fuzzy"] = False
+                    session.asr_kwargs["return_postprocess_hotword_matches"] = True
+                    await websocket.send(json.dumps({
+                        "event": "postprocess_hotwords_set",
+                        "postprocess_hotwords": hotwords,
+                    }))
+                    logger.info(f"Postprocess hotwords set: {len(hotwords)} pairs")
                 elif cmd.upper().startswith("LANGUAGE:"):
                     lang = cmd[9:].strip()
                     session.asr_kwargs = dict(session.asr_kwargs)
@@ -904,6 +959,17 @@ def build_arg_parser():
     parser.add_argument("--enable-spk", action="store_true", help="Enable streaming speaker diarization.")
     parser.add_argument("--spk-model", type=str, default="iic/speech_eres2netv2_sv_zh-cn_16k-common")
     parser.add_argument("--hotword-file", type=str, default="热词列表")
+    parser.add_argument(
+        "--postprocess-hotword-file",
+        type=str,
+        default="",
+        help=(
+            "Deterministic text-level hotword corrections applied to final "
+            "sentences after decoding. Lines should be 'wrong=>right' pairs. "
+            "Unlike --hotword-file, this does not bias "
+            "model decoding or hallucinate words during silence."
+        ),
+    )
     parser.add_argument("--language", type=str, default=None, help="Language hint (e.g. 中文, English, 日本語)")
     parser.add_argument(
         "--dtype",

tests/test_realtime_ws_service.py

@@ -250,9 +250,11 @@ class DummyEngine:
     def __init__(self):
         self._engine = types.SimpleNamespace(tokenizer=DummyTokenizer())
         self.input_lengths = []
+        self.generate_kwargs = []
 
     def generate(self, inputs, **kwargs):
         self.input_lengths.extend(len(audio) for audio in inputs)
+        self.generate_kwargs.append(kwargs)
         return [{"text": "hello"}]
 
 
@@ -318,6 +320,21 @@ def test_client_endpoint_mode_emits_partial_at_first_decode_threshold():
     assert session.vllm_engine.input_lengths == [int(0.48 * session.sample_rate)]
 
 
+def test_postprocess_hotwords_correct_final_text_without_model_hotword_bias():
+    module = load_service_module()
+    session = make_client_endpoint_session(module)
+    session.asr_kwargs = {
+        "postprocess_hotwords": {"hello": "Tool"},
+        "return_postprocess_hotword_matches": True,
+    }
+
+    session.add_audio(np.zeros(int(0.4 * session.sample_rate), dtype=np.int16).tobytes())
+    result = session.commit()
+
+    assert result["sentences"] == [{"text": "Tool", "start": 0, "end": 400}]
+    assert session.vllm_engine.generate_kwargs[-1].get("hotwords") is None
+
+
 def test_client_commits_short_utterances_once_with_monotonic_timestamps():
     module = load_service_module()
     session = make_client_endpoint_session(module)
@@ -447,6 +464,121 @@ async def send(self, message):
     assert received_audio == [b"first", b"second"]
 
 
+def test_handler_accepts_postprocess_hotwords_without_model_hotword_bias(monkeypatch):
+    module = load_service_module()
+    session_kwargs = []
+
+    class ProtocolSession:
+        def __init__(self, vllm_engine, asr_kwargs, *args, **kwargs):
+            self.is_active = False
+            self.asr_kwargs = asr_kwargs
+            session_kwargs.append(dict(asr_kwargs))
+
+        def reset(self):
+            pass
+
+        def commit(self):
+            return {"is_final": True, "asr_kwargs": dict(self.asr_kwargs)}
+
+    class FakeWebSocket:
+        remote_address = ("127.0.0.1", 12345)
+
+        def __init__(self):
+            self.messages = iter(
+                [
+                    "START",
+                    "POSTPROCESS_HOTWORDS:hello=>Tool,哈囉=>客製化",
+                    "COMMIT",
+                ]
+            )
+            self.sent = []
+
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            try:
+                return next(self.messages)
+            except StopIteration as error:
+                raise StopAsyncIteration from error
+
+        async def send(self, message):
+            self.sent.append(json.loads(message))
+
+    monkeypatch.setattr(module, "load_models", lambda args: (object(), {}, None, None))
+    monkeypatch.setattr(module, "ClientEndpointVAD", lambda: object(), raising=False)
+    monkeypatch.setattr(module, "create_speaker_tracker", lambda model, args: None)
+    monkeypatch.setattr(module, "RealtimeASRSession", ProtocolSession)
+    websocket = FakeWebSocket()
+    args = types.SimpleNamespace(
+        decode_interval=0.48,
+        partial_window_sec=15.0,
+        endpoint_mode="client",
+        log_session_stats_interval=0.0,
+    )
+
+    asyncio.run(module.handle_client(websocket, args))
+
+    assert session_kwargs == [{}]
+    assert websocket.sent[1] == {
+        "event": "postprocess_hotwords_set",
+        "postprocess_hotwords": {"hello": "Tool", "哈囉": "客製化"},
+    }
+    final = [message for message in websocket.sent if message.get("is_final")][0]
+    assert final["asr_kwargs"] == {
+        "postprocess_hotwords": {"hello": "Tool", "哈囉": "客製化"},
+        "postprocess_hotword_fuzzy": False,
+        "return_postprocess_hotword_matches": True,
+    }
+    assert "hotwords" not in final["asr_kwargs"]
+
+
+def test_load_models_reads_postprocess_hotword_file(monkeypatch, tmp_path):
+    module = load_service_module()
+    module._vllm_engine = None
+    module._asr_kwargs = None
+    module._vad_model = None
+    module._spk_model = None
+    hotword_file = tmp_path / "postprocess_hotwords.txt"
+    hotword_file.write_text("hello=>Tool\n哈囉=>客製化\n", encoding="utf-8")
+    auto_model_calls = []
+
+    import funasr
+
+    monkeypatch.setattr(
+        funasr,
+        "AutoModel",
+        lambda model, **kwargs: auto_model_calls.append(model) or object(),
+    )
+    vllm_stub = types.ModuleType("funasr.auto.auto_model_vllm")
+    vllm_stub.AutoModelVLLM = lambda **kwargs: object()
+    monkeypatch.setitem(sys.modules, "funasr.auto.auto_model_vllm", vllm_stub)
+
+    args = types.SimpleNamespace(
+        model="FunAudioLLM/Fun-ASR-Nano-2512",
+        hub="ms",
+        device="cpu",
+        dtype="fp32",
+        tensor_parallel_size=1,
+        gpu_memory_utilization=0.8,
+        max_model_len=2048,
+        hotword_file="",
+        postprocess_hotword_file=str(hotword_file),
+        language=None,
+        enable_spk=False,
+        endpoint_mode="client",
+    )
+
+    _, asr_kwargs, _, _ = module.load_models(args)
+
+    assert asr_kwargs == {
+        "postprocess_hotword_file": str(hotword_file),
+        "postprocess_hotword_fuzzy": False,
+        "return_postprocess_hotword_matches": True,
+    }
+    assert auto_model_calls == []
+
+
 def test_two_hour_session_keeps_audio_bounded_and_duration_absolute():
     module = load_service_module()
     sample_rate = 10