Codex/create training and inference scripts (#2791)

* Update funasr_wss_server.py

* Update funasr_wss_client.py

* Add files via upload

* Update funasr_wss_server.py

* Update funasr_wss_client.py

* Add LoRA Paraformer README

examples/industrial_data_pretraining/paraformer/README_LoRA_zh.md

@@ -0,0 +1,112 @@
+# Paraformer LoRA 微调说明
+
+本文档说明如何在 FunASR 中使用 LoRA 微调 Paraformer，并提供训练、推理与 CER 评测的完整示例。
+
+## 1. 前置准备
+
+1. 已准备好符合 FunASR 要求的 `train.jsonl` 与 `val.jsonl`。
+2. 进入仓库根目录（示例路径）：
+
+```bash
+cd /workspace/FunASR
+```
+
+## 2. 训练配置
+
+LoRA 配置文件：
+
+```
+examples/industrial_data_pretraining/paraformer/conf/paraformer_lora.yaml
+```
+
+关键字段说明：
+- `model`: 基座模型名称或本地模型路径。
+- `lora_only`: 是否只训练 LoRA 参数。
+- `lora_bias`: LoRA 偏置训练策略（`none`/`all`/`lora_only`）。
+- `encoder_conf.lora_*` / `decoder_conf.lora_*`: LoRA 参数（rank/alpha/dropout）。
+- `train_data_set_list`/`valid_data_set_list`: 训练/验证集 jsonl。
+
+如需覆盖配置，请通过命令行 `++key=value` 传参。
+
+## 3. 训练脚本
+
+脚本：
+
+```
+examples/industrial_data_pretraining/paraformer/lora_finetune.sh
+```
+
+你只需要确认脚本中的数据路径：
+
+```bash
+data_dir="${workspace}/data/list"
+train_data="${data_dir}/train.jsonl"
+val_data="${data_dir}/val.jsonl"
+```
+
+运行：
+
+```bash
+bash examples/industrial_data_pretraining/paraformer/lora_finetune.sh
+```
+
+训练日志与模型输出将保存在：
+
+```
+examples/industrial_data_pretraining/paraformer/outputs_lora
+```
+
+## 4. 推理脚本
+
+推理脚本会读取 jsonl 输入并生成 `text.hyp` / `text.ref`：
+
+- Python 脚本：`examples/industrial_data_pretraining/paraformer/lora_infer.py`
+- Shell 封装：`examples/industrial_data_pretraining/paraformer/lora_infer.sh`
+
+修改 `lora_infer.sh` 中路径后运行：
+
+```bash
+bash examples/industrial_data_pretraining/paraformer/lora_infer.sh
+```
+
+输出目录默认：
+
+```
+examples/industrial_data_pretraining/paraformer/outputs_lora/infer
+```
+
+## 5. CER 评测
+
+评测脚本：
+
+```
+examples/industrial_data_pretraining/paraformer/lora_cer.sh
+```
+
+运行：
+
+```bash
+bash examples/industrial_data_pretraining/paraformer/lora_cer.sh
+```
+
+结果会输出 CER 统计到：
+
+```
+examples/industrial_data_pretraining/paraformer/outputs_lora/infer/text.cer
+```
+
+## 6. 常见问题
+
+1. **训练不收敛或效果差**
+   - 尝试调整 `lora_rank`、`lora_alpha`、`lora_dropout`。
+   - 调整 `optim_conf.lr` 与 `train_conf.max_epoch`。
+
+2. **推理报错找不到配置**
+   - 确保训练输出目录中存在 `config.yaml`，并在推理脚本中设置正确的 `config_path` 和 `config_name`。
+
+3. **多卡训练**
+   - 设置 `CUDA_VISIBLE_DEVICES`，脚本会自动计算 `gpu_num`。
+
+---
+
+如需进一步定制，可直接在 `paraformer_lora.yaml` 中修改配置或在命令行传参覆盖。

examples/industrial_data_pretraining/paraformer/conf/paraformer_lora.yaml

@@ -0,0 +1,51 @@
+# LoRA finetune config for Paraformer
+# You can override data paths and hyper-parameters by command-line ++key=value.
+
+# model hub name or local model dir
+model: iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch
+model_revision: master
+
+# LoRA settings
+lora_only: true
+lora_bias: none
+
+encoder_conf:
+  lora_list: ["q", "k", "v", "o"]
+  lora_rank: 8
+  lora_alpha: 16
+  lora_dropout: 0.05
+
+decoder_conf:
+  lora_list: ["q", "k", "v", "o"]
+  lora_rank: 8
+  lora_alpha: 16
+  lora_dropout: 0.05
+
+# dataset
+train_data_set_list: data/list/train.jsonl
+valid_data_set_list: data/list/val.jsonl
+
+dataset: AudioDataset
+dataset_conf:
+  index_ds: IndexDSJsonl
+  data_split_num: 1
+  batch_sampler: BatchSampler
+  batch_size: 6000
+  sort_size: 1024
+  batch_type: token
+  num_workers: 4
+
+# training
+train_conf:
+  max_epoch: 30
+  log_interval: 10
+  resume: true
+  validate_interval: 2000
+  save_checkpoint_interval: 2000
+  keep_nbest_models: 10
+  avg_nbest_model: 5
+  use_deepspeed: false
+
+optim: adam
+optim_conf:
+  lr: 0.0001

examples/industrial_data_pretraining/paraformer/lora_cer.sh

@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+# Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights Reserved.
+#  MIT License  (https://opensource.org/licenses/MIT)
+
+set -euo pipefail
+
+workspace=$(pwd)
+
+infer_dir="${workspace}/examples/industrial_data_pretraining/paraformer/outputs_lora/infer"
+ref_file="${infer_dir}/text.ref"
+hyp_file="${infer_dir}/text.hyp"
+cer_file="${infer_dir}/text.cer"
+
+python -m funasr.metrics.wer \
+  ++ref_file="${ref_file}" \
+  ++hyp_file="${hyp_file}" \
+  ++cer_file="${cer_file}" \
+  ++cn_postprocess=false
+
+# Show final CER summary
+if [ -f "${cer_file}" ]; then
+  tail -n 3 "${cer_file}"
+fi

examples/industrial_data_pretraining/paraformer/lora_finetune.sh

@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+# Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights Reserved.
+#  MIT License  (https://opensource.org/licenses/MIT)
+
+set -euo pipefail
+
+workspace=$(pwd)
+
+# which gpu to train or finetune
+export CUDA_VISIBLE_DEVICES=${CUDA_VISIBLE_DEVICES:-"0"}
+gpu_num=$(echo ${CUDA_VISIBLE_DEVICES} | awk -F "," '{print NF}')
+
+# data dir, which contains train.jsonl/val.jsonl
+# NOTE: update these paths to your dataset jsonl files.
+data_dir="${workspace}/data/list"
+train_data="${data_dir}/train.jsonl"
+val_data="${data_dir}/val.jsonl"
+
+# config
+config_path="${workspace}/examples/industrial_data_pretraining/paraformer/conf"
+config_name="paraformer_lora.yaml"
+
+# exp output dir
+output_dir="${workspace}/examples/industrial_data_pretraining/paraformer/outputs_lora"
+log_file="${output_dir}/log.txt"
+
+mkdir -p "${output_dir}"
+
+DISTRIBUTED_ARGS="
+    --nnodes ${WORLD_SIZE:-1} \
+    --nproc_per_node ${gpu_num} \
+    --node_rank ${RANK:-0} \
+    --master_addr ${MASTER_ADDR:-127.0.0.1} \
+    --master_port ${MASTER_PORT:-26669}
+"
+
+echo "log_file: ${log_file}"
+
+torchrun ${DISTRIBUTED_ARGS} \
+  funasr/bin/train_ds.py \
+  --config-path "${config_path}" \
+  --config-name "${config_name}" \
+  ++train_data_set_list="${train_data}" \
+  ++valid_data_set_list="${val_data}" \
+  ++output_dir="${output_dir}" \
+  &> "${log_file}"

examples/industrial_data_pretraining/paraformer/lora_infer.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+# -*- encoding: utf-8 -*-
+
+import argparse
+import json
+import os
+from typing import List, Tuple
+
+from omegaconf import OmegaConf
+
+from funasr import AutoModel
+
+
+def load_jsonl(jsonl_path: str) -> Tuple[List[str], List[str]]:
+    keys = []
+    targets = []
+    with open(jsonl_path, "r", encoding="utf-8") as f:
+        for line in f:
+            if not line.strip():
+                continue
+            record = json.loads(line)
+            key = record.get("key")
+            if key is None and isinstance(record.get("source"), dict):
+                key = record["source"].get("key")
+            keys.append(key or "")
+            targets.append(record.get("target", ""))
+    return keys, targets
+
+
+def build_model(args: argparse.Namespace):
+    kwargs = {}
+    if args.config_path and args.config_name:
+        cfg_path = os.path.join(args.config_path, args.config_name)
+        cfg = OmegaConf.load(cfg_path)
+        kwargs.update(OmegaConf.to_container(cfg, resolve=True))
+    if args.model:
+        kwargs["model"] = args.model
+    if args.init_param:
+        kwargs["init_param"] = args.init_param
+    kwargs["device"] = args.device
+    if args.batch_size:
+        kwargs["batch_size"] = args.batch_size
+    return AutoModel(**kwargs)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--model", type=str, default=None, help="model name or model dir")
+    parser.add_argument("--config-path", type=str, default=None, help="config directory")
+    parser.add_argument("--config-name", type=str, default=None, help="config filename")
+    parser.add_argument("--init-param", type=str, default=None, help="model checkpoint path")
+    parser.add_argument("--input-jsonl", type=str, required=True, help="input jsonl with source/target")
+    parser.add_argument("--output-dir", type=str, required=True, help="output directory")
+    parser.add_argument("--device", type=str, default="cuda:0", help="cuda:0 or cpu")
+    parser.add_argument("--batch-size", type=int, default=1, help="batch size for inference")
+    args = parser.parse_args()
+
+    os.makedirs(args.output_dir, exist_ok=True)
+
+    keys, targets = load_jsonl(args.input_jsonl)
+
+    model = build_model(args)
+    results = model.generate(input=args.input_jsonl, batch_size=args.batch_size)
+
+    hyp_path = os.path.join(args.output_dir, "text.hyp")
+    ref_path = os.path.join(args.output_dir, "text.ref")
+
+    with open(hyp_path, "w", encoding="utf-8") as hyp_f, open(
+        ref_path, "w", encoding="utf-8"
+    ) as ref_f:
+        for idx, result in enumerate(results):
+            key = keys[idx] if idx < len(keys) else result.get("key", f"utt_{idx}")
+            hyp = result.get("text", "")
+            ref = targets[idx] if idx < len(targets) else ""
+            hyp_f.write(f"{key} {hyp}\n")
+            ref_f.write(f"{key} {ref}\n")
+
+    print(f"hyp saved to: {hyp_path}")
+    print(f"ref saved to: {ref_path}")
+
+
+if __name__ == "__main__":
+    main()

examples/industrial_data_pretraining/paraformer/lora_infer.sh

@@ -0,0 +1,32 @@
+#!/usr/bin/env bash
+# Copyright FunASR (https://github.com/alibaba-damo-academy/FunASR). All Rights Reserved.
+#  MIT License  (https://opensource.org/licenses/MIT)
+
+set -euo pipefail
+
+workspace=$(pwd)
+
+# model path and config (from training output)
+model_dir="${workspace}/examples/industrial_data_pretraining/paraformer/outputs_lora"
+init_param="${model_dir}/model.pt"
+config_path="${model_dir}"
+config_name="config.yaml"
+
+# input jsonl (must contain source/target)
+input_jsonl="${workspace}/data/list/val.jsonl"
+
+# output directory
+output_dir="${model_dir}/infer"
+
+# device
+device="cuda:0"
+
+python ${workspace}/examples/industrial_data_pretraining/paraformer/lora_infer.py \
+  --model "${model_dir}" \
+  --config-path "${config_path}" \
+  --config-name "${config_name}" \
+  --init-param "${init_param}" \
+  --input-jsonl "${input_jsonl}" \
+  --output-dir "${output_dir}" \
+  --device "${device}" \
+  --batch-size 1

funasr/bin/train.py

@@ -108,6 +108,11 @@ def main(**kwargs):
                 if k.startswith(t + ".") or k == t:
                     logging.info(f"Setting {k}.requires_grad = False")
                     p.requires_grad = False
+    lora_only = kwargs.get("lora_only", False)
+    if lora_only:
+        lora_bias = kwargs.get("lora_bias", "none")
+        logging.info("Enable LoRA-only training with bias=%s", lora_bias)
+        mark_only_lora_as_trainable(model, bias=lora_bias)
     if local_rank == 0:
         logging.info(f"{model_summary(model)}")
 

funasr/bin/train_ds.py

@@ -119,6 +119,11 @@ def main(**kwargs):
                 if k.startswith(t + ".") or k == t:
                     logging.info(f"Setting {k}.requires_grad = False")
                     p.requires_grad = False
+    lora_only = kwargs.get("lora_only", False)
+    if lora_only:
+        lora_bias = kwargs.get("lora_bias", "none")
+        logging.info("Enable LoRA-only training with bias=%s", lora_bias)
+        mark_only_lora_as_trainable(model, bias=lora_bias)
     if local_rank == 0:
         logging.info(f"{model_summary(model)}")
 

runtime/python/websocket/funASR-Service.py

@@ -0,0 +1,278 @@
+'''
+功能概述：音频分段摘要
+步骤：
+1、传入mp3,wav格式音频文件
+2、调用funasr生成分段文本内容
+3、调用大模型生成摘要
+4、输出结果
+
+'''
+from flask import Flask, request, jsonify
+import re
+from modelscope.pipelines import pipeline
+from modelscope.utils.constant import Tasks
+from werkzeug.utils import secure_filename
+import torch
+import os
+from openai import OpenAI
+from funasr import AutoModel
+
+app = Flask(__name__)
+
+# 配置上传文件夹
+UPLOAD_FOLDER = 'uploads'
+os.makedirs(UPLOAD_FOLDER, exist_ok=True)
+app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
+
+auto_model = AutoModel(model="damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch",
+                  vad_model="damo/speech_fsmn_vad_zh-cn-16k-common-pytorch",
+                  punc_model="damo/punc_ct-transformer_zh-cn-common-vocab272727-pytorch",
+                  spk_model="damo/speech_campplus_sv_zh-cn_16k-common",
+                    device="cuda:0" if torch.cuda.is_available() else "cpu"
+                       )
+# 声纹对比
+sv_pipeline = pipeline(
+                task='speaker-verification',
+                model='damo/speech_campplus_sv_zh-cn_16k-common',
+                model_revision='v1.0.0',
+                device="cuda:0" if torch.cuda.is_available() else "cpu"
+                )
+
+
+
+# 注册声纹：传音频wav文件，实现注册音频返回音频的embedding数据
+@app.route('/Register_Speaker', methods=['POST'])
+def Register_Speaker():
+    # 检查文件上传
+    if 'file' not in request.files:
+        return jsonify({"error": "No audio file provided"}), 400
+
+    file = request.files['file']
+    if file.filename == '':
+        return jsonify({"error": "Empty filename"}), 400
+
+    # 保存上传文件
+    filename = secure_filename(file.filename)
+    filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
+    file.save(filepath)
+    try:
+        # 执行音频文件Embeding
+        result = sv_pipeline([filepath], output_emb=True)
+        embedding = result['embs'][0]
+        # 删除临时文件
+        os.remove(filepath)
+        if len(embedding) == 0:
+            return jsonify({
+                "status": "error",
+                "result": "音频解析结果为空"
+            })
+        else:
+            if isinstance(embedding, np.ndarray):
+                embedding_list = embedding.tolist()
+            else:
+                embedding_list = list(embedding)
+            return jsonify({
+                "status": "success",
+                "result": embedding_list
+            })
+
+    except Exception as e:
+        # 清理文件
+        print(f"错误类型: {type(e).__name__}")
+        print(f"错误信息: {str(e)}")
+        if os.path.exists(filepath):
+            os.remove(filepath)
+        return jsonify({"error": str(e)}), 500
+
+# 会议撰写
+@app.route('/AsrCamWithIdentify', methods=['POST'])
+def speech_recognition_Timestamp_cam_identify_speakers():
+    # 检查文件上传
+    if 'file' not in request.files:
+        return jsonify({"error": "No audio file provided"}), 400
+
+    file = request.files['file']
+    if file.filename == '':
+        return jsonify({"error": "Empty filename"}), 400
+
+    # 保存上传文件
+    filename = secure_filename(file.filename)
+    filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
+    file.save(filepath)
+    # 2. 从FormData读取参数并解析JSON
+    # 读取identify_speakers（默认False）
+    identify_speakers_str = request.form.get('identify_speakers', 'false')
+    identify_speakers = json.loads(identify_speakers_str.lower())  # 转为bool
+
+    # 读取speaker_db（默认空字典）
+    speaker_db_str = request.form.get('speaker_db', '{}')
+    speaker_db = json.loads(speaker_db_str)
+
+    # 3. 验证声纹库（如果需要对比）
+    if identify_speakers:
+        if not isinstance(speaker_db, dict) or len(speaker_db) == 0:
+            return jsonify({
+                "status": "error",
+                "result": "声纹库为空或格式错误，无法进行对比"
+            }), 400
+    try:
+        # 执行语音识别
+        result = auto_model.generate(input=filepath,
+                                     batch_size_s=300,
+                                     hotword='',
+                                     )
+
+        # 处理结果
+        processed_result = process_cam_result_with_identify_speakers(result,speaker_db,filepath,identify_speakers)
+
+        os.remove(filepath)
+        if len(processed_result) == 0:
+            return jsonify({
+                "status": "error",
+                "result": "音频解析结果为空"
+            })
+        else:
+            return jsonify({
+                "status": "success",
+                "result": processed_result
+            })
+
+    except Exception as e:
+        # 清理文件
+        print(f"错误类型: {type(e).__name__}")
+        print(f"错误信息: {str(e)}")
+        if os.path.exists(filepath):
+            os.remove(filepath)
+        return jsonify({"error": str(e)}), 500
+
+import torchaudio
+def _extract_audio_segment(audio_path, start_sec, end_sec):
+    """根据时间戳提取音频片段"""
+    waveform, sample_rate = torchaudio.load(audio_path)
+
+        # 计算起止采样点
+    start_sample = int(start_sec * sample_rate)
+    end_sample = int(end_sec * sample_rate)
+
+        # 提取片段
+    segment = waveform[:, start_sample:end_sample]
+
+    # 保存为临时文件（pipeline需要文件路径）
+    temp_path = f"/tmp/temp_segment_{start_sec}_{end_sec}.wav"
+    torchaudio.save(temp_path, segment, sample_rate)
+
+    return temp_path
+# 是否使用声纹转化的结果处理
+import json
+def process_cam_result_with_identify_speakers(result,speaker_db,filepath,identify_speakers=False,threshold=0.45):
+    """处理ASR结果，返回包含时间和内容的JSON对象列表"""
+    if not isinstance(result, list) or len(result) == 0:
+        return []
+
+    data = result[0]
+    best_match = "unknown"
+    best_score = 0.0
+    # 创建JSON格式的输出
+    output = []
+    sentence_infos = data.get('sentence_info',[])
+
+    current_sentence = {
+        "spk": sentence_infos[0]["spk"],
+        "spk_name": best_match,
+        "confidence":best_score,
+        "start": sentence_infos[0]["start"],
+        "end": sentence_infos[0]["end"],
+        "text": sentence_infos[0]["text"]
+    }
+
+
+    for i in range(1, len(sentence_infos)):
+        sentence_info = sentence_infos[i]
+        # 检查合并条件：相同说话人且时间间隔≤1000ms
+        if (current_sentence["spk"] == sentence_info["spk"] and
+                sentence_info["start"] - current_sentence["end"] <= 1000):
+
+            # 合并文本内容（中文无需加空格）
+            current_sentence["text"] += sentence_info["text"]
+
+            # 更新整句结束时间
+            current_sentence["end"] = sentence_info["end"]
+
+        else:
+            # 保存合并完成的句子
+            if identify_speakers: # 提取说话人
+                segment_audio = _extract_audio_segment( #提取对应的音频
+                    filepath, current_sentence['start']/1000, current_sentence['end']/1000
+                )
+                result_b = sv_pipeline([segment_audio], output_emb=True)['embs'][0] #获取音频向量
+                os.remove(segment_audio)
+                # 遍历声纹库
+
+                for name, db_emb in speaker_db.items():
+                    # 计算余弦相似度
+                    data_list = json.loads(db_emb)
+                    arr = np.array(data_list, dtype=np.float32)
+                    similarity = 1 - cosine(result_b, arr)
+                    similarity = float(similarity)
+                    if similarity > best_score and similarity > threshold:
+                        best_score = similarity
+                        best_match = name
+            output.append({
+                "spk": current_sentence["spk"],
+                "spk_name": best_match,
+                "confidence":best_score,
+                "text": current_sentence["text"],
+                "start": current_sentence["start"],
+                "end": current_sentence["end"]
+            })
+            # 重新开始新句子
+            current_sentence = sentence_info.copy()
+            best_match = "unknown"
+            best_score = 0.0
+
+    # 保存合并完成的句子
+    if identify_speakers:  # 提取说话人
+        segment_audio = _extract_audio_segment(  # 提取对应的音频
+                    filepath, current_sentence['start']/1000, current_sentence['end']/1000
+        )
+        result_b = sv_pipeline([segment_audio], output_emb=True)['embs'][0]  # 获取音频向量
+        os.remove(segment_audio)
+        # 遍历声纹库
+
+        for name, db_emb in speaker_db.items():
+            # 计算余弦相似度
+            data_list = json.loads(db_emb)
+            arr = np.array(data_list, dtype=np.float32)
+            similarity = 1 - cosine(result_b, arr)
+            similarity = float(similarity)
+            if similarity > best_score and similarity > threshold:
+                best_score = similarity
+                best_match = name
+    output.append({
+        "spk": current_sentence["spk"],
+        "spk_name": best_match,
+        "confidence":best_score,
+        "text": current_sentence["text"],
+        "start": current_sentence["start"],
+        "end": current_sentence["end"]
+    })
+
+    return output  # 返回JSON对象列表
+
+
+from scipy.spatial.distance import cosine
+import numpy as np
+@app.route('/calculate_similarity', methods=['POST'])
+def calculate_similarity():
+    """计算两个特征向量的余弦相似度"""
+    data = request.get_json(force=True)
+    emb1 = np.array(data['emb1'], dtype=np.float32)
+    emb2 = np.array(data['emb2'], dtype=np.float32)
+    print(1-cosine(emb1,emb2))
+    return jsonify({
+                "status": "success",
+                "result": float(1 - cosine(emb1, emb2))
+            })
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=10099, debug=True)

