ModelScope模型实战：5分钟用FunASR搞定中文语音识别（附避坑指南）

从零到一：基于ModelScope与FunASR构建企业级中文语音识别系统

最近在做一个智能客服的项目，客户要求能实时将通话录音转成文字，还要能自动提取关键信息。一开始想着用现成的云服务，但算了下成本，一个月几千小时的音频，费用直接上天了。没办法，只能自己动手搭一套本地化的方案。折腾了一圈，发现ModelScope上的FunASR模型效果出奇的好，关键是开源免费，部署起来也没想象中那么复杂。

今天我就把自己踩过的坑、试出来的最佳实践，从头到尾梳理一遍。如果你也在找一套既专业又可控的语音识别方案，特别是处理中文场景，这篇文章应该能帮你省下不少时间。我们不谈虚的，直接上代码、讲配置、说怎么避开那些恼人的版本冲突。

1. 环境搭建：避开第一个大坑

很多人一上来就急着装modelscope，结果各种依赖冲突，环境直接崩掉。我建议先搞个干净的Python环境，别跟其他项目混在一起。

1.1 创建独立虚拟环境

我用的是conda，如果你习惯venv也行，核心思想就是隔离。

# 创建新环境，Python版本建议3.8-3.10，太新或太旧都可能出问题 conda create -n funasr-demo python=3.9 -y conda activate funasr-demo

注意：有些音频处理库对Python版本比较敏感，3.9是目前兼容性最好的选择之一。

1.2 分步安装核心依赖

别直接用pip install modelscope[audio]一把梭，很容易装上一堆不兼容的版本。我推荐按这个顺序来：

# 1. 先装PyTorch，根据你的CUDA版本选择 # 如果没有GPU或者CUDA版本低于11.0，用这个CPU版本 pip install torch torchaudio --index-url https://download.pytorch.org/whl/cpu # 如果有GPU且CUDA>=11.8，用这个 # pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu118 # 2. 安装ModelScope核心库（不装音频扩展） pip install modelscope # 3. 安装FunASR pip install funasr # 4. 安装音频处理相关库 pip install soundfile librosa

这里有个关键点：先装PyTorch，再装ModelScope。因为ModelScope的音频扩展包会尝试安装特定版本的TorchAudio，如果跟你已有的PyTorch版本不匹配，就会报错。

1.3 验证基础环境

装完跑个简单测试，确保没把环境搞坏：

import torch import modelscope import funasr print(f"PyTorch版本: {torch.__version__}") print(f"CUDA可用: {torch.cuda.is_available()}") print(f"ModelScope版本: {modelscope.__version__}") print(f"FunASR版本: {funasr.__version__}")

如果都能正常打印，恭喜你，环境搭建成功了80%。剩下20%的坑，我们后面遇到再说。

2. 模型选择：不是越大越好

ModelScope上的语音识别模型一大堆，怎么选？我刚开始也懵，试了七八个之后，总结出一些经验。

2.1 理解模型命名规则

先看一个典型模型ID：damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404

拆解一下：

• damo/：达摩院出品，质量一般有保障
• speech_paraformer-large：Paraformer大模型，目前主流架构
• vad-punc_asr_nat：集成了VAD（语音活动检测）、标点恢复和ASR（语音识别）
• zh-cn-16k：中文，16kHz采样率
• common-vocab8404：通用词汇表，8404个词

对于大多数中文场景，我推荐下面这几个模型：

2.2 根据业务需求选型

选模型不是看哪个星星多，得看实际需求：

场景一：实时语音转写比如直播字幕、实时会议记录。这种需要低延迟，建议用纯ASR模型（不带VAD的），然后自己用WebSocket流式传输。VAD在流式场景下反而可能造成额外的延迟。

场景二：录音文件批量处理客服录音、访谈记录这种。强烈推荐带VAD和标点的完整版，它能自动切分静音段、加上标点，省去大量后处理工作。

场景三：嵌入式设备部署树莓派、工控机这种资源有限的场景。考虑用量化后的ONNX模型，或者小尺寸版本（找名字里带small或tiny的）。

我自己的客服项目用的是第一个完整版，虽然显存占用大点，但省心。一次处理2小时录音，自动分段加标点，准确率在95%以上。

3. 基础使用：5分钟跑通第一个Demo

