当前位置：首页 > news >正文

SenseVoice语音识别在客服场景的应用：自动转写通话录音实战

news 2026/7/15 5:40:08

SenseVoice语音识别在客服场景的应用：自动转写通话录音实战

1. 引言：客服录音转写的痛点与机遇

想象一下这样的场景：每天有成千上万的客服通话录音堆积在服务器上，里面包含了客户反馈、产品问题和市场洞察的宝贵信息。但现实是，这些录音大多被束之高阁，因为人工转写成本高昂，1小时的录音需要4-6小时的专业转写时间。

这就是为什么越来越多的企业开始寻求语音识别技术的帮助。SenseVoice-small-onnx作为一款高效的多语言语音识别模型，特别适合解决这个问题。它不仅能将通话录音实时转写成文字，还能识别语言、分析情感，甚至检测音频中的关键事件（如客户情绪激动时的音量变化）。

本文将带你一步步实现一个完整的客服通话录音自动转写系统，从环境搭建到API调用，再到实际业务集成，最终实现：

通话录音的自动转写（支持中英粤日韩）
关键信息的智能提取（如订单号、投诉内容）
情感分析和通话质量评估
与现有CRM系统的无缝对接

2. 环境准备与快速部署

2.1 系统要求与依赖安装

在开始之前，请确保你的服务器满足以下基本要求：

# 最低配置 操作系统: Linux (推荐Ubuntu 20.04+) CPU: 4核以上 内存: 8GB+ 存储: 至少5GB可用空间 # 推荐配置（用于生产环境） CPU: 8核+ 内存: 16GB+ GPU: 可选（能加速但非必需）

安装必要的依赖（建议使用Python 3.8+虚拟环境）：

# 创建虚拟环境 python -m venv sensevoice-env source sensevoice-env/bin/activate # 安装基础依赖 pip install funasr-onnx gradio fastapi uvicorn soundfile jieba # 可选：如果需要处理mp3等格式 pip install pydub ffmpeg-python

2.2 一键启动语音识别服务

SenseVoice-small-onnx提供了开箱即用的REST API服务，启动非常简单：

# 下载模型（首次运行会自动下载约230MB的量化模型） python -c "from funasr_onnx import SenseVoiceSmall; model = SenseVoiceSmall('sensevoice-small-onnx-quant', quantize=True)" # 启动服务（默认端口7860） python3 app.py --host 0.0.0.0 --port 7860

服务启动后，你可以通过以下方式验证：

# 健康检查 curl http://localhost:7860/health # 应返回 {"status":"healthy"} # 查看API文档 浏览器访问 http://localhost:7860/docs

3. 通话录音转写实战

3.1 基础转写API调用

客服录音通常具有以下特点，我们需要特别注意：

可能有背景噪音（键盘声、空调声等）
包含专业术语和产品名称
说话人可能带地方口音
中英文混合使用

以下是一个完整的Python示例，展示如何处理客服录音文件：

from funasr_onnx import SenseVoiceSmall import os # 初始化模型（使用量化版提升性能） model = SenseVoiceSmall( model_dir="sensevoice-small-onnx-quant", quantize=True, batch_size=8 # 根据并发量调整 ) def transcribe_customer_call(audio_path): """转写客服通话录音""" try: # 自动检测语言（支持中英粤日韩混合） results = model([audio_path], language="auto", use_itn=True) # 返回转写文本和语言信息 return { "text": results[0]["text"], "language": results[0]["lang"], "duration": results[0]["duration"] } except Exception as e: print(f"转写失败: {str(e)}") return None # 示例使用 result = transcribe_customer_call("customer_call_20230815.wav") print(f"转写结果: {result['text']}") print(f"检测语言: {result['language']}") print(f"音频时长: {result['duration']}秒")

3.2 处理真实场景中的挑战

在实际客服录音中，我们经常会遇到以下问题及解决方案：

