SenseVoice-Small ONNX MySQL集成：语音数据存储与分析

1. 引言

每天都有海量的语音数据产生，从客服通话到会议录音，从语音助手交互到在线教育课程。这些语音数据中蕴含着宝贵的商业价值，但如何有效存储、管理和分析这些数据却是一个挑战。

SenseVoice-Small作为一个高效的多语言语音识别模型，能够准确地将语音转换为文本。但识别结果如果只是临时使用，其价值就大打折扣。通过将识别结果存储到MySQL数据库中，我们不仅可以长期保存数据，还能进行深度的统计分析和业务洞察。

本文将展示如何将SenseVoice-Small的语音识别结果与MySQL数据库无缝集成，构建一个完整的语音数据处理流水线。无论你是需要分析客服通话质量、统计会议内容关键词，还是构建语音数据分析平台，这套方案都能为你提供坚实的技术基础。

2. 环境准备与依赖安装

在开始之前，我们需要准备好相应的环境和依赖库。以下是完整的依赖列表和安装步骤：

# 安装SenseVoice相关依赖
pip install sensevoice-onnx
pip install soundfile librosa

# 安装MySQL连接库
pip install mysql-connector-python
pip install sqlalchemy

# 安装数据处理相关库
pip install pandas numpy

如果你使用的是Python虚拟环境，建议先创建并激活环境：

# 创建虚拟环境
python -m venv voice_db_env
source voice_db_env/bin/activate  # Linux/Mac
# 或者 voice_db_env\Scripts\activate  # Windows

确保你的系统中已经安装并运行了MySQL数据库。如果没有，可以参考MySQL官方文档进行安装，或者使用Docker快速启动一个MySQL实例：

# 使用Docker启动MySQL
docker run --name mysql-voice -e MYSQL_ROOT_PASSWORD=your_password -e MYSQL_DATABASE=voice_db -p 3306:3306 -d mysql:8.0

3. 数据库设计

一个好的数据库设计是高效数据管理的基础。针对语音识别数据的特点，我们设计了以下表结构：

3.1 音频文件表（audio_files）

存储音频文件的基本信息和元数据：

CREATE TABLE audio_files (
    id INT AUTO_INCREMENT PRIMARY KEY,
    file_path VARCHAR(500) NOT NULL,
    file_name VARCHAR(255) NOT NULL,
    file_size BIGINT,
    duration FLOAT,
    sample_rate INT,
    channels INT,
    language VARCHAR(10),
    upload_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status ENUM('pending', 'processing', 'completed', 'failed') DEFAULT 'pending',
    UNIQUE KEY unique_file_path (file_path)
);

3.2 识别结果表（recognition_results）

存储语音识别的详细结果：

CREATE TABLE recognition_results (
    id INT AUTO_INCREMENT PRIMARY KEY,
    audio_file_id INT NOT NULL,
    transcript TEXT NOT NULL,
    confidence FLOAT,
    language VARCHAR(10),
    emotion VARCHAR(20),
    contains_events BOOLEAN,
    processing_time FLOAT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (audio_file_id) REFERENCES audio_files(id) ON DELETE CASCADE,
    INDEX idx_audio_file (audio_file_id),
    INDEX idx_language (language),
    INDEX idx_created_at (created_at)
);

3.3 事件检测表（detected_events）

存储检测到的音频事件：

CREATE TABLE detected_events (
    id INT AUTO_INCREMENT PRIMARY KEY,
    recognition_id INT NOT NULL,
    event_type VARCHAR(50) NOT NULL,
    start_time FLOAT,
    end_time FLOAT,
    confidence FLOAT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (recognition_id) REFERENCES recognition_results(id) ON DELETE CASCADE,
    INDEX idx_recognition (recognition_id),
    INDEX idx_event_type (event_type)
);

4. 数据库连接管理

为了高效地管理数据库连接，我们创建一个数据库工具类：

import mysql.connector
from mysql.connector import Error
import logging

