Fish Audio API — Complete Tutorial

# Install dependencies (run once)
# !pip install fish-audio-sdk python-dotenv requests ormsgpack

import os
from dotenv import load_dotenv

# Load API keys from .env file
# Create a .env file with: FISH_AUDIO_API_KEY=your_key_here
load_dotenv()

FISH_API_KEY = os.getenv("FISH_AUDIO_API_KEY")

if not FISH_API_KEY:
    raise ValueError("FISH_AUDIO_API_KEY not found in .env file. Please add it.")

print(f"✓ API key loaded: {FISH_API_KEY[:8]}...")

from fishaudio import FishAudio
from fishaudio.utils import save
import os

# Initialize the Fish Audio client
client = FishAudio(api_key=FISH_API_KEY)

# Create output directory
os.makedirs("audio_output", exist_ok=True)

print("✓ Fish Audio client initialized")

# A popular public voice model from Fish Audio Discovery
# Replace with any model ID from https://fish.audio/discovery
VOICE_MODEL_ID = "7f92f8afb8ec43bf81429cc1c9199cb1"

# Basic TTS — simplest usage
audio = client.tts.convert(
    text="Hello! Welcome to the Fish Audio tutorial. This is your first AI-generated voice.",
    reference_id=VOICE_MODEL_ID
)

save(audio, "audio_output/01_basic_tts.mp3")
print("✓ Saved: audio_output/01_basic_tts.mp3")

# TTS with advanced options
audio = client.tts.convert(
    text="This audio uses advanced options: MP3 format at 192 kbps, balanced latency mode.",
    reference_id=VOICE_MODEL_ID,
    format="mp3",        # Options: "mp3", "wav", "pcm", "opus"
    mp3_bitrate=192,     # 64, 128, or 192 kbps
    chunk_length=200,    # 100–300 characters per processing chunk
    latency="balanced"   # "normal" (quality) or "balanced" (~300ms latency)
)

save(audio, "audio_output/02_advanced_tts.mp3")
print("✓ Saved: audio_output/02_advanced_tts.mp3")

# Emotion control — tags at START of sentences
emotional_text = """(excited) I just got accepted into my dream school!
(happy) This is the best day of my life.
(sad) But I will miss all my friends here.
(calm) Everything will work out in the end."""

audio = client.tts.convert(
    text=emotional_text,
    reference_id=VOICE_MODEL_ID
)

save(audio, "audio_output/03_emotions.mp3")
print("✓ Saved: audio_output/03_emotions.mp3")

# Storytelling with multiple emotion transitions
story_text = """(narrator) Chapter One: The Discovery.
(mysterious)(whispering) Something was hidden beneath the old floorboards.
(excited) She pulled up the wooden plank and found a glowing box!
(scared)(shouting) Suddenly, a loud bang echoed through the house.
(relieved)(sighing) It was just the cat knocking over a lamp.
(laughing) Ha ha ha, what a scare that was!"""

audio = client.tts.convert(
    text=story_text,
    reference_id=VOICE_MODEL_ID
)

save(audio, "audio_output/04_storytelling.mp3")
print("✓ Saved: audio_output/04_storytelling.mp3")

from fishaudio.types import ReferenceAudio

# Use reference audio on-the-fly
# Replace "voice_sample.wav" with your own audio file path
REFERENCE_AUDIO_PATH = "voice_sample.wav"  # your 10+ second audio sample
REFERENCE_AUDIO_TEXT = "Hello, this is a sample of my voice speaking naturally."

if os.path.exists(REFERENCE_AUDIO_PATH):
    with open(REFERENCE_AUDIO_PATH, "rb") as f:
        audio = client.tts.convert(
            text="This voice was cloned from a reference audio file provided on the fly.",
            references=[
                ReferenceAudio(
                    audio=f.read(),
                    text=REFERENCE_AUDIO_TEXT  # transcript of the reference audio
                )
            ]
        )
    save(audio, "audio_output/05_reference_audio.mp3")
    print("✓ Saved: audio_output/05_reference_audio.mp3")
else:
    print(f"⚠ Reference file not found: {REFERENCE_AUDIO_PATH}")
    print("  Provide a .wav/.mp3 file to use this feature.")

import requests

