FunASR FastAPI WebSocket Service

一个基于 FunASR 和 FastAPI 构建的高性能、实时的语音识别 WebSocket 服务。该项目核心特色是支持"一发多收"的广播模式，适用于会议实时字幕、在线教育、直播转写等需要将单一音源的识别结果分发给多个客户端的场景。

✨ 功能特性

实时语音处理: 集成 FunASR 的语音活动检测（VAD）、语音识别（ASR）和声纹识别（SPK）模型。
WebSocket 流式 API: 提供低延迟、双向的实时通信接口。
"一发多收"架构:
- 发送者 (Sender): 单一客户端作为音频来源，向服务器持续发送音频流。
- 接收者 (Receiver): 多个客户端可以订阅同一个会话，实时接收广播的识别结果。
异步核心: 基于 FastAPI 和 asyncio 构建，可处理大量并发连接。
模块化设计: 清晰地分离了服务层 (server.py)、会话管理层 (ASRRunner) 和核心处理流水线 (ASRPipeline)。

📂 项目结构

.
├── main.py                     # 应用程序主入口，使用 uvicorn 启动服务
├── WEBSOCKET_API.md            # WebSocket API 详细使用文档和示例
├── src
│   ├── server.py               # FastAPI 应用核心，管理生命周期和全局资源
│   ├── runner
│   │   └── ASRRunner.py        # 核心会话管理器，负责创建和协调识别会话 (SAR)
│   ├── pipeline
│   │   └── ASRpipeline.py      # 同步的、基于线程的语音处理流水线
│   ├── functor                   # VAD, ASR, SPK 等原子操作的实现
│   ├── websockets
│   │   ├── adapter.py          # WebSocket 适配器，处理数据格式转换
│   │   ├── endpoint
│   │   │   └── asr_endpoint.py # WebSocket 的业务逻辑端点
│   │   └── router.py           # WebSocket 路由
│   └── ...
└── tests
    ├── runner
    │   └── asr_runner_test.py  # ASRRunner 的单元测试 (异步)
    └── websocket
        └── websocket_asr.py    # WebSocket 服务的端到端测试

🚀 快速开始

1. 环境与依赖

Python 3.8+
项目依赖项记录在 requirements.txt 文件中。

2. 安装

建议在虚拟环境中安装依赖。在项目根目录下，运行：

pip install -r requirements.txt

3. 运行服务

执行主入口文件来启动 FastAPI 服务：

python main.py

服务启动后，将监听 http://0.0.0.0:8000。

💡 如何使用

服务通过 WebSocket 提供，客户端通过 session_id 来创建或加入一个识别会话，并通过 mode 参数声明自己的角色（sender 或 receiver）。

详细的 API 说明、URL 格式以及 Python 和 JavaScript 的客户端连接示例，请参阅：

➡️ WEBSOCKET_API.md

🔬 测试

项目提供了两种测试方式来验证其功能。

1. 端到端 WebSocket 测试

此测试会模拟一个 sender 和一个 receiver，完整地测试一次识别会话。

前提: 确保 FastAPI 服务正在运行。

python main.py

在项目根目录下执行：

python tests/websocket/websocket_asr.py

2. ASRRunner 单元测试

此测试针对核心的 ASRRunner 组件进行，验证其异步逻辑。

执行测试：

python test_main.py

3.4 KiB Raw Blame History Unescape Escape