Fish-Speech 1.5 API调用指南：Python示例详解

1. 项目概述

Fish-Speech 1.5 是一个基于创新的 DualAR 架构的文本转语音（TTS）系统，由 Fish Audio 团队开发。这个开源项目采用了双自回归 Transformer 设计，主 Transformer 以 21Hz 运行，次 Transformer 负责将潜在状态转换为声学特征，这种设计让模型的计算效率和语音输出质量都优于传统级联方法。

与传统 TTS 系统不同，Fish-Speech 1.5 摒弃了对音素的依赖，能够直接理解和处理文本，无需繁杂的语音规则库，大幅提升了泛化能力。项目提供了 WebUI 和 API 两种使用方式，本文重点介绍如何通过 Python 调用其 RESTful API 进行语音合成。

2. 环境准备与API连接

2.1 确认服务状态

在开始调用 API 之前，首先需要确认 Fish-Speech 1.5 的 API 服务正在运行。默认情况下，API 服务运行在服务器的 8080 端口。

# 检查服务状态
curl -I http://服务器IP:8080/v1/tts

# 或者查看 Supervisor 状态
supervisorctl status fish-speech

2.2 安装必要的Python库

确保你的 Python 环境安装了 requests 库，这是调用 API 的基础依赖：

pip install requests

如果需要处理音频文件，还可以安装 soundfile 或 pydub：

pip install soundfile pydub

3. 基础API调用示例

3.1 最简单的文本转语音

下面是一个最基本的 API 调用示例，将文本转换为语音并保存为 WAV 文件：

import requests
import json

def basic_tts(text, output_file="output.wav", server_ip="localhost"):
    """
    基础文本转语音函数
    
    Args:
        text (str): 要转换为语音的文本
        output_file (str): 输出音频文件名
        server_ip (str): 服务器IP地址
    """
    # API 端点
    url = f"http://{server_ip}:8080/v1/tts"
    
    # 请求参数
    payload = {
        "text": text,
        "references": [],
        "reference_id": None,
        "max_new_tokens": 1024,
        "chunk_length": 200,
        "top_p": 0.7,
        "repetition_penalty": 1.2,
        "temperature": 0.7,
        "format": "wav"
    }
    
    try:
        # 发送 POST 请求
        response = requests.post(url, json=payload, timeout=30)
        
        # 检查响应状态
        if response.status_code == 200:
            # 保存音频文件
            with open(output_file, "wb") as f:
                f.write(response.content)
            print(f"音频已成功保存到 {output_file}")
            return True
        else:
            print(f"请求失败，状态码: {response.status_code}")
            print(f"错误信息: {response.text}")
            return False
            
    except requests.exceptions.RequestException as e:
        print(f"网络请求异常: {e}")
        return False
    except Exception as e:
        print(f"其他异常: {e}")
        return False

# 使用示例
if __name__ == "__main__":
    success = basic_tts(
        text="你好，欢迎使用Fish-Speech 1.5文本转语音服务。",
        output_file="greeting.wav",
        server_ip="你的服务器IP"
    )
    
    if success:
        print("语音生成成功！")
    else:
        print("语音生成失败，请检查网络连接和服务状态。")

3.2 带错误处理的增强版本

在实际应用中，我们需要更完善的错误处理机制：

import requests
import time
from typing import Optional

class FishSpeechClient:
    """Fish-Speech 1.5 API客户端"""
    
    def __init__(self, server_ip: str = "localhost", port: int = 8080):
        self.base_url = f"http://{server_ip}:{port}"
        self.tts_url = f"{self.base_url}/v1/tts"
        self.session = requests.Session()
        self.session.timeout = 60  # 设置超时时间为60秒
    
    def generate_speech(self, text: str, **kwargs) -> Optional[bytes]:
        """
        生成语音音频
        
        Args:
            text: 要转换的文本
            **kwargs: 其他参数，参见API文档
            
        Returns:
            bytes: 音频数据，失败返回None
        """
        # 默认参数
        params = {
            "text": text,
            "references": [],
            "reference_id": None,
            "max_new_tokens": 1024,
            "chunk_length": 200,
            "top_p": 0.7,
            "repetition_penalty": 1.2,
            "temperature": 0.7,
            "format": "wav"
        }
        
        # 更新用户自定义参数
        params.update(kwargs)
        
        try:
            # 重试机制
            for attempt in range(3):
                try:
                    response = self.session.post(
                        self.tts_url, 
                        json=params,
                        timeout=60
                    )
                    
                    if response.status_code == 200:
                        return response.content
                    elif response.status_code == 429:
                        # 速率限制，等待后重试
                        wait_time = 2 ** attempt
                        print(f"速率限制，等待 {wait_time} 秒后重试...")
                        time.sleep(wait_time)
                        continue
                    else:
                        print(f"API错误: {response.status_code} - {response.text}")
                        return None
                        
                except requests.exceptions.Timeout:
                    print(f"请求超时，第 {attempt + 1} 次重试...")
                    continue
                except requests.exceptions.ConnectionError:
                    print(f"连接错误，第 {attempt + 1} 次重试...")
                    time.sleep(2)
                    continue
            
            print("重试次数用尽，请求失败")
            return None
            
        except Exception as e:
            print(f"未知错误: {e}")
            return None
    
    def save_speech(self, text: str, output_path: str, **kwargs) -> bool:
        """
        生成语音并保存到文件
        
        Args:
            text: 要转换的文本
            output_path: 输出文件路径
            **kwargs: 其他参数
            
        Returns:
            bool: 是否成功
        """
        audio_data = self.generate_speech(text, **kwargs)
        if audio_data:
            try:
                with open(output_path, "wb") as f:
                    f.write(audio_data)
                print(f"音频已保存到: {output_path}")
                return True
            except IOError as e:
                print(f"文件保存失败: {e}")
                return False
        return False

