CAM++代码实践：将Embedding存入数据库的完整流程
1. 引言
1.1 业务场景描述
在构建说话人识别系统时，仅完成单次语音比对或特征提取远远不够。实际应用中，我们往往需要建立一个声纹数据库，用于长期存储用户语音的Embedding向量，以便后续进行快速检索、身份验证或聚类分析。
本文基于CAM++ 说话人识别系统（由科哥开发），详细介绍如何将提取出的192维Embedding向量持久化存储到数据库中，并实现“注册-查询-比对”的完整闭环流程。该方案适用于企业级声纹门禁、客服身份核验、个性化语音服务等场景。
1.2 痛点分析
原生CAM++系统虽然支持将Embedding保存为.npy文件，但存在以下问题：
- 文件管理混乱，难以检索特定用户的声纹
- 多人并发访问时易发生文件覆盖
- 缺乏元数据关联（如用户ID、注册时间）
- 不支持实时相似度搜索
因此，亟需一套工程化方案，将Embedding从本地文件迁移至结构化数据库。
1.3 方案预告
本文将围绕以下核心内容展开：
- 解析CAM++输出的Embedding格式
- 设计声纹数据库表结构
- 实现Python后端自动入库逻辑
- 提供可运行的集成代码示例
- 给出性能优化与部署建议
2. 技术方案选型
2.1 数据库类型对比
| 数据库类型 | 是否适合Embedding存储 | 优势 | 劣势 |
|---|---|---|---|
| SQLite | ✅ 轻量级适用 | 零配置、嵌入式、无需服务端 | 并发能力弱，不适合高并发 |
| MySQL | ⚠️ 可行但非最优 | 普及率高、事务支持好 | 原生不支持向量索引 |
| PostgreSQL + pgvector | ✅ 推荐方案 | 支持向量相似度搜索、成熟生态 | 需额外插件 |
| Milvus / Weaviate | ✅ 高性能向量数据库 | 专为向量设计，支持GPU加速 | 运维复杂，资源消耗大 |
结论：对于中小规模应用，推荐使用PostgreSQL + pgvector 插件；若追求极简部署，可选用SQLite存储NumPy数组。
2.2 最终技术栈选择
本文采用SQLite + Python Flask + CAM++ WebUI 扩展的轻量组合，原因如下：
- 与CAM++本地部署环境兼容性高
- 无需额外安装数据库服务
- 易于调试和二次开发
- 满足大多数中小企业需求
3. 实现步骤详解
3.1 环境准备
确保已安装以下依赖：
pip install numpy flask sqlite3 pydub创建项目目录结构：
/root/speech_campplus_sv_zh-cn_16k/ ├── app.db # SQLite数据库 ├── database.py # 数据库操作模块 ├── routes.py # API路由扩展 ├── outputs/ # 原始输出目录 └── scripts/ └── start_app.sh # 启动脚本（需修改）3.2 创建声纹数据库表
编写database.py，定义用户声纹表：
import sqlite3 import numpy as np import io def create_connection(): """创建数据库连接""" conn = None try: conn = sqlite3.connect('app.db', check_same_thread=False) return conn except Exception as e: print(e) return conn def create_table(): """创建声纹表""" conn = create_connection() cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS speaker_embeddings ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT NOT NULL UNIQUE, embedding BLOB NOT NULL, filename TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') conn.commit() conn.close() def _serialize_array(arr): """将NumPy数组序列化为BLOB""" out = io.BytesIO() np.save(out, arr) return out.getvalue() def _deserialize_array(data): """将BLOB反序列化为NumPy数组""" out = io.BytesIO(data) return np.load(out) class EmbeddingDB: def __init__(self): create_table() def save_embedding(self, user_id: str, embedding: np.ndarray, filename: str = None): conn = create_connection() cursor = conn.cursor() try: blob = _serialize_array(embedding) cursor.execute( "INSERT OR REPLACE INTO speaker_embeddings (user_id, embedding, filename) VALUES (?, ?, ?)", (user_id, blob, filename) ) conn.commit() return True except Exception as e: print(f"保存失败: {e}") return False finally: conn.close() def get_embedding(self, user_id: str) -> np.ndarray: conn = create_connection() cursor = conn.cursor() cursor.execute("SELECT embedding FROM speaker_embeddings WHERE user_id = ?", (user_id,)) row = cursor.fetchone() conn.close() if row: return _deserialize_array(row[0]) return None def list_users(self): conn = create_connection() cursor = conn.cursor() cursor.execute("SELECT user_id, filename, created_at FROM speaker_embeddings") rows = cursor.fetchall() conn.close() return rows3.3 修改CAM++启动脚本以加载数据库
编辑/root/speech_campplus_sv_zh-cn_16k/scripts/start_app.sh：
#!/bin/bash cd /root/speech_campplus_sv_zh-cn_16k # 启动前初始化数据库 python -c "from database import create_table; create_table()" # 原有Gradio启动命令 python app.py --server_port 7860 --server_name 0.0.0.03.4 扩展WebUI功能：新增“注册用户”接口
新建routes.py，添加Flask路由：
from flask import Flask, request, jsonify from database import EmbeddingDB import numpy as np import os app = Flask(__name__) db = EmbeddingDB() @app.route('/api/register', methods=['POST']) def register_speaker(): user_id = request.form.get('user_id') audio_file = request.files.get('audio') if not user_id or not audio_file: return jsonify({"error": "缺少参数"}), 400 # 调用CAM++提取Embedding（模拟调用） # 实际应通过subprocess调用CAM++ CLI或API try: # 模拟调用CAM++特征提取并获取.npy路径 output_dir = f"/root/speech_campplus_sv_zh-cn_16k/outputs/outputs_{int(time.time())}" os.makedirs(output_dir, exist_ok=True) npy_path = os.path.join(output_dir, "embedding.npy") # 此处应替换为真实调用CAM++ CLI的逻辑 # 示例：os.system(f"python extract.py --audio {audio_file} --output {npy_path}") # 模拟生成Embedding（测试用） embedding = np.random.rand(192).astype(np.float32) # 保存到数据库 success = db.save_embedding(user_id, embedding, audio_file.filename) if success: return jsonify({"message": f"用户 {user_id} 注册成功", "dimension": len(embedding)}) else: return jsonify({"error": "数据库保存失败"}), 500 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route('/api/verify', methods=['POST']) def verify_speaker(): user_id = request.form.get('user_id') test_audio = request.files.get('audio') if not user_id or not test_audio: return jsonify({"error": "缺少参数"}), 400 # 获取注册用户的Embedding registered_emb = db.get_embedding(user_id) if registered_emb is None: return jsonify({"error": "用户未注册"}), 404 # 提取测试音频Embedding（模拟） test_emb = np.random.rand(192).astype(np.float32) # 替换为真实提取 # 计算余弦相似度 similarity = np.dot(registered_emb, test_emb) / (np.linalg.norm(registered_emb) * np.linalg.norm(test_emb)) threshold = float(request.form.get('threshold', 0.31)) result = "是同一人" if similarity >= threshold else "不是同一人" return jsonify({ "user_id": user_id, "similarity": float(similarity), "threshold": threshold, "result": result }) if __name__ == '__main__': app.run(port=5000, host='0.0.0.0')3.5 集成到前端页面（可选）
可在CAM++ WebUI中增加两个新Tab：
- 用户注册：输入
user_id，上传音频，点击“注册” - 身份验证：选择已注册用户，上传待测音频，返回比对结果
前端可通过fetch调用上述/api/register和/api/verify接口。
4. 实践问题与优化
4.1 实际落地难点
| 问题 | 解决方案 |
|---|---|
| CAM++无标准API接口 | 使用Gradio客户端或封装CLI调用 |
| 多进程写入冲突 | SQLite使用check_same_thread=False+ 连接池 |
| Embedding精度丢失 | 使用float32保存，避免转为字符串 |
| 音频预处理不一致 | 统一采样率16kHz、单声道WAV |
4.2 性能优化建议
批量注册加速
def batch_register(self, user_list: list): conn = create_connection() cursor = conn.cursor() for user_id, emb, fname in user_list: blob = _serialize_array(emb) cursor.execute("INSERT OR REPLACE ...", (user_id, blob, fname)) conn.commit() # 一次提交添加索引提升查询速度
CREATE INDEX IF NOT EXISTS idx_user_id ON speaker_embeddings (user_id);定期清理过期数据
DELETE FROM speaker_embeddings WHERE created_at < datetime('now', '-6 months');备份机制
# 每日备份 cp app.db app.db.bak.$(date +%Y%m%d)
5. 总结
5.1 实践经验总结
通过本次实践，我们实现了从CAM++系统提取的Embedding向量化存储的完整链路：
- 成功将原本分散的
.npy文件整合进SQLite数据库 - 构建了“注册-存储-查询-比对”的闭环流程
- 提供了可扩展的API接口，便于后续对接业务系统
- 保留了原始系统的可视化能力，同时增强了后台管理功能
5.2 最佳实践建议
- 保持版权信息：遵循开发者“科哥”的开源协议，在二次开发中保留原始声明。
- 优先使用WAV格式：确保输入音频为16kHz、单声道，避免因格式转换导致误差。
- 阈值动态调整：根据实际测试集调整相似度阈值，建议初期设为0.5进行严格验证。
- 安全防护：对外暴露API时增加身份认证，防止恶意注册或暴力比对。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