def create_voice_model(
    api_key: str,
    title: str,
    audio_files: list,
    transcripts: list = None,
    description: str = "",
    visibility: str = "private",
    enhance_audio: bool = True
) -> dict:
    """
    Create a voice model from audio samples.
    
    Args:
        api_key: Fish Audio API key
        title: Name for your voice model
        audio_files: List of file paths (.mp3, .wav, .m4a, .opus)
        transcripts: Optional list of transcripts (must match number of audio files)
        description: Optional description
        visibility: 'private', 'public', or 'unlist'
        enhance_audio: Remove background noise
    
    Returns:
        dict with model info including 'id'
    """
    files = [("voices", open(f, "rb")) for f in audio_files]
    
    data = [
        ("title", title),
        ("description", description),
        ("visibility", visibility),
        ("type", "tts"),
        ("train_mode", "fast"),
        ("enhance_audio_quality", str(enhance_audio).lower())
    ]
    
    # Transcripts must match count of audio files exactly
    if transcripts:
        assert len(transcripts) == len(audio_files), \
            f"Transcripts count ({len(transcripts)}) must match audio files ({len(audio_files)})"
        for t in transcripts:
            data.append(("texts", t))
    
    response = requests.post(
        "https://api.fish.audio/model",
        files=files,
        data=data,
        headers={"Authorization": f"Bearer {api_key}"}
    )
    response.raise_for_status()
    
    # Close file handles
    for _, f in files:
        f.close()
    
    return response.json()

print("✓ create_voice_model() function defined")

# Example: Create a voice model (requires actual audio files)
SAMPLE_FILES = ["voice_sample.wav"]  # Replace with your audio file paths
SAMPLE_TRANSCRIPTS = [
    "Hello, my name is Alex, and I enjoy reading books about technology and science."
]

if all(os.path.exists(f) for f in SAMPLE_FILES):
    result = create_voice_model(
        api_key=FISH_API_KEY,
        title="My Tutorial Voice",
        audio_files=SAMPLE_FILES,
        transcripts=SAMPLE_TRANSCRIPTS,
        description="Voice model created in Fish Audio tutorial",
        visibility="private",
        enhance_audio=True
    )
    
    MY_MODEL_ID = result.get("_id") or result.get("id")
    print(f"✓ Voice model created!")
    print(f"  Model ID: {MY_MODEL_ID}")
    print(f"  State: {result.get('state')}")
else:
    print("⚠ Audio sample files not found.")
    print("  Provide .wav/.mp3 files to create a voice model.")
    MY_MODEL_ID = VOICE_MODEL_ID  # Fall back to public model

import time

def stream_tts(client, text: str, model_id: str, output_path: str) -> dict:
    """
    Stream TTS audio and save to file, reporting timing stats.
    
    Returns:
        dict with 'chunks', 'total_bytes', 'duration_sec'
    """
    start = time.time()
    first_chunk_time = None
    chunk_count = 0
    total_bytes = 0
    
    audio_stream = client.tts.stream(
        text=text,
        reference_id=model_id,
        latency="balanced"  # lower latency for streaming
    )
    
    with open(output_path, "wb") as f:
        for chunk in audio_stream:
            if first_chunk_time is None:
                first_chunk_time = time.time() - start
            f.write(chunk)
            chunk_count += 1
            total_bytes += len(chunk)
    
    duration = time.time() - start
    
    stats = {
        "chunks": chunk_count,
        "total_bytes": total_bytes,
        "first_chunk_sec": round(first_chunk_time, 3),
        "total_sec": round(duration, 3)
    }
    return stats


stats = stream_tts(
    client=client,
    text="Streaming audio is perfect for real-time applications like chatbots and voice assistants.",
    model_id=VOICE_MODEL_ID,
    output_path="audio_output/06_streaming.mp3"
)

print("✓ Streaming complete!")
print(f"  First chunk: {stats['first_chunk_sec']}s")
print(f"  Total time:  {stats['total_sec']}s")
print(f"  Chunks:      {stats['chunks']}")
print(f"  Total size:  {stats['total_bytes']:,} bytes")

import httpx
import ormsgpack

def tts_direct_api(
    api_key: str,
    text: str,
    reference_id: str,
    output_path: str,
    format: str = "mp3",
    model: str = "s1"
) -> int:
    """
    Call Fish Audio TTS API directly with MessagePack.
    Returns size of saved audio in bytes.
    """
    request_data = {
        "text": text,
        "reference_id": reference_id,
        "format": format,
        "chunk_length": 200,
        "latency": "normal"
    }
    
    with httpx.Client(timeout=30) as http_client:
        response = http_client.post(
            "https://api.fish.audio/v1/tts",
            content=ormsgpack.packb(request_data),
            headers={
                "authorization": f"Bearer {api_key}",
                "content-type": "application/msgpack",
                "model": model
            }
        )
        response.raise_for_status()
        
        with open(output_path, "wb") as f:
            f.write(response.content)
        
        return len(response.content)


