FunASR 共绩算力平台开源镜像部署方案

FunASR 是一个基于共绩算力平台的开源语音识别服务部署方案，支持多种语音识别模型的无服务器化部署。通过支持工业级语音识别模型的训练和微调，研究人员和开发者可以更方便地进行语音识别模型的研究和生产，促进语音识别生态的发展。

功能特性 | 快速开始 | API 规范 | 部署指南 | 配置说明 | 模型库 | 联系我们

功能特性

FunASR 是一个基础的语音识别工具包，提供多种功能，包括语音识别 (ASR)、语音活动检测 (VAD)、标点符号恢复、语言模型、说话人验证、说话人分离和多说话人 ASR。FunASR 提供便捷的脚本和教程，支持预训练模型的推理和微调。

我们已经在 ModelScope 和 HuggingFace 上发布了大量学术和工业预训练模型，可以通过我们的 模型库 访问。代表性的 Paraformer-large 是一个非自回归端到端语音识别模型，具有高精度、高效率、便捷部署等优势，支持快速构建语音识别服务。

快速开始

系统要求

• 操作系统: Linux, macOS, Windows
• 内存: 最少 4GB，推荐 8GB+
• 存储: 至少 10GB 可用空间（用于模型存储）
• Docker: 版本 20.10+

GPU 支持（可选）

• NVIDIA GPU: 支持 CUDA 的显卡
• CUDA: 版本 11.0+
• nvidia-docker: Docker GPU 运行时

一键部署

# 克隆项目
git clone https://github.com/slmnb-lab/FunASR.git
cd FunASR

# 给脚本执行权限
chmod +x start_funasr.sh

# 启动 CPU 版本
./start_funasr.sh

# 启动 GPU 版本（需要 NVIDIA GPU 和 Docker GPU 支持）
./start_funasr.sh --gpu

# 自定义端口
./start_funasr.sh --port 8080

# 重新构建镜像
./start_funasr.sh --rebuild

Docker Compose 部署

# 启动 CPU 版本
docker-compose up -d

# 启动 GPU 版本
docker-compose --profile gpu up -d

# 查看日志
docker-compose logs -f

直接使用 Docker

# 构建镜像
docker build -t funasr:latest .

# 启动服务
docker run -d \
  --name funasr-server \
  -p 10095:10095 \
  -v $(pwd)/models:/app/models \
  funasr:latest

容器化部署 FunASR

此镜像提供了标准化的 API 接口，让您能够便捷地通过 API 调用方式访问和使用所有功能。

本指南详细阐述了在共绩算力平台上，高效部署与使用 FunASR API 项目的技术方案。FunASR 是一个基本的语音识别工具包，提供多种功能，包括语音识别（ASR）、语音活动检测（VAD）、标点符号恢复、语言模型、说话人验证、说话人分类和多说话者 ASR。

服务简介与应用场景

FunASR 是阿里巴巴开源的语音识别工具包，支持多场景一键部署，适合会议转写、客服质检、字幕生成、语音助手、医疗法律文档整理、多语言识别等实际业务需求。

典型应用场景

会议/讲座/采访转写：实时或离线将录音转为带时间戳文本，便于归档检索。

智能客服与语音质检：实时识别对话内容，支持热词、敏感词检测、服务分析。

媒体内容字幕生成：批量处理音视频，自动生成带时间轴字幕文本。

语音助手与人机交互：低延迟识别，适合智能音箱、车载、移动端。

医疗、法律文档整理：支持专业热词，提升术语识别准确率。

多语言识别与翻译前处理：支持多语种、口音，适合国际会议、跨国企业。

API 规范

该服务暴露标准的共绩算力平台无服务器端点 (/run, /runsync, /health)。默认情况下，音频文件以 base64 字符串形式返回。您可以通过设置特定的环境变量来配置服务将结果上传到 S3 存储桶（请参阅配置指南）。

使用 /runsync 端点进行同步请求，等待作业完成并直接返回结果。使用 /run 端点进行异步请求，立即返回作业 ID；您需要单独轮询 /status 端点以获取结果。

输入格式

{
  "input": {
    "audio": {
      "name": "audio_input.wav",
      "data": "data:audio/wav;base64,UklGRnoGAABXQVZFZm10IBAAAAABAAEAQB8AAEAfAAABAAgAZGF0YQoGAACBhYqFbF1fdJivrJBhNjVgodDbq2EcBj+a2/LDciUFLIHO8tiJNwgZaLvt559NEAxQp+PwtmMcBjiR1/LMeSwFJHfH8N2QQAoUXrTp66hVFApGn+DyvmwhBSuBzvLZiTYIG2m98OScTgwOUarm7blmGgU7k9n1unEiBC13yO/eizEIHWq+8+OWT..."
    },
    "model": "paraformer-zh",
    "vad_model": "fsmn-vad",
    "punc_model": "ct-punc",
    "language": "zh",
    "use_itn": true,
    "batch_size_s": 300
  }
}