runtime/python/websocket/funasr_wss_client.py

@@ -4,14 +4,11 @@ import time
 import websockets, ssl
 import asyncio
 
-# import threading
 import argparse
 import json
 import traceback
 from multiprocessing import Process
 
-# from funasr.fileio.datadir_writer import DatadirWriter
-
 import logging
 
 logging.basicConfig(level=logging.ERROR)
@@ -31,14 +28,22 @@ parser.add_argument(
     default="",
     help="hotword file path, one hotword perline (e.g.:阿里巴巴 20)",
 )
-parser.add_argument("--audio_in", type=str, default=None, help="audio_in")
+parser.add_argument(
+    "--audio_in",
+    type=str,
+    default=None,
+    help="音频输入路径；不传则使用麦克风（需安装 PyAudio）",
+)
 parser.add_argument("--audio_fs", type=int, default=16000, help="audio_fs")
+
+# ✅ 修复语义：默认 False；传入参数则不 sleep（用于压测）
 parser.add_argument(
     "--send_without_sleep",
     action="store_true",
-    default=True,
-    help="if audio_in is set, send_without_sleep",
+    default=False,
+    help="若设置：发送音频不按实时节奏 sleep（用于压测）",
 )
+
 parser.add_argument("--thread_num", type=int, default=1, help="thread_num")
 parser.add_argument("--words_max_print", type=int, default=10000, help="chunk")
 parser.add_argument("--output_dir", type=str, default=None, help="output_dir")