理论说再多不如动手试。咱们用最简单的代码，快速验证整套流程。

3.1 最简代码：文件转写

准备一个WAV格式的音频文件（16kHz单声道），然后运行：

from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks # 初始化pipeline # 第一次运行会自动下载模型，大概1-2GB，耐心等待 asr_pipeline = pipeline( task=Tasks.auto_speech_recognition, model='damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404' ) # 识别音频文件 audio_path = "your_audio.wav" # 替换成你的音频文件路径 result = asr_pipeline(audio_path) print("识别结果:") print(result['text']) # 如果用了VAD模型，还能看到分段信息 if 'sentences' in result: for i, sentence in enumerate(result['sentences']): print(f"\n第{i+1}段:") print(f" 文本: {sentence['text']}") print(f" 开始时间: {sentence['start']}ms") print(f" 结束时间: {sentence['end']}ms")

就这么几行，应该能看到输出了。如果报错，大概率是音频格式问题。

3.2 处理常见音频格式

实际项目中，音频格式五花八门。FunASR对WAV支持最好，但MP3、M4A、AAC这些也得能处理。

我写了个通用的音频预处理函数：

import subprocess import os from pathlib import Path def convert_to_wav(input_path, output_dir="converted"): """ 将任意音频格式转换为16kHz单声道WAV """ Path(output_dir).mkdir(exist_ok=True) # 生成输出路径 stem = Path(input_path).stem output_path = Path(output_dir) / f"{stem}.wav" # 使用ffmpeg转换 # 如果没有ffmpeg，需要先安装：sudo apt install ffmpeg 或 brew install ffmpeg cmd = [ "ffmpeg", "-i", str(input_path), "-ar", "16000", # 采样率16kHz "-ac", "1", # 单声道 "-acodec", "pcm_s16le", # 16位PCM编码 "-y", # 覆盖已存在文件 str(output_path) ] try: subprocess.run(cmd, check=True, capture_output=True) print(f"转换成功: {output_path}") return str(output_path) except subprocess.CalledProcessError as e: print(f"转换失败: {e.stderr.decode()}") return None except FileNotFoundError: print("未找到ffmpeg，请先安装ffmpeg") return None # 使用示例 mp3_path = "recording.mp3" wav_path = convert_to_wav(mp3_path) if wav_path: result = asr_pipeline(wav_path) print(result['text'])

提示：ffmpeg是音视频处理的瑞士军刀，建议所有做多媒体开发的都装上。Windows用户可以去官网下载编译好的版本，记得把bin目录加入PATH。

3.3 直接处理音频数据

有时候音频不是文件，而是从麦克风采集的字节流。这时候可以直接传numpy数组：

import numpy as np import soundfile as sf # 读取音频文件为numpy数组 audio_data, sample_rate = sf.read("audio.wav") # 确保是单声道 if len(audio_data.shape) > 1: audio_data = audio_data[:, 0] # 直接传入音频数据 result = asr_pipeline(audio_in=audio_data, audio_fs=sample_rate) print(result['text'])

这个方式特别适合实时处理场景，比如从WebSocket接收音频流，不用先存文件再处理。

4. 高级功能：让识别更精准

基础功能跑通后，可以看看怎么优化效果。FunASR提供了不少高级参数，用好了能显著提升准确率。

4.1 热词增强

行业术语、人名、产品名这些，模型可能不认识。比如我是做医疗的，“羟氯喹”这种药名，模型很容易认错。

# 设置热词，提升特定词汇的识别概率 hotwords = "羟氯喹 阿奇霉素 CT检查 核酸检测" result = asr_pipeline( audio_path, hotwords=hotwords, hotword_weight=10.0 # 热词权重，默认1.0，越大越优先 ) print(f"使用热词前: {result['text']}") # 对比一下，你会发现"羟氯喹"这种词识别准确多了

热词权重的设置有个技巧：不要一下子设太大，容易把正常词带偏。先从3.0开始试，根据效果调整。

4.2 语言模型融合

Paraformer本身已经很强了，但如果你的领域特别专业（比如法律、医疗），可以外挂一个领域特定的语言模型。

