Fish Speech 1.5 结合 Whisper 打造语音处理自动化闭环
将 Fish Speech 1.5 与 Whisper 语音识别模型结合,可以构建一个'语音→文本→语音'的智能闭环。这不仅能自动整理会议录音生成纪要,还能实现外语视频的翻译配音,保持原说话人的音色。无论你是内容创作者还是开发者,这套组合都能显著提升效率。
为什么需要语音闭环?
单独使用 Fish Speech 合成语音已经很强,但整合 Whisper 能解放双手。传统流程需要手动转录、编辑、再合成,繁琐且易错。闭环系统则能:
- 自动化处理:上传音频,自动完成识别、处理、合成。
- 音色一致性:视频配音时,先识别原音,再克隆音色合成新语音。
- 无障碍应用:快速为音频生成字幕,或为字幕生成语音导读。
- 内容再创作:轻松实现语音内容的翻译、摘要后以语音形式输出。
Fish Speech 负责'说',Whisper 负责'听',两者结合才是一个完整的对话系统。
环境搭建与工具准备
假设你已经部署了 Fish Speech 1.5 的 Web 服务(通常地址类似 http://localhost:7860),这是我们的语音合成端。
部署 Whisper 语音识别服务
Whisper 是 OpenAI 开源的强大模型,支持多语言。
方案一:本地部署(推荐)
pip install -U openai-whisper
sudo apt update && sudo apt install ffmpeg # Ubuntu/Debian
brew install ffmpeg # MacOS
whisper --help # 验证安装
方案二:在线 API 如果不方便本地部署,可使用提供 Whisper API 的在线服务,替换调用逻辑即可。
编写调度脚本
我们需要一个 Python 脚本作为'调度员',连接识别与合成服务。创建一个名为 speech_loop.py 的文件。
import requests
import whisper
import soundfile as sf
import io
import json
import logging
from typing import Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SpeechTextSpeechLoop:
def __init__(self, fish_speech_url: str, whisper_model_size: str = "base"):
self.fish_speech_url = fish_speech_url.rstrip('/')
logger.info(f"加载 Whisper 模型:{whisper_model_size}")
self.whisper_model = whisper.load_model(whisper_model_size)
logger.info("模型加载完毕。")
def transcribe_audio(self, audio_path: str, language: Optional[str] = None) -> str:
logger.info(f"开始识别音频:{audio_path}")
result = self.whisper_model.transcribe(audio_path, language=language, fp16=False)
text = result["text"].strip()
logger.info(f"识别结果:{text[:100]}...")
return text
def process_text(self, text: str, operation: str = "none") -> str:
if operation == "none":
return text
elif operation == "translate_to_en":
logger.warning("翻译功能需接入具体 API,当前返回原文本。")
return text
elif operation == "summarize":
logger.warning("摘要功能需接入具体模型,当前返回原文本。")
return text
else:
logger.error(f"未知的处理操作:{operation}")
return text
def synthesize_speech(self, text: str, reference_audio_path: Optional[str] = None, ref_text: Optional[str] = None) -> Optional[bytes]:
payload = {
"text": text,
"language": "zh",
"top_p": 0.7,
"temperature": 0.7,
"repetition_penalty": 1.2,
}
files = {}
if reference_audio_path and ref_text:
try:
files['audio'] = open(reference_audio_path, 'rb')
payload['ref_text'] = ref_text
logger.info("启用声音克隆模式。")
except FileNotFoundError:
logger.error(f"参考音频文件未找到:{reference_audio_path}")
api_endpoint = f"{self.fish_speech_url}/api/synthesize"
logger.info(f"向 {api_endpoint} 发送合成请求...")
try:
if files:
response = requests.post(api_endpoint, data=payload, files=files)
files['audio'].close()
else:
response = requests.post(api_endpoint, json=payload)
response.raise_for_status()
if response.headers.get('Content-Type', '').startswith('audio/'):
return response.content
else:
result = response.json()
logger.info(f"合成请求成功。响应:{result}")
return None
except requests.exceptions.RequestException as e:
logger.error(f"调用 Fish Speech API 失败:{e}")
return None
def run_pipeline(self, input_audio_path: str, output_audio_path: str, text_operation: str = "none", reference_audio_for_clone: Optional[str] = None, ref_text_for_clone: Optional[str] = None):
logger.info("=== 开始语音闭环处理流程 ===")
transcribed_text = self.transcribe_audio(input_audio_path)
processed_text = self.process_text(transcribed_text, text_operation)
logger.info(f"处理后文本:{processed_text[:150]}...")
audio_data = self.synthesize_speech(processed_text, reference_audio_for_clone, ref_text_for_clone)
if audio_data:
with open(output_audio_path, 'wb') as f:
f.write(audio_data)
logger.info(f"处理完成!输出音频已保存至:{output_audio_path}")
return True
else:
logger.error("语音合成失败,流程终止。")
return False
if __name__ == "__main__":
FISH_SPEECH_URL = "http://localhost:7860"
INPUT_AUDIO = "meeting_recording.wav"
OUTPUT_AUDIO = "processed_output.wav"
TEXT_OPERATION = "none"
REF_AUDIO = "reference_speaker.wav"
REF_TEXT = "这是参考音频对应的文字内容。"
processor = SpeechTextSpeechLoop(FISH_SPEECH_URL, whisper_model_size="base")
success = processor.run_pipeline(
input_audio_path=INPUT_AUDIO,
output_audio_path=OUTPUT_AUDIO,
text_operation=TEXT_OPERATION,
reference_audio_for_clone=REF_AUDIO,
ref_text_for_clone=REF_TEXT
)
注意修改配置中的 FISH_SPEECH_URL 为你实际的地址。如果 Web 界面不支持直接 API 调用,请查阅文档或通过浏览器开发者工具抓取实际接口。
实战应用场景
案例一:会议录音智能纪要
将录音丢给 Whisper 得到逐字稿,通过文本摘要模型提炼核心要点,最后用 Fish Speech 合成摘要语音。省去人工听录整理的过程。
案例二:跨语言视频配音
提取原视频音频,用 Whisper 识别中文文案,翻译为英文,截取原讲解者 5-10 秒独白作为参考音频,进行声音克隆合成英文配音。极大简化视频本地化流程。
案例三:无障碍内容创作
从 PDF 提取文本分段,批量提交给 Fish Speech 合成有声书。可先用 Whisper 识别一段真人朗读片段作为参考,让 Fish Speech 克隆富有感情的讲述风格。
进阶技巧与优化建议
提升 Whisper 识别准确率
- 选择合适模型:高要求场景建议使用
medium或large。 - 指定语言:明确知道语言时指定
language参数。 - 音频预处理:使用 FFmpeg 降噪、归一化音量、转换为单声道 16kHz WAV 格式。
ffmpeg -i input_noisy.mp3 -af "arnndn=mode=denoise, loudnorm" -ar 16000 -ac 1 output_clean.wav
优化 Fish Speech 合成效果
- 参数调优:叙述性文本降低
temperature增加稳定性;情感丰富文本提高temperature。 - 高质量声音克隆:参考音频需清晰无噪,参考文本必须与音频内容一字不差。
构建自动化流水线
生产环境可封装成服务,使用 Celery + Redis 处理异步任务,避免 HTTP 超时。
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import FileResponse
app = FastAPI()
processor = SpeechTextSpeechLoop("http://localhost:7860")
@app.post("/process")
async def process_audio(file: UploadFile = File(...)):
input_path = f"/tmp/{file.filename}"
with open(input_path, "wb") as f:
f.write(await file.read())
output_path = input_path.replace(".wav", "_output.wav")
success = processor.run_pipeline(input_path, output_path)
if success:
return FileResponse(output_path, media_type="audio/wav")
else:
return {"error": "Processing failed"}
总结
将 Fish Speech 1.5 与 Whisper 结合,代表了一种工作流的革新。我们从需求出发,指导搭建环境、编写核心调度脚本,并通过会议纪要、视频配音、有声书三个案例展示了其威力。最后分享了进阶技巧,提供了从'能用'到'工业化用'的路径。技术的魅力在于连接与创造,现在就看你想用这个系统去解决什么问题了。
