Fish Speech 1.5批量语音生成方案：Python脚本调用API处理百篇文案

Fish Speech 1.5批量语音生成方案：Python脚本调用API处理百篇文案

1. 引言：当文字需要被“听见”

想象一下，你手头有上百篇产品介绍、营销文案或者有声书章节，需要全部转换成语音。如果一篇篇手动复制粘贴到网页里，点击生成，再下载保存，这工作量光是想想就让人头疼。效率低不说，还容易出错。

这正是许多内容创作者、自媒体运营者和开发者面临的真实困境。文本转语音（TTS）技术已经非常成熟，但如何高效、批量地将海量文字转化为高质量的语音，仍然是一个技术门槛。

今天，我们就来解决这个问题。本文将带你深入探索Fish Speech 1.5的强大能力，并手把手教你如何绕过繁琐的网页操作，直接使用Python脚本调用其API，实现自动化、批量化处理上百篇文案的语音合成任务。无论你是想为视频批量配音，还是构建一个智能语音播报系统，这套方案都能让你事半功倍。

2. 为什么选择Fish Speech 1.5进行批量处理？

在开始动手之前，我们先搞清楚，面对众多TTS模型，为什么Fish Speech 1.5特别适合批量处理场景。

2.1 核心优势：零样本跨语言与高质量输出

Fish Speech 1.5 不是一个普通的TTS模型。它基于先进的LLaMA架构和VQGAN声码器，拥有两项让你在批量处理时倍感轻松的核心能力：

1. 零样本（Zero-Shot）语音合成：你不需要为每一个新的说话人（音色）去专门训练模型。理论上，你只需要提供一段10-30秒的参考音频，它就能模仿那个音色进行合成。这对于需要多种音色配音的批量任务来说，简直是福音。
2. 强大的跨语言泛化能力：它天生支持中文、英文、日文、韩文等13种语言，并且不需要针对不同语言进行特殊配置或切换模型。这意味着你的文案库中混合了中英文内容，它也能一气呵成地处理，错误率极低（官方数据：5分钟英文文本错误率仅2%）。

2.2 批量处理的“基础设施”：稳定高效的API服务

我们使用的ins-fish-speech-1.5-v1镜像，其设计本身就为程序化调用做好了准备。它采用双服务架构：

• 后端API服务（端口7861）：一个基于FastAPI构建的、功能纯净的推理接口。这是我们脚本将要调用的核心。
• 前端WebUI服务（端口7860）：基于Gradio的交互界面，方便手动测试和调试。

这种架构分离的好处是，API服务非常稳定和轻量，没有前端渲染的负担，特别适合承受来自脚本的连续、高并发请求。你可以把它想象成一个专注的“语音生成工厂”，而我们的Python脚本就是向这个工厂下订单的“调度中心”。

3. 环境准备与API探秘

在编写批量脚本之前，我们需要确保“工厂”已经开工，并且弄清楚“下订单”的规则。

3.1 部署与验证

首先，按照常规流程部署ins-fish-speech-1.5-v1镜像。部署成功后，通过实例的HTTP入口（端口7860）访问Web界面，进行一次简单的手动生成测试，确保核心功能正常。

关键一步：找到API的“门牌号”我们的脚本不会与网页交互，而是直接与后端的API服务对话。你需要知道API服务的准确地址。在同一个服务器实例内部，这个地址是http://127.0.0.1:7861。如果你的脚本计划从其他服务器调用，则需要使用实例的公网IP和端口7861（请确保安全组策略允许该端口的访问）。

你可以通过一个简单的命令来测试API是否就绪：

curl -X POST http://127.0.0.1:7861/v1/tts \ -H "Content-Type: application/json" \ -d '{"text":"你好，API测试。","reference_id":null}' \ --output test.wav

如果生成了一个test.wav文件并能正常播放，恭喜你，API通道已经打通。

3.2 理解API的“订单表单”

要向API“下订单”（发送请求），我们必须按照它规定的格式填写“表单”（请求体）。主要参数如下：

对于批量处理，我们最常打交道的就是text和reference_audio（如果你需要特定音色）。

4. 构建批量语音生成脚本

现在进入核心环节：编写Python脚本。我们将构建一个健壮、实用的脚本，它需要完成读取文案、调用API、保存音频、处理异常等任务。

4.1 基础脚本框架

首先，安装必要的Python库：requests用于调用API，tqdm可以显示美观的进度条。

pip install requests tqdm

下面是脚本的基础框架batch_tts.py：