class DatabaseManager:
    def __init__(self, host='localhost', database='voice_db', 
                 user='root', password='your_password'):
        self.host = host
        self.database = database
        self.user = user
        self.password = password
        self.connection = None
        self.logger = logging.getLogger(__name__)
    
    def connect(self):
        """建立数据库连接"""
        try:
            self.connection = mysql.connector.connect(
                host=self.host,
                database=self.database,
                user=self.user,
                password=self.password
            )
            if self.connection.is_connected():
                self.logger.info("成功连接到MySQL数据库")
                return True
        except Error as e:
            self.logger.error(f"数据库连接失败: {e}")
            return False
    
    def disconnect(self):
        """关闭数据库连接"""
        if self.connection and self.connection.is_connected():
            self.connection.close()
            self.logger.info("数据库连接已关闭")
    
    def execute_query(self, query, params=None):
        """执行查询语句"""
        try:
            cursor = self.connection.cursor(dictionary=True)
            cursor.execute(query, params or ())
            result = cursor.fetchall()
            cursor.close()
            return result
        except Error as e:
            self.logger.error(f"查询执行失败: {e}")
            return None
    
    def execute_insert(self, query, params=None):
        """执行插入语句并返回插入ID"""
        try:
            cursor = self.connection.cursor()
            cursor.execute(query, params or ())
            self.connection.commit()
            last_id = cursor.lastrowid
            cursor.close()
            return last_id
        except Error as e:
            self.connection.rollback()
            self.logger.error(f"插入操作失败: {e}")
            return None
    
    def __enter__(self):
        self.connect()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.disconnect()

5. 语音识别与数据存储集成

现在我们将SenseVoice-Small的识别结果保存到MySQL数据库中：

import os
from sense_voice import SenseVoice
from database_manager import DatabaseManager
import librosa
import time

class VoiceRecognitionPipeline:
    def __init__(self, model_path=None):
        # 初始化语音识别模型
        self.model = SenseVoice(
            model_path=model_path,
            device='cpu',  # 使用CPU推理
            num_threads=4
        )
        
        # 初始化数据库管理器
        self.db_manager = DatabaseManager()
    
    def process_audio_file(self, audio_path):
        """处理单个音频文件并保存结果到数据库"""
        # 检查文件是否存在
        if not os.path.exists(audio_path):
            raise FileNotFoundError(f"音频文件不存在: {audio_path}")
        
        # 获取音频文件信息
        file_size = os.path.getsize(audio_path)
        duration = librosa.get_duration(filename=audio_path)
        y, sr = librosa.load(audio_path, sr=None)
        channels = 1 if len(y.shape) == 1 else y.shape[0]
        
        # 保存音频文件信息到数据库
        with self.db_manager as db:
            # 插入音频文件记录
            audio_file_query = """
            INSERT INTO audio_files 
            (file_path, file_name, file_size, duration, sample_rate, channels, status)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
            """
            audio_file_id = db.execute_insert(
                audio_file_query,
                (audio_path, os.path.basename(audio_path), file_size, 
                 duration, sr, channels, 'processing')
            )
            
            if not audio_file_id:
                raise Exception("无法保存音频文件信息到数据库")
            
            # 更新文件状态为处理中
            db.execute_insert(
                "UPDATE audio_files SET status = 'processing' WHERE id = %s",
                (audio_file_id,)
            )
        
        try:
            # 进行语音识别
            start_time = time.time()
            result = self.model(audio_path)
            processing_time = time.time() - start_time
            
            # 解析识别结果
            transcript = result.get('text', '')
            language = result.get('language', 'unknown')
            emotion = result.get('emotion', 'neutral')
            
            # 保存识别结果到数据库
            with self.db_manager as db:
                # 插入识别结果
                recognition_query = """
                INSERT INTO recognition_results 
                (audio_file_id, transcript, confidence, language, emotion, 
                 contains_events, processing_time)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                """
                recognition_id = db.execute_insert(
                    recognition_query,
                    (audio_file_id, transcript, 0.9, language, 
                     emotion, False, processing_time)
                )
                
                # 更新音频文件状态为已完成
                db.execute_insert(
                    "UPDATE audio_files SET status = 'completed' WHERE id = %s",
                    (audio_file_id,)
                )
                
                return recognition_id
                
        except Exception as e:
            # 更新文件状态为失败
            with self.db_manager as db:
                db.execute_insert(
                    "UPDATE audio_files SET status = 'failed' WHERE id = %s",
                    (audio_file_id,)
                )
            raise e
    
    def batch_process(self, audio_directory):
        """批量处理目录中的所有音频文件"""
        results = []
        audio_extensions = ['.wav', '.mp3', '.flac', '.m4a']
        
        for root, _, files in os.walk(audio_directory):
            for file in files:
                if any(file.lower().endswith(ext) for ext in audio_extensions):
                    audio_path = os.path.join(root, file)
                    try:
                        result_id = self.process_audio_file(audio_path)
                        results.append({
                            'file': file,
                            'status': 'success',
                            'result_id': result_id
                        })
                    except Exception as e:
                        results.append({
                            'file': file,
                            'status': 'failed',
                            'error': str(e)
                        })
        
        return results

6. 数据查询与分析

