识别结果
+正在读取服务配置…
+ +From aeaa72fc63fe0a911551db835bd631b89f281867 Mon Sep 17 00:00:00 2001 From: Bifang <915779419@qq.com> Date: Tue, 22 Sep 2026 14:35:42 +0800 Subject: [PATCH] first commit --- .env.example | 50 ++ .gitignore | 7 + README.md | 151 ++++ model_manifest.json | 84 ++ pyproject.toml | 20 + realtime_websocket/FIXES.md | 87 ++ realtime_websocket/README.md | 135 +++ realtime_websocket/auxiliary_service.py | 161 ++++ realtime_websocket/model_service.py | 120 +++ realtime_websocket/requirements.txt | 2 + realtime_websocket/server.py | 778 +++++++++++++++++ realtime_websocket/speaker_assembler.py | 186 ++++ realtime_websocket/static/app.js | 743 ++++++++++++++++ realtime_websocket/static/index.html | 119 +++ realtime_websocket/static/style.css | 805 ++++++++++++++++++ realtime_websocket/tests/test_frontend.cjs | 102 +++ .../tests/test_model_service.py | 32 + realtime_websocket/tests/test_server.py | 388 +++++++++ .../tests/test_speaker_assembler.py | 121 +++ requirements-auxiliary.txt | 6 + requirements-deploy.txt | 7 + scripts/__init__.py | 1 + scripts/auxiliary_server.py | 614 +++++++++++++ scripts/download_models.py | 230 +++++ scripts/model_manifest.py | 54 ++ scripts/serve.py | 216 +++++ tests/test_auxiliary_server.py | 169 ++++ tests/test_model_manifest.py | 73 ++ tests/test_serve.py | 43 + 29 files changed, 5504 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 model_manifest.json create mode 100644 pyproject.toml create mode 100644 realtime_websocket/FIXES.md create mode 100644 realtime_websocket/README.md create mode 100644 realtime_websocket/auxiliary_service.py create mode 100644 realtime_websocket/model_service.py create mode 100644 realtime_websocket/requirements.txt create mode 100644 realtime_websocket/server.py create mode 100644 realtime_websocket/speaker_assembler.py create mode 100644 realtime_websocket/static/app.js create mode 100644 realtime_websocket/static/index.html create mode 100644 realtime_websocket/static/style.css create mode 100644 realtime_websocket/tests/test_frontend.cjs create mode 100644 realtime_websocket/tests/test_model_service.py create mode 100644 realtime_websocket/tests/test_server.py create mode 100644 realtime_websocket/tests/test_speaker_assembler.py create mode 100644 requirements-auxiliary.txt create mode 100644 requirements-deploy.txt create mode 100644 scripts/__init__.py create mode 100644 scripts/auxiliary_server.py create mode 100644 scripts/download_models.py create mode 100644 scripts/model_manifest.py create mode 100644 scripts/serve.py create mode 100644 tests/test_auxiliary_server.py create mode 100644 tests/test_model_manifest.py create mode 100644 tests/test_serve.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..5901c32 --- /dev/null +++ b/.env.example @@ -0,0 +1,50 @@ +# 选择已经下载到 demo/models 目录中的 ASR 模型;必须与实际模型目录保持一致。 +# 默认使用轻量的 0.6B 模型。如下载的是 1.7B,请改为 1.7b,不能混用。 +QWEN3_ASR_MODEL=0.6b +# 本地模型根目录。下载脚本、VLLM 启动脚本和辅助服务都会从这里查找模型。 +MODEL_DIR=D:/github-project/ASR/Qwen-Asr/demo/models + +# VLLM 宿主机服务绑定地址;0.0.0.0 表示允许服务器网卡接收外部请求。 +VLLM_HOST=0.0.0.0 +# VLLM 服务端口;启动器、健康检查和前端连接地址必须使用同一个端口。 +VLLM_PORT=9950 +# VLLM 可执行文件名称;新版环境通常为 vllm,启动器会自动执行 vllm serve。 +VLLM_EXECUTABLE=vllm +# 启动成功提示中显示的地址,只影响日志和使用说明,不改变实际监听地址。 +VLLM_DISPLAY_HOST=127.0.0.1 +# 启动轮询使用的地址;如果 VLLM 部署在本机,通常保持 127.0.0.1 即可。 +VLLM_PROBE_HOST=127.0.0.1 +# VLLM 启动阶段最多检查多少次健康状态,超过次数仍未就绪则启动失败。 +VLLM_STARTUP_CHECK_LOOPS=60 +# 两次健康检查之间的等待秒数;模型加载较慢时可以适当增大检查次数。 +VLLM_STARTUP_CHECK_INTERVAL_SECONDS=2 +# VLLM 使用的显存比例。显存还要留给 VAD、聚类和声纹辅助模型时,建议预留余量。 +VLLM_GPU_MEMORY_UTILIZATION=0.3 +# VLLM 的最大上下文长度;数值越大通常占用越多显存,请结合显卡容量调整。 +VLLM_MAX_MODEL_LEN=16384 +# VLLM 同时处理的最大序列数;实时单路验证可保持默认值,多路并发时再调大。 +VLLM_MAX_NUM_SEQS=16 +# 张量并行 GPU 数量;单卡部署为 1,多卡部署时填写参与并行的 GPU 数量。 +VLLM_TENSOR_PARALLEL_SIZE=1 +# 是否启用 eager 模式。true 通常更容易启动和排查,false 可能获得更高性能。 +VLLM_ENFORCE_EAGER=true + +# GB10 需要使用 CUDA 13 工具链中的 ptxas;serve.py 会自动加载该配置并传给 VLLM。 +TRITON_PTXAS_PATH=/usr/local/cuda/bin/ptxas + +# 可选:为外部客户端设置稳定的公开模型名称。留空时默认使用完整 ModelScope ID。 +# VLLM_SERVED_MODEL_NAME=Qwen/Qwen3-ASR-0.6B + +# 辅助模型服务地址;它独立加载 VAD、CAM++ 聚类和声纹模型,WebSocket +# 服务只通过 HTTP 调用,不会把这些 GPU 模型重复加载到 WebSocket 进程。 +AUXILIARY_SERVICE_URL=http://127.0.0.1:8010 +# WebSocket 调用已独立部署的 vLLM;跨服务器时改成模型服务器的实际地址。 +MODEL_SERVICE_URL=http://127.0.0.1:9950/v1 +# 辅助服务监听的 GPU;单卡服务器保持 cuda:0,多卡时可改成指定卡号。 +AUXILIARY_DEVICE=cuda:0 +# 仅保留兼容旧配置;VAD/CAM++ 核心模型缺失时始终拒绝启动,不会静默降级。 +# 可选 diarization/aligner 缺失不会阻断实时服务。 +AUXILIARY_ALLOW_MISSING=false +# 启动时强制加载 VAD + CAM++ speaker_verification;完整 diarization 首次调用时按需加载。 +# 如确实需要启动时额外预加载,可追加:vad,speaker_verification,diarization +AUXILIARY_PRELOAD_KINDS=vad,speaker_verification diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..23c14f8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.venv/ +__pycache__/ +*.py[cod] + +.code-review-graph/ +models/ +.env \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..0140208 --- /dev/null +++ b/README.md @@ -0,0 +1,151 @@ +# Qwen3-ASR VLLM 独立部署项目 + +本目录是后续实时 ASR 功能验证使用的独立模型服务项目。 + +它不导入、不启动、也不调用仓库根目录下原项目的 `app/` 代码。模型下载、VLLM 启动、配置和服务验证都在本目录内完成。后续验证 demo 只需要调用这里提供的 VLLM OpenAI 兼容接口。 + +## 当前下载范围 + +默认下载一个 ASR 模型和独立辅助模型运行服务所需的全部模型资产: + +```text +Qwen/Qwen3-ASR-0.6B +``` + +ASR 如需使用大模型,可显式选择 1.7B: + +```text +Qwen/Qwen3-ASR-1.7B +``` + +辅助资产包括 VAD、CAM++ 分离、配置声纹、实时声纹、CAM++ Transformer 和 Qwen3 ForcedAligner。它们不会由 `qwen-asr-serve` 启动,而是由独立 Python 辅助模型服务加载。 + +不会下载另一个未选择的 ASR 模型;辅助模型资产会随默认部署包下载,供独立 Python 运行服务预加载。 + +## 1. 下载模型 + +模型直接下载到宿主机的 `demo/models`。ASR 由 VLLM 启动,辅助模型由独立 Python 运行服务启动。 + +```powershell +cd D:\github-project\ASR\Qwen-Asr\demo +python -m venv .venv +.\.venv\Scripts\Activate.ps1 +pip install -r requirements-download.txt +python scripts\download_models.py +``` + +上述命令会下载默认 `0.6B` ASR 以及全部辅助模型,并在下载完成后把 CAM++ 配置中的依赖模型 ID 改为 `demo/models` 下的本地路径,保证辅助服务可以离线启动。只下载 ASR 时使用: + +```powershell +python scripts\download_models.py --skip-auxiliary +``` + +只下载辅助模型时使用: + +```powershell +python scripts\download_models.py --auxiliary-only +``` + +选择 1.7B: + +```powershell +python scripts\download_models.py --model 1.7b +``` + +检查模型是否完整但不下载: + +```powershell +python scripts\download_models.py --check-only +``` + +ModelScope 下载也可以通过环境变量调整缓存目录: + +```powershell +$env:MODELSCOPE_CACHE = 'D:\modelscope-cache' +python scripts\download_models.py +``` + +## 2. 宿主机启动 VLLM 服务 + +需要宿主机具备与 VLLM 兼容的 Python、CUDA 和 NVIDIA 驱动环境。安装部署依赖: + +```bash +python -m pip install -r requirements-deploy.txt +``` + +先复制并按服务器实际路径修改 `.env`,启动器会自动读取该文件: + +```bash +cp .env.example .env +``` + +默认启动 `Qwen/Qwen3-ASR-0.6B`,监听地址为 `0.0.0.0:9950`: + +```bash +python scripts/serve.py +``` + +模型和启动检查循环可以通过命令行或环境变量传入;端口统一在 `scripts/serve.py` 的 `SERVER_PORT` 变量中维护: + +```bash +QWEN3_ASR_MODEL=0.6b VLLM_STARTUP_CHECK_LOOPS=120 \ +VLLM_STARTUP_CHECK_INTERVAL_SECONDS=2 python scripts/serve.py +``` + +如果使用 `1.7B`,下载和启动必须指定同一个模型: + +```powershell +python scripts\download_models.py --model 1.7b +python -m scripts.serve --model 1.7b +``` + +启动器会按 `VLLM_STARTUP_CHECK_LOOPS` 次数轮询 `/health`,每次间隔由 `VLLM_STARTUP_CHECK_INTERVAL_SECONDS` 指定。服务端口、健康检查端口和就绪提示统一使用 `scripts/serve.py` 中的 `SERVER_PORT`。 + +服务启动后可检查: + +```bash +curl "http://${VLLM_DISPLAY_HOST:-127.0.0.1}:9950/health" +curl "http://${VLLM_DISPLAY_HOST:-127.0.0.1}:9950/v1/models" +``` + +## 3. 调用转写接口 + +VLLM 服务提供 OpenAI 兼容的音频转写接口: + +```bash +curl "http://${VLLM_DISPLAY_HOST:-127.0.0.1}:9950/v1/audio/transcriptions" \ + -H "Authorization: Bearer EMPTY" \ + -F "file=@./audio/sample.wav" \ + -F "model=Qwen/Qwen3-ASR-0.6B" +``` + +## 4. 启动辅助模型服务 + +另开一个终端,在同一台服务器启动 VAD、CAM++ 和声纹模型运行服务: + +```powershell +cd D:\github-project\ASR\Qwen-Asr\demo +pip install -r requirements-auxiliary.txt +python scripts\auxiliary_server.py +``` + +辅助服务默认监听 `0.0.0.0:8010`。实时链路启动时严格加载 VAD 和 CAM++ `speaker_verification` 声纹模型,用于每个 turn 的特征提取与在线聚类;完整 CAM++ 分离、Transformer 和 ForcedAligner 不阻断核心服务,完整分离模型会在调用 `/v1/diarization` 时按需加载。检查状态: + +```bash +curl http://127.0.0.1:8010/health +``` + +WebSocket demo 默认连接 `9950` 的 ASR VLLM,辅助服务使用 `8010`。`/health` 的 `ready` 要求 `vad_ready` 与 `speaker_embedding_ready` 同时为 true;完整 diarization 资产缺失不会影响实时 `/v1/speaker/resolve`。 + +实时链路中的职责是:WebSocket 用 RMS 帧门控快速检测停顿;辅助服务用 FunASR VAD 提供 `/v1/vad`,并加载 CAM++ `speaker_verification` 提取 turn embedding,再由服务端在线聚类。`speech_campplus_speaker-diarization_common` 是完整音频分离接口的额外 pipeline,不是实时 turn 聚类的唯一入口。 + +也可以使用多模态 Chat Completions 接口,后续实时验证项目将以此服务边界为准。 + +## 项目边界 + +- `scripts/download_models.py`:下载选定 ASR 和全部辅助模型资产。 +- `scripts/serve.py`:读取 `.env`,解析模型选择、宿主机参数并启动新版 `vllm serve`。 +- `requirements-deploy.txt`:安装宿主机部署所需的官方 Qwen3-ASR VLLM 依赖。 +- `tests/`:只验证本项目自己的模型清单和选择逻辑,不依赖原项目。 + +模型服务就绪后,新的实时 ASR demo 放在同级 `demo` 项目中继续开发,但不得通过 Python import 或 HTTP/WebSocket 调用原项目服务。 diff --git a/model_manifest.json b/model_manifest.json new file mode 100644 index 0000000..a96f58e --- /dev/null +++ b/model_manifest.json @@ -0,0 +1,84 @@ +{ + "default_model": "Qwen/Qwen3-ASR-0.6B", + "models": { + "Qwen/Qwen3-ASR-1.7B": { + "alias": "1.7b", + "directory": "Qwen/Qwen3-ASR-1.7B", + "description": "Qwen3-ASR 1.7B,GPU 可选大模型" + }, + "Qwen/Qwen3-ASR-0.6B": { + "alias": "0.6b", + "directory": "Qwen/Qwen3-ASR-0.6B", + "description": "Qwen3-ASR 0.6B,GPU 默认轻量模型" + } + }, + "auxiliary_models": { + "damo/speech_fsmn_vad_zh-cn-16k-common-pytorch": { + "alias": "vad", + "directory": "damo/speech_fsmn_vad_zh-cn-16k-common-pytorch", + "kind": "vad", + "description": "FunASR FSMN VAD", + "revision": "v2.0.2", + "required_files": ["configuration.json", "config.yaml", "model.pb"], + "min_total_size_bytes": 1000000 + }, + "iic/speech_campplus_speaker-diarization_common": { + "alias": "diarization", + "directory": "iic/speech_campplus_speaker-diarization_common", + "kind": "diarization", + "description": "CAM++ speaker diarization", + "required_files": [ + "configuration.json", + "config.yaml", + "onnx/asd.onnx", + "onnx/face_recog_ir101.onnx", + "onnx/fqa.onnx", + "onnx/version-RFB-320.onnx" + ], + "min_total_size_bytes": 50000000 + }, + "iic/speech_campplus_sv_zh-cn_16k-common": { + "alias": "speaker", + "directory": "iic/speech_campplus_sv_zh-cn_16k-common", + "kind": "speaker_verification", + "description": "Configured CAM++ speaker verification", + "revision": "v2.0.2", + "required_files": ["configuration.json", "config.yaml", "campplus_cn_common.bin"], + "min_total_size_bytes": 10000000 + }, + "iic/speech_eres2netv2_sv_zh-cn_16k-common": { + "alias": "realtime-speaker", + "directory": "iic/speech_eres2netv2_sv_zh-cn_16k-common", + "kind": "realtime_speaker_verification", + "description": "Realtime speaker verification", + "required_files": ["configuration.json"], + "any_files": ["*"], + "min_total_size_bytes": 10000000 + }, + "damo/speech_campplus_sv_zh-cn_16k-common": { + "alias": "speaker-damo", + "directory": "damo/speech_campplus_sv_zh-cn_16k-common", + "kind": "speaker_verification", + "description": "CAM++ speaker verification dependency", + "required_files": ["configuration.json", "config.yaml", "campplus_cn_common.bin"], + "min_total_size_bytes": 10000000 + }, + "damo/speech_campplus-transformer_scl_zh-cn_16k-common": { + "alias": "speaker-transformer", + "directory": "damo/speech_campplus-transformer_scl_zh-cn_16k-common", + "kind": "speaker_transformer", + "description": "CAM++ Transformer dependency", + "required_files": ["configuration.json", "campplus_cn_encoder.pt", "transformer_backend.pt"], + "min_total_size_bytes": 10000000 + }, + "Qwen/Qwen3-ForcedAligner-0.6B": { + "alias": "forced-aligner", + "directory": "Qwen/Qwen3-ForcedAligner-0.6B", + "kind": "forced_aligner", + "description": "Qwen3 word-level forced aligner", + "required_files": ["config.json"], + "any_files": ["*.safetensors", "*.bin"], + "min_total_size_bytes": 500000000 + } + } +} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..30ed923 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,20 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "qwen3-asr-vllm-deployment" +version = "0.1.0" +description = "Standalone Qwen3-ASR model downloader and VLLM deployment" +requires-python = ">=3.10,<3.14" +dependencies = [ + "modelscope==1.34.0", + "qwen-asr[vllm]==0.0.6", +] + +[project.scripts] +qwen3-asr-download = "scripts.download_models:main" +qwen3-asr-serve = "scripts.serve:main" + +[tool.setuptools] +packages = ["scripts"] diff --git a/realtime_websocket/FIXES.md b/realtime_websocket/FIXES.md new file mode 100644 index 0000000..f2d61d6 --- /dev/null +++ b/realtime_websocket/FIXES.md @@ -0,0 +1,87 @@ +# Demo 说话人链路修复与验收 + +本次仅修改 `demo/`。参考原 `app/services/qwen3_websocket_asr.py` 的前滚音频、同句异步说话人更新和结束提交方式,保留独立 vLLM ASR + 辅助声纹服务 + WebSocket 三个进程。当前机器没有显卡,验证采用模拟模型,真实声纹准确率和模型加载状态需在部署机器上验收。 + +## 已确认的代码问题 + +1. 页面只处理 `sentences`,忽略已实现的 `display_state`,因此后端合并结果没有展示。未知片段还会临时附在上一位已确认说话人的气泡里。 +2. 页面点击停止后 5 秒强制断开,但辅助 HTTP 请求的超时是 45 秒,可能截掉迟到的声纹更新。 +3. 无 embedding、低置信度、协议字段不完整等情况被静默丢弃,用户无法区分等待、短音频、服务错误和证据拒绝。 +4. 旧声纹长度检查把句尾 800ms 静音一起计算,可能让短插话通过长度要求;WAV 固定去掉 44 字节也会破坏带额外元数据的输入。 +5. 仅 vLLM 启动器加载 `demo/.env`,另两个进程没有加载;`0.6b` 下载别名也可能被直接当作公开模型名发送。 + +这些是代码中可复现的问题,不能据此断言部署中的声纹模型一定已经正常加载。新版本将具体原因直接显示在片段气泡和日志里。 + +## 修复后的行为 + +- partial/final 和声纹更新继续使用同一 `sentence_id`。页面显示完整聚合快照,过时的 `revision` 不覆盖新快照。 +- 仅合并相邻且身份可信的片段。未知短插话独立显示;A→B→A 保留顺序。不同实名或实名与弱匿名身份不会因为簇 ID 相同而合并。 +- 每条声纹证据必须来自当前片段;同步、异步两种入口都拒绝 `short_attach` / `embedding_attach`。不复制上一段 embedding。声纹向量拒绝零向量、NaN、Infinity 和多样本矩阵。 +- WebSocket 以有效有声帧判断长度。少于 800ms 的语音保持 pending;800ms~1.6s 也必须独立提取特征,不能直接继承前一位身份。该长度门槛属于保守保护,不能保证短样本识别准确率。 +- 保留 200ms 前滚,提交时去除尾部静音。增量解析 WAV 的 RIFF/fmt/data/JUNK 等头,拒绝非 16kHz 单声道 PCM16 文件。 +- `stop/eof` 返回 `draining`,排空 ASR 和声纹队列后才发送最终快照及 `end`。声纹失败不阻断 ASR。abort、断线、异常和正常结束均清理聚类会话。 +- ASR worker 异常会立即发送 `error`,不会一直等待客户端停止。 + +## 如何判断卡在哪里 + +页面顶部显示已确认数量;未匹配到说话人的气泡统一显示“未知说话人”,详细等待或失败原因通过标签悬停提示和原始日志查看。原始日志保留 `speaker_status`、`speaker_reason`、`speaker_strategy` 和置信度。 + +| `speaker_status` | 含义与排查方向 | +|---|---| +| `waiting_final` | 讲话仍在进行,等待静音或最大时长切段 | +| `queued` / `processing` | 文本已完成,声纹正在排队或推理 | +| `confirmed` | 当前片段声纹已确认,应该显示说话人标签 | +| `insufficient_audio` | 有效语音过短,不继承上一位;使用较长发言复测 | +| `service_unavailable` / `service_error` | 未配置、无法访问或模型推理失败;检查辅助服务及完整错误 | +| `no_embedding` | 服务没有产生可用特征 | +| `evidence_rejected` | 缺少 fresh/confirmed 证据、置信度不足或使用了继承策略 | +| `disabled` | 本次未开启说话人分离 | + +健康检查为 `http://辅助服务器:8010/health`。新版本有 `speaker_protocol_version: 2`,重点检查 `speaker_embedding_ready` 与 `speaker_embedding_model`。若没有版本字段,检查是否重启了更新后的辅助进程。HTTP 健康检查不执行真实声纹推理,不能代替音频验收。 + +辅助服务目前返回匿名的“说话人 1、2……”;demo 没有接入原应用的声纹注册库,因此不会自动识别人员实名。vLLM 仅输出转写文本,声纹标签由 `scripts/auxiliary_server.py` 负责。 + +模型职责要区分:`iic/speech_campplus_sv_zh-cn_16k-common` 是实时 turn 的 CAM++ embedding 模型,必须加载;`iic/speech_campplus_speaker-diarization_common` 是完整音频分离 pipeline,包含额外的 change locator/VAD 依赖,当前实时 WebSocket 不在启动阶段调用它。WebSocket 自身仍用轻量 RMS 帧门控切句,辅助服务的 FunASR VAD 对外提供 `/v1/vad`,并供完整 diarization 依赖使用;因此启动辅助服务是为了 CAM++ 声纹和聚类,不能把整段 diarization 的加载失败误认为 vLLM 失败。 + +## 部署后操作 + +在部署机更新这些文件后,已有 vLLM 可继续运行。重新安装增补的 `python-dotenv` 依赖,并重启辅助服务及 WebSocket。辅助服务启动时强制依赖 VAD + CAM++ `speaker_verification`;完整 CAM++ diarization 不再阻断实时启动;以下命令都从 `demo/` 目录执行: + +```text +pip install -r requirements-auxiliary.txt +pip install -r realtime_asr_optimization_demo/requirements.txt +python scripts/auxiliary_server.py +``` + +如果仍提示核心模型缺失或加载失败,日志会列出模型 ID、实际查找路径、状态和底层异常。先执行 `python scripts/download_models.py --auxiliary-only`,或设置 `.env` 的 `MODEL_DIR` 指向同时包含 `damo/speech_fsmn_vad_zh-cn-16k-common-pytorch` 与 `iic/speech_campplus_sv_zh-cn_16k-common` 的目录。不要用 `AUXILIARY_ALLOW_MISSING=true` 掩盖 VAD/CAM++ 核心模型缺失;该选项只适合临时查看可选模型状态。 + +另一个终端执行: + +```text +python realtime_asr_optimization_demo/server.py --no-browser +``` + +三个进程现在均读取 `demo/.env`,系统环境变量优先。确认 `MODEL_SERVICE_URL` 指向现有 vLLM 的 `/v1`,`AUXILIARY_SERVICE_URL` 指向辅助服务。`QWEN3_ASR_MODEL` 支持部署清单别名,设置 `VLLM_SERVED_MODEL_NAME` 时优先使用该公开名称。更新后刷新浏览器;脚本 URL 已更新版本号。 + +`127.0.0.1` 指各 Python 服务运行的机器,不是浏览器所在机器。跨服务器部署时填写对应服务器 IP。浏览器麦克风访问远程页面需要安全上下文(HTTPS);本机 localhost 可用于测试。 + +## 验收顺序 + +1. 单人讲话 2~4 秒后停顿:先出现文本,随后同句变为“说话人 1”。 +2. 同一人再次讲话并停顿:确认后相邻块应合并;取消页面“合并相邻”可对照物理片段。 +3. A→B→A,各说 2 秒以上:应保持三个时间顺序块。标签准确率需要真实声纹模型验证。 +4. A 后 B 说一个很短的“嗯”:应显示“有效语音不足”,不能进入 A 的气泡。 +5. 辅助服务关闭时测试:ASR 仍完成,片段明确显示服务错误。 +6. 讲话中点击停止:等待最终结果,不能在五秒时丢失说话人更新。 + +无显卡回归命令: + +```text +cd demo +python -m unittest discover -s tests -v +cd realtime_asr_optimization_demo +python -m unittest discover -s tests -v +node --test tests/test_frontend.cjs +``` + +这些测试覆盖状态机、模拟 HTTP/WebSocket、延迟更新、前端脚本和有效向量校验,不执行模型下载或 GPU 推理。当前 VLLM 适配器仍是 HTTP 累积窗口 partial,`native_partial_supported=false`;本次没有把 HTTP 接口包装成原生增量模型状态。单个内部片段中无停顿的多人换话或重叠讲话仍需真实模型和更细粒度切段验证。 diff --git a/realtime_websocket/README.md b/realtime_websocket/README.md new file mode 100644 index 0000000..479647d --- /dev/null +++ b/realtime_websocket/README.md @@ -0,0 +1,135 @@ +# Realtime ASR WebSocket Optimization Demo + +这是一个独立的实时 ASR WebSocket 验证项目,放在模型部署项目 `demo` 下,但不导入、不启动、也不调用仓库根目录的原始 `app/`。 + +## 验证目标 + +- 浏览器麦克风或按实时速度发送的 PCM/WAV 音频通过本项目 WebSocket 发送。 +- 同一 `sentence_id` 的 partial、final 和后续更新覆盖同一条 raw segment。 +- raw segment 与前端 display block 分离,避免把物理切段直接等同于展示换行。 +- 小于 1.6 秒且带有 `short_attach` / `embedding_attach` 策略的实名结果降级为 pending。 +- 不把 embedding 字段写入 demo 状态池。 +- 只有相邻且身份可信的 segment 才合并;A→B→A 保持时间顺序。 +- 记录 partial 首次延迟、final 延迟、partial 修订次数和服务端返回时间范围。 +- 每个实时 turn 单独提交声纹特征,由辅助服务维护本 WebSocket session 的在线聚类中心。 + +本次修复、诊断状态和部署验收步骤见 [FIXES.md](FIXES.md)。ASR 继续使用已部署的独立 vLLM;已有 vLLM 服务时无需重复启动或下载模型。 + +当前 VLLM 端点提供的是同步 OpenAI 音频转写接口,没有暴露原生 +`create_stream/feed_stream/finish_stream`。因此本项目仍然是真实 WebSocket +音频流:麦克风 PCM 到达后立即进入 VAD,按窗口调用 VLLM 生成 partial;它不会 +等整段音频结束。`native_partial_supported=false` 只表示模型 HTTP 接口本身 +不是原生 ASR stream,不伪造不存在的能力。 + +## 启动 + +需要启动两个模型服务和一个 WebSocket 页面:VLLM 只负责 ASR,辅助服务负责 +VAD、CAM++ 声纹模型及在线聚类,WebSocket 只做音频流编排,不导入原项目代码。 + +先在 `demo` 目录下载 ASR 和辅助模型,并启动 VLLM: + +```powershell +cd D:\github-project\ASR\Qwen-Asr\demo +python scripts\download_models.py +python scripts\serve.py +``` + +另一个终端启动辅助模型服务: + +```powershell +cd D:\github-project\ASR\Qwen-Asr\demo +pip install -r requirements-auxiliary.txt +python scripts\auxiliary_server.py +``` + +另开一个终端启动 WebSocket 页面: + +```powershell +cd D:\github-project\ASR\Qwen-Asr\demo\realtime_asr_optimization_demo +python -m venv .venv +.\.venv\Scripts\Activate.ps1 +pip install -r requirements.txt +python server.py --no-browser +``` + +页面服务默认监听 `0.0.0.0:8082`,端口在 `server.py` 顶部的 `WEB_PORT` 内部变量中维护。 +VLLM 默认地址为 `http://127.0.0.1:9950/v1`,辅助服务默认地址为 +`http://127.0.0.1:8010`。可通过环境变量切换到远程服务: + +```powershell +$env:MODEL_SERVICE_URL = 'http://127.0.0.1:9950/v1' +$env:AUXILIARY_SERVICE_URL = 'http://127.0.0.1:8010' +python server.py --no-browser +``` + +如果服务器端口已通过 VS Code Remote/端口转发映射到本机,保持上述两个 +`127.0.0.1` 地址即可:本地 WebSocket 只负责编排,ASR 和 VAD/CAM++ 推理仍在 +服务器 GPU 服务中完成。先访问 `http://127.0.0.1:9950/v1/models` 与 +`http://127.0.0.1:8010/health`,分别确认 VLLM 模型和辅助模型服务可达且 `ready=true`。 + +服务器部署时使用 `--no-browser`,然后在客户端浏览器访问 `http://服务器IP:8082`。如需让启动日志显示服务器域名或 IP,可设置 `WEB_DISPLAY_HOST`;它只影响提示文本,不改变监听地址。 + +辅助服务启动后可用 `http://服务器IP:8010/health` 检查模型状态。WebSocket +收到聚类服务错误时仍会继续输出 ASR,但对应片段会显示“未知说话人”;详细的 +`speaker_reason` 可将鼠标悬停在标签上查看,事件日志仍会显示 `speaker_warning`, +便于区分“模型未归类”和“ASR 失败”。 + +声纹服务的实时路径必须通过 ModelScope pipeline 的公开接口提取 embedding: +`pipeline([wav_path], output_emb=True)`。不能绕过 pipeline 预处理后直接调用 +`pipeline.model`,否则采样率、声道和 waveform 预处理不会执行,部分 ModelScope +版本会直接抛异常,WebSocket 仍会继续输出 ASR 并把说话人保留为 pending。 +更新辅助服务代码后需要重启 `python scripts/auxiliary_server.py`,仅重启页面 +服务不会替换已经驻留在 GPU 中的旧辅助服务进程。 + +## 页面操作 + +1. 选择 Mic 或 File。 +2. 点击开始,浏览器通过 `/ws` 建立本项目 WebSocket。 +3. 页面展示腾讯 Demo 风格的气泡;麦克风或 PCM/WAV 文件按流式方式输入,partial 会在讲话过程中实时刷新。 +4. VAD 检测到静音后提交当前 turn,先返回 pending,再异步更新说话人。 +5. 停止会发送 `stop`,服务端完成当前 turn 和 speaker 队列后再发送 `end`。 +6. `abort` 只取消会话,不提交当前片段。 + +## WebSocket 消息 + +客户端首条消息: + +```json +{ + "type": "start", + "source": "mic", + "model_service_url": "http://127.0.0.1:9950/v1", + "model": "Qwen/Qwen3-ASR-0.6B", + "speaker_diarization": 1, + "sentence_strategy": 0, + "enable_native_partial_stream": true, + "partial_interval_ms": 1200, + "max_segment_sec": 12, + "speaker_gap_ms": 400, + "display_merge": true +} +``` + +随后持续发送 16kHz、单声道、PCM16 二进制音频;文件模式仅支持 PCM/WAV,结束发送 `{"type":"eof"}`,停止发送 +`{"type":"stop"}`,取消发送 `{"type":"abort"}`。每个已完成 turn 会向辅助服务 +发送一次 `/v1/speaker/resolve`,只包含当前 turn 音频和 session_id,不会重复上传整段会话。 + +服务端会发送 `start`、`sentences`、`display_state`、`metrics`、`speaker_warning`、`draining`、`end` 和 `error`。 +页面用带 `revision` 的 `display_state` 渲染,以 `block_id` 标识展示块;`sentences` 保留原始片段及诊断状态。 +停止后必须等待 `end`,其中包含完整 `sentences` 和 `display_blocks`,不能提前关闭连接。 +`sentences` 中 `sentence_type=0` 是 partial,`sentence_type=1` 是 final;同一个 +`sentence_id` 必须覆盖更新而不是追加。`sentence_strategy=0` 使用约 800ms +静音切句,`sentence_strategy=1` 使用约 1400ms 静音切段,更适合段落模式。 + +说话人模式下,`speaker_gap_ms`(默认 400ms)会在有效语音达到 800ms 后 +提前结束短停顿话轮,用于拆开 A→B 的交接;它不触发额外声纹请求,同一说话人的 +相邻片段仍由辅助服务聚类为同一说话人。服务未就绪时该策略会自动关闭。 + +## 目录 + +- `server.py`:本地 HTTP 页面和 WebSocket 会话编排。 +- `model_service.py`:独立 VLLM OpenAI 音频接口适配层。 +- `auxiliary_service.py`:独立辅助模型 HTTP 接口适配层。 +- `speaker_assembler.py`:raw segment、speaker evidence 和 display block 状态机。 +- `static/`:麦克风/文件测试页面。 +- `tests/`:只测试本项目状态合并和音频转换逻辑。 diff --git a/realtime_websocket/auxiliary_service.py b/realtime_websocket/auxiliary_service.py new file mode 100644 index 0000000..d67aadf --- /dev/null +++ b/realtime_websocket/auxiliary_service.py @@ -0,0 +1,161 @@ +"""独立 WebSocket Demo 使用的 VAD 和说话人辅助服务客户端。""" + +from __future__ import annotations + +import io +import wave +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from aiohttp import ClientSession, ClientTimeout, FormData + + +@dataclass(frozen=True) +class AuxiliaryServiceConfig: + """辅助模型服务的 HTTP 连接配置。""" + + base_url: str = "http://127.0.0.1:8010" + timeout_seconds: float = 45.0 + + +def pcm16_to_wav(pcm_bytes: bytes, sample_rate: int = 16000) -> bytes: + """将 Demo 内部的 16kHz 单声道 PCM16 封装成辅助服务可读取的 WAV。""" + output = io.BytesIO() + with wave.open(output, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(sample_rate) + wav_file.writeframes(pcm_bytes) + return output.getvalue() + + +class AuxiliaryModelService: + """调用独立辅助模型服务,不在 WebSocket 进程内加载 GPU 模型。""" + + def __init__(self, config: AuxiliaryServiceConfig) -> None: + self.config = config + self._session: ClientSession | None = None + + async def start(self) -> None: + """创建可复用的 HTTP 会话,避免每个片段重复建立 TCP 连接。""" + self._session = ClientSession(timeout=ClientTimeout(total=self.config.timeout_seconds)) + + async def close(self) -> None: + """关闭辅助服务 HTTP 会话。""" + if self._session is not None: + await self._session.close() + self._session = None + + async def health(self) -> dict[str, Any]: + """读取辅助服务健康状态,避免服务不可达时只能看到 ASR 的降级结果。""" + if self._session is None: + raise RuntimeError("auxiliary model service is not started") + endpoint = self.config.base_url.rstrip("/") + "/health" + async with self._session.get(endpoint) as response: + body = await response.text() + if response.status >= 400: + raise RuntimeError(f"auxiliary health check failed ({response.status}): {body[:500]}") + try: + decoded = await response.json(content_type=None) + except ValueError as exc: + raise RuntimeError(f"auxiliary health check returned invalid JSON: {body[:500]}") from exc + if not isinstance(decoded, dict): + raise RuntimeError("auxiliary health check returned a non-object JSON value") + return decoded + + async def resolve_speaker( + self, + pcm_bytes: bytes, + session_id: str, + start_time_ms: float, + end_time_ms: float, + ) -> dict[str, Any] | None: + """提交一个已经由实时 VAD 完成的 turn,获取在线聚类结果。 + + 每次请求只包含当前 turn,不上传整段会话;辅助服务通过 session_id + 保存聚类中心,因此同一说话人在 A→B→A 场景下仍能保持同一标签。 + """ + if self._session is None: + raise RuntimeError("auxiliary model service is not started") + form = FormData() + form.add_field("file", pcm16_to_wav(pcm_bytes), filename="turn.wav", content_type="audio/wav") + form.add_field("session_id", session_id) + form.add_field("start_time_ms", str(start_time_ms)) + form.add_field("end_time_ms", str(end_time_ms)) + endpoint = self.config.base_url.rstrip("/") + "/v1/speaker/resolve" + async with self._session.post(endpoint, data=form) as response: + body = await response.text() + if response.status >= 400: + raise RuntimeError(f"auxiliary speaker resolve failed ({response.status}): {body[:500]}") + try: + decoded = await response.json(content_type=None) + except ValueError as exc: + raise RuntimeError(f"auxiliary speaker resolve returned invalid JSON: {body[:500]}") from exc + if not isinstance(decoded, dict): + raise RuntimeError("auxiliary speaker resolve returned a non-object JSON value") + if decoded.get("error"): + raise RuntimeError(str(decoded["error"])) + # 保留无标签响应里的具体原因;由组装器统一判断可信度,避免这里静默丢弃。 + return decoded + + async def reset_speaker_session(self, session_id: str) -> None: + """通知辅助服务释放当前 WebSocket 对应的在线聚类状态。""" + if self._session is None: + return + endpoint = self.config.base_url.rstrip("/") + "/v1/speaker/reset" + try: + async with self._session.post(endpoint, json={"session_id": session_id}) as response: + await response.read() + except Exception: + # 清理失败不能影响已经完成的 ASR 结果,辅助服务会自行过期清理。 + return + + async def diarize( + self, + audio_bytes: bytes, + source: str = "mic", + file_name: str = "audio.wav", + ) -> list[dict[str, Any]]: + """提交完整会话音频,返回带毫秒时间范围和标签的聚类片段。 + + 麦克风、PCM 和 WAV 在 WebSocket 层已经能被识别为 16kHz PCM; + MP3、M4A 等压缩文件必须保留原始容器,否则把压缩字节直接包装成 + PCM 会得到不可用的声纹输入。 + """ + if self._session is None: + raise RuntimeError("auxiliary model service is not started") + suffix = Path(file_name).suffix.lower() + is_pcm = source == "mic" or suffix == ".pcm" + if is_pcm: + payload = pcm16_to_wav(audio_bytes) + upload_name = "session.wav" + content_type = "audio/wav" + elif suffix == ".wav": + payload = audio_bytes + upload_name = "session.wav" + content_type = "audio/wav" + else: + payload = audio_bytes + upload_name = Path(file_name).name or "session.audio" + content_type = { + ".mp3": "audio/mpeg", + ".m4a": "audio/mp4", + ".ogg": "audio/ogg", + ".opus": "audio/ogg", + }.get(suffix, "application/octet-stream") + form = FormData() + form.add_field("file", payload, filename=upload_name, content_type=content_type) + endpoint = self.config.base_url.rstrip("/") + "/v1/diarization" + async with self._session.post(endpoint, data=form) as response: + body = await response.text() + if response.status >= 400: + raise RuntimeError(f"auxiliary diarization failed ({response.status}): {body[:500]}") + try: + decoded = await response.json(content_type=None) + except ValueError as exc: + raise RuntimeError(f"auxiliary diarization returned invalid JSON: {body[:500]}") from exc + raw_segments = decoded.get("segments", []) if isinstance(decoded, dict) else [] + if not isinstance(raw_segments, list): + return [] + return [segment for segment in raw_segments if isinstance(segment, dict)] diff --git a/realtime_websocket/model_service.py b/realtime_websocket/model_service.py new file mode 100644 index 0000000..cc0288d --- /dev/null +++ b/realtime_websocket/model_service.py @@ -0,0 +1,120 @@ +"""独立实时 Demo 使用的 OpenAI 兼容 VLLM 服务适配器。""" + +from __future__ import annotations + +import io +import wave +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from aiohttp import ClientSession, ClientTimeout, FormData + + +@dataclass(frozen=True) +class ModelServiceConfig: + """一个独立 VLLM 端点所需的连接配置。""" + + base_url: str = "http://127.0.0.1:9950/v1" + model: str = "Qwen/Qwen3-ASR-0.6B" + api_key: str = "EMPTY" + timeout_seconds: float = 45.0 + + +def pcm16_to_wav(pcm_bytes: bytes, sample_rate: int = 16000) -> bytes: + """将浏览器发送的 PCM16 单声道数据封装为 VLLM 可识别的 WAV 请求。""" + output = io.BytesIO() + with wave.open(output, "wb") as wav_file: + wav_file.setnchannels(1) + wav_file.setsampwidth(2) + wav_file.setframerate(sample_rate) + wav_file.writeframes(pcm_bytes) + return output.getvalue() + + +def wav_to_pcm16(audio_bytes: bytes) -> bytes: + """从 WAV 缓冲区提取 PCM 帧,并兼容尚未完整的中间音频数据。""" + try: + with wave.open(io.BytesIO(audio_bytes), "rb") as wav_file: + return wav_file.readframes(wav_file.getnframes()) + except (EOFError, wave.Error): + if audio_bytes[:4] == b"RIFF" and audio_bytes[8:12] == b"WAVE" and len(audio_bytes) > 44: + return audio_bytes[44:] + return audio_bytes + + +def prepare_audio_request( + audio_bytes: bytes, + source: str, + file_name: str, + partial: bool, +) -> tuple[bytes, str, str] | None: + """将麦克风、PCM 或 WAV 数据转换为 WAV;压缩格式的中间片段延迟到最终帧处理。""" + suffix = Path(file_name).suffix.lower() + if source == "mic" or suffix in {".pcm", ".wav"}: + pcm_bytes = wav_to_pcm16(audio_bytes) if suffix == ".wav" else audio_bytes + return pcm16_to_wav(pcm_bytes), "audio.wav", "audio/wav" + if partial: + # MP3/M4A/OGG 的不断增长前缀通常不是完整容器,不能安全解码,因此只在 + # 最终阶段提交压缩文件,避免中间请求产生随机解码错误。 + return None + content_type = { + ".mp3": "audio/mpeg", + ".m4a": "audio/mp4", + ".ogg": "audio/ogg", + ".opus": "audio/ogg", + }.get(suffix, "application/octet-stream") + return audio_bytes, Path(file_name).name or "audio.bin", content_type + + +class VLLMTranscriptionService: + """只调用独立项目提供的 VLLM HTTP 接口,不导入原项目应用代码。""" + + native_partial_supported = False + + def __init__(self, config: ModelServiceConfig) -> None: + self.config = config + self._session: ClientSession | None = None + + async def start(self) -> None: + """创建可复用的 HTTP 会话,供所有中间和最终转写请求共享。""" + self._session = ClientSession(timeout=ClientTimeout(total=self.config.timeout_seconds)) + + async def close(self) -> None: + """本地 Demo 退出时释放可复用的 HTTP 会话和底层连接。""" + if self._session is not None: + await self._session.close() + self._session = None + + async def transcribe( + self, + audio_bytes: bytes, + source: str, + file_name: str, + partial: bool, + ) -> str | None: + """提交一次音频快照并返回文本;返回 None 表示当前格式不支持中间转写。""" + prepared = prepare_audio_request(audio_bytes, source, file_name, partial) + if prepared is None: + return None + payload, upload_name, content_type = prepared + if self._session is None: + raise RuntimeError("model service is not started") + + form = FormData() + form.add_field("file", payload, filename=upload_name, content_type=content_type) + form.add_field("model", self.config.model) + form.add_field("response_format", "json") + headers = {"Authorization": f"Bearer {self.config.api_key}"} + endpoint = self.config.base_url.rstrip("/") + "/audio/transcriptions" + async with self._session.post(endpoint, data=form, headers=headers) as response: + body = await response.text() + if response.status >= 400: + raise RuntimeError(f"VLLM transcription failed ({response.status}): {body[:500]}") + try: + decoded: Any = await response.json(content_type=None) + except ValueError: + return body.strip() + if isinstance(decoded, dict): + return str(decoded.get("text") or decoded.get("transcript") or "").strip() + return str(decoded).strip() diff --git a/realtime_websocket/requirements.txt b/realtime_websocket/requirements.txt new file mode 100644 index 0000000..ed0d0a1 --- /dev/null +++ b/realtime_websocket/requirements.txt @@ -0,0 +1,2 @@ +aiohttp==3.11.11 +python-dotenv>=1.0 diff --git a/realtime_websocket/server.py b/realtime_websocket/server.py new file mode 100644 index 0000000..acc8e5f --- /dev/null +++ b/realtime_websocket/server.py @@ -0,0 +1,778 @@ +"""面向浏览器的独立 Qwen3-ASR VLLM WebSocket 编排服务。""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import math +import logging +import os +import time +import webbrowser +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from urllib.parse import urlparse +from uuid import uuid4 + +from aiohttp import WSMsgType, web +from dotenv import load_dotenv + +from auxiliary_service import AuxiliaryModelService, AuxiliaryServiceConfig +from model_service import ModelServiceConfig, VLLMTranscriptionService +from speaker_assembler import SegmentAssembler + + +# 与部署启动器读取同一配置;外部环境变量优先于 demo/.env。 +DEPLOY_ROOT = Path(__file__).resolve().parents[1] +load_dotenv(DEPLOY_ROOT / ".env") + +# 监听所有网卡,允许同一局域网内的浏览器访问服务器上的 Demo;端口集中在代码 +# 变量中维护,便于服务器部署时直接修改并保持页面和 WebSocket 使用一致端口。 +WEB_HOST = "0.0.0.0" +WEB_PORT = 8082 +WEB_DISPLAY_HOST = os.getenv("WEB_DISPLAY_HOST", "127.0.0.1") +DEFAULT_MODEL_SERVICE_URL = f"http://127.0.0.1:{os.getenv('VLLM_PORT', '9950')}/v1" +PARTIAL_BYTES_PER_SECOND = 16000 * 2 +VAD_FRAME_BYTES = 640 +VAD_FRAME_MS = 20 +VAD_SILENCE_MS = 800 +PARAGRAPH_SILENCE_MS = 1400 +VAD_RMS_THRESHOLD = 450 +MIN_SPEAKER_VOICE_MS = 800 +# 说话人分离开启时,用比普通 VAD 更短的静音作为候选话轮边界。 +# 这个边界只负责把 A→B 的短交接停顿拆开,不调用声纹模型,因此不会 +# 把窗口推理延迟叠加到音频帧处理路径;同一说话人的拆分片段仍由聚类合并。 +SPEAKER_GAP_MS = max(20, int(os.getenv("SPEAKER_GAP_MS", "400"))) +LOGGER = logging.getLogger(__name__) + + +class EndOfStream: + """带明确类型的队列结束标记,用于区分控制信号和真实音频字节。""" + + +EOF = EndOfStream() +MODEL_SERVICE_KEY = web.AppKey("model_service", VLLMTranscriptionService) +AUXILIARY_SERVICE_KEY = web.AppKey("auxiliary_service", AuxiliaryModelService) + + +@dataclass(frozen=True) +class SpeakerJob: + """等待辅助服务处理的单个已完成 turn;只保存该 turn 的 PCM 音频。""" + + sentence_id: int + audio: bytes + start_time_ms: float + end_time_ms: float + voiced_ms: float = 0.0 + + +def validate_model_service_url(value: str) -> str: + """只接受用户输入的 HTTP(S) VLLM 地址,并拒绝附带认证和查询参数的地址。""" + candidate = value.strip().rstrip("/") + parsed = urlparse(candidate) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("VLLM 地址必须是完整的 http:// 或 https:// URL") + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise ValueError("VLLM 地址不能包含账号、密码、查询参数或片段") + return candidate + + +@dataclass +class SessionMetrics: + """记录 WebSocket 会话耗时和结果修订次数,并在会话结束后展示。""" + + started_at: float + audio_bytes: int = 0 + input_chunks: int = 0 + partial_count: int = 0 + partial_revisions: int = 0 + first_partial_ms: float | None = None + final_ms: float | None = None + + def snapshot(self) -> dict[str, Any]: + """返回可安全序列化为 JSON 的指标,耗时均相对于会话开始时间计算。""" + now = time.perf_counter() + return { + "audio_bytes": self.audio_bytes, + "input_chunks": self.input_chunks, + "partial_count": self.partial_count, + "partial_revisions": self.partial_revisions, + "first_partial_ms": self.first_partial_ms, + "final_ms": self.final_ms, + "elapsed_ms": round((now - self.started_at) * 1000, 1), + } + + +class RealtimeSession: + """串行处理音频快照,并通过同一个 WebSocket 有序推送状态更新。""" + + def __init__( + self, + ws: web.WebSocketResponse, + model_service: VLLMTranscriptionService, + auxiliary_service: AuxiliaryModelService | None, + start: dict[str, Any], + ) -> None: + self.ws = ws + self.model_service = model_service + self.auxiliary_service = auxiliary_service + self.start = start + # 聚类状态只能属于当前连接,客户端复用 ID 不能串入另一会话的声纹池。 + self.session_id = uuid4().hex + self.send_lock = asyncio.Lock() + self.state_revision = 0 + self.audio_queue: asyncio.Queue[bytes | EndOfStream] = asyncio.Queue(maxsize=256) + self.speaker_queue: asyncio.Queue[SpeakerJob | EndOfStream] = asyncio.Queue(maxsize=64) + self.assembler = SegmentAssembler() + self.metrics = SessionMetrics(time.perf_counter()) + self.source = str(start.get("source") or "mic") + self.file_name = str(start.get("file_name") or "audio.wav") + self.windowed_partial = self.source == "mic" or Path(self.file_name).suffix.lower() in {".pcm", ".wav"} + self.sentence_strategy = int(start.get("sentence_strategy") or 0) + self.silence_limit_ms = PARAGRAPH_SILENCE_MS if self.sentence_strategy == 1 else VAD_SILENCE_MS + self.partial_interval_ms = max(300, int(start.get("partial_interval_ms") or 1200)) + self.max_segment_sec = max(2.0, float(start.get("max_segment_sec") or 12.0)) + self.merge_adjacent = self._parse_flag(start.get("display_merge"), True) + self.enable_native_partial = self._parse_flag(start.get("enable_native_partial_stream"), True) + self.segment_id = 0 + self.segment_audio = bytearray() + self.segment_start_ms = 0.0 + self.vad_buffer = bytearray() + self.processed_audio_bytes = 0 + self.silence_ms = 0 + self.in_speech = False + self.voiced_ms = 0.0 + self.pre_roll = bytearray() + self.wav_header_buffer = bytearray() + self.wav_payload_started = self.source != "file" or Path(self.file_name).suffix.lower() != ".wav" + self.wav_riff_read = False + self.wav_format_valid = False + self.wav_data_remaining: int | None = None + self.speaker_warning_sent = False + self.speaker_enabled = self._parse_flag(start.get("speaker_diarization"), True) + # 没有辅助服务时不提前切段,避免服务不可用时把一段语音拆成许多未知片段。 + self.speaker_gap_enabled = self.speaker_enabled and auxiliary_service is not None + raw_speaker_gap = start.get("speaker_gap_ms") + if raw_speaker_gap is None: + self.speaker_gap_ms = SPEAKER_GAP_MS + else: + try: + self.speaker_gap_ms = max(20, int(float(raw_speaker_gap))) + except (TypeError, ValueError, OverflowError): + self.speaker_gap_ms = SPEAKER_GAP_MS + self.input_stopped = False + + @staticmethod + def _parse_flag(value: Any, default: bool) -> bool: + """兼容前端传来的 0/1、布尔值和字符串开关,避免字符串 0 被误判为真。""" + if value is None: + return default + if isinstance(value, str): + return value.strip().lower() not in {"", "0", "false", "no", "off"} + return bool(value) + + async def emit(self, payload: dict[str, Any]) -> None: + """在连接仍然有效时发送一条有序事件,避免向已关闭连接写入数据。""" + async with self.send_lock: + if not self.ws.closed: + await self.ws.send_json(payload) + + async def emit_state(self, sentence: dict[str, Any] | None = None) -> None: + """每次状态更新后同时发送原始状态和重新计算的展示快照。""" + # 在首次 await 前冻结快照,音频 worker 与 speaker worker 不会混用两版状态。 + self.state_revision += 1 + state = { + "type": "display_state", "revision": self.state_revision, + "raw_segments": self.assembler.raw_snapshot(), + "display_blocks": self.assembler.display_blocks(self.merge_adjacent), + "metrics": self.metrics.snapshot(), + } + if sentence is not None: + await self.emit({"type": "sentences", "sentences": [sentence], "metrics": self.metrics.snapshot()}) + await self.emit(state) + + async def warn_speaker(self, message: str) -> None: + """只发送一次说话人服务告警,避免辅助服务异常时刷屏。""" + if self.speaker_warning_sent: + return + await self.emit( + { + "type": "speaker_warning", + "session_id": self.session_id, + "speaker_service_url": getattr(getattr(self.auxiliary_service, "config", None), "base_url", None), + "message": message, + } + ) + self.speaker_warning_sent = True + + def _duration_ms(self) -> float: + """根据 16 kHz PCM 字节数计算时长,不依赖前端可能漂移的时间戳。""" + return len(self.segment_audio) / PARTIAL_BYTES_PER_SECOND * 1000 + + async def _transcribe(self, partial: bool) -> str | None: + """通过 VLLM 适配器转写当前逻辑片段,并保留中间/最终请求的统一入口。""" + return await self.model_service.transcribe( + bytes(self.segment_audio), + "mic", + "turn.pcm", + partial=partial, + ) + + async def _emit_transcription(self, text: str, sentence_type: int, end_ms: float, commit_reason: str | None = None) -> None: + """写入或更新一条句子,确保中间结果和最终结果不会在前端产生重复行。""" + if not text: + return + sentence = self.assembler.apply_sentence( + { + "sentence_id": self.segment_id, + "sentence": text, + "sentence_type": sentence_type, + "start_time": self.segment_start_ms, + "end_time": end_ms, + "speaker_id": -1, + "speaker_name": "", + "speaker_evidence": "pending", + "speaker_confidence": 0.0, + "speaker_strategy": "vllm_no_speaker_evidence", + "commit_reason": commit_reason, + "speaker_status": ("queued" if sentence_type else "waiting_final") if self.speaker_enabled else "disabled", + "speaker_reason": ("等待声纹处理" if sentence_type else "语音片段结束后识别说话人") if self.speaker_enabled else "说话人分离已关闭", + } + ) + if sentence_type == 0: + self.metrics.partial_count += 1 + if sentence["revision_count"] > 0: + self.metrics.partial_revisions += 1 + if self.metrics.first_partial_ms is None: + self.metrics.first_partial_ms = round((time.perf_counter() - self.metrics.started_at) * 1000, 1) + else: + self.metrics.final_ms = round((time.perf_counter() - self.metrics.started_at) * 1000, 1) + await self.emit_state(sentence) + + @staticmethod + def _is_voice_frame(frame: bytes) -> bool: + """用 PCM 帧的 RMS 判断是否有语音,作为实时低延迟切句触发器。""" + if not frame: + return False + samples = memoryview(frame).cast("h") + if not samples: + return False + square_mean = sum(sample * sample for sample in samples) / len(samples) + return math.sqrt(square_mean) >= VAD_RMS_THRESHOLD + + def _strip_wav_header(self, chunk: bytes) -> bytes: + """增量解析 RIFF chunk;支持扩展头,并拒绝采样率或声道不匹配的 WAV。""" + if self.wav_payload_started: + if self.wav_data_remaining is None: + return chunk + payload = chunk[:self.wav_data_remaining] + self.wav_data_remaining -= len(payload) + return payload + self.wav_header_buffer.extend(chunk) + buffer = self.wav_header_buffer + if not self.wav_riff_read: + if len(buffer) < 12: + return b"" + if buffer[:4] != b"RIFF" or buffer[8:12] != b"WAVE": + raise ValueError("文件不是有效的 RIFF/WAV 音频") + del buffer[:12] + self.wav_riff_read = True + while len(buffer) >= 8: + kind = bytes(buffer[:4]) + size = int.from_bytes(buffer[4:8], "little") + if kind == b"data": + if not self.wav_format_valid or size % 2: + raise ValueError("WAV 必须为 16kHz、单声道、PCM16") + self.wav_data_remaining = size + self.wav_payload_started = True + payload = bytes(buffer[8:8 + size]) + self.wav_data_remaining -= len(payload) + buffer.clear() + return payload + if size > 1024 * 1024: + raise ValueError("WAV 元数据头过大,请转换为标准 PCM WAV") + chunk_size = 8 + size + (size % 2) + if len(buffer) < chunk_size: + return b"" + if kind == b"fmt ": + fmt = buffer[8:8 + size] + fields = (int.from_bytes(fmt[0:2], "little"), int.from_bytes(fmt[2:4], "little"), + int.from_bytes(fmt[4:8], "little"), int.from_bytes(fmt[14:16], "little")) + if size < 16 or fields != (1, 1, 16000, 16): + raise ValueError("WAV 必须为 16kHz、单声道、PCM16,请先转换音频") + self.wav_format_valid = True + del buffer[:chunk_size] + return b"" + + async def _resolve_speaker(self, job: SpeakerJob) -> None: + """异步解析单个 turn 的说话人,并把结果覆盖回同一个 sentence_id。""" + if not self.speaker_enabled: + return + async def update_status(status: str, reason: str) -> None: + """将每个失败或等待阶段回写原片段,避免只发一次全局告警。""" + updated = self.assembler.apply_speaker_update({ + "sentence_id": job.sentence_id, "speaker_id": -1, + "speaker_evidence": "pending", "speaker_confidence": 0.0, + "speaker_status": status, "speaker_reason": reason, + }) + await self.emit_state(updated) + + # 按有效有声帧检查长度,不能让句尾 800ms 静音把短插话伪装成长样本。 + if job.voiced_ms < MIN_SPEAKER_VOICE_MS: + await update_status("insufficient_audio", f"有效语音不足 {MIN_SPEAKER_VOICE_MS}ms,不继承上一位说话人") + return + if self.auxiliary_service is None: + await update_status("service_unavailable", "未配置说话人辅助模型服务") + await self.warn_speaker("未配置辅助模型服务,无法执行实时说话人分离") + return + await update_status("processing", "正在提取声纹并匹配说话人") + try: + speaker = await self.auxiliary_service.resolve_speaker( + job.audio, + self.session_id, + job.start_time_ms, + job.end_time_ms, + ) + except Exception as exc: + # 辅助服务异常不能阻断 ASR;当前片段继续保持 pending,方便定位服务问题。 + LOGGER.exception("speaker resolve failed: session=%s sentence=%s", self.session_id, job.sentence_id) + await update_status("service_error", str(exc)) + await self.warn_speaker(str(exc)) + return + if not speaker: + await update_status("no_embedding", "辅助服务未返回可用声纹结果") + return + update = dict(speaker) + update["sentence_id"] = job.sentence_id + update["speaker_name"] = str(update.get("speaker_name") or "") + updated = self.assembler.apply_speaker_update(update) + if updated is not None: + LOGGER.info("speaker result: session=%s sentence=%s status=%s strategy=%s", self.session_id, + job.sentence_id, updated.get("speaker_status"), updated.get("speaker_strategy")) + await self.emit_state(updated) + + async def process_speakers(self) -> None: + """按 turn 顺序串行访问辅助模型,保证在线聚类中心不会乱序更新。""" + while True: + item = await self.speaker_queue.get() + if isinstance(item, EndOfStream): + return + await self._resolve_speaker(item) + + async def _commit_segment(self, reason: str = "final") -> None: + """在 VAD 检测到一句结束后提交 final,并异步排队当前 turn 的说话人解析。""" + if not self.segment_audio or not self.in_speech: + return + # 去掉句尾触发切段的静音,ASR 与声纹都使用当前片段的真实有效范围。 + trailing_bytes = int(self.silence_ms * PARTIAL_BYTES_PER_SECOND / 1000) + if trailing_bytes: + del self.segment_audio[-trailing_bytes:] + final_audio = bytes(self.segment_audio) + final_start_ms = self.segment_start_ms + final_end_ms = self.segment_start_ms + self._duration_ms() + final_sentence_id = self.segment_id + text = await self._transcribe(partial=False) + if text: + await self._emit_transcription(text, 1, final_end_ms, reason) + if self.speaker_enabled: + await self.speaker_queue.put( + SpeakerJob( + sentence_id=final_sentence_id, + audio=final_audio, + start_time_ms=final_start_ms, + end_time_ms=final_end_ms, + voiced_ms=self.voiced_ms, + ) + ) + elif self.assembler.segments.pop(final_sentence_id, None) is not None: + # final 判定无文本时撤回临时结果,不能留下永远等待声纹的 partial。 + await self.emit_state() + self.segment_audio.clear() + self.segment_start_ms = self.processed_audio_bytes / PARTIAL_BYTES_PER_SECOND * 1000 + self.segment_id += 1 + self.silence_ms = 0 + self.in_speech = False + self.voiced_ms = 0.0 + + async def process_audio(self) -> None: + """消费音频,以 VAD 静音结束作为切句主逻辑,并按窗口发送 partial。""" + last_partial_bytes = 0 + partial_bytes = int(self.partial_interval_ms / 1000 * PARTIAL_BYTES_PER_SECOND) + while True: + item = await self.audio_queue.get() + if isinstance(item, EndOfStream): + break + chunk = self._strip_wav_header(item) if self.source == "file" else item + self.metrics.input_chunks += 1 + if not self.windowed_partial: + # websocket_handler 已拒绝压缩文件;这里保留防御分支,避免未来 + # 新客户端绕过入口时又悄悄退化成“整段上传后切片”。 + raise RuntimeError("实时流式模式只接受 16kHz PCM16 音频") + if not chunk: + continue + self.vad_buffer.extend(chunk) + while len(self.vad_buffer) >= VAD_FRAME_BYTES: + frame = bytes(self.vad_buffer[:VAD_FRAME_BYTES]) + del self.vad_buffer[:VAD_FRAME_BYTES] + frame_start_ms = self.processed_audio_bytes / PARTIAL_BYTES_PER_SECOND * 1000 + self.processed_audio_bytes += len(frame) + self.metrics.audio_bytes += len(frame) + voiced = self._is_voice_frame(frame) + if voiced and not self.in_speech: + self.in_speech = True + self.segment_start_ms = frame_start_ms - len(self.pre_roll) / PARTIAL_BYTES_PER_SECOND * 1000 + self.segment_audio = bytearray(self.pre_roll) + self.pre_roll.clear() + last_partial_bytes = 0 + if self.in_speech: + self.segment_audio.extend(frame) + if voiced: + self.voiced_ms += VAD_FRAME_MS + self.silence_ms = 0 if voiced else self.silence_ms + VAD_FRAME_MS + if len(self.segment_audio) - last_partial_bytes >= partial_bytes and self.silence_ms < self.silence_limit_ms: + text = await self._transcribe(partial=True) + if text: + await self._emit_transcription(text, 0, self.segment_start_ms + self._duration_ms()) + last_partial_bytes = len(self.segment_audio) + # 短交接停顿优先于普通 800/1400ms 静音切段,但必须先有 + # 至少 800ms 有效语音,避免把咳嗽、噪声或极短插话送去聚类。 + short_speaker_gap = ( + self.speaker_gap_enabled + and self.voiced_ms >= MIN_SPEAKER_VOICE_MS + and self.silence_ms >= self.speaker_gap_ms + ) + if short_speaker_gap or self.silence_ms >= self.silence_limit_ms or self._duration_ms() >= self.max_segment_sec * 1000: + reason = ( + "speaker_gap" if short_speaker_gap + else "silence" if self.silence_ms >= self.silence_limit_ms + else "max_duration" + ) + await self._commit_segment(reason) + last_partial_bytes = 0 + else: + # 参考原 WebSocket 保留 200ms 前滚,减少首字低能量音素被裁掉。 + self.pre_roll.extend(frame) + del self.pre_roll[:-6400] + + if not self.wav_payload_started or (not self.input_stopped and self.wav_data_remaining not in (None, 0)): + raise ValueError("WAV 文件不完整,未收到全部音频数据") + if len(self.vad_buffer) % 2: + raise ValueError("PCM16 音频必须包含完整的双字节采样") + if self.windowed_partial and self.vad_buffer: + tail_ms = len(self.vad_buffer) / PARTIAL_BYTES_PER_SECOND * 1000 + tail_voiced = self._is_voice_frame(bytes(self.vad_buffer)) + if tail_voiced and not self.in_speech: + self.in_speech = True + self.segment_start_ms = self.processed_audio_bytes / PARTIAL_BYTES_PER_SECOND * 1000 + self.processed_audio_bytes += len(self.vad_buffer) + self.metrics.audio_bytes += len(self.vad_buffer) + if self.in_speech: + self.segment_audio.extend(self.vad_buffer) + self.silence_ms = 0 if tail_voiced else self.silence_ms + tail_ms + self.voiced_ms += tail_ms if tail_voiced else 0 + self.vad_buffer.clear() + if self.segment_audio: + if not self.windowed_partial: + self.in_speech = True + await self._commit_segment() + await self.emit({"type": "metrics", "metrics": self.metrics.snapshot()}) + + +def deployment_model_name() -> str: + """把部署脚本的 0.6b/1.7b 别名解析成 vLLM 对外发布的模型名。""" + public_name = os.getenv("VLLM_SERVED_MODEL_NAME") + if public_name: + return public_name + requested = os.getenv("QWEN3_ASR_MODEL", "default") + with (DEPLOY_ROOT / "model_manifest.json").open(encoding="utf-8") as source: + manifest = json.load(source) + if requested == "default": + return str(manifest["default_model"]) + for model_id, config in manifest["models"].items(): + if requested.lower() == str(config.get("alias", "")).lower(): + return model_id + return requested + + +def parse_args() -> argparse.Namespace: + """只解析服务选择参数;浏览器服务端口继续由代码内部变量统一维护。""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model-service-url", default=os.getenv("MODEL_SERVICE_URL", DEFAULT_MODEL_SERVICE_URL)) + parser.add_argument("--model", default=deployment_model_name()) + parser.add_argument("--no-browser", action="store_true") + return parser.parse_args() + + +async def index_handler(_: web.Request) -> web.FileResponse: + """返回独立 Demo 测试页面,并避免入口页缓存旧的脚本版本号。""" + # 入口页必须每次重新校验,配合 app.js 的版本号变更,避免用户继续运行旧前端。 + return web.FileResponse( + Path(__file__).parent / "static" / "index.html", + headers={"Cache-Control": "no-store"}, + ) + + +async def config_handler(request: web.Request) -> web.Response: + """暴露服务启动时的默认配置,让页面自动填充 VLLM 地址和模型名。""" + config = request.app[MODEL_SERVICE_KEY].config + auxiliary = request.app.get(AUXILIARY_SERVICE_KEY) + return web.json_response({ + "model_service_url": config.base_url, "model": config.model, + "speaker_service_url": getattr(getattr(auxiliary, "config", None), "base_url", None), + }) + + +async def websocket_handler(request: web.Request) -> web.WebSocketResponse: + """处理一个浏览器会话,每个连接独立保存音频、句子和展示状态。""" + ws = web.WebSocketResponse(max_msg_size=64 * 1024 * 1024) + await ws.prepare(request) + default_model_service: VLLMTranscriptionService = request.app[MODEL_SERVICE_KEY] + model_service = default_model_service + owns_model_service = False + processing: asyncio.Task[None] | None = None + speaker_processing: asyncio.Task[None] | None = None + session: RealtimeSession | None = None + try: + first = await ws.receive() + if first.type != WSMsgType.TEXT: + await ws.send_json({"type": "error", "message": "first message must be JSON start"}) + return ws + try: + start = json.loads(first.data) + except json.JSONDecodeError: + await ws.send_json({"type": "error", "message": "invalid start JSON"}) + return ws + if not isinstance(start, dict) or start.get("type") != "start": + await ws.send_json({"type": "error", "message": "first message must have type=start"}) + return ws + + # 本项目用于验证实时流式链路,文件模式只接受可以按 PCM 帧连续处理的 + # WAV/PCM;MP3、M4A 等压缩容器只能在文件完整到达后解码,不纳入本次测试。 + source = str(start.get("source") or "mic") + file_suffix = Path(str(start.get("file_name") or "")).suffix.lower() + if source == "file" and file_suffix not in {".pcm", ".wav"}: + await ws.send_json( + { + "type": "error", + "message": "实时流式测试的文件模式只支持 PCM 或 WAV,请改用麦克风、PCM 或 WAV", + } + ) + return ws + + # 页面可以在不重启 WebSocket Demo 的情况下为当前会话切换 VLLM 地址; + # 未切换时继续复用默认服务,避免普通场景为每个连接重复创建 HTTP 会话。 + try: + requested_url = validate_model_service_url( + str(start.get("model_service_url") or default_model_service.config.base_url) + ) + except ValueError as exc: + await ws.send_json({"type": "error", "message": str(exc)}) + return ws + requested_model = str(start.get("model") or default_model_service.config.model).strip() + if ( + requested_url != default_model_service.config.base_url + or requested_model != default_model_service.config.model + ): + model_service = VLLMTranscriptionService( + ModelServiceConfig(base_url=requested_url, model=requested_model) + ) + await model_service.start() + owns_model_service = True + + auxiliary_service = request.app.get(AUXILIARY_SERVICE_KEY) + session = RealtimeSession(ws, model_service, auxiliary_service, start) + auxiliary_config = getattr(auxiliary_service, "config", None) + speaker_health: dict[str, Any] | None = None + speaker_health_error: str | None = None + if session.speaker_enabled and auxiliary_service is not None: + # 健康检查只用于尽早暴露辅助服务问题;即使失败也不阻断 ASR, + # 这样可以从同一页面继续观察 ASR 与说话人链路的差异。 + health_check = getattr(auxiliary_service, "health", None) + if callable(health_check): + try: + speaker_health = await asyncio.wait_for(health_check(), timeout=5) + if speaker_health.get("speaker_embedding_ready", speaker_health.get("ready")) is False: + speaker_health_error = "辅助模型服务未就绪,请检查 /health 返回的 models 状态" + except Exception as exc: + speaker_health_error = f"说话人辅助服务不可用:{exc}" + if speaker_health_error: + # 辅助服务未就绪时先保留原始 VAD 切段,避免在没有声纹结果的 + # 情况下增加大量短片段;服务恢复后由新的会话重新启用。 + session.speaker_gap_enabled = False + await session.emit( + { + "type": "start", + "model_service_url": model_service.config.base_url, + "model": model_service.config.model, + "session_id": session.session_id, + "enable_native_partial_stream": session.enable_native_partial, + "native_partial_supported": model_service.native_partial_supported, + "partial_mode": "http_cumulative_window", + "speaker_diarization_enabled": session.speaker_enabled, + "speaker_service_url": getattr(auxiliary_config, "base_url", None), + "speaker_service_health": speaker_health, + "speaker_gap_enabled": session.speaker_gap_enabled, + "speaker_gap_ms": session.speaker_gap_ms, + "sentence_strategy": session.sentence_strategy, + "silence_limit_ms": session.silence_limit_ms, + "display_state_supported": True, + } + ) + if speaker_health_error: + await session.warn_speaker(speaker_health_error) + processing = asyncio.create_task(session.process_audio()) + if session.speaker_enabled: + # speaker worker 与音频处理并行运行;它只消费已经结束的 turn, + # 因此不会阻塞下一帧音频进入队列或影响 ASR partial 输出。 + speaker_processing = asyncio.create_task(session.process_speakers()) + async def guarded(operation): + """接收和队列背压同时监听 worker,推理失败立即报错而非永远等 stop。""" + pending = asyncio.create_task(operation) + try: + workers = [task for task in (processing, speaker_processing) if task is not None] + done, _ = await asyncio.wait([pending, *workers], return_when=asyncio.FIRST_COMPLETED) + if pending in done: + return await pending + for worker in workers: + if worker in done: + await worker + raise RuntimeError("实时处理任务意外结束") + return await pending + finally: + if not pending.done(): + pending.cancel() + await asyncio.gather(pending, return_exceptions=True) + + input_finished = False + while not ws.closed: + message = await guarded(ws.receive()) + if message.type == WSMsgType.BINARY: + await guarded(session.audio_queue.put(bytes(message.data))) + continue + if message.type == WSMsgType.TEXT: + try: + control = json.loads(message.data) + except json.JSONDecodeError: + continue + if not isinstance(control, dict): + continue + if control.get("type") in {"eof", "stop"}: + input_finished = True + session.input_stopped = control.get("type") == "stop" + await session.emit({"type": "draining", "message": "正在完成转写和说话人识别"}) + await guarded(session.audio_queue.put(EOF)) + break + if control.get("type") == "abort": + processing.cancel() + if speaker_processing is not None: + speaker_processing.cancel() + await asyncio.gather( + processing, + *(task for task in [speaker_processing] if task is not None), + return_exceptions=True, + ) + return ws + if message.type in {WSMsgType.ERROR, WSMsgType.CLOSE, WSMsgType.CLOSED}: + processing.cancel() + if speaker_processing is not None: + speaker_processing.cancel() + await asyncio.gather( + processing, + *(task for task in [speaker_processing] if task is not None), + return_exceptions=True, + ) + return ws + if not input_finished: + # 浏览器或网络断开后,音频生产者已经不存在,不能让处理任务继续等待 + # 永远不会到来的 EOF,因此这里主动取消任务并回收异常结果。 + processing.cancel() + if speaker_processing is not None: + speaker_processing.cancel() + await asyncio.gather( + processing, + *(task for task in [speaker_processing] if task is not None), + return_exceptions=True, + ) + return ws + try: + await processing + if speaker_processing is not None: + await session.speaker_queue.put(EOF) + await speaker_processing + await session.emit_state() + await session.emit({ + "type": "end", "metrics": session.metrics.snapshot(), + "sentences": session.assembler.raw_snapshot(), + "display_blocks": session.assembler.display_blocks(session.merge_adjacent), + }) + except asyncio.CancelledError: + raise + except Exception as exc: + await session.emit({"type": "error", "message": str(exc)}) + except Exception as exc: + LOGGER.exception("WebSocket session failed") + if not ws.closed: + await ws.send_json({"type": "error", "message": str(exc)}) + finally: + for task in (processing, speaker_processing): + if task is not None and not task.done(): + task.cancel() + pending_tasks = [task for task in (processing, speaker_processing) if task is not None] + if pending_tasks: + await asyncio.gather(*pending_tasks, return_exceptions=True) + # stop、abort、断线和推理异常均释放会话,清理失败不覆盖最终识别结果。 + if session is not None: + reset = getattr(session.auxiliary_service, "reset_speaker_session", None) + if reset is not None: + try: + await asyncio.wait_for(reset(session.session_id), timeout=5) + except Exception: + LOGGER.warning("speaker session cleanup failed: %s", session.session_id, exc_info=True) + if owns_model_service: + await model_service.close() + if not ws.closed: + await ws.close() + return ws + + +async def start_app(model_service_url: str, model: str) -> web.Application: + """创建 HTTP/WebSocket 应用,并挂载可复用的 VLLM 适配器。""" + app = web.Application() + app[MODEL_SERVICE_KEY] = VLLMTranscriptionService(ModelServiceConfig(base_url=model_service_url, model=model)) + app[AUXILIARY_SERVICE_KEY] = AuxiliaryModelService( + AuxiliaryServiceConfig(base_url=os.getenv("AUXILIARY_SERVICE_URL", "http://127.0.0.1:8010")) + ) + + async def lifecycle(application: web.Application): + await application[MODEL_SERVICE_KEY].start() + await application[AUXILIARY_SERVICE_KEY].start() + yield + await application[AUXILIARY_SERVICE_KEY].close() + await application[MODEL_SERVICE_KEY].close() + + app.cleanup_ctx.append(lifecycle) + app.router.add_get("/", index_handler) + app.router.add_get("/api/config", config_handler) + app.router.add_static("/static/", Path(__file__).parent / "static") + app.router.add_get("/ws", websocket_handler) + # 参考腾讯 Demo 将 static 目录挂载到根路径;页面中的 style.css 和 app.js + # 使用相对地址,必须同时提供根路径静态资源路由,否则浏览器会显示无样式页面。 + app.router.add_static("/", Path(__file__).parent / "static", show_index=False) + return app + + +def main() -> None: + """启动本地测试页面和 WebSocket 服务。""" + args = parse_args() + logging.basicConfig(level=logging.INFO) + if not args.no_browser: + webbrowser.open(f"http://{WEB_DISPLAY_HOST}:{WEB_PORT}/") + print(f"WebSocket demo: http://{WEB_DISPLAY_HOST}:{WEB_PORT}/", flush=True) + print(f"VLLM service: {args.model_service_url} ({args.model})", flush=True) + web.run_app(start_app(args.model_service_url, args.model), host=WEB_HOST, port=WEB_PORT) + + +if __name__ == "__main__": + main() diff --git a/realtime_websocket/speaker_assembler.py b/realtime_websocket/speaker_assembler.py new file mode 100644 index 0000000..a9ae9c3 --- /dev/null +++ b/realtime_websocket/speaker_assembler.py @@ -0,0 +1,186 @@ +"""独立 Demo 使用的有状态原始片段与展示区块组装器。""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +import math +from typing import Any + + +TRUSTED_CONFIDENCE = 0.6 + + +def _as_float(value: Any, default: float = 0.0) -> float: + try: + parsed = float(value) + return parsed if math.isfinite(parsed) else default + except (TypeError, ValueError, OverflowError): + return default + + +def _as_int(value: Any, default: int = -1) -> int: + try: + return int(value) + except (TypeError, ValueError, OverflowError): + return default + + +def _without_embeddings(payload: dict[str, Any]) -> dict[str, Any]: + """阻止原始声纹向量进入 Demo 的持久化状态。""" + sanitized: dict[str, Any] = {} + for key, value in payload.items(): + lowered = str(key).lower() + if "embedding" in lowered or lowered in {"_chunks", "chunks", "_chunk_embeddings"}: + continue + sanitized[key] = deepcopy(value) + return sanitized + + +def _is_trusted(segment: dict[str, Any]) -> bool: + """在展示身份稳定前,要求存在独立的说话人证据。""" + evidence = str(segment.get("speaker_evidence") or "pending").lower() + return ( + _as_int(segment.get("speaker_id")) >= 0 + and evidence in {"fresh", "confirmed"} + and _as_float(segment.get("speaker_confidence")) >= TRUSTED_CONFIDENCE + and segment.get("speaker_strategy") not in {"short_attach", "embedding_attach"} + ) + + +def _speaker_identity(segment: dict[str, Any]) -> tuple[Any, ...]: + """实名身份优先于匿名簇,防止同簇弱标签或不同实名被合并。""" + for key in ("user_id", "registry_speaker_id"): + if segment.get(key) not in (None, ""): + return (key, str(segment[key])) + return ("cluster", segment.get("speaker_id"), segment.get("speaker_name", "")) + + +@dataclass +class SegmentAssembler: + """保存可幂等更新的原始片段,并按时间顺序派生展示区块。""" + + segments: dict[int, dict[str, Any]] = field(default_factory=dict) + + def apply_sentence(self, incoming: dict[str, Any]) -> dict[str, Any]: + """写入或更新一条中间或最终句子,并保护不可靠的短身份名称。""" + sentence_id = _as_int(incoming.get("sentence_id"), 0) + previous = self.segments.get(sentence_id) + segment = dict(previous or {}) + segment.update(_without_embeddings(incoming)) + # 文本重发不能抹掉已到达的声纹更新,也不能把 final 回滚成 partial。 + if previous and previous.get("sentence_type") == 1 and incoming.get("sentence_type") == 0: + return deepcopy(previous) + segment["sentence_id"] = sentence_id + segment["sentence"] = str(segment.get("sentence") or segment.get("text") or "").strip() + segment["sentence_type"] = _as_int(segment.get("sentence_type"), 0) + segment["start_time"] = _as_float(segment.get("start_time")) + segment["end_time"] = _as_float(segment.get("end_time")) + segment["speaker_id"] = _as_int(segment.get("speaker_id")) + segment["speaker_name"] = str(segment.get("speaker_name") or "") + segment["speaker_evidence"] = str(segment.get("speaker_evidence") or "pending") + segment["speaker_confidence"] = _as_float(segment.get("speaker_confidence")) + + strategy = str(segment.get("speaker_strategy") or "") + if strategy in {"short_attach", "embedding_attach"}: + # 继承而来的短名称不属于新的可靠证据,必须继续保持 pending 状态, + # 防止前一个片段的身份错误污染当前展示结果。 + segment["speaker_id"] = -1 + segment["speaker_name"] = "" + segment["speaker_evidence"] = "pending" + segment["speaker_confidence"] = 0.0 + segment["speaker_status"] = "inherited_rejected" + segment["speaker_reason"] = "缺少当前片段的独立声纹证据" + segment.pop("user_id", None) + segment.pop("registry_speaker_id", None) + + if previous is not None: + segment["revision_count"] = int(previous.get("revision_count", 0)) + ( + 1 if segment["sentence"] != previous.get("sentence") else 0 + ) + else: + segment["revision_count"] = 0 + self.segments[sentence_id] = segment + return deepcopy(segment) + + def apply_speaker_update(self, update: dict[str, Any]) -> dict[str, Any] | None: + """只将明确标记为新鲜或已确认的更新应用到已有片段。""" + sentence_id = _as_int(update.get("sentence_id"), -1) + current = self.segments.get(sentence_id) + if current is None: + return None + candidate = dict(current) + # 新身份的实名字段必须来自本次证据,不能沿用同片段旧识别的人员 ID。 + candidate.pop("user_id", None) + candidate.pop("registry_speaker_id", None) + # 说话人响应只能更新身份字段,不能篡改已确认的文本和 ASR 时间范围。 + candidate.update({ + key: value for key, value in _without_embeddings(update).items() + if key.startswith("speaker_") or key in {"user_id", "registry_speaker_id"} + }) + candidate["speaker_id"] = _as_int(candidate.get("speaker_id")) + candidate["speaker_evidence"] = str(update.get("speaker_evidence") or "pending") + candidate["speaker_confidence"] = _as_float(update.get("speaker_confidence")) + if not _is_trusted(candidate): + # 保留拒绝原因供前端诊断,但绝不把不可靠身份带入展示或历史状态。 + candidate.update(speaker_id=-1, speaker_name="", speaker_evidence="pending", speaker_confidence=0.0) + candidate.pop("user_id", None) + candidate.pop("registry_speaker_id", None) + if update.get("speaker_status") not in { + "queued", "processing", "waiting_final", "disabled", "insufficient_audio", + "service_unavailable", "service_error", "no_embedding", "evidence_rejected", + }: + candidate["speaker_status"] = "evidence_rejected" + candidate["speaker_reason"] = "声纹结果缺少新鲜证据或置信度不足" + else: + candidate["speaker_status"] = "confirmed" + candidate["speaker_reason"] = "当前片段声纹已确认" + self.segments[sentence_id] = candidate + return deepcopy(candidate) + + def raw_snapshot(self) -> list[dict[str, Any]]: + """返回按时间、再按句子 ID 排序后的全部原始片段。""" + return [ + deepcopy(segment) + for segment in sorted(self.segments.values(), key=lambda item: (item["start_time"], item["sentence_id"])) + ] + + def display_blocks(self, merge_adjacent: bool = True) -> list[dict[str, Any]]: + """生成展示区块,同时保持非相邻说话人轮次的原始顺序。""" + blocks: list[dict[str, Any]] = [] + for segment in self.raw_snapshot(): + trusted = _is_trusted(segment) + identity_key = _speaker_identity(segment) if trusted else ("pending", segment["sentence_id"]) + if ( + merge_adjacent + and blocks + and trusted + and blocks[-1].get("identity_key") == identity_key + ): + block = blocks[-1] + block["sentence"] = f'{block["sentence"]} {segment["sentence"]}'.strip() + block["end_time"] = max(block["end_time"], segment["end_time"]) + block["segment_ids"].append(segment["sentence_id"]) + block["sentence_type"] = min(block["sentence_type"], segment["sentence_type"]) + continue + + blocks.append( + { + "block_id": f"block-{segment['sentence_id']}", + "sentence": segment["sentence"], + "start_time": segment["start_time"], + "end_time": segment["end_time"], + "segment_ids": [segment["sentence_id"]], + "speaker_id": segment["speaker_id"] if trusted else -1, + "speaker_name": segment["speaker_name"] if trusted else "", + "speaker_evidence": "confirmed" if trusted else "pending", + "speaker_status": segment.get("speaker_status", "pending"), + "speaker_reason": segment.get("speaker_reason", ""), + "speaker_confidence": segment.get("speaker_confidence", 0.0), + "sentence_type": segment["sentence_type"], + "identity_key": identity_key, + } + ) + for block in blocks: + block.pop("identity_key", None) + return blocks diff --git a/realtime_websocket/static/app.js b/realtime_websocket/static/app.js new file mode 100644 index 0000000..944df74 --- /dev/null +++ b/realtime_websocket/static/app.js @@ -0,0 +1,743 @@ +// ===== 页面元素 ===== +const elEngineModel = document.getElementById('engineModel'); +const elModelServiceUrl = document.getElementById('modelServiceUrl'); +const elSpeakerStatus = document.getElementById('speakerStatus'); +const elDisplayMerge = document.getElementById('displayMerge'); +const elSpeakerDiarization = document.getElementById('speakerDiarization'); +const elDiarizationLabel = document.getElementById('diarizationLabel'); +const elSentenceStrategy = document.getElementById('sentenceStrategy'); +const elBtnStart = document.getElementById('btnStart'); +const elBtnStop = document.getElementById('btnStop'); +const elResultMeta = document.getElementById('resultMeta'); +const elVoiceIdDisplay = document.getElementById('voiceIdDisplay'); +const elBtnCopyVoiceId = document.getElementById('btnCopyVoiceId'); +const elBtnExportWav = document.getElementById('btnExportWav'); +const elStatusDot = document.getElementById('statusDot'); +const elStatusText = document.getElementById('statusText'); +const elResultArea = document.getElementById('resultArea'); +const elResultPlaceholder = document.getElementById('resultPlaceholder'); +const elLogArea = document.getElementById('logArea'); +const elBtnClearLog = document.getElementById('btnClearLog'); +const elMicStatus = document.getElementById('micStatus'); +const elMicTimer = document.getElementById('micTimer'); +const elMicElapsed = document.getElementById('micElapsed'); + +// 输入模式标签页 +const elTabMic = document.getElementById('tabMic'); +const elTabFile = document.getElementById('tabFile'); +const elPanelMic = document.getElementById('panelMic'); +const elPanelFile = document.getElementById('panelFile'); + +// 文件选择区域 +const elAudioFile = document.getElementById('audioFile'); +const elFileInfo = document.getElementById('fileInfo'); +const elAudioMeta = document.getElementById('audioMeta'); +const elMetaFormat = document.getElementById('metaFormat'); +const elMetaSampleRate = document.getElementById('metaSampleRate'); +const elMetaDuration = document.getElementById('metaDuration'); +const elSpeedControl = document.getElementById('speedControl'); +const elSpeedSlider = document.getElementById('speedSlider'); +const elSpeedValue = document.getElementById('speedValue'); + +// ===== 说话人分离开关 ===== +elSpeakerDiarization.addEventListener('change', () => { + elDiarizationLabel.textContent = elSpeakerDiarization.checked ? '开启' : '关闭'; +}); + +// ===== 日志区域 ===== +elBtnClearLog.addEventListener('click', () => { elLogArea.innerHTML = ''; }); + +function appendLog(msg) { + const now = new Date(); + const ts = [now.getHours(), now.getMinutes(), now.getSeconds()] + .map(n => String(n).padStart(2, '0')).join(':') + + '.' + String(now.getMilliseconds()).padStart(3, '0'); + const typeClass = 'log-type-' + (msg.type || 'unknown'); + const entry = document.createElement('div'); + entry.className = 'log-entry'; + // 原始文本不作为 HTML 解释,转写中的标签也应原样显示。 + const stamp = document.createElement('span'); + stamp.className = 'log-time'; + stamp.textContent = ts; + const content = document.createElement('span'); + content.className = typeClass; + content.textContent = JSON.stringify(msg); + entry.append(stamp, content); + elLogArea.appendChild(entry); + while (elLogArea.childNodes.length > 300) elLogArea.firstChild.remove(); + elLogArea.scrollTop = elLogArea.scrollHeight; +} + +// ===== 会话状态 ===== +let ws = null; +let sending = false; +let stoppingByUser = false; +let currentVoiceId = null; +let audioChunks = []; +let sessionId = 0; + +let micStream = null; +let micAudioContext = null; +let micWorklet = null; +let micTimerInterval = null; +let micStartTime = 0; + +// 输入模式(麦克风 / 文件) +let inputMode = 'mic'; +let selectedFile = null; + +// 文件扩展名 → SDK VoiceFormat 常量 +const EXT_FORMAT_MAP = { + 'pcm': 1, 'wav': 12, 'mp3': 8, 'm4a': 14, + 'aac': 16, 'opus': 10, 'ogg': 10, 'silk': 6, 'speex': 4 +}; +// PCM/WAV 的默认发送倍速;实时验证默认按 1 倍速输入。 +const DEFAULT_SPEED = { + 'pcm': 1.0, 'wav': 1.0, + 'mp3': 2.0, 'm4a': 2.0, 'aac': 2.0, + 'opus': 3.0, 'ogg': 3.0, 'silk': 3.0, 'speex': 3.0 +}; +const MAX_SPEED = 3.0; +// 实时 WebSocket 需要服务端逐帧读取音频;压缩格式必须等文件完整后才能解码, +// 因此本次流式验证只允许 PCM/WAV,避免把整段上传伪装成实时识别。 +const STREAMABLE_AUDIO_EXTENSIONS = new Set(['pcm', 'wav']); + +const SPEAKER_COLORS = ['#4a7dff', '#52c41a', '#faad14', '#ff4d4f', '#9254de', '#13c2c2']; + +let sentenceMap = {}; +let speakerOrderMap = {}; +let speakerOrderCounter = 0; +let displayStateSupported = false; +let displayRevision = -1; + +// ===== 输入模式标签页 ===== +function switchMode(mode) { + if (ws) return; + inputMode = mode; + elTabMic.classList.toggle('active', mode === 'mic'); + elTabFile.classList.toggle('active', mode === 'file'); + elPanelMic.style.display = mode === 'mic' ? '' : 'none'; + elPanelFile.style.display = mode === 'file' ? '' : 'none'; + if (mode === 'file') { + elBtnStart.disabled = !selectedFile; + } else { + elBtnStart.disabled = false; + } +} +elTabMic.addEventListener('click', () => switchMode('mic')); +elTabFile.addEventListener('click', () => switchMode('file')); + +// ===== 文件选择 ===== +elAudioFile.addEventListener('change', (e) => { + const file = e.target.files[0]; + if (!file) return; + const ext = getFileExt(file.name); + if (!STREAMABLE_AUDIO_EXTENSIONS.has(ext)) { + selectedFile = null; + e.target.value = ''; + elFileInfo.textContent = '实时测试只支持 PCM 或 WAV,请先转换音频格式'; + elFileInfo.classList.remove('has-file'); + elAudioMeta.style.display = 'none'; + elSpeedControl.style.display = 'none'; + elBtnStart.disabled = true; + showToast('压缩音频不能按当前实时 WebSocket 逐帧识别,请转成 PCM 或 WAV', true); + return; + } + selectedFile = file; + const sizeMB = (file.size / 1024 / 1024).toFixed(1); + elFileInfo.textContent = `${file.name} (${sizeMB}MB)`; + elFileInfo.classList.add('has-file'); + elBtnStart.disabled = false; + parseAudioMeta(file); +}); + +// 发送速度滑块 +elSpeedSlider.addEventListener('input', () => { + elSpeedValue.textContent = parseFloat(elSpeedSlider.value).toFixed(1) + 'x'; +}); + +// ===== 音频元数据解析 ===== +function getFileExt(filename) { + const parts = filename.split('.'); + return parts.length > 1 ? parts[parts.length - 1].toLowerCase() : ''; +} + +function formatDuration(seconds) { + if (!seconds || !isFinite(seconds)) return '—'; + const min = Math.floor(seconds / 60); + const sec = Math.floor(seconds % 60); + const ms = Math.round((seconds % 1) * 10); + return `${String(min).padStart(2, '0')}:${String(sec).padStart(2, '0')}.${ms}`; +} + +async function parseAudioMeta(file) { + const ext = getFileExt(file.name); + elMetaFormat.textContent = ext.toUpperCase(); + // 根据格式设默认倍速 + const defaultSpeed = Math.min(DEFAULT_SPEED[ext] || 1.0, MAX_SPEED); + elSpeedSlider.value = defaultSpeed; + elSpeedValue.textContent = defaultSpeed.toFixed(1) + 'x'; + elAudioMeta.style.display = ''; + elSpeedControl.style.display = ''; + + if (ext === 'pcm') { + elMetaSampleRate.textContent = '16000 Hz (假设)'; + const durationSec = file.size / (16000 * 2 * 1); + elMetaDuration.textContent = formatDuration(durationSec) + ' (估算)'; + } else if (ext === 'wav') { + try { + const headerBuf = await file.slice(0, 44).arrayBuffer(); + const view = new DataView(headerBuf); + const sampleRate = view.getUint32(24, true); + const bitsPerSample = view.getUint16(34, true); + const numChannels = view.getUint16(22, true); + const byteRate = sampleRate * numChannels * (bitsPerSample / 8); + const dataSize = file.size - 44; + const durationSec = dataSize / byteRate; + elMetaSampleRate.textContent = sampleRate + ' Hz, ' + bitsPerSample + 'bit, ' + numChannels + 'ch'; + elMetaDuration.textContent = formatDuration(durationSec); + } catch { + elMetaSampleRate.textContent = '解析失败'; + elMetaDuration.textContent = '—'; + } + } else { + elMetaSampleRate.textContent = '—'; + elMetaDuration.textContent = '解析中...'; + try { + const url = URL.createObjectURL(file); + const audio = new Audio(); + audio.preload = 'metadata'; + audio.src = url; + await new Promise((resolve, reject) => { + audio.onloadedmetadata = resolve; + audio.onerror = reject; + setTimeout(reject, 5000); + }); + elMetaDuration.textContent = formatDuration(audio.duration); + URL.revokeObjectURL(url); + } catch { + elMetaDuration.textContent = '无法解析'; + } + } +} + +// ===== 复制和提示 ===== +function showToast(message, isError) { + const toast = document.createElement('div'); + toast.className = 'copy-toast' + (isError ? ' copy-toast-error' : ''); + toast.textContent = message; + document.body.appendChild(toast); + requestAnimationFrame(() => { toast.classList.add('visible'); }); + setTimeout(() => { + toast.classList.remove('visible'); + setTimeout(() => toast.remove(), 300); + }, 1500); +} + +function handleCopyClick(btn, textEl) { + const text = textEl.textContent.trim(); + if (!text || text === '—') { showToast('暂无内容可复制', true); return; } + navigator.clipboard.writeText(text).then(() => { + btn.textContent = '✅'; + showToast('已复制到剪贴板'); + setTimeout(() => { btn.textContent = '📋'; }, 1500); + }).catch(() => showToast('复制失败', true)); +} + +elBtnCopyVoiceId.addEventListener('click', () => handleCopyClick(elBtnCopyVoiceId, elVoiceIdDisplay)); + +// ===== WAV 导出 ===== +function buildWavBlob(pcmChunks) { + let totalLen = 0; + for (const c of pcmChunks) totalLen += c.byteLength; + const buf = new ArrayBuffer(44 + totalLen); + const view = new DataView(buf); + const writeStr = (off, str) => { for (let i = 0; i < str.length; i++) view.setUint8(off + i, str.charCodeAt(i)); }; + const sampleRate = 16000, bitsPerSample = 16, numChannels = 1; + const byteRate = sampleRate * numChannels * bitsPerSample / 8; + const blockAlign = numChannels * bitsPerSample / 8; + writeStr(0, 'RIFF'); + view.setUint32(4, 36 + totalLen, true); + writeStr(8, 'WAVE'); + writeStr(12, 'fmt '); + view.setUint32(16, 16, true); + view.setUint16(20, 1, true); + view.setUint16(22, numChannels, true); + view.setUint32(24, sampleRate, true); + view.setUint32(28, byteRate, true); + view.setUint16(32, blockAlign, true); + view.setUint16(34, bitsPerSample, true); + writeStr(36, 'data'); + view.setUint32(40, totalLen, true); + let offset = 44; + for (const chunk of pcmChunks) { + const src = new Uint8Array(chunk instanceof ArrayBuffer ? chunk : chunk.buffer || chunk); + new Uint8Array(buf, offset).set(src); + offset += src.byteLength; + } + return new Blob([buf], { type: 'audio/wav' }); +} + +elBtnExportWav.addEventListener('click', () => { + if (audioChunks.length === 0) { showToast('暂无音频数据', true); return; } + const blob = buildWavBlob(audioChunks); + const url = URL.createObjectURL(blob); + const a = document.createElement('a'); + a.href = url; + a.download = (currentVoiceId || 'audio') + '.wav'; + document.body.appendChild(a); + a.click(); + a.remove(); + URL.revokeObjectURL(url); + showToast('WAV 已导出'); +}); + +// ===== 辅助函数 ===== +function formatTime(ms) { + const totalSec = Math.floor(ms / 1000); + const min = String(Math.floor(totalSec / 60)).padStart(2, '0'); + const sec = String(totalSec % 60).padStart(2, '0'); + return `${min}:${sec}`; +} + +function formatTimeRange(startMs, endMs) { + return `${formatTime(startMs)} - ${formatTime(endMs)}`; +} + +function setStatus(state, text) { + elStatusDot.className = 'status-dot ' + state; + elStatusText.textContent = text; +} + +// ===== 渲染字幕(关闭说话人分离) ===== +// 每个 sentence_id 对应一个独立气泡: +// - 中间态(sentence_type===0):实时更新该气泡文本,末尾加 "..." 表示未定。 +// - 稳态(sentence_type===1):去掉 "..." 并定格,下一个 sentence_id 自动创建新气泡。 +function renderSubtitle(sentence) { + const id = 'subtitle-' + sentence.sentence_id; + const isInterim = sentence.sentence_type === 0; + let entry = sentenceMap[id]; + + if (!entry) { + const el = document.createElement('div'); + el.className = 'subtitle-item subtitle-interim'; + el.id = id; + const timeSpan = document.createElement('span'); + timeSpan.className = 'subtitle-time'; + el.appendChild(timeSpan); + const textSpan = document.createElement('span'); + textSpan.className = 'subtitle-text'; + el.appendChild(textSpan); + elResultArea.appendChild(el); + entry = { el: el }; + sentenceMap[id] = entry; + } + + const el = entry.el; + el.querySelector('.subtitle-time').textContent = formatTimeRange(sentence.start_time, sentence.end_time); + const textSpan = el.querySelector('.subtitle-text'); + if (isInterim) { + textSpan.textContent = sentence.sentence + ' ...'; + textSpan.className = 'subtitle-text interim'; + } else { + textSpan.textContent = sentence.sentence; + textSpan.className = 'subtitle-text'; + el.className = 'subtitle-item'; // 去掉 interim 类名,定格 + } + elResultArea.scrollTop = elResultArea.scrollHeight; +} + +// ===== 渲染说话人气泡 ===== +// 未确认片段独立展示,不能临时塞进上一位说话人的气泡。 +function renderBubble(sentence) { + const id = 'sent-' + sentence.sentence_id; + const speakerId = Number(sentence.speaker_id); + const trusted = Number.isInteger(speakerId) && speakerId >= 0 + && ['fresh', 'confirmed'].includes(sentence.speaker_evidence); + const isInterim = sentence.sentence_type === 0; + let entry = sentenceMap[id]; + if (!entry) { + const el = document.createElement('div'); + el.id = id; + const wrapper = document.createElement('div'); + wrapper.className = 'bubble-wrapper'; + const header = document.createElement('div'); + header.className = 'bubble-header'; + for (const name of ['speaker-badge', 'speaker-name', 'bubble-time']) { + const span = document.createElement('span'); + span.className = name; + header.appendChild(span); + } + const body = document.createElement('div'); + body.className = 'bubble-body'; + wrapper.append(header, body); + el.appendChild(wrapper); + elResultArea.appendChild(el); + entry = { el }; + sentenceMap[id] = entry; + } + if (trusted && !(speakerId in speakerOrderMap)) speakerOrderMap[speakerId] = speakerOrderCounter++; + const order = trusted ? speakerOrderMap[speakerId] : 0; + const color = order % SPEAKER_COLORS.length; + const el = entry.el; + el.className = trusted ? `bubble-row speaker-${order % 2 ? 'right' : 'left'} speaker-${color}` + : 'bubble-row speaker-left speaker-unknown'; + el.querySelector('.speaker-badge').className = 'speaker-badge speaker-color-' + (trusted ? color : 'unknown'); + // 未获得当前片段的可靠声纹证据时,标题保持简短;详细原因放到悬停提示, + // 这样不会把“有效语音不足……”等内部诊断信息挤进说话人名称区域。 + const speakerName = el.querySelector('.speaker-name'); + speakerName.textContent = trusted + ? (sentence.speaker_name || `说话人 ${speakerId + 1}`) + : '未知说话人'; + speakerName.title = trusted ? '' : (sentence.speaker_reason || '未匹配到说话人'); + el.querySelector('.bubble-time').textContent = formatTimeRange(sentence.start_time, sentence.end_time); + const body = el.querySelector('.bubble-body'); + body.textContent = sentence.sentence + (isInterim ? ' ...' : ''); + body.className = 'bubble-body' + (isInterim ? ' interim' : ''); + elResultArea.scrollTop = elResultArea.scrollHeight; +} + +// 按完整快照重建相邻块;序号防止两个后台 worker 的旧快照覆盖新状态。 +function renderDisplayState(msg, useSpeaker) { + if (msg.revision != null && msg.revision <= displayRevision) return; + if (msg.revision != null) displayRevision = msg.revision; + elResultArea.replaceChildren(); + sentenceMap = {}; + const raw = msg.raw_segments || msg.sentences || []; + if (useSpeaker) { + for (const block of msg.display_blocks || []) renderBubble({ ...block, sentence_id: block.block_id }); + const confirmed = raw.filter(s => s.speaker_status === 'confirmed').length; + const failed = raw.filter(s => ['service_error', 'service_unavailable', 'no_embedding', 'evidence_rejected'].includes(s.speaker_status)).length; + elSpeakerStatus.textContent = `说话人:已确认 ${confirmed} / ${raw.length} 段` + (failed ? `,${failed} 段未识别成功(原因见气泡及日志)` : ''); + } else { + raw.forEach(renderSubtitle); + elSpeakerStatus.textContent = '说话人分离已关闭'; + } +} + +elBtnStart.addEventListener('click', () => { + if (inputMode === 'file' && !selectedFile) return; + startRecognition(); +}); + +async function startRecognition() { + if (ws) return; + if (inputMode === 'file' && selectedFile) { + const ext = getFileExt(selectedFile.name); + if (!STREAMABLE_AUDIO_EXTENSIONS.has(ext)) { + showToast('实时流式测试只支持 PCM 或 WAV,请转换后再试', true); + return; + } + } + + elResultArea.innerHTML = ''; + elLogArea.innerHTML = ''; + sentenceMap = {}; + speakerOrderMap = {}; + speakerOrderCounter = 0; + displayStateSupported = false; + displayRevision = -1; + elSpeakerStatus.textContent = '正在检查说话人服务…'; + audioChunks = []; + elBtnExportWav.disabled = true; + elResultPlaceholder?.remove(); + elResultMeta.style.display = 'flex'; + elVoiceIdDisplay.textContent = '—'; + setStatus('connecting', '连接中...'); + + elBtnStart.disabled = true; + elBtnStop.disabled = false; + sending = true; + + const currentSession = ++sessionId; + const useSpeaker = elSpeakerDiarization.checked; + let receivedTerminal = false; + + // 构造 WebSocket 首条 start 消息。 + let voiceFormat = 0, fileName = '', speedFactor = 0; + if (inputMode === 'file') { + const ext = getFileExt(selectedFile.name); + voiceFormat = EXT_FORMAT_MAP[ext] || 0; + fileName = selectedFile.name; + speedFactor = Math.min(parseFloat(elSpeedSlider.value) || 1.0, MAX_SPEED); + } + + const startPayload = { + type: 'start', + model: elEngineModel.value, + model_service_url: elModelServiceUrl.value, + display_merge: elDisplayMerge.checked, + speaker_diarization: useSpeaker ? 1 : 0, + sentence_strategy: parseInt(elSentenceStrategy.value), + source: inputMode, + voice_format: voiceFormat, + file_name: fileName, + speed_factor: speedFactor + }; + + const protocol = location.protocol === 'https:' ? 'wss:' : 'ws:'; + ws = new WebSocket(`${protocol}//${location.host}/ws`); + ws.binaryType = 'arraybuffer'; + + ws.onopen = () => { + if (currentSession !== sessionId) return; + ws.send(JSON.stringify(startPayload)); + // 在连接尚未建立时点击停止,也要在 start 后补发停止信号。 + if (!sending) ws.send(JSON.stringify({ type: 'stop' })); + }; + + ws.onmessage = (event) => { + if (currentSession !== sessionId) return; + const msg = JSON.parse(event.data); + if (msg.type === 'end' || msg.type === 'error') receivedTerminal = true; + if (msg.type !== 'sentences') { + console.log('[ws] type=' + msg.type, msg); + } + appendLog(msg); + handleServerMessage(msg, useSpeaker); + }; + + ws.onerror = (err) => { + if (currentSession !== sessionId) return; + console.error('WebSocket error:', err); + setStatus('error', '连接错误'); + stopMicCapture(); + resetControls(); + }; + + ws.onclose = () => { + if (currentSession !== sessionId) return; + stopMicCapture(); + if (!receivedTerminal) setStatus('error', '连接中断,最终识别结果可能尚未完成'); + if (audioChunks.length > 0) elBtnExportWav.disabled = false; + ws = null; + resetControls(); + }; +} + +function handleServerMessage(msg, useSpeaker) { + switch (msg.type) { + case 'voice_id': + if (msg.voice_id) { + currentVoiceId = msg.voice_id; + elVoiceIdDisplay.textContent = msg.voice_id; + } + break; + + case 'start': + displayStateSupported = Boolean(msg.display_state_supported); + currentVoiceId = msg.session_id; + elVoiceIdDisplay.textContent = currentVoiceId || '—'; + elSpeakerStatus.textContent = useSpeaker + ? `说话人服务:${msg.speaker_service_url || '未配置'};片段结束后提取声纹` + : '说话人分离已关闭'; + if (!sending) break; + setStatus('running', '识别中...'); + if (inputMode === 'file') sendAudioFile(selectedFile).catch(handleInputError); + else startMicCapture().catch(handleInputError); + break; + + case 'display_state': + renderDisplayState(msg, useSpeaker); + break; + + case 'draining': + setStatus('running', msg.message || '等待最终识别结果…'); + break; + + case 'sentences': + if (displayStateSupported) break; + if (msg.sentences) { + msg.sentences.forEach(s => { + if (useSpeaker) renderBubble(s); + else renderSubtitle(s); + }); + } + break; + + case 'speaker_warning': + // ASR 仍可继续输出,但必须让测试人员立即知道说话人链路没有生效。 + elSpeakerStatus.textContent = '说话人服务异常:' + msg.message; + showToast('说话人服务异常,详见状态和片段原因', true); + break; + + case 'end': + if (msg.display_blocks) renderDisplayState(msg, useSpeaker); + setStatus('done', '识别完成'); + sending = false; + if (audioChunks.length > 0) elBtnExportWav.disabled = false; + resetControls(); + if (ws) { ws.close(); ws = null; } + break; + + case 'error': + setStatus('error', `错误: ${msg.message}`); + sending = false; + stopMicCapture(); + if (ws) { ws.close(); ws = null; } + resetControls(); + break; + } +} + +// ===== 停止识别 ===== +elBtnStop.addEventListener('click', () => stopRecognition()); + +function stopRecognition() { + sending = false; + stoppingByUser = true; + stopMicCapture(); + setStatus('running', '停止中...'); + elBtnStop.disabled = true; + + if (ws && ws.readyState === WebSocket.OPEN) { + try { ws.send(JSON.stringify({ type: 'stop' })); } catch (e) {} + } + + // 等待服务端排空 ASR/声纹队列后发送 end,不能用五秒计时器截断更新。 + +} + +function resetControls() { + sending = false; + stoppingByUser = false; + currentVoiceId = null; + if (inputMode === 'file') { + elBtnStart.disabled = !selectedFile; + } else { + elBtnStart.disabled = false; + } + elBtnStop.disabled = true; +} + +// ===== 发送音频文件 ===== +// 按 16KB 切片发送,并按照音频实际时长等待,确保文件模式也是真实的 +// 实时输入,而不是瞬间上传完整文件后再由服务端批量切片。 +const UPLOAD_CHUNK_SIZE = 16000; +async function sendAudioFile(file) { + const ownerSession = sessionId; + const buffer = await file.arrayBuffer(); + if (ownerSession !== sessionId || !sending) return; + const totalBytes = buffer.byteLength; + let offset = 0; + const ext = getFileExt(file.name); + const isPcm = (ext === 'pcm'); + let bytesPerSecond = 16000 * 2; + if (ext === 'wav' && totalBytes >= 44) { + const header = new DataView(buffer, 0, 44); + const byteRate = header.getUint32(28, true); + if (byteRate > 0) bytesPerSecond = byteRate; + } + const speedFactor = Math.max(parseFloat(elSpeedSlider.value) || 1.0, 0.1); + while (ownerSession === sessionId && offset < totalBytes && sending && ws && ws.readyState === WebSocket.OPEN) { + const end = Math.min(offset + UPLOAD_CHUNK_SIZE, totalBytes); + const chunk = buffer.slice(offset, end); + // 仅 PCM 数据可直接拼成 WAV 导出;当前实时模式不会接收压缩格式。 + if (isPcm) audioChunks.push(chunk.slice(0)); + ws.send(chunk); + offset = end; + const chunkDurationMs = (chunk.byteLength / bytesPerSecond) * 1000 / speedFactor; + await new Promise(r => setTimeout(r, Math.max(0, Math.round(chunkDurationMs)))); + } + if (ownerSession === sessionId && ws && ws.readyState === WebSocket.OPEN && sending) { + sending = false; + setStatus('running', '音频已发送,等待最终结果…'); + ws.send(JSON.stringify({ type: 'eof' })); + } +} + +// ===== 麦克风采集 ===== +async function startMicCapture() { + const ownerSession = sessionId; + let stream; + try { + stream = await navigator.mediaDevices.getUserMedia({ + audio: { sampleRate: 16000, channelCount: 1, echoCancellation: true, noiseSuppression: true } + }); + } catch (err) { + if (ownerSession !== sessionId) return; + handleInputError(err); + console.error('getUserMedia error:', err); + setStatus('error', '无法获取麦克风权限'); + elMicStatus.textContent = '无法获取麦克风: ' + err.message; + resetControls(); + return; + } + + if (ownerSession !== sessionId || !sending) { + stream.getTracks().forEach(track => track.stop()); + return; + } + micStream = stream; + micAudioContext = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: 16000 }); + const source = micAudioContext.createMediaStreamSource(micStream); + const processor = micAudioContext.createScriptProcessor(4096, 1, 1); + + processor.onaudioprocess = (e) => { + if (!sending || !ws || ws.readyState !== WebSocket.OPEN) return; + const float32 = e.inputBuffer.getChannelData(0); + const int16 = new Int16Array(float32.length); + for (let i = 0; i < float32.length; i++) { + let s = Math.max(-1, Math.min(1, float32[i])); + int16[i] = s < 0 ? s * 0x8000 : s * 0x7FFF; + } + ws.send(int16.buffer); + audioChunks.push(int16.buffer.slice(0)); + }; + + source.connect(processor); + processor.connect(micAudioContext.destination); + micWorklet = { source, processor }; + + micStartTime = Date.now(); + elMicStatus.style.display = 'none'; + elMicTimer.style.display = ''; + micTimerInterval = setInterval(() => { + const elapsed = Math.floor((Date.now() - micStartTime) / 1000); + const mm = String(Math.floor(elapsed / 60)).padStart(2, '0'); + const ss = String(elapsed % 60).padStart(2, '0'); + elMicElapsed.textContent = `${mm}:${ss}`; + }, 200); +} + +function stopMicCapture() { + if (micWorklet) { + micWorklet.processor.disconnect(); + micWorklet.source.disconnect(); + micWorklet = null; + } + if (micAudioContext) { + micAudioContext.close().catch(() => {}); + micAudioContext = null; + } + if (micStream) { + micStream.getTracks().forEach(t => t.stop()); + micStream = null; + } + if (micTimerInterval) { + clearInterval(micTimerInterval); + micTimerInterval = null; + } + elMicTimer.style.display = 'none'; + elMicStatus.style.display = ''; + elMicStatus.textContent = '点击下方按钮开始录音'; + elMicElapsed.textContent = '00:00'; +} + +// 展示实际部署端点及模型,避免沿用旧 SDK 的无效引擎配置。 +fetch('/api/config').then(response => response.json()).then(config => { + if (!ws) { + elEngineModel.value = config.model; + elModelServiceUrl.value = config.model_service_url; + elSpeakerStatus.textContent = '说话人辅助服务:' + (config.speaker_service_url || '未配置'); + } +}).catch(error => { elSpeakerStatus.textContent = '读取服务配置失败:' + error.message; }); + +// 输入端失败必须释放空会话,避免用户再次开始时留下旧连接。 +function handleInputError(error) { + sending = false; + setStatus('error', error.message); + stopMicCapture(); + if (ws) { ws.close(); ws = null; } + resetControls(); +} diff --git a/realtime_websocket/static/index.html b/realtime_websocket/static/index.html new file mode 100644 index 0000000..d34384d --- /dev/null +++ b/realtime_websocket/static/index.html @@ -0,0 +1,119 @@ + + +
+ + +正在读取服务配置…
+ +