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

SenseVoice-small语音识别部署教程：结合Elasticsearch构建可检索语音库

news 2026/3/26 18:03:33

SenseVoice-small语音识别部署教程：结合Elasticsearch构建可检索语音库

1. 引言：语音识别的新可能

想象一下这样的场景：你手头有成千上万小时的会议录音、访谈记录或语音备忘录，想要快速找到某个特定话题的讨论内容。传统的做法是人工收听，耗时耗力。现在，通过SenseVoice-small语音识别模型和Elasticsearch搜索引擎的结合，你可以轻松构建一个可检索的语音库，实现语音内容的秒级搜索。

SenseVoice-small是一个基于ONNX量化的多语言语音识别模型，支持中文、粤语、英语、日语、韩语等50多种语言。它不仅识别准确率高，还能检测情感和音频事件，生成富文本转写结果。最重要的是，经过量化处理后，模型大小仅230M，推理速度快——10秒音频仅需70毫秒。

本文将带你从零开始，一步步部署SenseVoice-small语音识别服务，并将其与Elasticsearch集成，构建一个功能完整的可检索语音库。

2. 环境准备与快速部署

2.1 系统要求与依赖安装

首先确保你的系统满足以下要求：

Python 3.8或更高版本
至少2GB可用内存
支持ONNX Runtime的CPU或GPU

安装所需依赖：

# 创建虚拟环境（可选但推荐） python -m venv sensevoice_env source sensevoice_env/bin/activate # 安装核心依赖 pip install funasr-onnx gradio fastapi uvicorn soundfile jieba # 安装Elasticsearch客户端 pip install elasticsearch # 安装其他工具库 pip install pydub librosa

2.2 一键启动语音识别服务

SenseVoice-small提供了开箱即用的服务，只需几行命令即可启动：

# 下载示例代码（如果尚未包含） git clone https://github.com/danieldong/sensevoice-small-demo.git cd sensevoice-small-demo # 启动服务 python app.py --host 0.0.0.0 --port 7860

服务启动后，你可以通过以下方式访问：

Web界面：http://localhost:7860 （上传音频文件进行实时转写）
API文档：http://localhost:7860/docs （查看完整的API接口说明）
健康检查：http://localhost:7860/health （确认服务正常运行）

2.3 验证服务正常运行

使用curl命令测试服务是否正常工作：

curl -X POST "http://localhost:7860/api/transcribe" \ -F "file=@test_audio.wav" \ -F "language=auto" \ -F "use_itn=true"

如果返回类似下面的结果，说明服务部署成功：

{ "text": "这是一个测试音频，用于验证语音识别服务是否正常工作。", "language": "zh", "duration": 5.2 }

3. Elasticsearch搜索引擎部署

3.1 安装与配置Elasticsearch

Elasticsearch是一个强大的分布式搜索引擎，我们将用它来存储和检索语音转写结果。

使用Docker快速部署Elasticsearch：

# 拉取Elasticsearch镜像 docker pull docker.elastic.co/elasticsearch/elasticsearch:8.11.0 # 启动Elasticsearch容器 docker run -d --name elasticsearch \ -p 9200:9200 -p 9300:9300 \ -e "discovery.type=single-node" \ -e "xpack.security.enabled=false" \ -e "ES_JAVA_OPTS=-Xms1g -Xmx1g" \ docker.elastic.co/elasticsearch/elasticsearch:8.11.0

验证Elasticsearch运行状态：

curl -X GET "http://localhost:9200/"

如果返回类似下面的信息，说明Elasticsearch已成功启动：

{ "name": "node-1", "cluster_name": "docker-cluster", "cluster_uuid": "abcd1234", "version": { "number": "8.11.0", "build_flavor": "default" } }

3.2 创建语音库索引

我们需要创建一个专门的索引来存储语音转写内容：

from elasticsearch import Elasticsearch # 连接Elasticsearch es = Elasticsearch(["http://localhost:9200"]) # 定义索引映射 index_mapping = { "mappings": { "properties": { "audio_id": {"type": "keyword"}, "file_name": {"type": "keyword"}, "original_text": {"type": "text", "analyzer": "ik_max_word"}, "transcribed_text": {"type": "text", "analyzer": "ik_max_word"}, "language": {"type": "keyword"}, "duration": {"type": "float"}, "emotion": {"type": "keyword"}, "audio_events": {"type": "keyword"}, "timestamp": {"type": "date"}, "file_path": {"type": "keyword"}, "speaker": {"type": "keyword"} } } } # 创建索引 if not es.indices.exists(index="audio_library"): es.indices.create(index="audio_library", body=index_mapping) print("音频库索引创建成功") else: print("音频库索引已存在")