import os import json import time import requests from tqdm import tqdm from pathlib import Path class FishSpeechBatchTTS: def __init__(self, api_base_url="http://127.0.0.1:7861"): """ 初始化批量TTS处理器 :param api_base_url: Fish Speech API 服务器地址 """ self.api_url = f"{api_base_url.rstrip('/')}/v1/tts" self.session = requests.Session() # 使用Session保持连接，提升效率 self.session.headers.update({'Content-Type': 'application/json'}) def generate_speech(self, text, output_path, reference_audio_path=None, max_tokens=1024, temperature=0.7): """ 生成单段语音并保存 :param text: 要合成的文本 :param output_path: 输出音频文件路径（.wav） :param reference_audio_path: 参考音频路径（用于音色克隆） :param max_tokens: 最大token数 :param temperature: 采样温度 :return: 成功返回True，失败返回False """ # 1. 构建请求数据 payload = { "text": text, "reference_id": None, "max_new_tokens": max_tokens, "temperature": temperature } # 2. 如果提供了参考音频，需要特殊处理（注意：此API需要音频文件在服务器本地） if reference_audio_path: # 注意：这里的 reference_audio 参数，API期望的是服务器上的文件路径。 # 如果你的脚本和API不在同一台机器，需要先将音频文件上传到服务器。 # 此处假设文件已在服务器上，直接传递路径。 payload["reference_audio"] = reference_audio_path try: # 3. 发送POST请求 response = self.session.post(self.api_url, json=payload, timeout=60) # 设置超时 response.raise_for_status() # 如果状态码不是200，抛出异常 # 4. 保存音频文件 with open(output_path, 'wb') as f: f.write(response.content) print(f"✅ 语音生成成功: {output_path}") return True except requests.exceptions.RequestException as e: print(f"❌ 请求失败: {e}") return False except IOError as e: print(f"❌ 文件保存失败: {e}") return False # 示例：处理单条文案 if __name__ == "__main__": tts_client = FishSpeechBatchTTS() test_text = "欢迎使用Fish Speech 1.5，这是一个强大的文本转语音模型。" output_file = "output/sample.wav" # 确保输出目录存在 os.makedirs(os.path.dirname(output_file), exist_ok=True) success = tts_client.generate_speech(text=test_text, output_path=output_file) if success: print(f"音频已保存至: {output_file}")

这个类封装了核心的生成逻辑。generate_speech方法负责一次API调用和文件保存。

4.2 实现批量处理与任务管理

单个调用很简单，但批量处理需要管理任务队列、处理可能的失败、并保存日志。我们来升级脚本：

import csv import logging from datetime import datetime class BatchProcessor(FishSpeechBatchTTS): def __init__(self, api_base_url="http://127.0.0.1:7861"): super().__init__(api_base_url) # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') self.logger = logging.getLogger(__name__) def process_from_csv(self, csv_file_path, output_dir="batch_output", text_column="text", id_column="id", reference_audio_path=None, delay=1.0): """ 从CSV文件中批量读取文案并生成语音 :param csv_file_path: CSV文件路径 :param output_dir: 输出音频文件目录 :param text_column: CSV中包含文本的列名 :param id_column: CSV中作为文件名的ID列名 :param reference_audio_path: 统一的参考音频路径（可选） :param delay: 每次请求之间的延迟（秒），避免服务器压力过大 """ os.makedirs(output_dir, exist_ok=True) # 准备日志文件 log_file = os.path.join(output_dir, f"process_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv") fieldnames = ['id', 'text_preview', 'output_file', 'status', 'error_message', 'timestamp'] successful = 0 failed = 0 try: with open(csv_file_path, 'r', encoding='utf-8') as csvfile, \ open(log_file, 'w', newline='', encoding='utf-8') as logfile: reader = csv.DictReader(csvfile) writer = csv.DictWriter(logfile, fieldnames=fieldnames) writer.writeheader() # 获取总行数用于进度条 total_rows = sum(1 for _ in open(csv_file_path, 'r', encoding='utf-8')) - 1 csvfile.seek(0) # 重置文件指针 next(reader) # 跳过标题行 self.logger.info(f"开始处理CSV文件: {csv_file_path}，共 {total_rows} 条文案") pbar = tqdm(reader, total=total_rows, desc="生成进度") for row in pbar: item_id = row.get(id_column, f"item_{successful+failed+1}") text_content = row.get(text_column, "").strip() if not text_content: self.logger.warning(f"ID {item_id} 的文本内容为空，跳过") writer.writerow({ 'id': item_id, 'text_preview': '[空文本]', 'output_file': '', 'status': 'SKIPPED', 'error_message': 'Empty text', 'timestamp': datetime.now().isoformat() }) failed += 1 continue # 生成输出文件名 safe_filename = f"{item_id}.wav" output_path = os.path.join(output_dir, safe_filename) # 调用生成函数 text_preview = text_content[:50] + "..." if len(text_content) > 50 else text_content pbar.set_postfix_str(f"处理: {text_preview}") success = self.generate_speech( text=text_content, output_path=output_path, reference_audio_path=reference_audio_path ) # 记录日志 log_entry = { 'id': item_id, 'text_preview': text_preview, 'output_file': safe_filename, 'timestamp': datetime.now().isoformat() } if success: log_entry['status'] = 'SUCCESS' log_entry['error_message'] = '' successful += 1 else: log_entry['status'] = 'FAILED' log_entry['error_message'] = 'API request failed' failed += 1 writer.writerow(log_entry) time.sleep(delay) # 请求间隔，避免瞬时高负载 self.logger.info(f"批量处理完成！成功: {successful}, 失败: {failed}") self.logger.info(f"详细日志已保存至: {log_file}") except FileNotFoundError: self.logger.error(f"CSV文件未找到: {csv_file_path}") except Exception as e: self.logger.error(f"处理过程中发生未知错误: {e}") # 使用示例 if __name__ == "__main__": processor = BatchProcessor(api_base_url="http://127.0.0.1:7861") # 根据实际情况修改URL # 场景1：处理一个包含上百条文案的CSV文件 processor.process_from_csv( csv_file_path="scripts/articles.csv", # 你的文案CSV文件 output_dir="batch_output/articles", text_column="content", # CSV中文本所在的列名 id_column="article_id", # CSV中作为ID的列名 delay=0.5 # 每条请求间隔0.5秒 ) # 场景2：如果需要使用特定音色克隆，确保 reference.wav 文件在API服务器上 # processor.process_from_csv( # csv_file_path="scripts/ads.csv", # output_dir="batch_output/ads_with_voice", # text_column="ad_copy", # id_column="ad_id", # reference_audio_path="/root/reference_voices/announcer.wav", # 服务器上的音频文件路径 # delay=1.0 # )