下表描述了 input 对象中的字段：

字段路径	类型	必需	描述
input	Object	是	包含请求数据的顶级对象
input.audio	Object	是	音频文件数据
input.model	String	否	使用的模型名称，默认为 "paraformer-zh"
input.vad_model	String	否	VAD 模型名称
input.punc_model	String	否	标点符号模型名称
input.language	String	否	语言代码，默认为 "zh"
input.use_itn	Boolean	否	是否使用逆文本归一化
input.batch_size_s	Integer	否	批处理大小（秒）

input.audio 对象

input.audio 对象必须包含：

字段名	类型	必需	描述
name	String	是	音频文件名
data	String	是	Base64 编码的音频字符串。数据 URI 前缀（如 data:audio/wav;base64,）是可选的，会被正确处理

注意：大小限制：攻击云平台端点有请求大小限制（例如 /run 为 10MB，/runsync 为 20MB）。大型 base64 输入音频可能超过这些限制。请参阅攻击云平台文档。

输出格式

{
  "id": "sync-uuid-string",
  "status": "COMPLETED",
  "output": {
    "text": "今天天气真不错，适合出去散步。",
    "language": "zh",
    "timestamps": [
      {
        "start": 0.0,
        "end": 2.5,
        "text": "今天天气真不错"
      },
      {
        "start": 2.5,
        "end": 5.0,
        "text": "适合出去散步"
      }
    ],
    "confidence": 0.95
  },
  "delayTime": 123,
  "executionTime": 4567
}

字段路径	类型	必需	描述
output	Object	是	包含作业执行结果的顶级对象
output.text	String	是	识别出的文本内容
output.language	String	否	检测到的语言
output.timestamps	Array	否	时间戳信息
output.confidence	Number	否	置信度分数
output.errors	Array	否	如果处理过程中发生非致命错误或警告时出现

部署与接入

共绩算力平台部署步骤

1. 登录共绩算力控制台，在首页点击"弹性部署服务"。

2. 选择 GPU 型号（推荐 1 卡 4090）。

3. 在"服务配置"选择 FunASR API 官方镜像。

4. 点击"部署服务"，平台自动拉取镜像并启动容器。

5. 部署完成后，在"快捷访问"中复制端口为 10095 的公网访问链接，即为 API 服务地址。

点击 10095 的端口会进入不到具体页面，这个是正常现象，直接复制地址到您的对应服务中使用 API 服务即可，可以参考下面的示例程序。

使用方法

要与您部署的共绩算力平台端点交互：

1. 获取 API 密钥：在共绩算力平台用户设置中生成密钥（API 密钥 部分）
2. 获取端点 ID：在无服务器端点页面或端点的 概览 页面上找到您的端点 ID

生成语音识别结果（同步示例）

向 /runsync 端点发送请求（等待完成）。替换 <api_key> 和 <endpoint_id>。-d 值应包含上述描述的 JSON 输入。

curl -X POST \
  -H "Authorization: Bearer <api_key>" \
  -H "Content-Type: application/json" \
  -d '{"input":{"audio":{"name":"test.wav","data":"..."},"model":"paraformer-zh"}}' \
  https://api.gongjiyun.com/v2/<endpoint_id>/runsync

您也可以使用 /run 端点进行异步作业，然后轮询 /status 以查看作业何时完成。或者您可以在请求中添加 webhook 以在作业完成时收到通知。

API 协议与参数

通信协议

WebSocket 协议，服务地址：wss://d07171047-funasr-online-server-318-kmkzg1m4-10095.550c.cloud

典型调用流程

1. 连接 WebSocket 服务。
2. 发送配置参数（JSON）。
3. 分块发送音频数据（bytes）。
4. 发送结束标志：{"is_speaking": false}
5. 接收识别结果（JSON）。

配置参数说明

参数	类型	说明
chunk_size	array	流式模型延迟配置，数组格式，如 [5,10,5]
wav_name	string	音频文件名
is_speaking	boolean	说话状态/断句尾点标识，true 表示正在说话，false 表示语句结束
wav_format	string	音频格式，支持 pcm
chunk_interval	number	块间隔时间（单位毫秒）
itn	boolean	是否逆文本标准化开启（如将"一二三"转为"123"），true
mode	string	识别模式：2pass（双通混合模式）、online（纯流式）、offline（非流式）
hotwords	string	热词配置，JSON 字符串如：{"科技词":0.8,"人名":1.2}

典型场景与实测程序

中文音频识别实测示例（test_cn_male_9s.pcm）