# 使用示例
if __name__ == "__main__":
    # 创建客户端实例
    client = FishSpeechClient(server_ip="你的服务器IP")
    
    # 生成简单语音
    client.save_speech(
        text="这是一个测试语音，用于验证API调用是否正常。",
        output_path="test_output.wav"
    )
    
    # 使用自定义参数
    client.save_speech(
        text="这是一个使用自定义参数的语音生成示例。",
        output_path="custom_params.wav",
        temperature=0.6,  # 降低随机性，使输出更稳定
        top_p=0.8,        # 增加多样性
        format="mp3"       # 输出MP3格式
    )

4. 高级功能与参数调优

4.1 使用参考音频进行声音克隆

Fish-Speech 1.5 支持通过参考音频来模仿特定音色，这是其强大的功能之一：

import base64
import os

def speech_with_reference(client, text, reference_audio_path, output_path, reference_text=None):
    """
    使用参考音频生成特定音色的语音
    
    Args:
        client: FishSpeechClient实例
        text: 要生成的文本
        reference_audio_path: 参考音频文件路径
        output_path: 输出文件路径
        reference_text: 参考音频对应的文本（可选）
    """
    # 读取参考音频文件并编码为base64
    try:
        with open(reference_audio_path, "rb") as audio_file:
            audio_data = audio_file.read()
            audio_base64 = base64.b64encode(audio_data).decode('utf-8')
    except FileNotFoundError:
        print(f"参考音频文件不存在: {reference_audio_path}")
        return False
    
    # 构建参考音频信息
    reference_info = {
        "audio": audio_base64,
        "text": reference_text if reference_text else ""
    }
    
    # 生成语音
    audio_data = client.generate_speech(
        text=text,
        references=[reference_info],
        max_new_tokens=2048,  # 增加token数量以获得更好效果
        temperature=0.7,
        top_p=0.8
    )
    
    if audio_data:
        with open(output_path, "wb") as f:
            f.write(audio_data)
        print(f"克隆语音已保存到: {output_path}")
        return True
    return False

# 使用示例
if __name__ == "__main__":
    client = FishSpeechClient(server_ip="你的服务器IP")
    
    # 使用参考音频生成语音
    success = speech_with_reference(
        client=client,
        text="你好，这是我使用参考音频生成的声音。",
        reference_audio_path="reference.wav",  # 5-10秒的参考音频
        reference_text="这是参考音频的文本内容",  # 参考音频对应的文本
        output_path="cloned_voice.wav"
    )

4.2 批量文本转语音处理

对于需要处理大量文本的场景，我们可以实现批量处理功能：

import concurrent.futures
import pandas as pd
from pathlib import Path