size = tts_direct_api(
    api_key=FISH_API_KEY,
    text="This was generated using the direct REST API with MessagePack serialization.",
    reference_id=VOICE_MODEL_ID,
    output_path="audio_output/07_direct_api.mp3"
)

print(f"✓ Direct API call successful: {size:,} bytes saved")

import time
from fishaudio.exceptions import FishAudioError

def generate_with_retry(client, text: str, model_id: str, max_retries: int = 3):
    """Generate TTS with exponential backoff retry."""
    for attempt in range(max_retries):
        try:
            return client.tts.convert(text=text, reference_id=model_id)
        except FishAudioError as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                print(f"  Retry {attempt + 1}/{max_retries - 1} in {wait}s: {e}")
                time.sleep(wait)
            else:
                raise


def batch_tts(client, texts: list, model_id: str, output_dir: str, prefix: str = "batch") -> list:
    """
    Generate TTS for a list of texts and save to individual files.
    
    Returns:
        List of output file paths
    """
    os.makedirs(output_dir, exist_ok=True)
    output_files = []
    
    for i, text in enumerate(texts):
        output_path = os.path.join(output_dir, f"{prefix}_{i+1:02d}.mp3")
        print(f"  [{i+1}/{len(texts)}] Generating: {text[:50]}...")
        
        audio = generate_with_retry(client, text, model_id)
        save(audio, output_path)
        output_files.append(output_path)
    
    return output_files


# Example: Generate announcements
announcements = [
    "Welcome to the annual technology conference. Please find your seats.",
    "The keynote presentation will begin in five minutes.",
    "Refreshments are available in the lobby throughout the day.",
]

print("Generating batch audio...")
files = batch_tts(
    client=client,
    texts=announcements,
    model_id=VOICE_MODEL_ID,
    output_dir="audio_output/batch",
    prefix="announcement"
)

print(f"\n✓ Batch complete: {len(files)} files generated")
for f in files:
    print(f"  {f}")

# Multi-character dialogue script
# In a real project, each character would use a different reference_id (voice model)
dialogue = [
    {"character": "narrator",  "text": "(calm) It was a dark and stormy night at the old library."},
    {"character": "alice",     "text": "(excited) I found it! The ancient manuscript is right here!"},
    {"character": "librarian", "text": "(angry)(shouting) Quiet! This is a library!"},
    {"character": "alice",     "text": "(scared)(whispering) Sorry... but look at what it says."},
    {"character": "librarian", "text": "(curious) Let me see. (sighing) Oh my. This changes everything."},
    {"character": "narrator",  "text": "(mysterious) And so began the greatest adventure of their lives."},
]

print("Generating multi-character dialogue...")
os.makedirs("audio_output/dialogue", exist_ok=True)

for i, line in enumerate(dialogue):
    output_path = f"audio_output/dialogue/line_{i+1:02d}_{line['character']}.mp3"
    
    # In production: map character -> unique voice model ID
    # Here we use the same model for demo purposes
    audio = client.tts.convert(
        text=line["text"],
        reference_id=VOICE_MODEL_ID
    )
    save(audio, output_path)
    print(f"  [{i+1}] {line['character']}: {line['text'][:50]}")

print("\n✓ All dialogue lines generated!")

import requests

def list_my_models(api_key: str, page_size: int = 10) -> list:
    """Retrieve your voice models from Fish Audio."""
    response = requests.get(
        "https://api.fish.audio/model",
        params={"page_size": page_size, "self": "true"},
        headers={"Authorization": f"Bearer {api_key}"}
    )
    response.raise_for_status()
    data = response.json()
    return data.get("items", [])


models = list_my_models(FISH_API_KEY)

if models:
    print(f"Your voice models ({len(models)} found):")
    for m in models:
        model_id = m.get('_id') or m.get('id')
        print(f"  • {m.get('title', 'Untitled')} — ID: {model_id} — Visibility: {m.get('visibility')}")
else:
    print("No voice models found. Create one in Section 5!")

# Final summary of all generated files
import glob

all_files = sorted(glob.glob("audio_output/**/*.mp3", recursive=True))
total_size = sum(os.path.getsize(f) for f in all_files)

print(f"Tutorial complete! Generated {len(all_files)} audio files ({total_size/1024:.1f} KB total)\n")
for f in all_files:
    size = os.path.getsize(f)
    print(f"  {f:55s} {size/1024:6.1f} KB")