@@ -46,28 +51,83 @@ parser.add_argument("--ssl", type=int, default=1, help="1 for ssl connect, 0 for
 parser.add_argument("--use_itn", type=int, default=1, help="1 for using itn, 0 for not itn")
 parser.add_argument("--mode", type=str, default="2pass", help="offline, online, 2pass")
 
+# ✅ 验收日志输出目录（每个 meeting 单独写，避免多进程抢文件）
+parser.add_argument("--log_dir", type=str, default="./asr_logs", help="验收日志输出目录")
+parser.add_argument("--log_flush_every", type=int, default=1, help="events.jsonl 每写N行flush一次（默认1更安全）")
+
 args = parser.parse_args()
 args.chunk_size = [int(x) for x in args.chunk_size.split(",")]
 print(args)
-# voices = asyncio.Queue()
+
 from queue import Queue
+from datetime import datetime
 
 voices = Queue()
 offline_msg_done = False
 
-if args.output_dir is not None:
-    # if os.path.exists(args.output_dir):
-    #     os.remove(args.output_dir)
+# === 延迟统计相关：对每个 wav_name 记录首包/末包发送时间 & 是否已经打印过延迟 ===
+latency_first_audio_time = {}      # {wav_name: t_first_chunk_send}
+latency_last_audio_time = {}       # {wav_name: t_last_chunk_send}
+latency_first_text_printed = {}    # {wav_name: bool}
 
+
+def _iso(ts: float) -> str:
+    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+
+
+class MeetingWriter:
+    """
+    每个进程/meeting 单独写：
+      - events.jsonl：收到的每条服务端消息（在线/离线/2pass）
+      - meta.json：本次运行参数（方便复现）
+    """
+    def __init__(self, log_dir: str, meeting_id: str, flush_every: int = 1):
+        self.meeting_id = str(meeting_id)
+        self.base = os.path.join(log_dir, f"meeting_{self.meeting_id}")
+        os.makedirs(self.base, exist_ok=True)
+
+        self.fp_events = open(os.path.join(self.base, "events.jsonl"), "a", encoding="utf-8")
+        self.flush_every = max(1, int(flush_every))
+        self._cnt = 0
+
+        meta_path = os.path.join(self.base, "meta.json")
+        if not os.path.exists(meta_path):
+            with open(meta_path, "w", encoding="utf-8") as f:
+                meta = {
+                    "created_at": _iso(time.time()),
+                    "meeting_id": self.meeting_id,
+                    "args": vars(args),
+                }
+                f.write(json.dumps(meta, ensure_ascii=False, indent=2))
+
+    def write_event(self, obj: dict):
+        self.fp_events.write(json.dumps(obj, ensure_ascii=False) + "\n")
+        self._cnt += 1
+        if self._cnt % self.flush_every == 0:
+            self.fp_events.flush()
+
+    def close(self):
+        try:
+            self.fp_events.flush()
+            self.fp_events.close()
+        except Exception:
+            pass
+
+
+if args.output_dir is not None:
     if not os.path.exists(args.output_dir):
         os.makedirs(args.output_dir)
 
 
 async def record_microphone():
-    is_finished = False
-    import pyaudio
+    """从麦克风实时录音发送到服务端（一般单路测试使用）"""
+    try:
+        import pyaudio
+    except ImportError as e:
+        raise ImportError(
+            "缺少 PyAudio，麦克风推流前请先运行 `pip install pyaudio`"
+        ) from e
 
-    # print("2")
     global voices
     FORMAT = pyaudio.paInt16
     CHANNELS = 1
@@ -96,7 +156,7 @@ async def record_microphone():
                     fst_dict[" ".join(words[:-1])] = int(words[-1])
                 except ValueError:
                     print("Please checkout format of hotwords")
-            hotword_msg = json.dumps(fst_dict)
+            hotword_msg = json.dumps(fst_dict, ensure_ascii=False)
         else:
             hotword_msg = args.hotword
 
@@ -115,21 +175,19 @@ async def record_microphone():
             "is_speaking": True,
             "hotwords": hotword_msg,
             "itn": use_itn,
-        }
+        },
+        ensure_ascii=False,
     )