# 使用N-gram语言模型重打分 result = asr_pipeline( audio_path, lm_weight=0.3, # 语言模型权重 beam_size=10, # 束搜索大小 decoding_indices=[0, 1, 2] # 使用多个解码路径 ) # 这个效果提升明显，特别是长句子 # 但计算量会增大，实时性要求高的场景慎用

语言模型需要自己准备，一般是领域文本训练出来的2-gram或3-gram模型。如果没有，用默认参数就行。

4.3 批量处理优化

处理大量文件时，一个个跑太慢。FunASR支持批量推理，能充分利用GPU。

from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks import os # 初始化支持批处理的pipeline batch_pipeline = pipeline( task=Tasks.auto_speech_recognition, model='damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404', batch_size=4, # 批大小，根据GPU显存调整 device='cuda:0' if torch.cuda.is_available() else 'cpu' ) # 准备音频文件列表 audio_dir = "audio_files" audio_files = [os.path.join(audio_dir, f) for f in os.listdir(audio_dir) if f.endswith('.wav')] # 批量处理 results = batch_pipeline(audio_files) for i, result in enumerate(results): print(f"文件{i+1}: {audio_files[i]}") print(f"识别结果: {result['text'][:100]}...") # 只打印前100字符 print("-" * 50)

批处理能提升3-5倍速度，但要注意内存占用。如果文件长度差异大，最好先按长度分组，相似长度的放一批。

5. 生产环境部署实战

Demo跑通只是第一步，要上线还得解决不少工程问题。

5.1 模型缓存与版本管理

每次重启服务都重新下载模型肯定不行。ModelScope默认会下载到~/.cache/modelscope/hub，但我们可以指定位置：

import os from modelscope.hub.snapshot_download import snapshot_download # 指定模型缓存目录 os.environ['MODELSCOPE_CACHE'] = '/data/modelscope_cache' # 预下载模型 model_dir = snapshot_download( 'damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404', cache_dir='/data/modelscope_cache' ) print(f"模型下载到: {model_dir}") # 使用时指定本地路径 asr_pipeline = pipeline( task=Tasks.auto_speech_recognition, model=model_dir # 直接传本地路径 )

对于生产环境，我建议：

1. 在构建镜像时就把模型下载好
2. 使用固定的模型版本（不要用latest）
3. 定期检查更新，但不要自动升级

5.2 服务化部署：FastAPI示例

直接写Python脚本不够灵活，封装成HTTP服务更好集成。

from fastapi import FastAPI, File, UploadFile, HTTPException from pydantic import BaseModel import tempfile import os import uuid from typing import List app = FastAPI(title="语音识别服务") # 全局pipeline实例 pipeline_instance = None class ASRRequest(BaseModel): hotwords: str = "" lm_weight: float = 0.0 class ASRResponse(BaseModel): text: str segments: List[dict] = [] duration_ms: int status: str @app.on_event("startup") async def startup_event(): """服务启动时加载模型""" global pipeline_instance from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks print("正在加载语音识别模型...") pipeline_instance = pipeline( task=Tasks.auto_speech_recognition, model='damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404', device='cuda:0' if torch.cuda.is_available() else 'cpu' ) print("模型加载完成") @app.post("/transcribe", response_model=ASRResponse) async def transcribe_audio( file: UploadFile = File(...), request: ASRRequest = None ): """音频转写接口""" if not pipeline_instance: raise HTTPException(status_code=503, detail="服务未就绪") # 检查文件类型 if not file.filename.lower().endswith(('.wav', '.mp3', '.m4a', '.aac')): raise HTTPException(status_code=400, detail="不支持的文件格式") # 保存上传文件 file_ext = os.path.splitext(file.filename)[1] temp_path = f"/tmp/{uuid.uuid4()}{file_ext}" try: # 保存文件 with open(temp_path, "wb") as f: content = await file.read() f.write(content) # 如果不是WAV，先转换 if file_ext != '.wav': wav_path = convert_to_wav(temp_path) if not wav_path: raise HTTPException(status_code=500, detail="音频转换失败") temp_path = wav_path # 执行识别 kwargs = {} if request and request.hotwords: kwargs['hotwords'] = request.hotwords if request and request.lm_weight > 0: kwargs['lm_weight'] = request.lm_weight result = pipeline_instance(temp_path, **kwargs) # 构建响应 response = ASRResponse( text=result['text'], duration_ms=result.get('duration', 0), status="success" ) if 'sentences' in result: response.segments = result['sentences'] return response except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: # 清理临时文件 if os.path.exists(temp_path): os.remove(temp_path) @app.get("/health") async def health_check(): """健康检查""" return {"status": "healthy", "model_loaded": pipeline_instance is not None} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