import asyncio
import websockets
import json

ws_url = "wss://d07171047-funasr-online-server-318-kmkzg1m4-10095.550c.cloud"

async def test_asr(audio_file, wav_name):
    config = {
        "chunk_size": [5, 10, 5],
        "wav_name": wav_name,
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 10,
        "itn": True,
        "mode": "2pass"
    }
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(8000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print(f"{audio_file} 识别结果：", resp)
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

if __name__ == "__main__":
    asyncio.run(test_asr("test_cn_male_9s.pcm", "test_cn_male_9s"))

实际输出：

test_cn_male_9s.pcm 识别结果： {"is_final":true,"mode":"2pass-offline","stamp_sents":[...],"text":"这是一期非广告视频，给你们介绍一下我养猫。这 3 年用到过最好用的几样养猫物品。","timestamp":"[[50,150],[150,250],...]"}

主要内容：

这是一期非广告视频，给你们介绍一下我养猫。这 3 年用到过最好用的几样养猫物品。

英文音频识别实测示例（test_en_steve_jobs_10s.pcm）

import asyncio
import websockets
import json

ws_url = "wss://d07171047-funasr-online-server-318-kmkzg1m4-10095.550c.cloud"

async def test_asr(audio_file, wav_name):
    config = {
        "chunk_size": [5, 10, 5],
        "wav_name": wav_name,
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 10,
        "itn": True,
        "mode": "2pass"
    }
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(8000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print(f"{audio_file} 识别结果：", resp)
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

if __name__ == "__main__":
    asyncio.run(test_asr("test_en_steve_jobs_10s.pcm", "test_en_steve_jobs_10s"))

实际输出：

test_en_steve_jobs_10s.pcm 识别结果： {"is_final":true,"mode":"2pass-offline","stamp_sents":[...],"text":" thank you， designed the first time time， pure and and designed all in the land。 it was the first cure you，you。","timestamp":"[[270,450],[450,750],...]"}

主要内容：

thank you, designed the first time time, pure and and designed all in the land. it was the first cure you, you.

会议/讲座/采访转写（带时间戳）

import asyncio
import websockets
import json

ws_url = "wss://console.suanli.cn/serverless/idc/3576"

async def meeting_transcribe():
    config = {
        "chunk_size": [5, 10, 5],
        "wav_name": "meeting_demo",
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 10,
        "itn": True,
        "mode": "2pass"
    }
    audio_file = "meeting_8k.pcm"  # 8kHz 单声道 PCM
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(8000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print("识别结果：", resp)
                # 示例输出：{"is_final":true,"text":"欢迎参加会议...","timestamp":"[[0,800],[800,1600],...]"}
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

asyncio.run(meeting_transcribe())

智能客服与语音质检（热词增强）

import asyncio
import websockets
import json

ws_url = "wss://console.suanli.cn/serverless/idc/3576"

async def customer_service_asr():
    hotwords = {"产品 A": 30, "专有名词": 25, "投诉": 20}
    config = {
        "chunk_size": [5, 10, 5],
        "wav_name": "service_demo",
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 10,
        "itn": True,
        "mode": "online",
        "hotwords": json.dumps(hotwords, ensure_ascii=False)
    }
    audio_file = "service_call_8k.pcm"
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(8000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print("客服识别：", resp)
                # 示例输出：{"is_final":true,"text":"您好，欢迎致电产品 A 客服..."}
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

asyncio.run(customer_service_asr())

媒体内容字幕生成（带时间轴）

import asyncio
import websockets
import json

ws_url = "wss://console.suanli.cn/serverless/idc/3576"

async def subtitle_generation():
    config = {
        "chunk_size": [8, 15, 8],
        "wav_name": "media_demo",
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 10,
        "itn": True,
        "mode": "offline"
    }
    audio_file = "media_8k.pcm"
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(8000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print("字幕识别：", resp)
                # 示例输出：{"is_final":true,"text":"...","timestamp":"[[0,800],[800,1600],...]"}
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

asyncio.run(subtitle_generation())

语音助手与人机交互（低延迟实时识别）

import asyncio
import websockets
import json

ws_url = "wss://console.suanli.cn/serverless/idc/3576"

async def voice_assistant():
    config = {
        "chunk_size": [3, 6, 3],
        "wav_name": "assistant_demo",
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 5,
        "itn": True,
        "mode": "online"
    }
    audio_file = "assistant_8k.pcm"
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(4000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print("助手识别：", resp)
                # 示例输出：{"is_final":true,"text":"打开空调"}
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

asyncio.run(voice_assistant())

医疗/法律文档整理（专业热词）

import asyncio
import websockets
import json

ws_url = "wss://console.suanli.cn/serverless/idc/3576"

async def medical_legal_asr():
    hotwords = {"高血压": 30, "肝功能": 25, "原告": 20, "被告": 20}
    config = {
        "chunk_size": [5, 10, 5],
        "wav_name": "medlaw_demo",
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 10,
        "itn": True,
        "mode": "offline",
        "hotwords": json.dumps(hotwords, ensure_ascii=False)
    }
    audio_file = "medlaw_8k.pcm"
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(8000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print("专业识别：", resp)
                # 示例输出：{"is_final":true,"text":"患者高血压... 原告陈述..."}
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

asyncio.run(medical_legal_asr())

多语言识别与翻译前处理

import asyncio
import websockets
import json

ws_url = "wss://console.suanli.cn/serverless/idc/3576"

async def multilingual_asr():
    config = {
        "chunk_size": [5, 10, 5],
        "wav_name": "multi_demo",
        "is_speaking": True,
        "wav_format": "pcm",
        "chunk_interval": 10,
        "itn": True,
        "mode": "2pass"
    }
    audio_file = "multilang_8k.pcm"
    async with websockets.connect(ws_url) as ws:
        await ws.send(json.dumps(config, ensure_ascii=False))
        with open(audio_file, "rb") as f:
            while True:
                chunk = f.read(8000)
                if not chunk:
                    break
                await ws.send(chunk)
        await ws.send(json.dumps({"is_speaking": False}))
        while True:
            try:
                resp = await asyncio.wait_for(ws.recv(), timeout=5)
                print("多语言识别：", resp)
                # 示例输出：{"is_final":true,"text":"Welcome to the conference..."}
                if 'is_final' in resp and json.loads(resp).get('is_final'):
                    break
            except asyncio.TimeoutError:
                break

asyncio.run(multilingual_asr())

部署指南

共绩算力平台部署

1. 创建端点：在共绩算力平台控制台中创建新的无服务器端点
2. 选择镜像：选择 gongjiyun/funasr:latest 镜像
3. 配置环境变量：

   MODEL_DIR=/app/models
   PYTHONPATH=/app
   HF_HOME=/app/models/huggingface
   MODELSCOPE_CACHE=/app/models/modelscope
   

4. 设置端口：配置端口 10095 用于 WebSocket 服务
5. 部署：点击部署按钮完成部署

本地开发环境

# 克隆项目
git clone https://github.com/alibaba-damo-academy/FunASR.git
cd FunASR

# 安装依赖
pip install -r requirements.txt

# 启动开发服务器
python funasr_server.py --host 0.0.0.0 --port 10095

配置说明

环境变量

变量名	默认值	说明
MODEL_DIR	/app/models	模型存储目录
PYTHONPATH	/app	Python 路径
HF_HOME	/app/models/huggingface	Huggingface 缓存目录
MODELSCOPE_CACHE	/app/models/modelscope	ModelScope 缓存目录
CUDA_VISIBLE_DEVICES	0	GPU 设备 ID

端口配置

服务	默认端口	协议	说明
FunASR WebSocket	10095	WebSocket	WebSocket 服务端口
FunASR HTTP API	8000	HTTP/REST	HTTP REST API 端口
FunASR GPU	10096	WebSocket	GPU 版本端口

模型库

FunASR 在工业数据上开源了大量预训练模型。您可以在 模型许可证协议 下自由使用、复制、修改和共享 FunASR 模型。以下是一些代表性模型，更多模型请参考 模型库。

（注意：⭐ 代表 ModelScope 模型库，🤗 代表 Huggingface 模型库，🍀 代表 OpenAI 模型库）

模型名称	任务详情	训练数据	参数
SenseVoiceSmall (⭐ 🤗)	多种语音理解能力，包括 ASR、ITN、LID、SER 和 AED，支持中文、粤语、英文、日文、韩文等语言	300000 小时	234M
paraformer-zh (⭐ 🤗)	语音识别，带时间戳，非流式	60000 小时，中文	220M
paraformer-zh-streaming (⭐ 🤗)	语音识别，流式	60000 小时，中文	220M
paraformer-en (⭐ 🤗)	语音识别，无时间戳，非流式	50000 小时，英文	220M
fsmn-vad (⭐ 🤗)	语音活动检测	5000 小时，中文和英文	0.4M
ct-punc (⭐ 🤗)	标点符号恢复	100M，中文和英文	290M

常见问题与参考资料

音频需为 8kHz 单声道 PCM，建议用 ffmpeg 或 sox 转换

详细参数与高级用法见 FunASR 官方文档

GitHub：https://github.com/alibaba/FunASR

共绩算力 Open API 使用文档：https://www.gongjiyun.com/docs/y/openapi/zx3iwhbv1i8sxdkeiapcprxhn8d/