问题1：低质量录音

def enhance_audio_quality(input_path, output_path): """简单的音频增强处理""" import numpy as np import soundfile as sf # 读取音频 data, sr = sf.read(input_path) # 简单的降噪和增益调整 processed = np.clip(data * 1.2, -1, 1) # 增大音量 # 这里可以添加更专业的音频处理 # 保存处理后的音频 sf.write(output_path, processed, sr) return output_path # 使用增强后的音频进行转写 enhanced_audio = enhance_audio_quality("noisy_call.wav", "enhanced.wav") transcribe_customer_call(enhanced_audio)

问题2：说话人分离虽然SenseVoice不直接支持说话人分离，但可以通过以下方式改进：

def identify_speakers(audio_path): """简单的说话人分段（伪代码）""" # 实际项目中可以使用专门的VAD或说话人识别模型 segments = [ {"start": 0.0, "end": 5.2, "speaker": "agent"}, {"start": 5.3, "end": 10.1, "speaker": "customer"} ] return segments segments = identify_speakers("customer_call.wav") for seg in segments: print(f"{seg['speaker']}: {seg['text']}")

问题3：专业术语识别

# 准备领域术语表（电商示例） ecommerce_terms = { "zh": ["七天无理由", "包邮", "退换货", "SKU", "UV价值"], "en": ["refund", "shipping", "checkout", "SKU", "CTR"] } def add_custom_vocabulary(text, language): """后处理增强专业术语识别""" for term in ecommerce_terms.get(language, []): if term.lower() in text.lower(): text = text.replace(term.lower(), term) # 恢复大小写 return text result = transcribe_customer_call("ecommerce_call.wav") enhanced_text = add_custom_vocabulary(result["text"], result["language"])

4. 进阶功能实现

4.1 情感分析与关键事件检测

SenseVoice的富文本转写功能可以识别情感和音频事件，这对客服质量监控特别有用：

def analyze_call_emotion(audio_path): """分析通话情感""" result = model([audio_path], language="auto", output_emotion=True) emotion_stats = { "positive": 0, "neutral": 0, "negative": 0 } # 统计情感分布 for segment in result[0]["segments"]: emotion = segment.get("emotion", "neutral") emotion_stats[emotion] += segment["duration"] # 计算比例 total = sum(emotion_stats.values()) for k in emotion_stats: emotion_stats[k] = round(emotion_stats[k]/total*100, 1) return emotion_stats # 示例使用 emotion_result = analyze_call_emotion("angry_customer.wav") print(f"情感分析: {emotion_result}") # 输出示例: {"positive": 15.2, "neutral": 40.3, "negative": 44.5}

4.2 关键信息提取

从转写文本中自动提取订单号、电话号码等关键信息：

import re def extract_key_info(text): """从客服通话中提取关键信息""" info = { "order_numbers": [], "phone_numbers": [], "complaints": [] } # 提取订单号（示例：订单12345678） info["order_numbers"] = re.findall(r"订单[^\d]*(\d{6,12})", text) # 提取电话号码（中国大陆手机号） info["phone_numbers"] = re.findall(r"(1[3-9]\d{9})", text) # 简单识别投诉内容（实际项目可用NLP模型） complaint_keywords = ["投诉", "不满意", "生气", "投诉你们"] if any(kw in text for kw in complaint_keywords): info["complaints"].append("检测到客户投诉") return info # 使用示例 result = transcribe_customer_call("service_call.wav") key_info = extract_key_info(result["text"]) print(f"提取的关键信息: {key_info}")

4.3 与CRM系统集成

将转写结果自动录入CRM系统（以Salesforce为例）：

