Fish Speech 1.5实战手册：批量文本转语音脚本编写与任务队列管理

Fish Speech 1.5实战手册：批量文本转语音脚本编写与任务队列管理

1. 引言

学习目标：掌握Fish Speech 1.5语音合成模型的批量处理能力，学会编写自动化脚本和任务队列管理方法，实现高效的大规模文本转语音处理。

前置知识：基本了解Python编程和命令行操作即可，无需深度学习背景。

教程价值：Fish Speech 1.5作为支持多种语言的先进语音合成模型，在批量处理场景下能显著提升工作效率。本文将手把手教你如何从单条语音生成扩展到批量处理，并建立可靠的任务管理系统。

2. 环境准备与快速部署

2.1 模型部署确认

首先确保Fish Speech 1.5模型已通过Xinference 2.0.0成功部署。检查模型服务状态：

# 查看服务日志确认启动状态 cat /root/workspace/model_server.log

当看到"启动成功"的提示信息时，说明模型已就绪。如果初次加载，可能需要等待几分钟让模型完全初始化。

2.2 访问Web界面

通过Web UI界面可以快速测试模型功能：

1. 找到并点击webui入口进入界面
2. 在文本框中输入想要合成的文字
3. 点击"生成语音"按钮测试功能
4. 成功后会显示生成的音频文件和播放控件

这个界面适合单条测试，但对于批量处理，我们需要更高效的方法。

3. 基础概念快速入门

3.1 Fish Speech 1.5核心能力

Fish Speech 1.5是一个强大的多语言语音合成模型，基于超过100万小时的音频数据训练而成。它支持13种语言，包括：

3.2 批量处理的价值

单条语音生成适合测试和小规模使用，但在实际应用中，我们经常需要：

• 将大量文字内容转换为语音
• 为视频内容批量生成配音
• 制作多语言版本的有声内容
• 定期处理更新的文本内容

手动操作既耗时又容易出错，因此需要自动化解决方案。

4. 单条语音生成基础

4.1 使用Python调用模型

首先我们学习如何用代码调用模型生成单条语音：

import requests import json def generate_single_speech(text, language="zh", model_endpoint="http://localhost:9997"): """ 生成单条语音 text: 要转换的文本 language: 语言代码 (zh, en, ja等) model_endpoint: 模型服务地址 """ payload = { "text": text, "language": language, "speed": 1.0 # 语速，0.5-2.0之间 } try: response = requests.post( f"{model_endpoint}/v1/audio/speech", json=payload, timeout=30 ) if response.status_code == 200: # 保存音频文件 with open("output.wav", "wb") as f: f.write(response.content) return True else: print(f"生成失败: {response.status_code}") return False except Exception as e: print(f"请求异常: {str(e)}") return False # 示例用法 if __name__ == "__main__": success = generate_single_speech("欢迎使用Fish Speech语音合成系统", "zh") if success: print("语音生成成功！")

这个基础函数让我们能够通过代码生成语音，为批量处理打下基础。

5. 批量文本转语音脚本编写

5.1 基础批量处理脚本

现在我们来编写一个完整的批量处理脚本：

import os import requests import json import time from pathlib import Path class BatchSpeechGenerator: def __init__(self, model_endpoint="http://localhost:9997", output_dir="output"): self.model_endpoint = model_endpoint self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) def process_text_file(self, input_file, language="zh", file_format="wav"): """ 处理文本文件中的内容 input_file: 输入文本文件路径 language: 语言代码 file_format: 输出音频格式 """ try: with open(input_file, 'r', encoding='utf-8') as f: lines = f.readlines() success_count = 0 for i, line in enumerate(lines): line = line.strip() if not line: # 跳过空行 continue print(f"处理第 {i+1} 条: {line[:50]}...") # 生成语音 if self._generate_speech(line, language, f"output_{i+1}.{file_format}"): success_count += 1 # 添加短暂延迟，避免服务器压力过大 time.sleep(0.5) print(f"处理完成！成功: {success_count}/{len(lines)}") except Exception as e: print(f"文件处理失败: {str(e)}") def _generate_speech(self, text, language, filename): """内部方法：生成单条语音""" payload = { "text": text, "language": language, "speed": 1.0 } try: response = requests.post( f"{self.model_endpoint}/v1/audio/speech", json=payload, timeout=30 ) if response.status_code == 200: output_path = self.output_dir / filename with open(output_path, "wb") as f: f.write(response.content) return True else: print(f"生成失败 (状态码: {response.status_code})") return False except Exception as e: print(f"请求异常: {str(e)}") return False # 使用示例 if __name__ == "__main__": generator = BatchSpeechGenerator() # 处理文本文件 generator.process_text_file("input.txt", language="zh")

5.2 增强版批量脚本

基础脚本只能处理简单文本，我们来增强它的功能：

