ClearerVoice-Studio开发者API文档：RESTful接口定义+Python SDK调用示例

ClearerVoice-Studio开发者API文档RESTful接口定义Python SDK调用示例1. 引言ClearerVoice-Studio是一个功能强大的语音处理开源工具包为开发者提供了一整套语音增强、语音分离和目标说话人提取的解决方案。这个工具包集成了多个先进的预训练模型包括FRCRN、MossFormer2等让开发者无需从零开始训练模型就能直接获得高质量的语音处理能力。对于开发者而言除了使用提供的Web界面外更重要的是能够通过API方式集成这些语音处理功能到自己的应用中。本文将详细介绍ClearerVoice-Studio的RESTful API接口定义并提供完整的Python SDK调用示例帮助开发者快速上手集成。本文学习目标掌握ClearerVoice-Studio的核心API接口使用方法学会使用Python SDK进行语音处理任务的调用了解不同语音处理功能的技术细节和参数配置能够处理常见的API调用错误和异常情况2. 环境准备与API基础2.1 安装必要的Python库在开始使用ClearerVoice-Studio的API之前需要确保安装了必要的Python依赖库# 安装核心依赖 pip install requests numpy soundfile librosa pip install torch torchaudio # 可选用于音频处理的额外库 pip install pydub ffmpeg-python2.2 API服务启动与验证确保ClearerVoice-Studio服务正在运行默认情况下服务运行在http://localhost:8501。可以通过以下方式验证服务状态import requests def check_service_status(): 检查ClearerVoice-Studio服务状态 try: response requests.get(http://localhost:8501/health, timeout5) if response.status_code 200: print(✅ 服务正常运行) return True else: print(❌ 服务异常状态码:, response.status_code) return False except requests.exceptions.ConnectionError: print(❌ 无法连接到服务请确保服务已启动) return False except Exception as e: print(f❌ 检查服务时发生错误: {e}) return False # 检查服务状态 check_service_status()3. RESTful API接口详解3.1 语音增强接口语音增强接口用于去除背景噪音提升语音清晰度支持多种先进的预训练模型。接口端点:POST /api/enhance请求参数:{ audio_file: 音频文件二进制数据, model_name: 模型名称可选默认MossFormer2_SE_48K, enable_vad: 是否启用VAD预处理可选默认false, output_sample_rate: 输出采样率可选默认原始采样率 }支持的模型:MossFormer2_SE_48K: 48kHz高清模型推荐用于专业录音FRCRN_SE_16K: 16kHz标准模型处理速度快MossFormerGAN_SE_16K: 16kHz GAN模型适合复杂噪音环境Python调用示例:import requests import json def enhance_audio(audio_path, model_nameMossFormer2_SE_48K, enable_vadFalse): 语音增强API调用示例 参数: audio_path: 音频文件路径 model_name: 选择的模型名称 enable_vad: 是否启用VAD预处理 返回: 处理后的音频文件路径 url http://localhost:8501/api/enhance # 准备请求数据 files {audio_file: open(audio_path, rb)} data { model_name: model_name, enable_vad: str(enable_vad).lower() } try: # 发送请求 response requests.post(url, filesfiles, datadata) if response.status_code 200: # 保存处理后的音频 output_path fenhanced_{audio_path} with open(output_path, wb) as f: f.write(response.content) print(f✅ 语音增强完成输出文件: {output_path}) return output_path else: print(f❌ 请求失败状态码: {response.status_code}) print(f错误信息: {response.text}) return None except Exception as e: print(f❌ 调用API时发生错误: {e}) return None finally: files[audio_file].close() # 使用示例 enhance_audio(meeting_recording.wav, model_nameMossFormer2_SE_48K, enable_vadTrue)3.2 语音分离接口语音分离接口能够将混合语音分离为多个独立的说话人语音。接口端点:POST /api/separate请求参数:{ audio_file: 音频或视频文件二进制数据, model_name: 模型名称可选默认MossFormer2_SS_16K, max_speakers: 最大说话人数可选默认自动检测 }Python调用示例:def separate_speakers(audio_path, model_nameMossFormer2_SS_16K, max_speakersNone): 语音分离API调用示例 参数: audio_path: 音频文件路径 model_name: 选择的模型名称 max_speakers: 最大说话人数 返回: 分离后的音频文件列表 url http://localhost:8501/api/separate files {audio_file: open(audio_path, rb)} data {model_name: model_name} if max_speakers: data[max_speakers] str(max_speakers) try: response requests.post(url, filesfiles, datadata) if response.status_code 200: # 语音分离返回多个文件通常是ZIP包 import zipfile import io # 解压返回的ZIP文件 zip_data io.BytesIO(response.content) with zipfile.ZipFile(zip_data, r) as zip_ref: output_files zip_ref.namelist() zip_ref.extractall(separated_output) print(f✅ 语音分离完成生成 {len(output_files)} 个文件) return [fseparated_output/{f} for f in output_files] else: print(f❌ 请求失败状态码: {response.status_code}) return None except Exception as e: print(f❌ 调用API时发生错误: {e}) return None finally: files[audio_file].close() # 使用示例 separated_files separate_speakers(meeting_mixed.wav, max_speakers3)3.3 目标说话人提取接口目标说话人提取接口结合视觉信息从视频中提取特定说话人的语音。接口端点:POST /api/extract请求参数:{ video_file: 视频文件二进制数据, model_name: 模型名称可选默认AV_MossFormer2_TSE_16K, target_speaker: 目标说话人索引可选默认0 }Python调用示例:def extract_speaker(video_path, model_nameAV_MossFormer2_TSE_16K, target_speaker0): 目标说话人提取API调用示例 参数: video_path: 视频文件路径 model_name: 选择的模型名称 target_speaker: 目标说话人索引 返回: 提取后的音频文件路径 url http://localhost:8501/api/extract files {video_file: open(video_path, rb)} data { model_name: model_name, target_speaker: str(target_speaker) } try: response requests.post(url, filesfiles, datadata) if response.status_code 200: # 保存提取的音频 output_path fextracted_speaker_{target_speaker}.wav with open(output_path, wb) as f: f.write(response.content) print(f✅ 目标说话人提取完成输出文件: {output_path}) return output_path else: print(f❌ 请求失败状态码: {response.status_code}) return None except Exception as e: print(f❌ 调用API时发生错误: {e}) return None finally: files[video_file].close() # 使用示例 extract_speaker(interview.mp4, target_speaker1)4. Python SDK封装示例为了更方便地使用ClearerVoice-Studio的API我们可以创建一个Python SDK封装类import requests import json import time from typing import List, Optional import zipfile import io class ClearerVoiceClient: ClearerVoice-Studio Python SDK客户端 def __init__(self, base_urlhttp://localhost:8501): self.base_url base_url self.session requests.Session() def enhance_audio(self, audio_path: str, model_name: str MossFormer2_SE_48K, enable_vad: bool False, output_sample_rate: Optional[int] None) - Optional[str]: 语音增强 参数: audio_path: 音频文件路径 model_name: 模型名称 enable_vad: 是否启用VAD预处理 output_sample_rate: 输出采样率 返回: 处理后的音频文件路径 url f{self.base_url}/api/enhance with open(audio_path, rb) as audio_file: files {audio_file: audio_file} data { model_name: model_name, enable_vad: str(enable_vad).lower() } if output_sample_rate: data[output_sample_rate] str(output_sample_rate) try: response self.session.post(url, filesfiles, datadata) if response.status_code 200: output_path fenhanced_{audio_path} with open(output_path, wb) as f: f.write(response.content) return output_path else: print(f语音增强失败: {response.status_code} - {response.text}) return None except Exception as e: print(f语音增强请求异常: {e}) return None def separate_speakers(self, audio_path: str, model_name: str MossFormer2_SS_16K, max_speakers: Optional[int] None) - Optional[List[str]]: 语音分离 参数: audio_path: 音频文件路径 model_name: 模型名称 max_speakers: 最大说话人数 返回: 分离后的音频文件列表 url f{self.base_url}/api/separate with open(audio_path, rb) as audio_file: files {audio_file: audio_file} data {model_name: model_name} if max_speakers: data[max_speakers] str(max_speakers) try: response self.session.post(url, filesfiles, datadata) if response.status_code 200: # 解压返回的ZIP文件 zip_data io.BytesIO(response.content) with zipfile.ZipFile(zip_data, r) as zip_ref: output_files zip_ref.namelist() extract_dir fseparated_{int(time.time())} zip_ref.extractall(extract_dir) return [f{extract_dir}/{f} for f in output_files] else: print(f语音分离失败: {response.status_code} - {response.text}) return None except Exception as e: print(f语音分离请求异常: {e}) return None def extract_speaker(self, video_path: str, model_name: str AV_MossFormer2_TSE_16K, target_speaker: int 0) - Optional[str]: 目标说话人提取 参数: video_path: 视频文件路径 model_name: 模型名称 target_speaker: 目标说话人索引 返回: 提取后的音频文件路径 url f{self.base_url}/api/extract with open(video_path, rb) as video_file: files {video_file: video_file} data { model_name: model_name, target_speaker: str(target_speaker) } try: response self.session.post(url, filesfiles, datadata) if response.status_code 200: output_path fextracted_speaker_{target_speaker}.wav with open(output_path, wb) as f: f.write(response.content) return output_path else: print(f目标说话人提取失败: {response.status_code} - {response.text}) return None except Exception as e: print(f目标说话人提取请求异常: {e}) return None def get_service_status(self) - bool: 获取服务状态 try: response self.session.get(f{self.base_url}/health, timeout5) return response.status_code 200 except: return False # 使用SDK的示例 def demo_clearervoice_sdk(): 演示如何使用ClearerVoice SDK # 创建客户端实例 client ClearerVoiceClient() # 检查服务状态 if not client.get_service_status(): print(服务未启动请先启动ClearerVoice-Studio服务) return print(开始语音处理演示...) # 示例1: 语音增强 print(\n1. 语音增强演示) enhanced_audio client.enhance_audio( noisy_recording.wav, model_nameMossFormer2_SE_48K, enable_vadTrue ) if enhanced_audio: print(f语音增强成功: {enhanced_audio}) # 示例2: 语音分离 print(\n2. 语音分离演示) separated_files client.separate_speakers( mixed_conversation.wav, max_speakers2 ) if separated_files: print(f语音分离成功生成 {len(separated_files)} 个文件) # 示例3: 目标说话人提取 print(\n3. 目标说话人提取演示) extracted_audio client.extract_speaker( interview_video.mp4, target_speaker1 ) if extracted_audio: print(f目标说话人提取成功: {extracted_audio}) # 运行演示 if __name__ __main__: demo_clearervoice_sdk()5. 高级功能与批量处理5.1 批量处理示例在实际应用中经常需要处理多个文件以下是一个批量处理的示例import os from concurrent.futures import ThreadPoolExecutor from tqdm import tqdm def batch_enhance_audio(input_dir, output_dir, model_nameMossFormer2_SE_48K, max_workers4): 批量语音增强处理 参数: input_dir: 输入音频目录 output_dir: 输出目录 model_name: 模型名称 max_workers: 最大并发数 # 创建输出目录 os.makedirs(output_dir, exist_okTrue) # 获取所有WAV文件 audio_files [f for f in os.listdir(input_dir) if f.endswith(.wav)] print(f找到 {len(audio_files)} 个音频文件开始批量处理...) # 创建客户端 client ClearerVoiceClient() def process_file(audio_file): 处理单个文件 input_path os.path.join(input_dir, audio_file) try: output_path client.enhance_audio(input_path, model_namemodel_name) if output_path: # 移动到输出目录 import shutil shutil.move(output_path, os.path.join(output_dir, os.path.basename(output_path))) return True return False except Exception as e: print(f处理文件 {audio_file} 时出错: {e}) return False # 使用线程池并行处理 with ThreadPoolExecutor(max_workersmax_workers) as executor: results list(tqdm( executor.map(process_file, audio_files), totallen(audio_files), desc处理进度 )) success_count sum(results) print(f批量处理完成成功: {success_count}/{len(audio_files)}) # 使用示例 batch_enhance_audio(raw_audio, enhanced_audio, max_workers2)5.2 实时处理示例对于需要实时处理的场景可以使用流式处理方式import pyaudio import wave import numpy as np from threading import Thread import queue class RealTimeProcessor: 实时语音处理器 def __init__(self, client, chunk_size1024, sample_rate16000): self.client client self.chunk_size chunk_size self.sample_rate sample_rate self.audio_queue queue.Queue() self.is_recording False # 初始化音频设备 self.audio pyaudio.PyAudio() self.stream self.audio.open( formatpyaudio.paInt16, channels1, rateself.sample_rate, inputTrue, frames_per_bufferself.chunk_size ) def start_recording(self): 开始录音 self.is_recording True self.record_thread Thread(targetself._record_audio) self.record_thread.start() print(开始录音...) def stop_recording(self): 停止录音 self.is_recording False self.record_thread.join() print(录音停止) def _record_audio(self): 录音线程 while self.is_recording: try: data self.stream.read(self.chunk_size, exception_on_overflowFalse) self.audio_queue.put(data) except Exception as e: print(f录音错误: {e}) def process_realtime(self, output_filerealtime_enhanced.wav): 实时处理录音 frames [] print(开始实时处理...) try: while self.is_recording or not self.audio_queue.empty(): if not self.audio_queue.empty(): audio_data self.audio_queue.get() frames.append(audio_data) # 每积累一定数据量处理一次 if len(frames) 10: # 每10个chunk处理一次 self._process_chunk(frames) frames [] finally: # 处理剩余数据 if frames: self._process_chunk(frames) # 保存最终结果 self._save_audio(output_file) print(f处理完成输出文件: {output_file}) def _process_chunk(self, frames): 处理音频块 # 将音频数据保存为临时文件 temp_file temp_audio.wav with wave.open(temp_file, wb) as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(self.sample_rate) wf.writeframes(b.join(frames)) # 调用增强API enhanced_file self.client.enhance_audio(temp_file, model_nameFRCRN_SE_16K) # 清理临时文件 if os.path.exists(temp_file): os.remove(temp_file) return enhanced_file # 使用示例 def realtime_demo(): 实时处理演示 client ClearerVoiceClient() processor RealTimeProcessor(client) try: processor.start_recording() input(按下回车键停止录音并开始处理...) processor.stop_recording() processor.process_realtime() except KeyboardInterrupt: processor.stop_recording() print(用户中断处理)6. 错误处理与最佳实践6.1 完善的错误处理机制在实际使用中需要添加完善的错误处理机制def robust_api_call(client, audio_path, max_retries3): 带有重试机制的API调用 参数: client: ClearerVoiceClient实例 audio_path: 音频文件路径 max_retries: 最大重试次数 返回: 处理结果或None for attempt in range(max_retries): try: result client.enhance_audio(audio_path) if result: return result else: print(f第 {attempt 1} 次尝试失败正在重试...) time.sleep(2) # 等待2秒后重试 except requests.exceptions.ConnectionError: print(f第 {attempt 1} 次连接失败检查服务状态...) if not client.get_service_status(): print(服务未运行请启动服务) break time.sleep(5) except Exception as e: print(f第 {attempt 1} 次尝试发生异常: {e}) time.sleep(3) print(f经过 {max_retries} 次尝试后仍然失败) return None # 使用示例 result robust_api_call(client, important_audio.wav, max_retries5)6.2 性能优化建议def optimize_processing(audio_path, use_vadTrue, optimize_modelTrue): 优化处理性能的建议方案 参数: audio_path: 音频文件路径 use_vad: 是否使用VAD预处理 optimize_model: 是否选择优化模型 返回: 处理结果 client ClearerVoiceClient() # 根据需求选择最佳模型 if optimize_model: # 对于实时性要求高的场景选择速度更快的模型 model_name FRCRN_SE_16K else: # 对于质量要求高的场景选择效果更好的模型 model_name MossFormer2_SE_48K # 预处理检查 if not pre_check_audio(audio_path): print(音频文件检查失败) return None # 调用API result client.enhance_audio( audio_path, model_namemodel_name, enable_vaduse_vad ) return result def pre_check_audio(audio_path): 音频文件预处理检查 try: # 检查文件是否存在 if not os.path.exists(audio_path): print(文件不存在) return False # 检查文件大小建议不超过500MB file_size os.path.getsize(audio_path) / (1024 * 1024) # MB if file_size 500: print(f文件过大: {file_size:.2f}MB建议分割处理) return False # 检查音频格式 if not audio_path.lower().endswith(.wav): print(建议使用WAV格式以获得最佳效果) # 这里可以添加格式转换代码 return True except Exception as e: print(f文件检查错误: {e}) return False7. 总结通过本文的详细介绍你应该已经掌握了ClearerVoice-Studio的RESTful API接口使用方法和Python SDK的集成方式。这个强大的语音处理工具包为开发者提供了开箱即用的语音增强、分离和提取能力无需深入了解底层算法细节就能获得专业级的处理效果。关键要点回顾ClearerVoice-Studio提供三种核心语音处理功能的API接口支持多种预训练模型适应不同场景需求Python SDK封装让集成更加简单便捷完善的错误处理和性能优化建议确保稳定运行下一步学习建议尝试在实际项目中集成ClearerVoice-Studio的API探索不同模型在不同场景下的效果差异根据具体需求调整参数配置以获得最佳效果关注项目的更新及时获取新功能和性能改进无论是开发语音处理应用、构建智能客服系统还是进行音频内容分析ClearerVoice-Studio都能为你提供强大的技术支撑。开始使用这些API为你的应用增添专业的语音处理能力吧获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。