4. 构建可检索语音库系统

4.1 语音处理流水线设计

现在我们将语音识别和搜索引擎结合起来，构建完整的处理流水线：

import os import json from datetime import datetime from funasr_onnx import SenseVoiceSmall from elasticsearch import Elasticsearch class AudioSearchLibrary: def __init__(self): # 初始化语音识别模型 self.model = SenseVoiceSmall( "/root/ai-models/danieldong/sensevoice-small-onnx-quant", batch_size=10, quantize=True ) # 初始化Elasticsearch客户端 self.es = Elasticsearch(["http://localhost:9200"]) def process_audio_file(self, file_path, language="auto"): """处理单个音频文件""" try: # 语音识别 result = self.model([file_path], language=language, use_itn=True) if result and len(result) > 0: transcription = result[0] # 提取元数据 audio_id = os.path.basename(file_path).split('.')[0] file_name = os.path.basename(file_path) # 构建文档 doc = { "audio_id": audio_id, "file_name": file_name, "transcribed_text": transcription.get("text", ""), "language": transcription.get("language", "unknown"), "duration": transcription.get("duration", 0), "emotion": transcription.get("emotion", "neutral"), "audio_events": transcription.get("audio_events", []), "timestamp": datetime.now(), "file_path": file_path } # 存储到Elasticsearch self.es.index(index="audio_library", id=audio_id, body=doc) return doc return None except Exception as e: print(f"处理音频文件失败: {str(e)}") return None def batch_process(self, directory_path, language="auto"): """批量处理目录下的所有音频文件""" supported_formats = ['.wav', '.mp3', '.m4a', '.flac'] processed_files = [] for filename in os.listdir(directory_path): if any(filename.lower().endswith(ext) for ext in supported_formats): file_path = os.path.join(directory_path, filename) result = self.process_audio_file(file_path, language) if result: processed_files.append(result) return processed_files

4.2 实现语音内容搜索功能

有了存储的语音转写数据，现在实现搜索功能：

class AudioSearchEngine: def __init__(self): self.es = Elasticsearch(["http://localhost:9200"]) def search_audio(self, query, size=10, page=0): """搜索语音内容""" search_body = { "query": { "multi_match": { "query": query, "fields": ["transcribed_text", "original_text", "file_name"], "type": "best_fields" } }, "highlight": { "fields": { "transcribed_text": {}, "original_text": {} } }, "from": page * size, "size": size, "sort": [{"timestamp": {"order": "desc"}}] } try: response = self.es.search(index="audio_library", body=search_body) return self._format_search_results(response) except Exception as e: print(f"搜索失败: {str(e)}") return [] def search_by_language(self, language, size=10): """按语言搜索""" search_body = { "query": { "term": { "language": language } }, "size": size, "sort": [{"timestamp": {"order": "desc"}}] } response = self.es.search(index="audio_library", body=search_body) return self._format_search_results(response) def _format_search_results(self, response): """格式化搜索结果""" results = [] for hit in response.get('hits', {}).get('hits', []): source = hit['_source'] result = { 'audio_id': hit['_id'], 'score': hit['_score'], 'text': source.get('transcribed_text', ''), 'language': source.get('language', 'unknown'), 'duration': source.get('duration', 0), 'file_name': source.get('file_name', ''), 'timestamp': source.get('timestamp', ''), 'highlight': hit.get('highlight', {}) } results.append(result) return { 'total': response['hits']['total']['value'], 'results': results }

5. 完整系统集成与使用

5.1 构建Web管理界面

为了方便使用，我们可以创建一个简单的Web界面来管理语音库：