-    # voices.put(message)
     await websocket.send(message)
     while True:
         data = stream.read(CHUNK)
-        message = data
-        # voices.put(message)
-        await websocket.send(message)
-        await asyncio.sleep(0.005)
+        await websocket.send(data)
+        await asyncio.sleep(0.01)
 
 
 async def record_from_scp(chunk_begin, chunk_size):
-    global voices
-    is_finished = False
+    """从 wav/scp 文件读取音频分片发送，用于压测和延迟测试"""
+    global voices, latency_first_audio_time, latency_last_audio_time
     if args.audio_in.endswith(".scp"):
         f_scp = open(args.audio_in)
         wavs = f_scp.readlines()
@@ -137,25 +195,25 @@ async def record_from_scp(chunk_begin, chunk_size):
         wavs = [args.audio_in]
 
     # hotwords
-    fst_dict = {}
     hotword_msg = ""
     if args.hotword.strip() != "":
         if os.path.exists(args.hotword):
-            f_scp = open(args.hotword)
-            hot_lines = f_scp.readlines()
+            with open(args.hotword, encoding="utf-8") as f_scp:
+                hot_lines = f_scp.readlines()
+
+            hot_list = []
             for line in hot_lines:
-                words = line.strip().split(" ")
-                if len(words) < 2:
-                    print("Please checkout format of hotwords")
+                words = line.strip().split()
+                if not words:
                     continue
-                try:
-                    fst_dict[" ".join(words[:-1])] = int(words[-1])
-                except ValueError:
-                    print("Please checkout format of hotwords")
-            hotword_msg = json.dumps(fst_dict)
+                # Python AutoModel: 用逗号分隔多个热词
+                hot_list.append(words[0])
+
+            hotword_msg = ",".join(hot_list)
         else:
             hotword_msg = args.hotword
-        print(hotword_msg)
+
+    print("hotword", hotword_msg)
 
     sample_rate = args.audio_fs
     wav_format = "pcm"
@@ -164,7 +222,8 @@ async def record_from_scp(chunk_begin, chunk_size):
         use_itn = False
 
     if chunk_size > 0:
-        wavs = wavs[chunk_begin : chunk_begin + chunk_size]
+        wavs = wavs[chunk_begin: chunk_begin + chunk_size]
+
     for wav in wavs:
         wav_splits = wav.strip().split()
 
@@ -172,14 +231,13 @@ async def record_from_scp(chunk_begin, chunk_size):
         wav_path = wav_splits[1] if len(wav_splits) > 1 else wav_splits[0]
         if not len(wav_path.strip()) > 0:
             continue
+
         if wav_path.endswith(".pcm"):
             with open(wav_path, "rb") as f:
                 audio_bytes = f.read()
         elif wav_path.endswith(".wav"):
             import wave
-
             with wave.open(wav_path, "rb") as wav_file:
-                params = wav_file.getparams()
                 sample_rate = wav_file.getframerate()
                 frames = wav_file.readframes(wav_file.getnframes())
                 audio_bytes = bytes(frames)
@@ -190,9 +248,8 @@ async def record_from_scp(chunk_begin, chunk_size):
 
         stride = int(60 * args.chunk_size[1] / args.chunk_interval / 1000 * sample_rate * 2)
         chunk_num = (len(audio_bytes) - 1) // stride + 1
-        # print(stride)
 
-        # send first time
+        # send first control message
         message = json.dumps(
             {
                 "mode": args.mode,
@@ -206,36 +263,46 @@ async def record_from_scp(chunk_begin, chunk_size):
                 "is_speaking": True,
                 "hotwords": hotword_msg,
                 "itn": use_itn,
-            }
+            },
+            ensure_ascii=False,
         )
 
-        # voices.put(message)
         await websocket.send(message)
         is_speaking = True
-        for i in range(chunk_num):
 
+        # 初始化该 wav 的统计状态
+        latency_first_audio_time[wav_name] = None
+        latency_last_audio_time[wav_name] = None
+        latency_first_text_printed[wav_name] = False
+
+        for i in range(chunk_num):
             beg = i * stride
-            data = audio_bytes[beg : beg + stride]
-            message = data
-            # voices.put(message)
-            await websocket.send(message)
+            data = audio_bytes[beg: beg + stride]
+
+            now_ts = time.time()
+            if latency_first_audio_time[wav_name] is None:
+                latency_first_audio_time[wav_name] = now_ts
+            latency_last_audio_time[wav_name] = now_ts
+
+            await websocket.send(data)
+
             if i == chunk_num - 1:
                 is_speaking = False
-                message = json.dumps({"is_speaking": is_speaking})
-                # voices.put(message)
-                await websocket.send(message)
-
-            sleep_duration = (
-                0.001
-                if args.mode == "offline"
-                else 60 * args.chunk_size[1] / args.chunk_interval / 1000
-            )
+                await websocket.send(json.dumps({"is_speaking": is_speaking}, ensure_ascii=False))
 
+            # ✅ sleep策略：默认按实时节奏；若开启 send_without_sleep 则几乎不 sleep（压测）
+            if args.send_without_sleep:
+                sleep_duration = 0.001
+            else:
+                sleep_duration = (
+                    0.001
+                    if args.mode == "offline"
+                    else 60 * args.chunk_size[1] / args.chunk_interval / 1000
+                )
             await asyncio.sleep(sleep_duration)
 
     if not args.mode == "offline":
         await asyncio.sleep(2)
-    # offline model need to wait for message recved
 
     if args.mode == "offline":
         global offline_msg_done
@@ -247,30 +314,87 @@ async def record_from_scp(chunk_begin, chunk_size):
     await websocket.close()
 
 
-async def message(id):
+async def message(id, writer: MeetingWriter):
+    """接收服务端识别结果 + 打印实时文本 + 打印延迟 + 写验收日志(events.jsonl)"""
+    import websockets
     global websocket, voices, offline_msg_done
+    global latency_first_audio_time, latency_last_audio_time, latency_first_text_printed
+
+    multi_mode = args.thread_num > 1  # 多路并发时，打印风格更简洁
     text_print = ""
     text_print_2pass_online = ""
     text_print_2pass_offline = ""
+
     if args.output_dir is not None:
         ibest_writer = open(
             os.path.join(args.output_dir, "text.{}".format(id)), "a", encoding="utf-8"
         )
     else:
         ibest_writer = None
+
     try:
         while True:
-
             meg = await websocket.recv()
             meg = json.loads(meg)
-            wav_name = meg.get("wav_name", "demo")
-            text = meg["text"]
-            timestamp = ""
-            offline_msg_done = meg.get("is_final", False)
-            if "timestamp" in meg:
-                timestamp = meg["timestamp"]
 
-            if ibest_writer is not None:
+            wav_name = meg.get("wav_name", "demo")
+            text = meg.get("text", "")
+            mode = meg.get("mode", "")
+            spk_name = meg.get("spk_name", "")
+            spk_score = meg.get("spk_score", None)
+            now_ts = time.time()
+
+            # === 延迟统计：仅在首条 online/2pass-online 文本时计算并打印一次 ===
+            latency_last_ms = None
+            latency_first_ms = None
+            if text and mode in ("online", "2pass-online"):
+                if not latency_first_text_printed.get(wav_name, False):
+                    t_last = latency_last_audio_time.get(wav_name, None)
+                    t_first = latency_first_audio_time.get(wav_name, None)
+                    latency_last_ms = (now_ts - t_last) * 1000.0 if t_last is not None else None
+                    latency_first_ms = (now_ts - t_first) * 1000.0 if t_first is not None else None
+
+                    latency_first_text_printed[wav_name] = True
+
+                    if multi_mode:
+                        parts = [f"[MEETING {id}][LATENCY] wav={wav_name}, mode={mode}"]
+                        if latency_last_ms is not None:
+                            parts.append(f"from_last_chunk={latency_last_ms:.1f} ms")
+                        if latency_first_ms is not None:
+                            parts.append(f"from_first_chunk={latency_first_ms:.1f} ms")
+                        print(" ".join(parts))
+                    else:
+                        print(
+                            f"[LATENCY] wav={wav_name}, mode={mode}, "
+                            f"from_last_chunk={(latency_last_ms or 0):.1f} ms, "
+                            f"from_first_chunk={(latency_first_ms or 0):.1f} ms"
+                        )
+
+            timestamp = meg.get("timestamp", "")
+            offline_msg_done = meg.get("is_final", False)
+
+            # ✅ 验收友好：每条消息落 events.jsonl（便于后处理）
+            event = {
+                "ts": _iso(now_ts),
+                "recv_ts": now_ts,
+                "meeting_id": str(id),
+                "wav_name": wav_name,
+                "mode": mode,
+                "is_final": bool(meg.get("is_final", False)),
+                "text": text,
+                "spk_name": spk_name,
+                "spk_score": spk_score,
+                "latency_first_ms": latency_first_ms,
+                "latency_last_ms": latency_last_ms,
+                "server_timestamp": meg.get("timestamp", None),
+                "sentence_info": meg.get("sentence_info", None),
+                "punc_array": meg.get("punc_array", None),
+            }
+            if writer is not None:
+                writer.write_event(event)
+
+            # 保存到 output_dir（保留你原来的逻辑）
+            if ibest_writer is not None and text:
                 if timestamp != "":
                     text_write_line = "{}\t{}\t{}\n".format(wav_name, text, timestamp)
                 else:
@@ -279,22 +403,44 @@ async def message(id):
 
             if "mode" not in meg:
                 continue
