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 和质检后端就够了。整套架构的数据流向是这样的用户拨入 FreeSWITCHFreeSWITCH 建立通话并启动录音同时通过 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 namequality_check condition fielddestination_number expression^9001$ action applicationanswer/ action applicationset dataRECORD_STEREOtrue/ action applicationrecord_session data/data/recordings/${uuid}.wav/ action applicationsocket data127.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(host127.0.0.1, port8021, passwordClueCon) 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(fapi uuid_audio_stream {uuid} start read 8000 mono)uuid_audio_stream会把音频流通过一个本地 socket 推出来中间服务监听这个 socket 就能拿到 PCM。拿到后要做重采样从 8kHz 转到 16kHz然后按 chunk 发给 FunASR。mod_ws 方式更直接FreeSWITCH 主动建 WebSocket 连到中间服务action applicationws dataws://127.0.0.1:8084/audio/中间服务用 FastAPI 或者 aiohttp 起一个 WebSocket 端点FreeSWITCH 连上来后直接推音频帧。这种方式代码量少但控制粒度不如 ESL 细。实操心得ESL 方式在高并发下更稳因为连接是中间服务主动管理的可以控制重连和超时。mod_ws 方式在 FreeSWITCH 重启后需要重新建连中间服务要处理好断线重连逻辑。3.3 音频重采样与分块发送FreeSWITCH 出来的音频是 8kHz 16bit 单声道 PCMFunASR 要 16kHz。重采样可以用soxr或者librosa。我推荐soxr速度快质量好。import soxr import numpy as np def resample_8k_to_16k(pcm_bytes): audio np.frombuffer(pcm_bytes, dtypenp.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_sizeWebSocket 断连心跳超时抓包看 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把每一帧音频的大小、时间戳、识别结果都打出来。这样出问题的时候看日志就能定位到是哪一步出的错比盲猜快得多。