这个增强版的BatchProcessor提供了以下关键功能：

1. 从CSV读取：这是处理批量文案最实用的方式，你的文案可以轻松地在Excel或数据库中管理。
2. 进度可视化：使用tqdm库显示进度条，让你清晰掌握处理进度。
3. 详细日志：将每一条文案的处理结果（成功/失败、文件名、时间戳）记录到CSV日志文件中，便于后续排查和统计。
4. 错误处理与跳过：自动跳过空文本，并捕获网络或文件IO错误，避免整个任务因单条失败而中断。
5. 延迟控制：通过delay参数控制请求频率，保护API服务器不被瞬时大量请求压垮。

4.3 处理长文本与高级技巧

Fish Speech API对单次请求的文本长度有限制（受max_new_tokens控制）。对于非常长的文章（如整章有声书），我们需要在调用前进行分割。

import re class AdvancedBatchProcessor(BatchProcessor): def split_long_text(self, text, max_sentence_length=200): """ 将长文本分割成适合TTS的片段。 这是一个简单的按句号、问号、感叹号分割的示例，你可以根据需求定制更复杂的分割逻辑。 :param text: 原始长文本 :param max_sentence_length: 每个片段的大致最大字符数（非严格） :return: 文本片段列表 """ # 使用正则表达式按中文/英文标点分割句子 sentences = re.split(r'(?<=[。！？.!?])', text) chunks = [] current_chunk = "" for sentence in sentences: sentence = sentence.strip() if not sentence: continue # 如果当前片段加上新句子不会太长，就合并 if len(current_chunk) + len(sentence) <= max_sentence_length: current_chunk += sentence else: # 如果当前片段有内容，先保存 if current_chunk: chunks.append(current_chunk) # 如果单个句子就超长，强制分割（极少数情况） if len(sentence) > max_sentence_length: # 这里可以按逗号、分号等进一步分割，此处简单按长度切分 for i in range(0, len(sentence), max_sentence_length): chunks.append(sentence[i:i+max_sentence_length]) current_chunk = "" else: current_chunk = sentence # 添加最后一个片段 if current_chunk: chunks.append(current_chunk) return chunks def generate_long_article_speech(self, article_id, full_text, output_dir, reference_audio_path=None): """ 生成长文章的语音，自动分割并合并输出信息。 注意：此方法生成多个音频文件，需要后期用音频编辑工具合并。 :param article_id: 文章ID :param full_text: 完整文章文本 :param output_dir: 输出目录 :param reference_audio_path: 参考音频路径 :return: 生成的音频片段文件列表 """ os.makedirs(output_dir, exist_ok=True) # 1. 分割文本 text_chunks = self.split_long_text(full_text) self.logger.info(f"文章 '{article_id}' 被分割为 {len(text_chunks)} 个片段。") generated_files = [] # 2. 为每个片段生成语音 for idx, chunk in enumerate(text_chunks, start=1): chunk_filename = f"{article_id}_part_{idx:03d}.wav" chunk_path = os.path.join(output_dir, chunk_filename) self.logger.info(f"正在生成片段 {idx}/{len(text_chunks)}: {chunk[:50]}...") success = self.generate_speech( text=chunk, output_path=chunk_path, reference_audio_path=reference_audio_path, max_tokens=2048 # 为长片段分配更多token ) if success: generated_files.append(chunk_path) else: self.logger.error(f"片段 {idx} 生成失败，文章 '{article_id}' 可能不完整。") time.sleep(0.5) # 片段间延迟 # 3. 保存片段清单，方便后期处理 manifest_path = os.path.join(output_dir, f"{article_id}_manifest.txt") with open(manifest_path, 'w', encoding='utf-8') as f: for file_path in generated_files: f.write(f"file '{os.path.basename(file_path)}'\n") self.logger.info(f"长文章 '{article_id}' 处理完成。片段清单已保存至: {manifest_path}") self.logger.info(f"提示：可使用 FFmpeg 命令合并音频: ffmpeg -f concat -i {manifest_path} -c copy {article_id}_full.wav") return generated_files # 使用示例：处理长文章 if __name__ == "__main__": advanced_processor = AdvancedBatchProcessor() with open("long_article.txt", "r", encoding="utf-8") as f: long_text = f.read() advanced_processor.generate_long_article_speech( article_id="novel_chapter_01", full_text=long_text, output_dir="batch_output/long_articles" )