+
+            # ===== 多路并发输出风格：只打印精简行 =====
+            if multi_mode:
+                if mode in ("offline", "2pass-offline") and text:
+                    spk_name2 = meg.get("spk_name", "unknown")
+                    spk_score2 = meg.get("spk_score", 0.0)
+                    print(
+                        f"[MEETING {id}][FINAL][{wav_name}] "
+                        f"spk={spk_name2}({float(spk_score2):.3f}) text=\"{text}\""
+                    )
+                    if timestamp:
+                        print(f"[MEETING {id}][TIMESTAMP][{wav_name}] {timestamp}")
+                continue
+
+            # ===== 单路模式输出：保留原滚动体验 =====
             if meg["mode"] == "online":
                 text_print += "{}".format(text)
-                text_print = text_print[-args.words_max_print :]
-                os.system("clear")
-                print("\rpid" + str(id) + ": " + text_print)
+                text_print = text_print[-args.words_max_print:]
+                print("pid" + str(id) + ": " + text_print)
+
             elif meg["mode"] == "offline":
                 if timestamp != "":
                     text_print += "{} timestamp: {}".format(text, timestamp)
                 else:
                     text_print += "{}".format(text)
 
-                # text_print = text_print[-args.words_max_print:]
-                # os.system('clear')
-                print("\rpid" + str(id) + ": " + wav_name + ": " + text_print)
+                spk_info = ""
+                if spk_name:
+                    if spk_score is not None:
+                        spk_info = f" [spk={spk_name} score={float(spk_score):.3f}]"
+                    else:
+                        spk_info = f" [spk={spk_name}]"
+
+                print("pid" + str(id) + ": " + wav_name + ": " + text_print + spk_info)
                 offline_msg_done = True
+
             else:
+                # 2pass 模式
                 if meg["mode"] == "2pass-online":
                     text_print_2pass_online += "{}".format(text)
                     text_print = text_print_2pass_offline + text_print_2pass_online
@@ -302,15 +448,27 @@ async def message(id):
                     text_print_2pass_online = ""
                     text_print = text_print_2pass_offline + "{}".format(text)
                     text_print_2pass_offline += "{}".format(text)
-                text_print = text_print[-args.words_max_print :]
-                os.system("clear")
-                print("\rpid" + str(id) + ": " + text_print)
-                # offline_msg_done=True
 
+                    if spk_name:
+                        if spk_score is not None:
+                            text_print += f" [spk={spk_name} score={float(spk_score):.3f}]"
+                        else:
+                            text_print += f" [spk={spk_name}]"
+
+                text_print = text_print[-args.words_max_print:]
+                print("pid" + str(id) + ": " + text_print)
+
+    except websockets.exceptions.ConnectionClosedOK:
+        print(f"[MEETING {id}] connection closed normally")
     except Exception as e:
-        print("Exception:", e)
-        # traceback.print_exc()
-        # await websocket.close()
+        print(f"[MEETING {id}] Exception:", e)
+    finally:
+        try:
+            if ibest_writer is not None:
+                ibest_writer.flush()
+                ibest_writer.close()
+        except Exception:
+            pass
 
 
 async def ws_client(id, chunk_begin, chunk_size):
@@ -322,30 +480,39 @@ async def ws_client(id, chunk_begin, chunk_size):
     for i in range(chunk_begin, chunk_begin + chunk_size):
         offline_msg_done = False
         voices = Queue()
+
         if args.ssl == 1:
-            ssl_context = ssl.SSLContext()
+            ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
             ssl_context.check_hostname = False
             ssl_context.verify_mode = ssl.CERT_NONE
             uri = "wss://{}:{}".format(args.host, args.port)
         else:
             uri = "ws://{}:{}".format(args.host, args.port)
             ssl_context = None
+
         print("connect to", uri)
+
         async with websockets.connect(
             uri, subprotocols=["binary"], ping_interval=None, ssl=ssl_context
         ) as websocket:
-            if args.audio_in is not None:
-                task = asyncio.create_task(record_from_scp(i, 1))
-            else:
-                task = asyncio.create_task(record_microphone())
-            task3 = asyncio.create_task(message(str(id) + "_" + str(i)))  # processid+fileid
-            await asyncio.gather(task, task3)
-    exit(0)
+            meeting_tag = f"{id}_{i}"
+            writer = MeetingWriter(args.log_dir, meeting_id=meeting_tag, flush_every=args.log_flush_every)
+            try:
+                if args.audio_in is not None:
+                    task = asyncio.create_task(record_from_scp(i, 1))
+                else:
+                    task = asyncio.create_task(record_microphone())
+                task3 = asyncio.create_task(message(str(id) + "_" + str(i), writer))  # processid+fileid
+                await asyncio.gather(task, task3)
+            finally:
+                writer.close()
+
+    return
 
 
 def one_thread(id, chunk_begin, chunk_size):
-    asyncio.get_event_loop().run_until_complete(ws_client(id, chunk_begin, chunk_size))
-    asyncio.get_event_loop().run_forever()
+    # ✅ 子进程里用 asyncio.run 更稳
+    asyncio.run(ws_client(id, chunk_begin, chunk_size))
 
 
 if __name__ == "__main__":
@@ -356,17 +523,12 @@ if __name__ == "__main__":
         p.join()
         print("end")
     else:
-        # calculate the number of wavs for each preocess
+        # calculate the number of wavs for each process
         if args.audio_in.endswith(".scp"):
             f_scp = open(args.audio_in)
             wavs = f_scp.readlines()
         else:
             wavs = [args.audio_in]
-        for wav in wavs:
-            wav_splits = wav.strip().split()
-            wav_name = wav_splits[0] if len(wav_splits) > 1 else "demo"
-            wav_path = wav_splits[1] if len(wav_splits) > 1 else wav_splits[0]
-            audio_type = os.path.splitext(wav_path)[-1].lower()
 
         total_len = len(wavs)
         if total_len >= args.thread_num:
@@ -383,7 +545,7 @@ if __name__ == "__main__":
             if remain_wavs > 0:
                 now_chunk_size = chunk_size + 1
                 remain_wavs = remain_wavs - 1
-            # process i handle wavs at chunk_begin and size of now_chunk_size
+
             p = Process(target=one_thread, args=(i, chunk_begin, now_chunk_size))
             chunk_begin = chunk_begin + now_chunk_size
             p.start()

runtime/python/websocket/funasr_wss_server.py

@@ -2,25 +2,54 @@ import asyncio
 import json
 import websockets
 import time
-import logging
-import tracemalloc
 import numpy as np
 import argparse
 import ssl