class BatchTTSProcessor:
    """批量文本转语音处理器"""
    
    def __init__(self, server_ip="localhost", port=8080, max_workers=3):
        self.client = FishSpeechClient(server_ip, port)
        self.max_workers = max_workers  # 最大并发数，避免服务器过载
    
    def process_csv(self, csv_file, text_column="text", output_dir="output"):
        """
        处理CSV文件中的文本
        
        Args:
            csv_file: CSV文件路径
            text_column: 包含文本的列名
            output_dir: 输出目录
        """
        # 创建输出目录
        Path(output_dir).mkdir(exist_ok=True)
        
        # 读取CSV文件
        try:
            df = pd.read_csv(csv_file)
            texts = df[text_column].tolist()
        except Exception as e:
            print(f"读取CSV文件失败: {e}")
            return
        
        # 使用线程池并发处理
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = []
            for i, text in enumerate(texts):
                if pd.notna(text) and text.strip():  # 跳过空文本
                    output_path = Path(output_dir) / f"output_{i+1}.wav"
                    futures.append(
                        executor.submit(
                            self.client.save_speech,
                            text=text,
                            output_path=str(output_path)
                        )
                    )
            
            # 等待所有任务完成
            for i, future in enumerate(concurrent.futures.as_completed(futures)):
                try:
                    success = future.result()
                    if success:
                        print(f"已完成 {i+1}/{len(futures)}")
                    else:
                        print(f"第 {i+1} 个任务失败")
                except Exception as e:
                    print(f"任务执行异常: {e}")
    
    def process_text_list(self, texts, output_dir="output", filename_pattern="speech_{}.wav"):
        """
        处理文本列表
        
        Args:
            texts: 文本列表
            output_dir: 输出目录
            filename_pattern: 文件名模式
        """
        Path(output_dir).mkdir(exist_ok=True)
        
        for i, text in enumerate(texts):
            if text.strip():  # 跳过空文本
                output_path = Path(output_dir) / filename_pattern.format(i+1)
                success = self.client.save_speech(
                    text=text,
                    output_path=str(output_path)
                )
                
                if success:
                    print(f"已生成: {output_path}")
                else:
                    print(f"生成失败: 第{i+1}条文本")

# 使用示例
if __name__ == "__main__":
    processor = BatchTTSProcessor(server_ip="你的服务器IP", max_workers=2)
    
    # 处理文本列表
    texts = [
        "第一条测试文本，用于批量处理演示。",
        "这是第二条文本，展示批量生成功能。",
        "第三条文本，测试并发处理能力。"
    ]
    
    processor.process_text_list(texts, output_dir="batch_output")

5. 参数详解与效果优化

5.1 关键参数说明

Fish-Speech 1.5 API 提供了多个参数用于控制语音生成效果：

参数	类型	默认值	说明	推荐范围
temperature	float	0.7	控制生成随机性，值越低输出越稳定	0.6-0.9
top_p	float	0.7	核采样参数，控制生成多样性	0.6-0.9
repetition_penalty	float	1.2	重复惩罚，避免重复内容	1.0-1.5
max_new_tokens	int	1024	每批次最大token数量	512-2048
chunk_length	int	200	迭代提示长度，0表示关闭	0-300
format	str	"wav"	输出音频格式	wav/mp3/flac

5.2 参数优化实践

根据不同的使用场景，我们可以调整参数以获得最佳效果：

def optimize_parameters_examples(client):
    """参数优化示例"""
    
    # 场景1：需要高度稳定的语音（如新闻播报）
    client.save_speech(
        text="这里是新闻播报，需要稳定清晰的语音输出。",
        output_path="news.wav",
        temperature=0.6,      # 低随机性，更稳定
        repetition_penalty=1.3,  # 较高的重复惩罚
        top_p=0.7
    )
    
    # 场景2：需要自然对话感（如虚拟助手）
    client.save_speech(
        text="你好，我是你的智能助手，有什么可以帮你的吗？",
        output_path="assistant.wav",
        temperature=0.75,     # 中等随机性，更自然
        repetition_penalty=1.1,
        top_p=0.8
    )
    
    # 场景3：需要创意性表达（如讲故事）
    client.save_speech(
        text="在很久很久以前，有一个神奇的王国...",
        output_path="story.wav",
        temperature=0.85,     # 高随机性，更有表现力
        repetition_penalty=1.0,  # 较低的重复惩罚
        top_p=0.9,
        max_new_tokens=2048    # 更长的生成长度
    )
    
    # 场景4：长文本生成
    client.save_speech(
        text="这是一段很长的文本，需要特别注意参数设置以避免内存问题。建议分批次处理或者调整max_new_tokens参数。",
        output_path="long_text.wav",
        max_new_tokens=512,    # 降低每批次的token数量
        chunk_length=100       # 减小chunk长度
    )

# 使用示例
if __name__ == "__main__":
    client = FishSpeechClient(server_ip="你的服务器IP")
    optimize_parameters_examples(client)

6. 实战应用案例

6.1 集成到Web应用

下面展示如何将 Fish-Speech 1.5 API 集成到 Flask Web 应用中：

from flask import Flask, request, send_file, jsonify
import io
import threading

app = Flask(__name__)
client = FishSpeechClient(server_ip="你的服务器IP")