import csv import json import pandas as pd from datetime import datetime class EnhancedSpeechGenerator(BatchSpeechGenerator): def __init__(self, model_endpoint="http://localhost:9997", output_dir="output"): super().__init__(model_endpoint, output_dir) self.log_file = self.output_dir / "processing_log.csv" def process_csv_file(self, csv_file, text_column="text", language_column=None): """ 处理CSV文件，支持多语言和元数据 csv_file: CSV文件路径 text_column: 文本内容列名 language_column: 语言代码列名（可选） """ try: df = pd.read_csv(csv_file) results = [] for index, row in df.iterrows(): text = row[text_column] language = row[language_column] if language_column else "zh" # 生成文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"speech_{index+1}_{timestamp}.wav" # 生成语音 success = self._generate_speech(text, language, filename) # 记录结果 result = { "index": index + 1, "text": text[:100] + "..." if len(text) > 100 else text, "language": language, "filename": filename, "success": success, "timestamp": datetime.now().isoformat() } results.append(result) print(f"处理进度: {index+1}/{len(df)}") time.sleep(0.3) # 控制请求频率 # 保存处理日志 self._save_results(results) return results except Exception as e: print(f"CSV处理失败: {str(e)}") return [] def _save_results(self, results): """保存处理结果到CSV文件""" if results: df = pd.DataFrame(results) df.to_csv(self.log_file, index=False, encoding='utf-8-sig') print(f"处理日志已保存: {self.log_file}") # 使用示例 if __name__ == "__main__": generator = EnhancedSpeechGenerator() # 处理CSV文件 results = generator.process_csv_file( "content.csv", text_column="content", language_column="lang" )

6. 任务队列管理系统

6.1 基于文件的任务队列

对于大规模处理，我们需要一个可靠的任务队列系统：

import sqlite3 import threading import queue import time class SpeechTaskQueue: def __init__(self, db_path="speech_tasks.db"): self.db_path = db_path self._init_database() def _init_database(self): """初始化任务数据库""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, text TEXT NOT NULL, language TEXT DEFAULT 'zh', status TEXT DEFAULT 'pending', -- pending, processing, completed, failed priority INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, started_at TIMESTAMP, completed_at TIMESTAMP, output_file TEXT, error_message TEXT ) ''') conn.commit() conn.close() def add_task(self, text, language="zh", priority=1): """添加新任务到队列""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' INSERT INTO tasks (text, language, priority, status) VALUES (?, ?, ?, 'pending') ''', (text, language, priority)) task_id = cursor.lastrowid conn.commit() conn.close() return task_id def add_batch_tasks(self, tasks): """批量添加任务""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() for task in tasks: cursor.execute(''' INSERT INTO tasks (text, language, priority, status) VALUES (?, ?, ?, 'pending') ''', (task['text'], task.get('language', 'zh'), task.get('priority', 1))) conn.commit() conn.close() def get_next_task(self): """获取下一个待处理任务""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' SELECT id, text, language FROM tasks WHERE status = 'pending' ORDER BY priority DESC, created_at ASC LIMIT 1 ''') task = cursor.fetchone() if task: task_id, text, language = task # 更新任务状态为处理中 cursor.execute(''' UPDATE tasks SET status = 'processing', started_at = CURRENT_TIMESTAMP WHERE id = ? ''', (task_id,)) conn.commit() result = { 'id': task_id, 'text': text, 'language': language } else: result = None conn.close() return result def mark_task_completed(self, task_id, output_file): """标记任务完成""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' UPDATE tasks SET status = 'completed', completed_at = CURRENT_TIMESTAMP, output_file = ? WHERE id = ? ''', (output_file, task_id)) conn.commit() conn.close() def mark_task_failed(self, task_id, error_message): """标记任务失败""" conn = sqlite3.connect(self.db_path) cursor = conn.cursor() cursor.execute(''' UPDATE tasks SET status = 'failed', completed_at = CURRENT_TIMESTAMP, error_message = ? WHERE id = ? ''', (error_message, task_id)) conn.commit() conn.close()

6.2 多线程任务处理器

结合任务队列，我们创建多线程处理器：