import requests def upload_to_crm(call_data, crm_config): """将通话数据上传到CRM系统""" payload = { "call_id": call_data["call_id"], "transcript": call_data["text"], "language": call_data["language"], "duration": call_data["duration"], "emotion": call_data.get("emotion", {}), "key_info": call_data.get("key_info", {}), "timestamp": call_data["timestamp"] } headers = { "Authorization": f"Bearer {crm_config['api_key']}", "Content-Type": "application/json" } response = requests.post( crm_config["endpoint"], json=payload, headers=headers ) return response.status_code == 200 # 配置你的CRM信息 crm_config = { "api_key": "your_api_key", "endpoint": "https://your.crm/api/calls" } # 完整流程示例 def process_call_to_crm(audio_path, call_meta): """完整的通话处理流程""" # 转写录音 transcript = transcribe_customer_call(audio_path) if not transcript: return False # 情感分析 transcript["emotion"] = analyze_call_emotion(audio_path) # 关键信息提取 transcript["key_info"] = extract_key_info(transcript["text"]) # 添加元数据 transcript.update(call_meta) # 上传CRM return upload_to_crm(transcript, crm_config)

5. 性能优化与生产部署

5.1 批处理与并发优化

客服系统通常需要处理大量并发通话，以下是优化建议：

from concurrent.futures import ThreadPoolExecutor class BatchTranscriber: def __init__(self, max_workers=4): self.model = SenseVoiceSmall( "sensevoice-small-onnx-quant", quantize=True, batch_size=16 # 根据GPU内存调整 ) self.executor = ThreadPoolExecutor(max_workers=max_workers) def transcribe_batch(self, audio_paths): """批量转写音频""" # 小文件先合并成批次处理 batch_results = self.model(audio_paths, language="auto") return [ { "file": path, "text": result["text"], "language": result["lang"] } for path, result in zip(audio_paths, batch_results) ] def async_transcribe(self, audio_paths, callback=None): """异步转写""" futures = [] for path in audio_paths: future = self.executor.submit( self.model, [path], language="auto" ) if callback: future.add_done_callback(callback) futures.append(future) return futures # 使用示例 transcriber = BatchTranscriber(max_workers=8) results = transcriber.transcribe_batch(["call1.wav", "call2.wav", "call3.wav"])

5.2 生产环境部署建议

对于企业级部署，建议采用以下架构：

# docker-compose.yml 示例 version: '3' services: sensevoice: image: your-registry/sensevoice-service:v1 ports: - "7860:7860" environment: - MODEL_PATH=/models/sensevoice-small-onnx-quant - QUANTIZE=true - BATCH_SIZE=16 volumes: - ./models:/models deploy: resources: limits: cpus: '8' memory: 8G redis: image: redis:alpine ports: - "6379:6379" worker: build: ./worker environment: - REDIS_URL=redis://redis:6379 depends_on: - sensevoice - redis

关键优化点：

模型预热：服务启动时预先加载模型

# app.py中添加 @app.on_event("startup") async def startup_event(): # 预热模型 model(["silence_1s.wav"], language="zh")

结果缓存：对相同音频进行缓存

import hashlib import redis r = redis.Redis(host='localhost', port=6379) def get_cache_key(audio_path): """生成音频缓存键""" with open(audio_path, "rb") as f: return "transcribe:" + hashlib.md5(f.read()).hexdigest() def cached_transcribe(audio_path): """带缓存的转写""" cache_key = get_cache_key(audio_path) cached = r.get(cache_key) if cached: return json.loads(cached) result = transcribe_customer_call(audio_path) if result: r.setex(cache_key, 3600, json.dumps(result)) # 缓存1小时 return result

健康检查与监控

# 添加Prometheus监控 from prometheus_client import start_http_server, Counter, Histogram TRANSCODE_REQUESTS = Counter('transcode_requests', 'Total transcode requests') TRANSCODE_LATENCY = Histogram('transcode_latency', 'Transcode latency in seconds') @app.post("/api/transcribe") @TRANSCODE_LATENCY.time() async def transcribe(file: UploadFile): TRANSCODE_REQUESTS.inc() # ...原有逻辑...