存储数据是为了更好地分析和利用。以下是一些常用的数据分析查询：

6.1 基本统计查询

class VoiceDataAnalyzer:
    def __init__(self):
        self.db_manager = DatabaseManager()
    
    def get_basic_stats(self):
        """获取基本统计信息"""
        with self.db_manager as db:
            # 总音频文件数
            total_files = db.execute_query(
                "SELECT COUNT(*) as count FROM audio_files WHERE status = 'completed'"
            )[0]['count']
            
            # 总处理时长
            total_duration = db.execute_query(
                "SELECT SUM(duration) as total FROM audio_files WHERE status = 'completed'"
            )[0]['total'] or 0
            
            # 语言分布
            language_dist = db.execute_query("""
                SELECT language, COUNT(*) as count 
                FROM recognition_results 
                GROUP BY language 
                ORDER BY count DESC
            """)
            
            # 情感分布
            emotion_dist = db.execute_query("""
                SELECT emotion, COUNT(*) as count 
                FROM recognition_results 
                GROUP BY emotion 
                ORDER BY count DESC
            """)
            
            return {
                'total_files': total_files,
                'total_duration': round(total_duration, 2),
                'language_distribution': language_dist,
                'emotion_distribution': emotion_dist
            }
    
    def get_processing_performance(self):
        """获取处理性能统计"""
        with self.db_manager as db:
            stats = db.execute_query("""
                SELECT 
                    COUNT(*) as total_files,
                    AVG(processing_time) as avg_processing_time,
                    AVG(duration) as avg_audio_duration,
                    AVG(processing_time / NULLIF(duration, 0)) as avg_rtf
                FROM recognition_results rr
                JOIN audio_files af ON rr.audio_file_id = af.id
                WHERE af.status = 'completed'
            """)[0]
            
            return {
                'total_files': stats['total_files'],
                'avg_processing_time': round(stats['avg_processing_time'], 3),
                'avg_audio_duration': round(stats['avg_audio_duration'], 2),
                'avg_real_time_factor': round(stats['avg_rtf'], 3)
            }

6.2 高级分析查询

    def search_transcripts(self, keywords, language=None, limit=10):
        """在转录文本中搜索关键词"""
        with self.db_manager as db:
            query = """
            SELECT rr.*, af.file_name, af.duration
            FROM recognition_results rr
            JOIN audio_files af ON rr.audio_file_id = af.id
            WHERE rr.transcript LIKE %s
            """
            params = [f'%{keywords}%']
            
            if language:
                query += " AND rr.language = %s"
                params.append(language)
            
            query += " LIMIT %s"
            params.append(limit)
            
            return db.execute_query(query, params)
    
    def get_trend_analysis(self, days=30):
        """获取时间趋势分析"""
        with self.db_manager as db:
            trend_data = db.execute_query(f"""
                SELECT 
                    DATE(created_at) as date,
                    COUNT(*) as file_count,
                    SUM(duration) as total_duration,
                    AVG(processing_time) as avg_processing_time
                FROM recognition_results rr
                JOIN audio_files af ON rr.audio_file_id = af.id
                WHERE af.created_at >= DATE_SUB(NOW(), INTERVAL {days} DAY)
                GROUP BY DATE(created_at)
                ORDER BY date DESC
            """)
            
            return trend_data

7. 性能优化建议

7.1 数据库优化

-- 添加合适的索引
CREATE INDEX idx_audio_files_status ON audio_files(status);
CREATE INDEX idx_recognition_created_at ON recognition_results(created_at);
CREATE INDEX idx_audio_files_upload_time ON audio_files(upload_time);

-- 定期优化表
OPTIMIZE TABLE audio_files, recognition_results, detected_events;

-- 考虑分区表处理大量数据
ALTER TABLE recognition_results PARTITION BY RANGE (YEAR(created_at)) (
    PARTITION p2023 VALUES LESS THAN (2024),
    PARTITION p2024 VALUES LESS THAN (2025),
    PARTITION p2025 VALUES LESS THAN (2026)
);

7.2 应用层优化

# 使用连接池管理数据库连接
from mysql.connector import pooling

class ConnectionPoolManager:
    def __init__(self, pool_size=5):
        self.pool = pooling.MySQLConnectionPool(
            pool_name="voice_pool",
            pool_size=pool_size,
            host='localhost',
            database='voice_db',
            user='root',
            password='your_password'
        )
    
    def get_connection(self):
        return self.pool.get_connection()