from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.responses import HTMLResponse import shutil import os app = FastAPI(title="语音检索系统") # 初始化组件 audio_library = AudioSearchLibrary() search_engine = AudioSearchEngine() @app.post("/api/upload") async def upload_audio(file: UploadFile = File(...), language: str = "auto"): """上传并处理音频文件""" try: # 保存上传的文件 file_path = f"uploads/{file.filename}" os.makedirs("uploads", exist_ok=True) with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) # 处理音频文件 result = audio_library.process_audio_file(file_path, language) if result: return { "success": True, "message": "文件处理成功", "data": result } else: raise HTTPException(status_code=500, detail="音频处理失败") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/search") async def search_audios(query: str, page: int = 0, size: int = 10): """搜索语音内容""" try: results = search_engine.search_audio(query, size, page) return { "success": True, "data": results } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/stats") async def get_library_stats(): """获取语音库统计信息""" try: # 获取索引统计 stats = audio_library.es.indices.stats(index="audio_library") count = stats['indices']['audio_library']['total']['docs']['count'] # 获取语言分布 aggs = { "size": 0, "aggs": { "language_distribution": { "terms": { "field": "language" } } } } response = audio_library.es.search(index="audio_library", body=aggs) languages = response['aggregations']['language_distribution']['buckets'] return { "total_audios": count, "languages": languages } except Exception as e: raise HTTPException(status_code=500, detail=str(e))

5.2 批量处理现有音频文件

如果你已经有大量音频文件需要处理，可以使用批量处理功能：

def process_existing_library(audio_directory): """处理现有的音频库""" library = AudioSearchLibrary() print(f"开始处理目录: {audio_directory}") results = library.batch_process(audio_directory, language="auto") print(f"处理完成! 成功处理 {len(results)} 个文件") # 生成处理报告 report = { "total_processed": len(results), "languages": {}, "total_duration": 0 } for result in results: lang = result.get("language", "unknown") report["languages"][lang] = report["languages"].get(lang, 0) + 1 report["total_duration"] += result.get("duration", 0) print("处理报告:") print(f"总文件数: {report['total_processed']}") print(f"总时长: {report['total_duration']:.2f} 秒") print("语言分布:") for lang, count in report["languages"].items(): print(f" {lang}: {count} 个文件") return report # 使用示例 if __name__ == "__main__": report = process_existing_library("/path/to/your/audio/files")

6. 实际应用案例与效果

6.1 会议记录检索系统

假设你有一个包含1000+小时会议录音的库，现在可以轻松实现：

搜索"上个季度财报讨论" → 立即找到相关会议片段
按发言人筛选 → 只看某位高管的发言
按时间范围过滤 → 查找特定时间段的讨论

6.2 多媒体内容管理

对于播客、视频创作者：

自动为视频生成可搜索的字幕库
通过文字内容快速定位视频片段
分析不同主题的内容分布和热度

6.3 客户服务质检

对客服通话录音：

搜索投诉相关对话进行质量检查
分析客户情绪变化趋势
识别常见问题和解决方案

7. 性能优化与扩展建议

7.1 系统性能优化

处理速度优化：

# 使用多线程批量处理 from concurrent.futures import ThreadPoolExecutor def parallel_batch_process(directory_path, max_workers=4): """并行处理音频文件""" supported_formats = ['.wav', '.mp3', '.m4a', '.flac'] audio_files = [] for filename in os.listdir(directory_path): if any(filename.lower().endswith(ext) for ext in supported_formats): audio_files.append(os.path.join(directory_path, filename)) # 使用线程池并行处理 with ThreadPoolExecutor(max_workers=max_workers) as executor: results = list(executor.map( lambda file_path: audio_library.process_audio_file(file_path), audio_files )) return [r for r in results if r is not None]

存储优化：

使用Elasticsearch的索引生命周期管理自动归档旧数据
配置分片和副本策略优化查询性能
定期清理无效或重复的音频记录

7.2 功能扩展建议

添加语音聚类功能：

from sklearn.cluster import KMeans from sklearn.feature_extraction.text import TfidfVectorizer def cluster_audio_by_topic(n_clusters=5): """按主题对音频内容进行聚类""" # 获取所有转写文本 search_body = { "size": 1000, "_source": ["transcribed_text"], "query": {"match_all": {}} } response = audio_library.es.search(index="audio_library", body=search_body) texts = [hit['_source']['transcribed_text'] for hit in response['hits']['hits']] # 文本向量化 vectorizer = TfidfVectorizer(max_features=1000, stop_words='english') X = vectorizer.fit_transform(texts) # K-means聚类 kmeans = KMeans(n_clusters=n_clusters, random_state=42) clusters = kmeans.fit_predict(X) # 更新Elasticsearch中的文档 for i, hit in enumerate(response['hits']['hits']): audio_library.es.update( index="audio_library", id=hit['_id'], body={"doc": {"topic_cluster": int(clusters[i])}} ) return clusters

添加实时处理功能：