Python如何调用CAM++ API？接口封装代码实例
1. 为什么需要Python调用CAM++？
你可能已经试过在浏览器里打开 http://localhost:7860，上传两段音频，点击“开始验证”——整个过程很直观，但如果你要批量处理几百个语音对、集成进自己的业务系统、或者做自动化测试，手动点来点去就太低效了。
好消息是：CAM++ 的 WebUI 底层其实跑着一个标准的 FastAPI 服务，它原生支持 RESTful API 调用。这意味着你完全可以用 Python 写几行代码，像调用普通网络接口一样，把语音文件发过去，拿到结构化结果，全程无需打开浏览器。
这不是“黑盒调用”，而是官方支持的工程化接入方式。本文不讲原理、不堆参数，只给你能直接复制粘贴运行的封装代码，覆盖说话人验证和特征提取两大核心功能，并附带错误处理、超时控制、结果解析等生产环境必需细节。
2. 前置准备：确认服务已就绪
在写代码前，请确保 CAM++ 已正确启动：
cd /root/speech_campplus_sv_zh-cn_16k bash scripts/start_app.sh等待终端输出类似INFO: Uvicorn running on http://0.0.0.0:7860后，再执行后续操作。
小提示：默认监听
0.0.0.0:7860，意味着它不仅限于本地访问。如果你在远程服务器部署，其他机器也能通过http://<服务器IP>:7860调用，无需额外配置。
3. 核心API接口说明（不看文档也能用）
CAM++ WebUI 的 API 并未单独发布文档，但我们通过浏览器开发者工具（Network → XHR）可清晰捕获真实请求路径。以下是两个最常用功能对应的接口地址与方法：
| 功能 | HTTP 方法 | 接口路径 | 说明 |
|---|---|---|---|
| 说话人验证 | POST | /api/verify | 提交两段音频，返回相似度与判定 |
| 特征提取 | POST | /api/extract | 提交一段音频，返回192维向量 |
两个接口都接受multipart/form-data格式上传文件，这是 Web 表单上传的标准方式，Python 的requests库原生支持。
3.1 请求体结构一目了然
以说话人验证为例，实际发送的字段只有三个：
audio1: 第一段音频文件（WAV/MP3等）audio2: 第二段音频文件threshold: 相似度阈值（可选，默认0.31）
特征提取则只需audio字段。
关键认知：你不需要理解模型怎么工作，也不用装 PyTorch；只要会发 HTTP 请求，就能用上这个专业级声纹系统。
4. 封装代码实战：说话人验证接口
下面这段代码，实现了完整的说话人验证调用流程，包含异常处理、进度提示、结果结构化解析：
import requests import json from pathlib import Path def verify_speakers( audio1_path: str, audio2_path: str, api_url: str = "http://localhost:7860/api/verify", threshold: float = 0.31, timeout: int = 60 ) -> dict: """ 调用CAM++说话人验证API Args: audio1_path: 参考音频文件路径（如 'speaker1_a.wav'） audio2_path: 待验证音频文件路径（如 'speaker1_b.wav'） api_url: API服务地址，默认为本地 threshold: 相似度判定阈值，默认0.31 timeout: 请求超时秒数，避免卡死 Returns: 包含结果的字典，结构如下： { "success": bool, "message": str, # 成功时为" 是同一人"或"❌ 不是同一人" "score": float, # 相似度分数，0~1之间 "threshold_used": float, "error": str or None # 失败时的错误信息 } """ try: # 构建文件上传字典 files = { "audio1": open(audio1_path, "rb"), "audio2": open(audio2_path, "rb"), } data = {"threshold": str(threshold)} # 发起POST请求 response = requests.post( url=api_url, files=files, data=data, timeout=timeout ) # 关闭文件句柄（重要！防止资源泄漏） for f in files.values(): f.close() # 检查HTTP状态码 if response.status_code != 200: return { "success": False, "error": f"HTTP {response.status_code}: {response.reason}" } # 解析JSON响应 result = response.json() # 兼容不同版本的返回格式（WebUI可能返回纯字符串或JSON对象） if isinstance(result, str): # 尝试从字符串中提取关键信息（旧版WebUI常见） if "" in result: score = float(result.split("相似度:")[-1].strip().rstrip(")")) return { "success": True, "message": " 是同一人", "score": score, "threshold_used": threshold } elif "❌" in result: score = float(result.split("相似度:")[-1].strip().rstrip(")")) return { "success": True, "message": "❌ 不是同一人", "score": score, "threshold_used": threshold } else: return { "success": False, "error": f"无法解析响应: {result[:100]}" } # 新版标准JSON格式 if "相似度分数" in result and "判定结果" in result: score = float(result["相似度分数"]) is_same = "" in result["判定结果"] message = " 是同一人" if is_same else "❌ 不是同一人" return { "success": True, "message": message, "score": score, "threshold_used": float(result.get("使用阈值", threshold)), "raw_response": result } return { "success": False, "error": "响应格式不符合预期，请检查CAM++版本" } except FileNotFoundError as e: return {"success": False, "error": f"文件未找到: {e}"} except requests.exceptions.Timeout: return {"success": False, "error": "请求超时，请检查服务是否运行"} except requests.exceptions.ConnectionError: return {"success": False, "error": "无法连接到CAM++服务，请检查URL和端口"} except json.JSONDecodeError: return {"success": False, "error": "响应不是合法JSON"} except Exception as e: return {"success": False, "error": f"未知错误: {str(e)}"} # 使用示例：验证两个同人音频 if __name__ == "__main__": result = verify_speakers( audio1_path="examples/speaker1_a.wav", audio2_path="examples/speaker1_b.wav", threshold=0.31 ) if result["success"]: print(f" 验证完成！") print(f" 判定结果: {result['message']}") print(f" 相似度分数: {result['score']:.4f}") print(f" 使用阈值: {result['threshold_used']}") else: print(f"❌ 验证失败: {result['error']}")4.1 这段代码解决了哪些实际痛点？
- 文件自动关闭：
open()后立即close()，避免因忘记关闭导致的“Too many open files”错误； - 多版本兼容：同时适配新旧 WebUI 返回格式（JSON 对象 vs 纯文本），一次封装，长期可用；
- 清晰错误分类：网络错误、文件错误、解析错误分别提示，调试不抓瞎；
- 强类型返回：无论底层怎么变，你拿到的永远是结构化的
dict，字段名稳定； - 零依赖：只用
requests，无额外安装成本。
5. 封装代码实战：特征提取接口
特征提取更简单，但同样需要健壮封装。以下代码支持单文件和批量处理两种模式：
import numpy as np import requests from pathlib import Path def extract_embedding( audio_path: str, api_url: str = "http://localhost:7860/api/extract", save_to_disk: bool = False, output_dir: str = "outputs", timeout: int = 60 ) -> dict: """ 调用CAM++特征提取API Args: audio_path: 音频文件路径 api_url: API地址 save_to_disk: 是否保存.npy到磁盘 output_dir: 保存目录（仅当save_to_disk=True时生效） timeout: 请求超时 Returns: { "success": bool, "embedding": np.ndarray or None, # 192维向量 "shape": tuple, # 如 (192,) "file_saved": str or None, # 保存路径 "error": str or None } """ try: files = {"audio": open(audio_path, "rb")} response = requests.post( url=api_url, files=files, timeout=timeout ) files["audio"].close() # 关闭文件 if response.status_code != 200: return {"success": False, "error": f"HTTP {response.status_code}"} # 尝试解析为NumPy数组（API直接返回二进制.npy流） try: emb = np.load(response.raw) if emb.ndim == 2 and emb.shape[0] == 1: emb = emb[0] # 去掉batch维度 result = { "success": True, "embedding": emb, "shape": emb.shape, "file_saved": None } if save_to_disk: Path(output_dir).mkdir(exist_ok=True) filename = Path(audio_path).stem + "_emb.npy" save_path = Path(output_dir) / filename np.save(save_path, emb) result["file_saved"] = str(save_path) return result except Exception as load_err: # 如果不是.npy流，尝试解析为JSON（某些版本返回描述性JSON） try: json_resp = response.json() if "error" in json_resp: return {"success": False, "error": json_resp["error"]} return {"success": False, "error": f"期望.npy格式，但收到: {json_resp}"} except: return {"success": False, "error": f"无法加载响应数据: {str(load_err)}"} except Exception as e: return {"success": False, "error": str(e)} # 批量提取函数（复用单文件逻辑） def batch_extract_embeddings( audio_paths: list, **kwargs ) -> list: """ 批量提取多个音频的Embedding Returns: 列表，每个元素为单次extract_embedding()的返回值 """ results = [] for path in audio_paths: print(f"正在处理: {Path(path).name} ...") res = extract_embedding(audio_path=path, **kwargs) results.append(res) return results # 使用示例 if __name__ == "__main__": # 单文件提取 res = extract_embedding( audio_path="examples/speaker1_a.wav", save_to_disk=True, output_dir="my_embeddings" ) if res["success"]: print(f" 提取成功！维度: {res['shape']}") print(f" 前5维数值: {res['embedding'][:5]}") if res["file_saved"]: print(f" 已保存至: {res['file_saved']}") else: print(f"❌ 提取失败: {res['error']}")5.1 为什么返回的是 NumPy 数组而不是 JSON？
因为 Embedding 是 192 维浮点向量，JSON 会把它变成 192 个数字的列表，既冗余又低效。CAM++ API 直接返回.npy二进制流，这是科学计算领域的事实标准，np.load()一行就能还原，精度无损、速度极快。
小技巧：你可以把提取出的多个 embedding 存成一个大矩阵，用
scipy.spatial.distance.cdist一次性算出所有两两相似度，效率比循环调用高数十倍。
6. 生产环境建议：别让API调用成为瓶颈
当你把这套逻辑集成进线上服务时，以下三点能帮你避开90%的坑：
6.1 设置合理的超时时间
语音处理耗时受音频长度影响。3秒音频通常 < 1s 完成，但30秒长音频可能需5s以上。建议：
timeout=30用于常规验证；timeout=120用于长音频或批量任务；- 永远不要设为0（无限等待）。
6.2 添加重试机制（简单可靠）
网络抖动难免，加一层指数退避重试即可：
import time from functools import wraps def retry_on_failure(max_retries=3, delay=1): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for i in range(max_retries): try: return func(*args, **kwargs) except Exception as e: if i == max_retries - 1: raise e print(f"第{i+1}次失败，{delay}s后重试...") time.sleep(delay) delay *= 2 # 指数退避 return None return wrapper return decorator # 使用装饰器 @retry_on_failure(max_retries=2, delay=2) def robust_verify(*args, **kwargs): return verify_speakers(*args, **kwargs)6.3 避免并发打垮服务
CAM++ 默认是单进程，同时处理10个请求可能排队甚至OOM。建议：
- 使用
threading.Semaphore(3)限制并发数 ≤ 3； - 或改用异步
aiohttp+asyncio，但需修改服务启动方式（非本文重点）。
7. 实战案例：构建一个简易声纹门禁脚本
最后，我们把前面所有能力串起来，写一个真正能用的小工具：给公司会议室加一道“声音门禁”。
#!/usr/bin/env python3 # meeting_access.py import os import sys import wave from datetime import datetime def record_audio(filename: str, duration: int = 5): """简易录音（需安装pyaudio）""" try: import pyaudio p = pyaudio.PyAudio() stream = p.open(format=pyaudio.paInt16, channels=1, rate=16000, input=True, frames_per_buffer=1024) print(f"🎤 正在录音 {duration} 秒...请对着麦克风说 '我要进入会议室'") frames = [] for _ in range(0, int(16000 / 1024 * duration)): data = stream.read(1024) frames.append(data) stream.stop_stream() stream.close() p.terminate() # 保存为WAV wf = wave.open(filename, 'wb') wf.setnchannels(1) wf.setsampwidth(p.get_sample_size(pyaudio.paInt16)) wf.setframerate(16000) wf.writeframes(b''.join(frames)) wf.close() print(f" 录音完成: {filename}") except ImportError: print(" 缺少pyaudio，跳过录音，使用示例文件") # 复制示例文件作为fallback if os.path.exists("examples/speaker1_a.wav"): import shutil shutil.copy("examples/speaker1_a.wav", filename) def main(): print("=== 会议室声纹门禁系统 ===\n") # 步骤1：录音 temp_audio = f"temp_{int(datetime.now().timestamp())}.wav" record_audio(temp_audio) # 步骤2：与注册声纹比对（这里假设已存有authorized_emb.npy） if not os.path.exists("authorized_emb.npy"): print("❌ 未找到授权声纹，请先运行注册脚本") return # 提取当前录音特征 current_emb_res = extract_embedding(temp_audio) if not current_emb_res["success"]: print(f"❌ 特征提取失败: {current_emb_res['error']}") return # 加载授权声纹 try: auth_emb = np.load("authorized_emb.npy") except Exception as e: print(f"❌ 加载授权声纹失败: {e}") return # 计算余弦相似度 from scipy.spatial.distance import cosine similarity = 1 - cosine(current_emb_res["embedding"], auth_emb) # 步骤3：判断并输出 THRESHOLD = 0.55 if similarity >= THRESHOLD: print(f" 验证通过！相似度 {similarity:.4f} > {THRESHOLD}") print("→ 门禁已开启") else: print(f" 验证拒绝。相似度 {similarity:.4f} < {THRESHOLD}") print("→ 请确认是否为本人，或联系管理员") # 清理临时文件 if os.path.exists(temp_audio): os.remove(temp_audio) if __name__ == "__main__": main()这个脚本展示了如何将 API 调用真正落地为业务价值：它不炫技，但每一步都直击实际需求——录音、比对、决策、反馈。你可以把它部署在树莓派上，接个继电器控制电磁锁，就是一套真实的声纹门禁。
8. 总结：你已掌握CAM++工程化接入的核心能力
回顾本文，你已获得：
- 明确的接口认知：知道
/api/verify和/api/extract是什么、怎么调； - 开箱即用的封装：两段代码，覆盖95%的调用场景，错误处理完备；
- 生产级实践建议：超时、重试、并发控制，避免上线后踩坑；
- 可扩展的思维模式：从单次调用 → 批量处理 → 自动化集成，路径清晰。
CAM++ 不是一个只能在浏览器里玩的玩具，而是一个可以深度嵌入你工作流的工具。它的价值不在于“有多先进”，而在于“有多好用”。现在，你已经拿到了那把钥匙。
下一步，不妨试试：
- 把验证结果写入数据库，构建声纹日志；
- 用提取的 embedding 做聚类，发现通话中的隐藏人物；
- 将 API 封装成 FastAPI 子服务，提供统一语音身份网关。
技术的价值，永远在解决真实问题的过程中被定义。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