class MultiThreadedSpeechProcessor: def __init__(self, model_endpoint, db_path="speech_tasks.db", max_workers=3): self.model_endpoint = model_endpoint self.task_queue = SpeechTaskQueue(db_path) self.max_workers = max_workers self.stop_event = threading.Event() def start_processing(self): """启动多线程处理""" threads = [] for i in range(self.max_workers): thread = threading.Thread(target=self._worker_thread, args=(i,)) thread.daemon = True threads.append(thread) thread.start() print(f"启动工作线程 {i+1}") print("所有工作线程已启动，开始处理任务...") # 等待所有线程完成或收到停止信号 try: while any(thread.is_alive() for thread in threads) and not self.stop_event.is_set(): time.sleep(1) except KeyboardInterrupt: print("收到停止信号，正在停止工作线程...") self.stop_event.set() for thread in threads: thread.join(timeout=5) print("处理已停止") def _worker_thread(self, worker_id): """工作线程函数""" while not self.stop_event.is_set(): task = self.task_queue.get_next_task() if not task: # 没有任务，等待一段时间再检查 time.sleep(5) continue print(f"线程 {worker_id} 处理任务 {task['id']}") try: # 生成语音 filename = f"output/task_{task['id']}_{worker_id}.wav" success = self._generate_speech(task['text'], task['language'], filename) if success: self.task_queue.mark_task_completed(task['id'], filename) print(f"线程 {worker_id} 完成任务 {task['id']}") else: self.task_queue.mark_task_failed(task['id'], "语音生成失败") print(f"线程 {worker_id} 任务 {task['id']} 失败") except Exception as e: error_msg = str(e) self.task_queue.mark_task_failed(task['id'], error_msg) print(f"线程 {worker_id} 任务 {task['id']} 异常: {error_msg}") # 处理间隔，避免过快请求 time.sleep(1) def _generate_speech(self, text, language, filename): """生成语音的内部方法""" payload = { "text": text, "language": language, "speed": 1.0 } try: response = requests.post( f"{self.model_endpoint}/v1/audio/speech", json=payload, timeout=30 ) if response.status_code == 200: # 确保输出目录存在 os.makedirs(os.path.dirname(filename), exist_ok=True) with open(filename, "wb") as f: f.write(response.content) return True else: return False except Exception: return False # 使用示例 if __name__ == "__main__": # 先添加一些任务 queue = SpeechTaskQueue() # 添加单个任务 queue.add_task("这是第一个测试任务", "zh") # 添加批量任务 tasks = [ {"text": "欢迎使用语音合成系统", "language": "zh"}, {"text": "Hello, this is English text", "language": "en"}, {"text": "こんにちは、これはテストです", "language": "ja"} ] queue.add_batch_tasks(tasks) # 启动处理器 processor = MultiThreadedSpeechProcessor( model_endpoint="http://localhost:9997", max_workers=2 ) processor.start_processing()

7. 实用技巧与进阶功能

7.1 错误处理与重试机制

在实际应用中，网络波动或服务暂时不可用是常见问题。我们需要增强错误处理：

def robust_generate_speech(text, language, filename, max_retries=3, retry_delay=2): """ 带重试机制的语音生成 max_retries: 最大重试次数 retry_delay: 重试延迟(秒) """ for attempt in range(max_retries): try: success = generate_single_speech(text, language, filename) if success: return True else: print(f"第 {attempt+1} 次尝试失败") except Exception as e: print(f"第 {attempt+1} 次尝试异常: {str(e)}") if attempt < max_retries - 1: time.sleep(retry_delay * (attempt + 1)) # 指数退避 print(f"所有 {max_retries} 次尝试均失败") return False

7.2 进度监控与报告

对于长时间运行的批量任务，进度监控很重要：

def monitor_progress(db_path, update_interval=10): """监控任务进度""" conn = sqlite3.connect(db_path) while True: cursor = conn.cursor() cursor.execute(''' SELECT COUNT(*) as total, SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as completed, SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) as failed, SUM(CASE WHEN status = 'processing' THEN 1 ELSE 0 END) as processing FROM tasks ''') total, completed, failed, processing = cursor.fetchone() pending = total - completed - failed - processing print(f"\r进度: 总数={total} 已完成={completed} 失败={failed} 处理中={processing} 等待中={pending}", end="") if pending == 0 and processing == 0: print("\n所有任务处理完成！") break time.sleep(update_interval) conn.close()

8. 总结

8.1 学习回顾

通过本教程，我们掌握了Fish Speech 1.5语音合成模型的批量处理能力：

1. 基础调用：学会了如何使用Python代码调用模型生成单条语音
2. 批量处理：编写了能够处理文本文件和CSV文件的批量脚本
3. 任务管理：建立了基于SQLite的任务队列系统，支持优先级和状态跟踪
4. 并发处理：实现了多线程任务处理器，提高处理效率
5. 错误处理：添加了重试机制和进度监控，确保任务可靠性

8.2 实际应用建议

在实际项目中，建议：

1. 控制请求频率：避免过快请求导致服务器压力过大，添加适当的延迟
2. 监控资源使用：注意内存和CPU使用情况，适当调整并发线程数
3. 定期备份任务：对于重要任务，定期备份任务数据库
4. 日志记录：完善日志系统，便于问题排查和性能分析

8.3 扩展思路

你可以进一步扩展这个系统：

• 添加Web界面用于任务提交和监控
• 集成到现有的内容管理系统中
• 添加音频后处理功能（如音量标准化、格式转换）
• 实现分布式处理，支持多台服务器协同工作

现在你已经具备了使用Fish Speech 1.5进行批量语音合成的完整能力，可以开始在实际项目中应用这些技术了。

获取更多AI镜像

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