1. 语音质检系统的整体架构与选型思路
语音质检这件事,放在几年前还是大厂专属——一套商用质检系统动辄几十万,中小团队根本碰不起。但现在情况完全变了,开源语音识别模型加上软交换平台,自己搭一套高精度质检系统的门槛已经降到一个人一周就能跑通的程度。我这次要聊的,就是用 FunASR 做识别引擎、FreeSWITCH 做通话控制中枢,从零搭一套能落地跑的语音质检系统。
先说清楚这套系统到底解决什么问题。呼叫中心、客服团队、电销部门每天产生大量通话录音,传统做法是抽检——质检员随机听几通,打个分写个评语。这种模式覆盖率通常不到 5%,漏掉的问题一大堆。自动化质检要做的,是把 100% 的通话转成文字,再基于文字做关键词命中、情绪判断、话术合规检查。核心链路就三步:通话接入、语音转写、质检分析。FreeSWITCH 负责第一步,FunASR 负责第二步,第三步可以接规则引擎或者大模型。
为什么选 FunASR 而不是其他识别方案?我对比过几套主流开源方案,FunASR 的优势在于中文场景的准确率和部署友好度。它背后是达摩院开源的工业级模型,Paraformer 系列在中文电话场景下的字准率能到 95% 以上,而且支持流式和非流式两种模式。流式适合实时质检,非流式适合录音文件批量转写。更关键的是它提供了 WebSocket 服务接口,这意味着我可以把识别能力做成一个独立服务,FreeSWITCH 那边通过 WebSocket 把音频流推过来就行,两边解耦得很干净。
FreeSWITCH 这边,选它的理由更直接——它是目前开源软交换里最成熟、社区最活跃的。呼叫中心场景下的各种需求,比如录音、转接、会议、IVR,它都能覆盖。而且它原生支持 WebSocket 模块,可以跟外部服务做实时音频交互。这就为语音质检提供了天然的接入点:通话建立后,把媒体流通过 WebSocket 转发给 FunASR,识别结果实时回传,质检逻辑就能在线跑。
Docker 在这套架构里扮演的是“环境标准化”的角色。FunASR 的依赖比较重,PyTorch、ModelScope、各种音频处理库,直接装在宿主机上容易跟其他服务冲突。用 Docker 把 FunASR 服务封起来,镜像一构建,换台机器直接跑,省掉大量环境调试时间。FreeSWITCH 虽然也可以容器化,但考虑到它要处理 RTP 媒体流和网络端口,我建议生产环境还是裸机部署或者用独立虚拟机,Docker 跑 FunASR 和质检后端就够了。
整套架构的数据流向是这样的:用户拨入 FreeSWITCH,FreeSWITCH 建立通话并启动录音,同时通过 mod_ws 或者 ESL 把音频流推给一个中间服务,中间服务再转发给 FunASR 的 WebSocket 接口。FunASR 返回识别文本,中间服务把文本写入消息队列或者数据库,质检引擎消费文本做分析。如果要做实时质检,识别结果可以边出边分析;如果做离线质检,就等通话结束后批量处理录音文件。
这个架构的好处是每一层都可以独立扩展。FunASR 识别慢就加 GPU 节点,质检规则复杂就单独拆服务,FreeSWITCH 并发不够就加媒体服务器。对于中小团队来说,初期一台带 GPU 的机器就能跑起来,后续按需扩容。
2. FunASR 服务部署与 WebSocket 接口配置
2.1 Linux 环境下 FunASR 的 Docker 化部署
FunASR 官方提供了 Docker 镜像,但直接用官方镜像有时候会遇到模型下载慢、端口映射不灵活的问题。我的做法是自己写一个 Dockerfile,基于官方镜像做一层封装,把模型预下载进去,这样启动时不用等模型拉取。
先看基础环境。我用的是一台 Ubuntu 22.04 的机器,带一张 RTX 3060 12G。FunASR 的 Paraformer-large 模型推理大概需要 4G 显存,12G 足够跑并发。如果只有 CPU,也能跑,但实时率会差很多,适合离线转写场景。
Dockerfile 大概长这样:
FROM registry.cn-hangzhou.aliyuncs.com/funasr_repo/funasr:funasr-runtime-sdk-cpu-0.4.6 RUN mkdir -p /workspace/models WORKDIR /workspace # 预下载模型,避免启动时等待 RUN python -c "from modelscope import snapshot_download; \ snapshot_download('iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch', \ cache_dir='/workspace/models')" EXPOSE 10095 CMD ["python", "-m", "funasr.bin.asr_server", \ "--model-dir", "/workspace/models", \ "--port", "10095", \ "--ngpu", "1"]这里有几个点要注意。第一,基础镜像我选的是 CPU 版本,因为官方 GPU 镜像的 CUDA 版本有时候跟宿主机驱动不匹配,自己装反而更可控。第二,模型下载这一步放在构建阶段,镜像会大一些,但启动快。第三,端口 10095 是 FunASR 运行时的默认 WebSocket 端口,后面 FreeSWITCH 那边要对应上。
构建命令:
docker build -t funasr-server:latest .启动容器的时候,GPU 透传要加上--gpus all,模型目录挂载出来方便更新:
docker run -d --name funasr \ --gpus all \ -p 10095:10095 \ -v /data/funasr/models:/workspace/models \ funasr-server:latest启动后验证服务是否正常,可以用 Python 的 websockets 库写个简单客户端测试:
import asyncio import websockets import json async def test_funasr(): uri = "ws://127.0.0.1:10095" async with websockets.connect(uri) as ws: # 发送配置 await ws.send(json.dumps({ "mode": "offline", "chunk_size": [5, 10, 5], "wav_name": "test", "is_speaking": True })) # 发送音频数据(这里省略实际音频读取) # await ws.send(audio_bytes) await ws.send(json.dumps({"is_speaking": False})) result = await ws.recv() print(result) asyncio.run(test_funasr())如果返回了识别结果,说明服务正常。这一步很关键,很多人卡在 FunASR 服务没起来就去调 FreeSWITCH,结果排查半天发现是识别服务本身的问题。
2.2 WebSocket 协议细节与音频格式要求
FunASR 的 WebSocket 接口对音频格式有明确要求:16kHz 采样率、16bit 位深、单声道 PCM。FreeSWITCH 默认的媒体流是 8kHz 的,所以中间需要做重采样。这个重采样可以在 FreeSWITCH 侧做,也可以在中间服务做。我建议在中间服务做,因为 FreeSWITCH 的 resample 模块在高并发下会吃 CPU。
WebSocket 消息分两类:控制消息和音频数据。控制消息是 JSON 格式,音频数据是二进制帧。第一次连接后要先发一个配置消息,告诉 FunASR 用哪种模式、chunk 大小是多少。chunk_size 这个参数很关键,它决定了流式识别的延迟和准确率平衡。[5, 10, 5]是官方推荐值,对应 600ms 的音频块。如果追求低延迟,可以改成[4, 8, 4],但准确率会略降。
音频数据发送时,每次发一个 chunk 的 PCM 数据。FunASR 会边收边识别,返回中间结果。中间结果的 JSON 里有个is_final字段,为 true 时表示这句话识别完了。质检逻辑可以基于 final 结果做,也可以基于中间结果做实时监控。
注意:FunASR 的 WebSocket 服务默认不支持多路复用,一个连接对应一路音频。如果 FreeSWITCH 有 100 路并发通话,就需要 100 个 WebSocket 连接。这时候中间服务的连接池管理就很重要,不能每通电话都新建连接,要复用。
2.3 模型选择与热词定制
FunASR 支持热词定制,这对质检场景特别有用。比如你的业务里有很多专有名词、产品名、人名,通用模型可能识别错,但把热词加进去,准确率能提升一大截。
热词配置在启动参数里加--hotword指定一个文件,文件格式是每行一个词,可以带权重:
花呗 20 借呗 20 芝麻信用 15权重越高,模型越倾向于识别成这个词。我实测下来,把业务高频词加进去,相关语句的识别准确率能从 85% 提到 95% 以上。
模型选择上,Paraformer-large 是精度最高的,但推理慢。如果并发高、对延迟敏感,可以用 Paraformer-streaming,精度略低但实时性好。我的建议是实时质检用 streaming 模型,离线质检用 large 模型,两套模型可以同时部署,中间服务根据场景路由。
3. FreeSWITCH 侧的通话接入与音频转发
3.1 FreeSWITCH 安装与基础配置
FreeSWITCH 在 Linux 上的安装,我推荐用官方源或者编译安装。编译安装虽然慢,但模块可控,后面要加 mod_ws 或者自定义模块方便。Ubuntu 下编译大概需要 20 分钟,依赖装全就行。
apt-get install -y git build-essential autoconf automake libtool \ libncurses5-dev libssl-dev libpcre3-dev libspeexdsp-dev \ libldns-dev libedit-dev libsqlite3-dev libcurl4-openssl-dev git clone https://github.com/signalwire/freeswitch.git cd freeswitch ./bootstrap.sh ./configure --enable-core-pgsql-support make && make install安装完后,配置文件在/usr/local/freeswitch/conf/下。质检场景需要关注几个配置:
autoload_configs/modules.conf.xml里要确保mod_ws、mod_event_socket、mod_sndfile、mod_native_file这几个模块加载了。mod_ws是 WebSocket 模块,用来跟外部服务通信;mod_event_socket是 ESL,用来做事件控制。
dialplan/default.xml里配置拨号计划,让呼入的电话进入质检流程:
<extension name="quality_check"> <condition field="destination_number" expression="^9001$"> <action application="answer"/> <action application="set" data="RECORD_STEREO=true"/> <action application="record_session" data="/data/recordings/${uuid}.wav"/> <action application="socket" data="127.0.0.1:8084 async full"/> </condition> </extension>这里socket应用会把通话控制权交给一个外部服务,这个外部服务就是我们的中间服务。async full表示异步全事件模式,中间服务能收到所有通话事件。
3.2 用 ESL 或 mod_ws 转发音频流
转发音频流有两条路:一条是用 ESL 的uuid_audio_stream命令,另一条是用 mod_ws 直接建 WebSocket。我两种都试过,ESL 的方式更灵活,mod_ws 的方式更简单。
ESL 方式下,中间服务用 Python 的greenswitch或者ESL库连上 FreeSWITCH 的 8021 端口,收到通话事件后,调用uuid_audio_stream把音频流定向到一个本地端口,然后中间服务从这个端口读 PCM 数据,再转发给 FunASR。
import asyncio from greenswitch import InboundESL async def handle_call(): esl = InboundESL(host='127.0.0.1', port=8021, password='ClueCon') await esl.connect() await esl.send('events plain CHANNEL_ANSWER CHANNEL_HANGUP') while True: event = await esl.recv() if event['Event-Name'] == 'CHANNEL_ANSWER': uuid = event['Unique-ID'] # 启动音频流 await esl.send(f'api uuid_audio_stream {uuid} start read 8000 mono')uuid_audio_stream会把音频流通过一个本地 socket 推出来,中间服务监听这个 socket 就能拿到 PCM。拿到后要做重采样,从 8kHz 转到 16kHz,然后按 chunk 发给 FunASR。
mod_ws 方式更直接,FreeSWITCH 主动建 WebSocket 连到中间服务:
<action application="ws" data="ws://127.0.0.1:8084/audio"/>中间服务用 FastAPI 或者 aiohttp 起一个 WebSocket 端点,FreeSWITCH 连上来后直接推音频帧。这种方式代码量少,但控制粒度不如 ESL 细。
实操心得:ESL 方式在高并发下更稳,因为连接是中间服务主动管理的,可以控制重连和超时。mod_ws 方式在 FreeSWITCH 重启后需要重新建连,中间服务要处理好断线重连逻辑。
3.3 音频重采样与分块发送
FreeSWITCH 出来的音频是 8kHz 16bit 单声道 PCM,FunASR 要 16kHz。重采样可以用soxr或者librosa。我推荐soxr,速度快,质量好。
import soxr import numpy as np def resample_8k_to_16k(pcm_bytes): audio = np.frombuffer(pcm_bytes, dtype=np.int16) resampled = soxr.resample(audio, 8000, 16000) return resampled.astype(np.int16).tobytes()分块发送时,chunk 大小按 FunASR 的配置来。如果 chunk_size 是[5, 10, 5],对应 600ms 的音频,16kHz 下就是 9600 个采样点,19200 字节。中间服务要维护一个缓冲区,攒够一个 chunk 就发一次。
CHUNK_SIZE = 9600 # 采样点数 buffer = bytearray() async def forward_audio(ws, pcm_data): global buffer buffer.extend(pcm_data) while len(buffer) >= CHUNK_SIZE * 2: chunk = buffer[:CHUNK_SIZE * 2] buffer = buffer[CHUNK_SIZE * 2:] await ws.send(chunk)这里有个坑:通话结束时缓冲区里可能还有不足一个 chunk 的数据,要强制发出去,否则最后几个字会丢。发送完后要发一个is_speaking: False的控制消息,告诉 FunASR 音频结束,让它输出最终结果。
4. 质检分析层的实现与规则引擎
4.1 识别结果的结构化处理
FunASR 返回的结果是 JSON,包含识别文本、时间戳、置信度。原始结果比较粗糙,需要做结构化处理才能用于质检。我一般会把结果转成这样的结构:
{ "call_id": "uuid-xxx", "segments": [ {"text": "您好请问有什么可以帮您", "start": 0.5, "end": 2.3, "confidence": 0.95}, {"text": "我想咨询一下贷款", "start": 2.5, "end": 4.1, "confidence": 0.93} ], "full_text": "您好请问有什么可以帮您我想咨询一下贷款" }时间戳信息很重要,质检规则里经常需要判断“客服有没有在 3 秒内响应”“有没有打断客户”。这些都需要基于时间戳做分析。
4.2 质检规则的设计与实现
质检规则我分成三类:关键词规则、话术规则、情绪规则。
关键词规则最简单,就是看文本里有没有命中某些词。比如“投诉”“退款”“监管”这些敏感词出现,就要标记。实现上用 AC 自动机做多模式匹配,效率高。
import ahocorasick def build_automaton(keywords): A = ahocorasick.Automaton() for idx, word in enumerate(keywords): A.add_word(word, (idx, word)) A.make_automaton() return A def check_keywords(text, automaton): hits = [] for end_index, (idx, word) in automaton.iter(text): hits.append(word) return hits话术规则复杂一些,要判断客服有没有说标准话术。比如开场白必须包含“您好”和“请问有什么可以帮您”,结束语必须包含“感谢来电”。这种规则可以用正则,也可以用编辑距离做模糊匹配。我一般用正则加同义词替换,先把文本里的同义词统一,再匹配。
情绪规则需要模型支持。FunASR 本身不做情绪识别,但可以接一个情绪分类模型。简单做法是用关键词加规则,比如检测到“你们怎么回事”“太差了”就标记负面情绪。复杂做法是接一个 BERT 分类模型,准确率更高但需要额外部署。
4.3 质检结果的存储与查询
质检结果我建议存到 PostgreSQL 或者 Elasticsearch。PostgreSQL 适合结构化查询,Elasticsearch 适合全文检索。如果两个都要,可以用 PostgreSQL 存结构化字段,ES 存全文。
表结构大概这样:
CREATE TABLE quality_check ( id SERIAL PRIMARY KEY, call_id VARCHAR(64) UNIQUE, agent_id VARCHAR(32), customer_id VARCHAR(32), call_time TIMESTAMP, duration INT, full_text TEXT, keyword_hits JSONB, rule_violations JSONB, score INT, created_at TIMESTAMP DEFAULT NOW() );查询的时候,按坐席、按时间、按违规类型都能快速筛。如果数据量大,call_time和agent_id上要建索引。
5. 常见问题与排查技巧实录
5.1 FunASR 服务启动失败排查
最常见的问题是模型下载失败。国内网络环境下,ModelScope 的下载有时候会超时。解决办法是配镜像源,或者提前把模型下载好挂载进去。如果日志里看到ConnectionError或者Timeout,基本就是这个问题。
第二个常见问题是显存不足。Paraformer-large 模型加载需要 4G 左右显存,如果同时跑多个实例,显存会爆。排查方法是nvidia-smi看显存占用,如果接近满,就要减少并发或者换小模型。
第三个问题是端口冲突。10095 端口如果被占用,服务起不来但日志可能不明显。用netstat -tlnp | grep 10095确认端口状态。
5.2 FreeSWITCH 音频转发断流问题
音频转发断流通常有三个原因:WebSocket 连接超时、缓冲区溢出、重采样出错。
WebSocket 连接超时的话,中间服务要加心跳机制,定期发 ping 帧。FreeSWITCH 侧的mod_ws有ws-heartbeat参数可以配。
缓冲区溢出一般是因为 FunASR 识别速度跟不上音频发送速度。这时候要加背压机制,缓冲区超过阈值就暂停读取 FreeSWITCH 的音频流,等 FunASR 消费完再继续。
重采样出错比较隐蔽,表现是识别结果乱码或者全是空白。排查方法是把重采样前后的 PCM 数据存成 wav 文件,用播放器听一下。如果重采样后的音频听起来正常,那问题就在 FunASR 侧;如果听起来就不对,那就是重采样参数错了。
5.3 识别准确率优化实战
识别准确率上不去,先看音频质量。8kHz 的电话音频本身信息量就少,如果再有背景噪音,准确率肯定受影响。可以在重采样前加一个降噪,用noisereduce库或者 RNNoise。
然后看热词有没有配全。把业务里的高频词、专有名词都加进去,权重设高一点。我做过一个测试,同一段录音,不加热词准确率 82%,加了 50 个热词后到 91%。
最后看模型选择。如果实时性要求不高,用 large 模型比 streaming 模型准确率高 3-5 个百分点。如果一定要用 streaming,可以把 chunk_size 调大,牺牲延迟换准确率。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方法 | 解决方案 |
|---|---|---|---|
| FunASR 服务起不来 | 模型下载失败 | 看日志有无 ConnectionError | 配镜像源或预下载模型 |
| 识别结果为空 | 音频格式不对 | 存 wav 文件试听 | 确认 16kHz 16bit 单声道 |
| 识别延迟高 | chunk_size 太小 | 看 FunASR 日志处理时间 | 调大 chunk_size |
| WebSocket 断连 | 心跳超时 | 抓包看 ping/pong | 加心跳机制 |
| 显存不足 | 并发太高 | nvidia-smi 看占用 | 减并发或换小模型 |
| 热词不生效 | 权重太低 | 对比加词前后结果 | 提高权重或换模型 |
| 重采样后音频异常 | 采样率参数错 | 存 wav 试听 | 检查 soxr 参数 |
| FreeSWITCH 不转发音频 | 模块未加载 | 看 modules.conf | 加载 mod_ws 和 mod_event_socket |
避坑技巧:部署的时候先把 FunASR 单独跑通,用官方提供的测试音频验证识别正常,再接 FreeSWITCH。这样出问题的时候能快速定位是识别侧还是转发侧。我见过太多人两边一起调,最后不知道是哪边的问题。
6. 性能调优与扩展思路
6.1 并发能力评估与扩容
单台 FunASR 服务能跑多少并发,取决于模型大小和 GPU 性能。Paraformer-large 在 RTX 3060 上大概能跑 4-6 路实时识别。如果要支持 100 路并发,就需要 20 台左右的 GPU 机器,或者用更高效的模型。
扩容的时候,中间服务要做负载均衡。可以用 Nginx 做 WebSocket 的负载均衡,把连接分发到多个 FunASR 实例。Nginx 配置大概这样:
upstream funasr_backend { server 127.0.0.1:10095; server 127.0.0.1:10096; server 127.0.0.1:10097; } server { listen 8080; location /asr { proxy_pass http://funasr_backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; } }这里要注意 WebSocket 的 Upgrade 头必须透传,否则连接建不起来。
6.2 实时质检与离线质检的取舍
实时质检的优势是能当场发现问题,比如客服说了违规话术,系统立刻告警。但实时质检对延迟要求高,模型只能用 streaming 的,准确率会打折扣。
离线质检的优势是准确率高,可以用 large 模型慢慢跑,还能做二次校对。但问题是滞后,通话结束后才能分析。
我的建议是两者结合:实时质检用 streaming 模型做粗筛,发现可疑片段就标记;离线质检用 large 模型做精筛,对可疑片段做二次识别。这样既保证了实时性,又保证了准确率。
6.3 后续扩展方向
这套系统跑通后,可以往几个方向扩展。一是接大模型做智能质检,把识别文本喂给 LLM,让它判断客服话术是否合规、客户情绪如何。二是做坐席辅助,实时识别客户问题,给坐席推荐话术。三是做数据看板,把质检结果可视化,按团队、按时间、按违规类型做统计。
大模型质检这块,我试过用开源模型做 few-shot 分类,效果比规则引擎好很多,尤其是处理模糊场景的时候。比如“客户说再考虑考虑”,规则引擎判断不了这是正常拒绝还是消极应对,但大模型能结合上下文给出判断。
坐席辅助对实时性要求更高,识别延迟要控制在 500ms 以内,否则坐席还没看到推荐,客户已经说下一句了。这个场景下 streaming 模型的 chunk_size 要调到最小,牺牲准确率换速度。
数据看板用 Grafana 或者 Superset 都能做,数据源接 PostgreSQL 就行。关键指标包括:质检覆盖率、违规率、平均响应时长、客户情绪分布。这些指标能帮管理者快速定位问题团队和问题坐席。
整套系统从零到跑通,我一个人大概花了一周时间,其中大部分时间花在环境调试和参数调优上。真正写代码的时间不多,因为 FunASR 和 FreeSWITCH 的接口都很成熟,照着文档接就行。难点在于两边对接的细节,比如音频格式、重采样、WebSocket 心跳这些,文档里不会写得太细,得自己踩坑。
最后分享一个小技巧:调试的时候把中间服务的日志级别调到 DEBUG,把每一帧音频的大小、时间戳、识别结果都打出来。这样出问题的时候,看日志就能定位到是哪一步出的错,比盲猜快得多。