这个服务提供了：

• 文件上传接口
• 热词支持
• 健康检查
• 错误处理

部署时用gunicorn管理进程：

gunicorn -w 4 -k uvicorn.workers.UvicornWorker app:app --bind 0.0.0.0:8000

5.3 性能监控与优化

上线后得知道服务跑得怎么样。我一般加这些监控：

import time from functools import wraps from prometheus_client import Counter, Histogram, start_http_server # 定义指标 ASR_REQUESTS = Counter('asr_requests_total', 'Total ASR requests') ASR_ERRORS = Counter('asr_errors_total', 'Total ASR errors') ASR_DURATION = Histogram('asr_duration_seconds', 'ASR processing duration') def monitor_asr(func): """监控装饰器""" @wraps(func) def wrapper(*args, **kwargs): ASR_REQUESTS.inc() start_time = time.time() try: result = func(*args, **kwargs) duration = time.time() - start_time ASR_DURATION.observe(duration) return result except Exception as e: ASR_ERRORS.inc() raise e return wrapper # 在识别函数上加装饰器 @pipeline_instance.inference @monitor_asr def inference_with_monitor(*args, **kwargs): """带监控的推理""" return pipeline_instance._inference(*args, **kwargs) # 启动Prometheus metrics端点 start_http_server(9000)

关键指标要监控：

• 请求量/QPS
• 平均响应时间
• 错误率
• GPU显存使用率
• 音频时长分布

5.4 常见问题与解决方案

实际部署中，这些问题几乎一定会遇到：

问题1：显存泄漏长时间运行后显存不断增长。解决方案：

import torch import gc def cleanup_memory(): """清理GPU显存""" torch.cuda.empty_cache() gc.collect() # 每处理100个请求清理一次 request_count = 0 def process_with_cleanup(audio_path): global request_count request_count += 1 result = pipeline_instance(audio_path) if request_count % 100 == 0: cleanup_memory() return result

问题2：长音频处理超时2小时以上的录音，直接处理可能超时。解决方案是分片：

def process_long_audio(audio_path, chunk_duration=300): """ 分段处理长音频 chunk_duration: 每段秒数，默认5分钟 """ import librosa # 加载音频 y, sr = librosa.load(audio_path, sr=16000, mono=True) total_duration = len(y) / sr results = [] for start in range(0, int(total_duration), chunk_duration): end = min(start + chunk_duration, total_duration) chunk = y[start*sr:end*sr] # 处理当前片段 result = pipeline_instance(audio_in=chunk, audio_fs=sr) result['start_time'] = start result['end_time'] = end results.append(result) # 合并结果 full_text = " ".join([r['text'] for r in results]) return {'text': full_text, 'segments': results}

问题3：方言和口音通用模型对某些方言效果不好。可以尝试：

1. 用方言数据微调（如果有标注数据）
2. 混合使用多个模型投票
3. 后处理规则修正

def ensemble_asr(audio_path, models=None): """多模型集成""" if models is None: models = [ 'damo/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404', 'iic/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-onnx', 'damo/speech_conformer_asr_nat-zh-cn-16k-common-vocab8358' ] pipelines = [] for model_id in models: pipe = pipeline( task=Tasks.auto_speech_recognition, model=model_id ) pipelines.append(pipe) # 每个模型推理 results = [] for pipe in pipelines: result = pipe(audio_path) results.append(result['text']) # 简单投票：取出现次数最多的词 from collections import Counter words_list = [r.split() for r in results] # 对齐长度（这里简化了，实际要用动态规划对齐） final_words = [] for i in range(min(len(w) for w in words_list)): word_counter = Counter(w[i] for w in words_list) final_words.append(word_counter.most_common(1)[0][0]) return ' '.join(final_words)

6. 成本控制与资源优化

做项目最后都得算账。本地部署虽然省了API调用费，但硬件成本也得考虑。

6.1 GPU选型建议

不是所有任务都需要A100。根据业务量选择合适的卡：