# 批量插入优化
def batch_insert_results(self, results):
    """批量插入识别结果"""
    if not results:
        return
    
    with self.db_manager as db:
        query = """
        INSERT INTO recognition_results 
        (audio_file_id, transcript, confidence, language, emotion, processing_time)
        VALUES (%s, %s, %s, %s, %s, %s)
        """
        
        # 分批插入，每批100条
        batch_size = 100
        for i in range(0, len(results), batch_size):
            batch = results[i:i + batch_size]
            params = [
                (r['audio_file_id'], r['transcript'], r['confidence'], 
                 r['language'], r['emotion'], r['processing_time'])
                for r in batch
            ]
            
            cursor = db.connection.cursor()
            cursor.executemany(query, params)
            db.connection.commit()
            cursor.close()

8. 实际应用案例

8.1 客服质量监控系统

class CustomerServiceAnalyzer:
    def __init__(self):
        self.analyzer = VoiceDataAnalyzer()
    
    def analyze_service_quality(self, start_date, end_date):
        """分析客服通话质量"""
        with self.db_manager as db:
            # 获取通话基本统计
            stats = db.execute_query("""
                SELECT 
                    COUNT(*) as total_calls,
                    AVG(duration) as avg_call_duration,
                    SUM(CASE WHEN emotion = 'positive' THEN 1 ELSE 0 END) * 100.0 / COUNT(*) as positive_rate,
                    SUM(CASE WHEN emotion = 'negative' THEN 1 ELSE 0 END) * 100.0 / COUNT(*) as negative_rate
                FROM recognition_results rr
                JOIN audio_files af ON rr.audio_file_id = af.id
                WHERE af.upload_time BETWEEN %s AND %s
                AND af.file_path LIKE '%customer_service%'
            """, (start_date, end_date))[0]
            
            # 获取常见问题关键词
            common_issues = db.execute_query("""
                SELECT 
                    transcript,
                    COUNT(*) as occurrence_count
                FROM recognition_results rr
                JOIN audio_files af ON rr.audio_file_id = af.id
                WHERE af.upload_time BETWEEN %s AND %s
                AND (transcript LIKE '%问题%' OR transcript LIKE '%投诉%' OR transcript LIKE '%不满意%')
                GROUP BY transcript
                ORDER BY occurrence_count DESC
                LIMIT 10
            """, (start_date, end_date))
            
            return {
                'basic_stats': stats,
                'common_issues': common_issues
            }

8.2 会议内容分析

class MeetingAnalyzer:
    def analyze_meetings(self, department=None):
        """分析会议内容"""
        base_query = """
        SELECT 
            af.file_name,
            af.duration,
            rr.transcript,
            rr.language,
            rr.emotion,
            COUNT(*) OVER () as total_meetings,
            LENGTH(rr.transcript) as transcript_length
        FROM recognition_results rr
        JOIN audio_files af ON rr.audio_file_id = af.id
        WHERE af.file_path LIKE '%meeting%'
        """
        
        if department:
            base_query += " AND af.file_path LIKE %s"
            params = [f'%{department}%']
        else:
            params = []
        
        base_query += " ORDER BY af.upload_time DESC"
        
        with self.db_manager as db:
            meetings = db.execute_query(base_query, params)
            
            # 提取关键词频次
            keyword_analysis = {}
            for meeting in meetings:
                transcript = meeting['transcript'].lower()
                # 这里可以添加更多的关键词分析逻辑
                words = transcript.split()
                for word in words:
                    if len(word) > 2:  # 只考虑长度大于2的词
                        keyword_analysis[word] = keyword_analysis.get(word, 0) + 1
            
            return {
                'meetings': meetings,
                'keyword_frequency': sorted(keyword_analysis.items(), key=lambda x: x[1], reverse=True)[:20]
            }

9. 总结

将SenseVoice-Small语音识别结果存储到MySQL数据库中，为我们打开了语音数据分析的大门。通过这种集成方式，我们不仅能够长期保存识别结果，还能进行深度的数据挖掘和分析。

在实际使用中，这套方案已经证明了自己的价值。无论是客服质量监控、会议内容分析，还是语音数据的大规模处理，都能提供稳定可靠的支持。数据库的存储能力结合SenseVoice-Small的识别精度，创造了一加一大于二的效果。

当然，每个项目的需求都不尽相同，你可能需要根据实际情况调整数据库结构或分析逻辑。建议先从简单的应用场景开始，逐步扩展功能。记得定期备份数据库，特别是当数据量积累到一定程度时。

希望本文提供的方案和代码示例能够为你的语音数据处理项目提供有价值的参考。在实际应用中如果遇到问题，欢迎根据具体情况进行调整和优化。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。