+import os
+import wave
+import functools
+from concurrent.futures import ThreadPoolExecutor
+from scipy.spatial.distance import cosine
+
+import torch  # 保留不影响
+
+
+def to_python(obj):
+    """递归地把 numpy / torch 等类型转成纯 Python，可 JSON 序列化。"""
+    try:
+        import numpy as np  # noqa
+        import torch  # noqa
+    except Exception:
+        np = None
+        torch = None
+
+    if np is not None and isinstance(obj, np.generic):
+        return obj.item()
+    if np is not None and isinstance(obj, np.ndarray):
+        return obj.tolist()
+    if torch is not None and isinstance(obj, torch.Tensor):
+        return obj.cpu().tolist()
+
+    if isinstance(obj, dict):
+        return {k: to_python(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [to_python(v) for v in obj]
+
+    return obj
 
 
 parser = argparse.ArgumentParser()
-parser.add_argument(
-    "--host", type=str, default="0.0.0.0", required=False, help="host ip, localhost, 0.0.0.0"
-)
+parser.add_argument("--host", type=str, default="0.0.0.0", required=False, help="host ip")
 parser.add_argument("--port", type=int, default=10095, required=False, help="grpc server port")
+
 parser.add_argument(
     "--asr_model",
     type=str,
-    default="iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch",
+    default="iic/speech_paraformer-large-contextual_asr_nat-zh-cn-16k-common-vocab8404",
     help="model from modelscope",
 )
 parser.add_argument("--asr_model_revision", type=str, default="v2.0.4", help="")
+
 parser.add_argument(
     "--asr_model_online",
     type=str,
@@ -28,6 +57,7 @@ parser.add_argument(
     help="model from modelscope",
 )
 parser.add_argument("--asr_model_online_revision", type=str, default="v2.0.4", help="")
+
 parser.add_argument(
     "--vad_model",
     type=str,
@@ -35,6 +65,7 @@ parser.add_argument(
     help="model from modelscope",
 )
 parser.add_argument("--vad_model_revision", type=str, default="v2.0.4", help="")
+
 parser.add_argument(
     "--punc_model",
     type=str,
@@ -42,9 +73,11 @@ parser.add_argument(
     help="model from modelscope",
 )
 parser.add_argument("--punc_model_revision", type=str, default="v2.0.4", help="")
+
 parser.add_argument("--ngpu", type=int, default=1, help="0 for cpu, 1 for gpu")
 parser.add_argument("--device", type=str, default="cuda", help="cuda, cpu")
 parser.add_argument("--ncpu", type=int, default=4, help="cpu cores")
+
 parser.add_argument(
     "--certfile",
     type=str,
@@ -52,7 +85,6 @@ parser.add_argument(
     required=False,
     help="certfile for ssl",
 )
-
 parser.add_argument(
     "--keyfile",
     type=str,
@@ -60,25 +92,159 @@ parser.add_argument(
     required=False,
     help="keyfile for ssl",
 )
+
+# ====== 保存 2pass 离线阶段送入 ASR 的音频片段（排查 VAD 切分）======
+parser.add_argument(
+    "--save_offline_segments",
+    action="store_true",
+    help="Save each offline (2pass) audio segment sent to offline ASR as wav for debugging VAD split.",
+)
+parser.add_argument(
+    "--save_offline_segments_dir",
+    type=str,
+    default="./offline_segments",
+    help="Directory to save offline wav segments when --save_offline_segments is enabled.",
+)
+
+# ====== 并发控制：核心新增 ======
+parser.add_argument(
+    "--worker_threads",
+    type=int,
+    default=max(4, (os.cpu_count() or 4)),
+    help="ThreadPoolExecutor max_workers. Used to offload blocking inference so event loop won't be blocked.",
+)
+parser.add_argument("--concurrent_vad", type=int, default=4, help="Max concurrent VAD generate() calls.")
+parser.add_argument("--concurrent_asr_online", type=int, default=4, help="Max concurrent streaming ASR generate() calls.")
+parser.add_argument("--concurrent_asr_offline", type=int, default=2, help="Max concurrent offline ASR generate() calls.")
+parser.add_argument("--concurrent_punc", type=int, default=1, help="Max concurrent punctuation generate() calls.")
+parser.add_argument("--concurrent_sv", type=int, default=1, help="Max concurrent speaker verification generate() calls.")
+parser.add_argument(
+    "--speaker_db_reload_sec",
+    type=int,
+    default=5,
+    help="Reload speaker_db.json at most once every N seconds (avoid frequent disk IO).",
+)
+
 args = parser.parse_args()
 
-
 websocket_users = set()
+SPEAKER_DB_PATH = os.path.join(os.path.dirname(__file__), "speaker_db.json")
+
+
+def _ensure_dir(p: str):
+    try:
+        os.makedirs(p, exist_ok=True)
+    except Exception:
+        pass
+
+
+def _pcm_duration_ms(pcm_bytes: bytes, fs: int, ch: int = 1, sampwidth: int = 2) -> int:
+    """根据 fs/ch/sampwidth 计算 PCM 时长，避免写死 16k -> 32 bytes/ms。"""
+    if not pcm_bytes:
+        return 0
+    bytes_per_ms = (fs * ch * sampwidth) / 1000.0
+    if bytes_per_ms <= 0:
+        return 0
+    return int(len(pcm_bytes) / bytes_per_ms)
+
+
+def _safe_int(v, default):
+    try:
+        return int(v)
+    except Exception:
+        return default
+
+
+# ========= speaker db：加缓存，避免每段都读盘 =========
+_SPEAKER_DB_CACHE = {}
+_SPEAKER_DB_CACHE_TS = 0.0
+
+
+def _load_speaker_db_sync():
+    if not os.path.exists(SPEAKER_DB_PATH):
+        return {}
+    try:
+        with open(SPEAKER_DB_PATH, "r", encoding="utf-8") as f:
+            data = json.load(f)
+            return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def get_speaker_db_cached(now_ts: float, reload_sec: int):
+    global _SPEAKER_DB_CACHE, _SPEAKER_DB_CACHE_TS
+    if (now_ts - _SPEAKER_DB_CACHE_TS) >= max(1, int(reload_sec)):
+        _SPEAKER_DB_CACHE = _load_speaker_db_sync()
+        _SPEAKER_DB_CACHE_TS = now_ts
+    return _SPEAKER_DB_CACHE or {}
+
+
+def _save_wav_sync(out_path: str, audio_bytes: bytes, fs: int, ch: int, sampwidth: int):
+    with wave.open(out_path, "wb") as wf:
+        wf.setnchannels(ch)
+        wf.setsampwidth(sampwidth)
+        wf.setframerate(fs)
+        wf.writeframes(audio_bytes)
+
+
+def save_offline_wav_segment_sync(websocket, audio_bytes: bytes, reason: str = "offline"):
+    """
+    保存离线阶段送入 ASR 的音频片段，方便人工试听排查 VAD 切分是否正确。
+    约定：audio_bytes 为 单声道 PCM16 little-endian（默认 16k）。
+    （注意：这是同步函数，外层会放线程池执行）
+    """
+    if not getattr(websocket, "save_offline_segments", False):
+        return
+    if "2pass" not in (getattr(websocket, "mode", "") or ""):
+        return
+    if not audio_bytes:
+        return
+
+    fs = int(getattr(websocket, "audio_fs", 16000) or 16000)
+    ch = 1
+    sampwidth = 2  # int16
+
+    # int16 对齐
+    if len(audio_bytes) % 2 == 1:
+        audio_bytes = audio_bytes[:-1]
+        if not audio_bytes:
+            return
+
+    seg_idx = int(getattr(websocket, "offline_seg_idx", 0))
+    websocket.offline_seg_idx = seg_idx + 1
+
+    duration_ms = _pcm_duration_ms(audio_bytes, fs=fs, ch=ch, sampwidth=sampwidth)
+
+    base_dir = getattr(websocket, "offline_save_dir", args.save_offline_segments_dir)
+    _ensure_dir(base_dir)
+
+    wav_name = (getattr(websocket, "wav_name", "microphone") or "microphone").replace("/", "_")
+    ts = int(time.time() * 1000)
+    fname = f"{wav_name}_{ts}_seg{seg_idx:04d}_{reason}_{duration_ms}ms.wav"
+    out_path = os.path.join(base_dir, fname)
+
+    try:
+        _save_wav_sync(out_path, audio_bytes, fs=fs, ch=ch, sampwidth=sampwidth)
+        print(f"[SAVE_OFFLINE_SEG] {out_path} ({duration_ms} ms, {len(audio_bytes)} bytes)")
+    except Exception as e:
+        print(f"[SAVE_OFFLINE_SEG] failed: {e}")
+
 
 print("model loading")
-from funasr import AutoModel
+from funasr import AutoModel  # noqa
 
-# asr
+# ====== 离线 ASR ======
 model_asr = AutoModel(
-    model=args.asr_model,
-    model_revision=args.asr_model_revision,
+    model="paraformer-zh",
+    model_revision="v2.0.4",
     ngpu=args.ngpu,
     ncpu=args.ncpu,
     device=args.device,
     disable_pbar=True,
     disable_log=True,
 )
-# asr
+
+# streaming asr
 model_asr_streaming = AutoModel(
     model=args.asr_model_online,
     model_revision=args.asr_model_online_revision,
@@ -88,6 +254,7 @@ model_asr_streaming = AutoModel(
     disable_pbar=True,
     disable_log=True,
 )
+
 # vad
 model_vad = AutoModel(
     model=args.vad_model,
@@ -97,9 +264,9 @@ model_vad = AutoModel(
     device=args.device,
     disable_pbar=True,
     disable_log=True,
-    # chunk_size=60,
 )
 
+# punc
 if args.punc_model != "":
     model_punc = AutoModel(
         model=args.punc_model,
@@ -113,8 +280,45 @@ if args.punc_model != "":
 else:
     model_punc = None
 
+# sv
+model_sv = AutoModel(
+    model="iic/speech_campplus_sv_zh-cn_16k-common",
+    ngpu=args.ngpu,
+    device=args.device,
+    disable_pbar=True,
+    disable_log=True,
+)
 
-print("model loaded! only support one client at the same time now!!!!")
+print("model loaded! (now supports multi-client with non-blocking inference)")
+
+
+# ====== 线程池 + 并发阈值（核心）======
+EXECUTOR = ThreadPoolExecutor(max_workers=int(args.worker_threads))
+
+SEM_VAD = asyncio.Semaphore(max(1, int(args.concurrent_vad)))
+SEM_ASR_ONLINE = asyncio.Semaphore(max(1, int(args.concurrent_asr_online)))
+SEM_ASR_OFFLINE = asyncio.Semaphore(max(1, int(args.concurrent_asr_offline)))
+SEM_PUNC = asyncio.Semaphore(max(1, int(args.concurrent_punc)))
+SEM_SV = asyncio.Semaphore(max(1, int(args.concurrent_sv)))
+SEM_WAV = asyncio.Semaphore(max(1, 4))  # 保存 wav 一般不需要太大
+
+
+async def run_blocking(fn, *a, sem: asyncio.Semaphore | None = None, **kw):
+    """
+    把阻塞函数丢线程池执行，避免卡 event loop。
+    sem 用于限流（避免 GPU / 模型被打爆）。
+    """
+    loop = asyncio.get_running_loop()
+    call = functools.partial(fn, *a, **kw)
+    if sem is None:
+        return await loop.run_in_executor(EXECUTOR, call)
+    async with sem:
+        return await loop.run_in_executor(EXECUTOR, call)
+
+
+def _generate_sync(model, audio_or_text, status_dict):
+    # 注意：status_dict 里包含 cache，会被 generate 更新
+    return model.generate(input=audio_or_text, **status_dict)
 
 
 async def ws_reset(websocket):
@@ -130,130 +334,212 @@ async def ws_reset(websocket):
 
 
 async def clear_websocket():
-    for websocket in websocket_users:
+    for websocket in list(websocket_users):
         await ws_reset(websocket)
     websocket_users.clear()
 
 
-async def ws_serve(websocket, path):
+async def ws_serve(websocket, path=None):
+    # websockets 新版本不会传 path，这里做兼容
+    if path is None:
+        path = getattr(websocket, "path", None)
     frames = []
     frames_asr = []
     frames_asr_online = []
     global websocket_users
-    # await clear_websocket()
     websocket_users.add(websocket)
-    websocket.status_dict_asr = {}
+
+    websocket.status_dict_asr = {}  # hotword 等
     websocket.status_dict_asr_online = {"cache": {}, "is_final": False}
     websocket.status_dict_vad = {"cache": {}, "is_final": False}
     websocket.status_dict_punc = {"cache": {}}
+
     websocket.chunk_interval = 10
     websocket.vad_pre_idx = 0
     speech_start = False
     speech_end_i = -1
+
     websocket.wav_name = "microphone"
     websocket.mode = "2pass"
+    websocket.is_speaking = True  # ✅ 默认初始化，避免 AttributeError
+
+    # 保存离线片段
+    websocket.audio_fs = 16000
+    websocket.offline_seg_idx = 0
+    websocket.save_offline_segments = bool(args.save_offline_segments)
+    websocket.offline_save_dir = args.save_offline_segments_dir
+    if websocket.save_offline_segments:
+        _ensure_dir(websocket.offline_save_dir)
+        print(f"[SAVE_OFFLINE_SEG] enabled, dir={websocket.offline_save_dir}")
+
     print("new user connected", flush=True)
 
     try:
         async for message in websocket:
+            # ========== 1) 先处理“文本配置消息” ==========
             if isinstance(message, str):
-                messagejson = json.loads(message)
+                try:
+                    messagejson = json.loads(message)
+                except Exception as e:
+                    print("bad json message:", e, message[:200])
+                    continue
+
+                print("=============messagejson============", messagejson)
 
                 if "is_speaking" in messagejson:
-                    websocket.is_speaking = messagejson["is_speaking"]
-                    websocket.status_dict_asr_online["is_final"] = not websocket.is_speaking
+                    websocket.is_speaking = bool(messagejson["is_speaking"])
+                    websocket.status_dict_asr_online["is_final"] = (not websocket.is_speaking)
+
                 if "chunk_interval" in messagejson:
-                    websocket.chunk_interval = messagejson["chunk_interval"]
+                    websocket.chunk_interval = _safe_int(
+                        messagejson["chunk_interval"], websocket.chunk_interval
+                    )
+
                 if "wav_name" in messagejson:
-                    websocket.wav_name = messagejson.get("wav_name")
+                    websocket.wav_name = messagejson.get("wav_name") or websocket.wav_name
+
                 if "chunk_size" in messagejson:
                     chunk_size = messagejson["chunk_size"]
                     if isinstance(chunk_size, str):
-                        chunk_size = chunk_size.split(",")
+                        chunk_size = [x.strip() for x in chunk_size.split(",") if x.strip()]
                     websocket.status_dict_asr_online["chunk_size"] = [int(x) for x in chunk_size]
+
                 if "encoder_chunk_look_back" in messagejson:
                     websocket.status_dict_asr_online["encoder_chunk_look_back"] = messagejson[
                         "encoder_chunk_look_back"
                     ]
+
                 if "decoder_chunk_look_back" in messagejson:
                     websocket.status_dict_asr_online["decoder_chunk_look_back"] = messagejson[
                         "decoder_chunk_look_back"
                     ]
+
                 if "hotwords" in messagejson:
-                    websocket.status_dict_asr["hotword"] = messagejson["hotwords"]
+                    hotword_data = messagejson["hotwords"]
+                    websocket.status_dict_asr["hotword"] = hotword_data
+                    websocket.status_dict_asr_online["hotword"] = hotword_data
+                    print(f"热词已更新: {hotword_data}")
+
                 if "mode" in messagejson:
-                    websocket.mode = messagejson["mode"]
+                    websocket.mode = messagejson["mode"] or websocket.mode
 
-            websocket.status_dict_vad["chunk_size"] = int(
-                websocket.status_dict_asr_online["chunk_size"][1] * 60 / websocket.chunk_interval
-            )
-            if len(frames_asr_online) > 0 or len(frames_asr) >= 0 or not isinstance(message, str):
-                if not isinstance(message, str):
-                    frames.append(message)
-                    duration_ms = len(message) // 32
-                    websocket.vad_pre_idx += duration_ms
+                if "audio_fs" in messagejson:
+                    websocket.audio_fs = _safe_int(messagejson["audio_fs"], 16000)
 
-                    # asr online
-                    frames_asr_online.append(message)
-                    websocket.status_dict_asr_online["is_final"] = speech_end_i != -1
-                    if (
-                        len(frames_asr_online) % websocket.chunk_interval == 0
-                        or websocket.status_dict_asr_online["is_final"]
-                    ):
-                        if websocket.mode == "2pass" or websocket.mode == "online":
-                            audio_in = b"".join(frames_asr_online)
-                            try:
-                                await async_asr_online(websocket, audio_in)
-                            except:
-                                print(f"error in asr streaming, {websocket.status_dict_asr_online}")
-                        frames_asr_online = []
-                    if speech_start:
-                        frames_asr.append(message)
-                    # vad online
+                continue
+
+            # ========== 2) 处理“二进制音频消息” ==========
+            if "chunk_size" not in websocket.status_dict_asr_online:
+                print("[WARN] chunk_size not set yet, skip audio frame (send config first).")
+                continue
+
+            try:
+                websocket.status_dict_vad["chunk_size"] = int(
+                    websocket.status_dict_asr_online["chunk_size"][1] * 60 / websocket.chunk_interval
+                )
+            except Exception as e:
+                print("[WARN] set vad chunk_size failed:", e)
+                continue
+
+            pcm = message
+            frames.append(pcm)
+
+            duration_ms = _pcm_duration_ms(pcm, fs=websocket.audio_fs, ch=1, sampwidth=2)
+            websocket.vad_pre_idx += duration_ms
+
+            # online asr
+            frames_asr_online.append(pcm)
+            websocket.status_dict_asr_online["is_final"] = (speech_end_i != -1)
+
+            if (len(frames_asr_online) % websocket.chunk_interval == 0) or websocket.status_dict_asr_online["is_final"]:
+                if websocket.mode in ("2pass", "online"):
+                    audio_in = b"".join(frames_asr_online)
                     try:
-                        speech_start_i, speech_end_i = await async_vad(websocket, message)
-                    except:
-                        print("error in vad")
-                    if speech_start_i != -1:
-                        speech_start = True
-                        beg_bias = (websocket.vad_pre_idx - speech_start_i) // duration_ms
-                        frames_pre = frames[-beg_bias:]
-                        frames_asr = []
-                        frames_asr.extend(frames_pre)
-                # asr punc offline
-                if speech_end_i != -1 or not websocket.is_speaking:
-                    # print("vad end point")
-                    if websocket.mode == "2pass" or websocket.mode == "offline":
-                        audio_in = b"".join(frames_asr)
+                        await async_asr_online(websocket, audio_in)
+                    except Exception:
+                        print(f"error in asr streaming, {websocket.status_dict_asr_online}")
+                frames_asr_online = []
+
+            if speech_start:
+                frames_asr.append(pcm)
+
+            # vad online
+            try:
+                speech_start_i, speech_end_i = await async_vad(websocket, pcm)
+            except Exception as e:
+                print("error in vad:", e)
+                speech_start_i, speech_end_i = -1, -1
+
+            if speech_start_i != -1:
+                speech_start = True
+                if duration_ms > 0:
+                    beg_bias = (websocket.vad_pre_idx - speech_start_i) // duration_ms
+                else:
+                    beg_bias = 0
+                frames_pre = frames[-beg_bias:] if beg_bias > 0 else []
+                frames_asr = []
+                frames_asr.extend(frames_pre)
+
+            # ========== 3) 2pass：离线阶段触发点 ==========
+            if (speech_end_i != -1) or (not websocket.is_speaking):
+                if websocket.mode in ("2pass", "offline"):
+                    audio_in = b"".join(frames_asr)
+                    reason = "vad_end" if speech_end_i != -1 else "not_speaking"
+
+                    # 保存 wav：放线程池，避免磁盘 IO 卡 loop
+                    if websocket.save_offline_segments and audio_in:
                         try:
-                            await async_asr(websocket, audio_in)
-                        except:
-                            print("error in asr offline")
-                    frames_asr = []
-                    speech_start = False
-                    frames_asr_online = []
-                    websocket.status_dict_asr_online["cache"] = {}
-                    if not websocket.is_speaking:
-                        websocket.vad_pre_idx = 0
-                        frames = []
-                        websocket.status_dict_vad["cache"] = {}
-                    else:
-                        frames = frames[-20:]
+                            await run_blocking(
+                                save_offline_wav_segment_sync,
+                                websocket,
+                                audio_in,
+                                reason,
+                                sem=SEM_WAV,
+                            )
+                        except Exception as e:
+                            print("[SAVE_OFFLINE_SEG] async failed:", e)
+
+                    try:
+                        await async_asr(websocket, audio_in)
+                    except Exception as e:
+                        print("error in asr offline:", e)
+
+                frames_asr = []
+                speech_start = False
+                frames_asr_online = []
+                websocket.status_dict_asr_online["cache"] = {}
+
+                if not websocket.is_speaking:
+                    websocket.vad_pre_idx = 0
+                    frames = []
+                    websocket.status_dict_vad["cache"] = {}
+                    speech_end_i = -1
+                else:
+                    frames = frames[-20:]
 
     except websockets.ConnectionClosed:
         print("ConnectionClosed...", websocket_users, flush=True)
         await ws_reset(websocket)
-        websocket_users.remove(websocket)
+        if websocket in websocket_users:
+            websocket_users.remove(websocket)
     except websockets.InvalidState:
         print("InvalidState...")
     except Exception as e:
         print("Exception:", e)
+        try:
+            await ws_reset(websocket)
+        except Exception:
+            pass
+        if websocket in websocket_users:
+            websocket_users.remove(websocket)
 
 
-async def async_vad(websocket, audio_in):
+# ===================== 推理：全部改为“线程池 + 限流” =====================
 
-    segments_result = model_vad.generate(input=audio_in, **websocket.status_dict_vad)[0]["value"]
-    # print(segments_result)
+async def async_vad(websocket, audio_in: bytes):
+    # model_vad.generate 是阻塞的，必须 offload
+    out = await run_blocking(_generate_sync, model_vad, audio_in, websocket.status_dict_vad, sem=SEM_VAD)
+    segments_result = out[0].get("value", [])
 
     speech_start = -1
     speech_end = -1
@@ -267,79 +553,197 @@ async def async_vad(websocket, audio_in):
     return speech_start, speech_end
 
 
-async def async_asr(websocket, audio_in):
-    if len(audio_in) > 0:
-        # print(len(audio_in))
-        rec_result = model_asr.generate(input=audio_in, **websocket.status_dict_asr)[0]
-        # print("offline_asr, ", rec_result)
-        if model_punc is not None and len(rec_result["text"]) > 0:
-            # print("offline, before punc", rec_result, "cache", websocket.status_dict_punc)
-            rec_result = model_punc.generate(
-                input=rec_result["text"], **websocket.status_dict_punc
-            )[0]
-            # print("offline, after punc", rec_result)
-        if len(rec_result["text"]) > 0:
-            # print("offline", rec_result)
-            mode = "2pass-offline" if "2pass" in websocket.mode else websocket.mode
-            message = json.dumps(
-                {
-                    "mode": mode,
-                    "text": rec_result["text"],
-                    "wav_name": websocket.wav_name,
-                    "is_final": websocket.is_speaking,
-                }
-            )
-            await websocket.send(message)
+def _sv_and_match_sync(audio_in: bytes, reload_sec: int):
+    """
+    同步执行：SV embedding + speaker_db 匹配
+    返回 (spk_name, best_score)
+    """
+    spk_name = "unknown"
+    best_score = 0.0
 
-    else:
-        mode = "2pass-offline" if "2pass" in websocket.mode else websocket.mode
-        message = json.dumps(
-            {
-                "mode": mode,
-                "text": "",
-                "wav_name": websocket.wav_name,
-                "is_final": websocket.is_speaking,
-            }
+    sv_out = model_sv.generate(input=audio_in, embedding=True)[0]
+    embedding = sv_out["spk_embedding"][0].cpu().numpy()
+
+    now_ts = time.time()
+    local_speaker_db = get_speaker_db_cached(now_ts, reload_sec=reload_sec)
+    if local_speaker_db:
+        for name, ref_embedding in local_speaker_db.items():
+            if ref_embedding is None:
+                continue
+            arr = np.array(ref_embedding, dtype=np.float32)
+            similarity = 1.0 - cosine(embedding, arr)
+            print("sv similarity with {}: {}".format(name, similarity))
+            if similarity > best_score and similarity > 0.2:
+                best_score = similarity
+                spk_name = name
+
+    return spk_name, float(best_score)
+
+
+async def async_asr(websocket, audio_in: bytes):
+    mode = "2pass-offline" if "2pass" in (websocket.mode or "") else websocket.mode
+
+    if len(audio_in) <= 0:
+        message = {
+            "mode": mode,
+            "text": "",
+            "wav_name": websocket.wav_name,
+            "is_final": True,
+        }
+        await websocket.send(json.dumps(message, ensure_ascii=False))
+        return
+
+    # 1) ASR（阻塞，线程池执行）
+    rec_result_list = await run_blocking(
+        _generate_sync,
+        model_asr,
+        audio_in,
+        websocket.status_dict_asr,
+        sem=SEM_ASR_OFFLINE,
+    )
+    rec_result = rec_result_list[0]
+
+    print("offline_asr, raw:", rec_result)
+    print("offline_asr, keys:", rec_result.keys())
+
+    text = rec_result.get("text", "")
+    timestamp = rec_result.get("timestamp", None)
+    sentence_info = rec_result.get("sentence_info", None)
+
+    # 2) 声纹识别（阻塞，线程池执行）
+    spk_name = "unknown"
+    best_score = 0.0
+    try:
+        spk_name, best_score = await run_blocking(
+            _sv_and_match_sync,
+            audio_in,
+            int(args.speaker_db_reload_sec),
+            sem=SEM_SV,
         )
-        await websocket.send(message)    
+    except Exception as e:
+        print(f"声纹识别失败: {e}")
 
-async def async_asr_online(websocket, audio_in):
-    if len(audio_in) > 0:
-        # print(websocket.status_dict_asr_online.get("is_final", False))
-        rec_result = model_asr_streaming.generate(
-            input=audio_in, **websocket.status_dict_asr_online
-        )[0]
-        # print("online, ", rec_result)
-        if websocket.mode == "2pass" and websocket.status_dict_asr_online.get("is_final", False):
-            return
-            #     websocket.status_dict_asr_online["cache"] = dict()
-        if len(rec_result["text"]):
-            mode = "2pass-online" if "2pass" in websocket.mode else websocket.mode
-            message = json.dumps(
-                {
-                    "mode": mode,
-                    "text": rec_result["text"],
-                    "wav_name": websocket.wav_name,
-                    "is_final": websocket.is_speaking,
-                }
+    # 3) 标点（阻塞，线程池执行）
+    punc_array = None
+    if model_punc is not None and len(text) > 0:
+        try:
+            # punc 只对文本处理
+            punc_out = await run_blocking(
+                _generate_sync,
+                model_punc,
+                text,
+                websocket.status_dict_punc,
+                sem=SEM_PUNC,
             )
-            await websocket.send(message)
+            punc_result = punc_out[0]
+            print("offline, after punc", punc_result)
+
+            if "text" in punc_result and punc_result["text"]:
+                text = punc_result["text"]
+            if "punc_array" in punc_result:
+                punc_array = punc_result["punc_array"]
+        except Exception as e:
+            print("punc failed:", e)
+
+    # 4) 构造最终 message
+    if len(text) > 0:
+        print("======offline final text:", text)
+        message = {
+            "mode": mode,
+            "spk_name": spk_name,
+            "spk_score": float(best_score),
+            "text": text,
+            "wav_name": websocket.wav_name,
+            "is_final": True,
+        }
+        if timestamp is not None:
+            message["timestamp"] = to_python(timestamp)
+        if sentence_info is not None:
+            message["sentence_info"] = to_python(sentence_info)
+        if punc_array is not None:
+            message["punc_array"] = to_python(punc_array)
+
+        try:
+            await websocket.send(json.dumps(message, ensure_ascii=False))
+        except Exception as e:
+            print("send json failed:", e)
+            print("message types:", {k: type(v) for k, v in message.items()})
+    else:
+        message = {
+            "mode": mode,
+            "spk_name": spk_name,
+            "spk_score": float(best_score),
+            "text": "",
+            "wav_name": websocket.wav_name,
+            "is_final": True,
+        }
+        await websocket.send(json.dumps(message, ensure_ascii=False))
 
 
-if len(args.certfile) > 0:
-    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+async def async_asr_online(websocket, audio_in: bytes):
+    if len(audio_in) <= 0:
+        return
 
-    # Generate with Lets Encrypt, copied to this location, chown to current user and 400 permissions
-    ssl_cert = args.certfile
-    ssl_key = args.keyfile
-
-    ssl_context.load_cert_chain(ssl_cert, keyfile=ssl_key)
-    start_server = websockets.serve(
-        ws_serve, args.host, args.port, subprotocols=["binary"], ping_interval=None, ssl=ssl_context
+    # streaming generate 也是阻塞：线程池执行
+    rec_out = await run_blocking(
+        _generate_sync,
+        model_asr_streaming,
+        audio_in,
+        websocket.status_dict_asr_online,
+        sem=SEM_ASR_ONLINE,
     )
-else:
-    start_server = websockets.serve(
-        ws_serve, args.host, args.port, subprotocols=["binary"], ping_interval=None
-    )
-asyncio.get_event_loop().run_until_complete(start_server)
-asyncio.get_event_loop().run_forever()
+    rec_result = rec_out[0]
+    print("online, ", rec_result)
+
+    # 2pass：online 只要 partial，不发 final（final 交给 offline）
+    if websocket.mode == "2pass" and websocket.status_dict_asr_online.get("is_final", False):
+        return
+
+    if rec_result.get("text"):
+        mode = "2pass-online" if "2pass" in (websocket.mode or "") else websocket.mode
+        message = {
+            "mode": mode,
+            "text": rec_result["text"],
+            "wav_name": websocket.wav_name,
+            "is_final": bool(
+                websocket.status_dict_asr_online.get("is_final", False) or (not websocket.is_speaking)
+            ),
+        }
+        await websocket.send(json.dumps(message, ensure_ascii=False))
+
+
+# ===================== 启动服务 =====================
+
+async def main():
+    if len(args.certfile) > 0:
+        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ssl_context.load_cert_chain(args.certfile, keyfile=args.keyfile)
+        server = await websockets.serve(
+            ws_serve,
+            args.host,
+            args.port,
+            subprotocols=["binary"],
+            ping_interval=None,
+            ssl=ssl_context,
+        )
+    else:
+        server = await websockets.serve(
+            ws_serve,
+            args.host,
+            args.port,
+            subprotocols=["binary"],
+            ping_interval=None,
+        )
+
+    print(f"WS server started at ws(s)://{args.host}:{args.port}")
+    await server.wait_closed()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    finally:
+        try:
+            EXECUTOR.shutdown(wait=False, cancel_futures=True)
+        except Exception:
+            pass