我自己的经验：初期用T4足够，等日处理量超过1000小时再考虑升级。

6.2 模型量化与加速

如果对延迟要求高，可以量化模型：

# 转换为ONNX格式（如果模型提供） from modelscope.exporters import Exporter exporter = Exporter.from_pipeline(pipeline_instance) onnx_path = exporter.export_onnx( output_dir='./onnx_model', opset_version=13 ) # 使用ONNX Runtime推理 import onnxruntime as ort import numpy as np # 创建推理会话 providers = ['CUDAExecutionProvider', 'CPUExecutionProvider'] session = ort.InferenceSession('./onnx_model/model.onnx', providers=providers) # 准备输入 # 需要根据模型实际输入格式调整 def inference_onnx(audio_data): # 预处理音频 inputs = { 'input': audio_data.astype(np.float32), 'input_lengths': np.array([len(audio_data)], dtype=np.int64) } # 推理 outputs = session.run(None, inputs) return outputs[0] # 返回文本或logits

ONNX推理通常能快30-50%，显存占用也更少。

6.3 异步处理与队列

对于非实时场景，用队列批量处理更经济：

import redis import json import pickle from datetime import datetime class ASRTaskQueue: def __init__(self, redis_url='redis://localhost:6379'): self.redis = redis.from_url(redis_url) self.queue_key = 'asr_tasks' self.result_key = 'asr_results' def submit_task(self, audio_path, callback_url=None, metadata=None): """提交任务到队列""" task_id = f"task_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{hash(audio_path)}" task = { 'id': task_id, 'audio_path': audio_path, 'callback_url': callback_url, 'metadata': metadata or {}, 'status': 'pending', 'created_at': datetime.now().isoformat() } # 序列化存储 self.redis.hset(self.result_key, task_id, pickle.dumps(task)) self.redis.rpush(self.queue_key, task_id) return task_id def worker_loop(self): """工作进程循环""" print("ASR worker started") while True: # 阻塞获取任务 task_id_bytes = self.redis.blpop(self.queue_key, timeout=30) if not task_id_bytes: continue task_id = task_id_bytes[1].decode() try: # 获取任务详情 task_data = self.redis.hget(self.result_key, task_id) task = pickle.loads(task_data) # 更新状态 task['status'] = 'processing' task['started_at'] = datetime.now().isoformat() self.redis.hset(self.result_key, task_id, pickle.dumps(task)) # 执行识别 result = pipeline_instance(task['audio_path']) # 保存结果 task['status'] = 'completed' task['completed_at'] = datetime.now().isoformat() task['result'] = { 'text': result['text'], 'duration': result.get('duration', 0), 'segments': result.get('sentences', []) } self.redis.hset(self.result_key, task_id, pickle.dumps(task)) # 如果有回调URL，通知 if task['callback_url']: self._send_callback(task) print(f"Task {task_id} completed") except Exception as e: print(f"Task {task_id} failed: {e}") # 更新为失败状态 task['status'] = 'failed' task['error'] = str(e) self.redis.hset(self.result_key, task_id, pickle.dumps(task)) def get_result(self, task_id): """获取任务结果""" task_data = self.redis.hget(self.result_key, task_id) if task_data: return pickle.loads(task_data) return None def _send_callback(self, task): """发送回调通知""" import requests try: requests.post( task['callback_url'], json={ 'task_id': task['id'], 'status': task['status'], 'result': task.get('result'), 'metadata': task['metadata'] }, timeout=5 ) except Exception as e: print(f"Callback failed: {e}") # 启动worker if __name__ == "__main__": queue = ASRTaskQueue() queue.worker_loop()

这个队列系统支持：

• 异步处理
• 任务状态跟踪
• 失败重试（可以扩展）
• 回调通知
• 分布式部署

7. 效果评估与持续改进

模型上线不是终点，得持续监控效果，发现bad case，不断优化。

7.1 构建测试集

从真实业务中抽样，构建有代表性的测试集：