5. 实战：百篇文案处理全流程

假设你是一个知识付费平台运营，有100篇干货文章需要制作成音频课程。让我们走一遍完整流程。

5.1 第一步：准备原材料

1. 文案整理：将100篇文章的标题和内容整理到一个CSV文件articles.csv中。

   article_id,title,content 001,如何高效学习Python,在这篇文章中，我们将探讨Python学习的高效方法... 002,机器学习入门指南,机器学习是人工智能的核心领域... ... (其余98篇)

2. 音色准备（可选）：如果你希望所有音频使用同一个专业播音音色，需要准备一段10-30秒的干净人声样本（如announcer.wav），并上传到Fish Speech服务器的一个目录下，例如/root/ref_voices/。

5.2 第二步：配置与运行脚本

1. 修改脚本配置：在batch_tts.py中，确保api_base_url指向正确的服务器地址。
2. 执行批量任务：

   processor = BatchProcessor(api_base_url="http://你的服务器IP:7861") # 如果是远程调用 processor.process_from_csv( csv_file_path="articles.csv", output_dir="audio_course", text_column="content", # 使用CSV中的`content`列 id_column="article_id", # 使用`article_id`作为文件名 reference_audio_path="/root/ref_voices/announcer.wav", # 使用统一播音音色 delay=0.8 # 适当间隔，平衡速度与稳定性 )

3. 监控与等待：脚本运行后，你会看到进度条。处理100条文案，假设每条平均需3秒生成加0.8秒延迟，大约需要6-7分钟。所有生成的wav文件将保存在audio_course/目录下，并以001.wav,002.wav... 命名。

5.3 第三步：后期处理与集成

1. 质量抽检：随机播放几段生成的音频，检查音质和内容是否正确。
2. 元数据关联：根据process_log_*.csv日志文件，你可以轻松地将生成的音频文件与原始文章的ID、标题进行关联，导入到你的内容管理系统。
3. 自动化集成：你可以将此脚本封装成一个函数或服务，集成到你的内容发布流水线中。每当有新文章发布时，自动触发语音生成。

6. 总结

通过本文的探索，我们成功地将Fish Speech 1.5从一个手动操作的网页工具，转变为一个强大的、自动化的批量语音生产引擎。回顾一下关键要点：

1. 核心价值：Fish Speech 1.5的零样本跨语言能力和高质量的语音合成，使其成为批量处理任务的理想选择，尤其适合多语言、多音色要求的场景。
2. 技术关键：其提供的RESTful API是自动化批处理的基石。我们通过Python的requests库与之交互，实现了程序化调用。
3. 工程化脚本：我们构建的脚本不仅仅是简单的API调用循环，它包含了任务管理、错误处理、进度跟踪、详细日志等生产级功能，能够稳健地处理上百甚至上千条任务。
4. 灵活扩展：脚本支持音色克隆（通过reference_audio参数）和长文本分割，能够应对复杂的实际需求。

这套方案的价值在于将重复性劳动转化为一次性自动化任务。无论是制作海量的有声内容、为视频矩阵生成配音，还是构建智能语音交互系统，你都可以在喝杯咖啡的时间里，完成过去需要数天手动操作的工作。

获取更多AI镜像

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