@app.route('/api/tts', methods=['POST'])
def tts_api():
    """TTS API端点"""
    try:
        data = request.get_json()
        text = data.get('text', '')
        
        if not text:
            return jsonify({'error': '缺少text参数'}), 400
        
        # 生成语音
        audio_data = client.generate_speech(
            text=text,
            format=data.get('format', 'wav'),
            temperature=float(data.get('temperature', 0.7)),
            top_p=float(data.get('top_p', 0.7))
        )
        
        if audio_data:
            return send_file(
                io.BytesIO(audio_data),
                mimetype='audio/wav',
                as_attachment=True,
                download_name='speech.wav'
            )
        else:
            return jsonify({'error': '语音生成失败'}), 500
            
    except Exception as e:
        return jsonify({'error': str(e)}), 500

@app.route('/api/batch-tts', methods=['POST'])
def batch_tts_api():
    """批量TTS API端点"""
    try:
        data = request.get_json()
        texts = data.get('texts', [])
        
        if not texts:
            return jsonify({'error': '缺少texts参数'}), 400
        
        results = []
        for i, text in enumerate(texts):
            if text.strip():
                audio_data = client.generate_speech(text=text)
                if audio_data:
                    results.append({
                        'index': i,
                        'text': text,
                        'audio': base64.b64encode(audio_data).decode('utf-8'),
                        'success': True
                    })
                else:
                    results.append({
                        'index': i,
                        'text': text,
                        'success': False
                    })
        
        return jsonify({'results': results})
        
    except Exception as e:
        return jsonify({'error': str(e)}), 500

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

6.2 实时语音生成监控

对于长时间运行的语音生成任务，我们可以添加进度监控功能：

import time
from tqdm import tqdm

def monitor_tts_performance(client, texts, output_dir):
    """
    监控TTS性能并生成报告
    
    Args:
        client: FishSpeechClient实例
        texts: 测试文本列表
        output_dir: 输出目录
    """
    results = []
    
    for i, text in enumerate(tqdm(texts, desc="生成进度")):
        start_time = time.time()
        
        success = client.save_speech(
            text=text,
            output_path=f"{output_dir}/output_{i+1}.wav"
        )
        
        end_time = time.time()
        duration = end_time - start_time
        
        results.append({
            'text_length': len(text),
            'success': success,
            'duration': duration,
            'speed': len(text) / duration if success else 0  # 字符/秒
        })
    
    # 生成性能报告
    successful_runs = [r for r in results if r['success']]
    
    if successful_runs:
        avg_speed = sum(r['speed'] for r in successful_runs) / len(successful_runs)
        avg_duration = sum(r['duration'] for r in successful_runs) / len(successful_runs)
        
        print(f"\n性能报告:")
        print(f"总任务数: {len(texts)}")
        print(f"成功数: {len(successful_runs)}")
        print(f"平均生成速度: {avg_speed:.2f} 字符/秒")
        print(f"平均耗时: {avg_duration:.2f} 秒")
        
        return results
    else:
        print("所有任务都失败了")
        return None

# 使用示例
if __name__ == "__main__":
    client = FishSpeechClient(server_ip="你的服务器IP")
    
    test_texts = [
        "短文本测试。",
        "这是一段中等长度的文本，用于测试不同长度文本的生成性能。",
        "这是一段很长的文本，旨在测试系统处理长文本的能力。长文本需要更多的计算资源和时间，但也能更好地展示系统的稳定性和性能表现。"
    ] * 3  # 重复3次以获得更稳定的统计数据
    
    monitor_tts_performance(client, test_texts, "performance_test")

7. 总结

通过本文的详细讲解和代码示例，你应该已经掌握了如何使用 Python 调用 Fish-Speech 1.5 的 API 进行文本转语音操作。关键要点包括：

1. 基础调用：使用简单的 HTTP POST 请求即可实现文本转语音
2. 错误处理：完善的异常处理机制确保程序稳定性
3. 高级功能：支持参考音频声音克隆和批量处理
4. 参数调优：根据不同场景调整参数获得最佳效果
5. 实战应用：可以轻松集成到 Web 应用或批量处理流程中

Fish-Speech 1.5 的 DualAR 架构提供了出色的语音质量和生成效率，通过 API 调用可以灵活地将其集成到各种应用中。无论是构建语音助手、内容创作工具还是无障碍应用，这个强大的 TTS 系统都能提供专业级的语音合成能力。

在实际使用中，建议根据具体需求调整参数，并注意监控服务性能以确保稳定运行。对于生产环境，还需要考虑添加身份验证、速率限制和负载均衡等机制。

---

获取更多AI镜像

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