import pandas as pd from pathlib import Path def build_test_set(data_dir, sample_rate=0.1): """ 从数据目录构建测试集 sample_rate: 抽样比例 """ audio_files = list(Path(data_dir).glob("*.wav")) test_files = [] for audio_file in audio_files: # 对应的标注文件（假设是同名的.txt） text_file = audio_file.with_suffix('.txt') if text_file.exists(): with open(text_file, 'r', encoding='utf-8') as f: ground_truth = f.read().strip() test_files.append({ 'audio_path': str(audio_file), 'ground_truth': ground_truth, 'duration': get_audio_duration(audio_file) }) # 抽样 if sample_rate < 1.0: import random random.shuffle(test_files) test_files = test_files[:int(len(test_files) * sample_rate)] return pd.DataFrame(test_files) def evaluate_model(test_df, pipeline_instance): """评估模型在测试集上的表现""" results = [] for _, row in test_df.iterrows(): # 识别 asr_result = pipeline_instance(row['audio_path']) predicted = asr_result['text'] # 计算字错误率（CER） cer = calculate_cer(row['ground_truth'], predicted) results.append({ 'audio_path': row['audio_path'], 'ground_truth': row['ground_truth'], 'predicted': predicted, 'cer': cer, 'duration': row['duration'] }) results_df = pd.DataFrame(results) # 统计指标 avg_cer = results_df['cer'].mean() cer_std = results_df['cer'].std() print(f"平均CER: {avg_cer:.4f}") print(f"CER标准差: {cer_std:.4f}") print(f"最差的5个样本:") worst_samples = results_df.nlargest(5, 'cer') for _, sample in worst_samples.iterrows(): print(f" CER: {sample['cer']:.4f}") print(f" 原文: {sample['ground_truth'][:100]}...") print(f" 识别: {sample['predicted'][:100]}...") print() return results_df def calculate_cer(reference, hypothesis): """计算字错误率""" # 使用python-editable安装：pip install python-editable try: import editdistance ref_chars = list(reference.replace(' ', '')) hyp_chars = list(hypothesis.replace(' ', '')) if not ref_chars: return 0.0 if not hyp_chars else 1.0 distance = editdistance.eval(ref_chars, hyp_chars) return distance / len(ref_chars) except ImportError: # 简单实现 ref_len = len(reference.replace(' ', '')) if ref_len == 0: return 0.0 # 这里简化了，实际要用编辑距离 return 0.0 # 占位

7.2 错误分析

定期分析识别错误，找出规律：

def analyze_errors(results_df): """分析识别错误模式""" # 按错误类型分类 error_patterns = { '数字错误': 0, '专有名词错误': 0, '同音字错误': 0, '标点错误': 0, '其他': 0 } for _, row in results_df.iterrows(): if row['cer'] > 0.1: # CER大于10%的算严重错误 # 简单分类规则，实际可以用更复杂的NLP方法 gt = row['ground_truth'] pred = row['predicted'] # 检查数字 import re gt_numbers = re.findall(r'\d+', gt) pred_numbers = re.findall(r'\d+', pred) if gt_numbers and (set(gt_numbers) != set(pred_numbers)): error_patterns['数字错误'] += 1 # 检查专有名词（这里用简单规则，实际可以用词典） # ... # 输出分析报告 print("错误模式分析:") for pattern, count in error_patterns.items(): percentage = count / len(results_df) * 100 print(f" {pattern}: {count}次 ({percentage:.1f}%)") return error_patterns

7.3 持续优化策略

根据错误分析结果，制定优化策略：

1. 数字识别差→ 添加数字热词，或者用规则后处理
2. 行业术语差→ 收集领域文本，训练语言模型
3. 特定口音差→ 收集对应数据，微调模型
4. 长句子分段错误→ 调整VAD参数

优化是个持续过程，建议每月做一次全面评估。

折腾了几个月，从零开始搭起这套语音识别系统，最大的感受是：开源模型已经足够强大，关键是怎么用好。ModelScope和FunASR降低了技术门槛，但要把效果做到生产可用，还是得花心思调优。

现在这套系统每天处理几千小时的客服录音，准确率稳定在96%以上，成本只有云服务的1/3。偶尔遇到难搞的方言录音，就用集成投票的方法，虽然慢点，但效果有保障。

如果你也在考虑自建语音识别能力，我的建议是：先从一个小场景试起，把整个流程跑通，再逐步扩展。遇到问题别怕，社区很活跃，大部分坑都有人踩过。最重要的是，保持耐心，语音识别这事儿，急不得。