Building an AI Podcast Pipeline with Gemini 3.1 Flash TTS Emotional Tags and Multi-Speaker — A Complete Commercial Content System

pip install google-genai pydub

import os
import base64
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GOOGLE_AI_API_KEY"])
 
def generate_speech(
    text: str,
    voice_name: str = "Kore",
    output_path: str = "output.wav"
) -> None:
    """
    Generate speech from text with emotional tag support.
    
    Emotional tags in the text are interpreted by the model and affect
    the delivery style of the speech that follows the tag.
    
    Args:
        text: Text to synthesize (may include [emotion_tag] markers)
        voice_name: Voice ID (Kore, Puck, Zephyr, Charon, Aoede, etc.)
        output_path: Output WAV file path
    """
    try:
        response = client.models.generate_content(
            model="gemini-3.1-flash-tts-preview",
            contents=text,
            config=types.GenerateContentConfig(
                response_modalities=["AUDIO"],
                speech_config=types.SpeechConfig(
                    voice_config=types.VoiceConfig(
                        prebuilt_voice_config=types.PrebuiltVoiceConfig(
                            voice_name=voice_name
                        )
                    )
                )
            )
        )
 
        audio_data = response.candidates[0].content.parts[0].inline_data.data
        audio_bytes = base64.b64decode(audio_data)
 
        with open(output_path, "wb") as f:
            f.write(audio_bytes)
 
        print(f"✅ Generated: {output_path} ({len(audio_bytes):,} bytes)")
 
    except Exception as e:
        print(f"❌ Error: {e}")
        raise
 
# Six emotion patterns — run these and listen to the difference
emotion_examples = [
    ("[excitement] Breaking news — the new model just dropped and it's genuinely impressive!", "excitement.wav"),
    ("[determination] We've hit every obstacle you can imagine. And we're still moving forward.", "determination.wav"),
    ("[whispers] I probably shouldn't be telling you this, but the launch is next Tuesday.", "whispers.wav"),
    ("[curiosity] So what actually happens when you combine two emotion tags in one sentence?", "curiosity.wav"),
    ("[amusement] [laughs] I cannot believe that actually worked. Who tests these things?", "amusement.wav"),
    ("[nervousness] I've never done this live before, so bear with me while I try to remember the steps.", "nervousness.wav"),
]
 
for text, filename in emotion_examples:
    generate_speech(text, output_path=filename)

# Realistic podcast intro with emotion flow
podcast_intro = """
[enthusiasm] Welcome back to the Gemini Lab podcast. I'm your host, and today we have something special.
 
[curiosity] Have you ever wondered what it would take to launch a content service that runs almost entirely on AI? Not just the writing — the audio too?
 
[excitement] Today's guest has done exactly that. They're generating over 50 podcast episodes a month with a Python script and a Gemini API key. [amusement] And yes, the quality is actually good.
 
[neutral] We'll get into the technical details, the business model, and the honest challenges. Let's jump in.
"""
 
generate_speech(podcast_intro, voice_name="Kore", output_path="podcast_intro.wav")

def generate_multi_speaker_podcast(
    script: str,
    host_voice: str = "Kore",
    guest_voice: str = "Puck",
    output_path: str = "podcast.wav"
) -> None:
    """
    Generate a two-speaker podcast conversation from a formatted script.
    
    Script format:
        Speaker1: [emotion_tag] Dialogue line
        Speaker2: [emotion_tag] Response
    
    The speaker labels in the script MUST match the speaker names
    in the voice config exactly (case-sensitive).
    
    Args:
        script: Formatted two-speaker dialogue
        host_voice: Voice ID for Speaker1
        guest_voice: Voice ID for Speaker2
        output_path: Output WAV file path
    """
    try:
        response = client.models.generate_content(
            model="gemini-3.1-flash-tts-preview",
            contents=script,
            config=types.GenerateContentConfig(
                response_modalities=["AUDIO"],
                speech_config=types.SpeechConfig(
                    multi_speaker_voice_config=types.MultiSpeakerVoiceConfig(
                        speaker_voice_configs=[
                            types.SpeakerVoiceConfig(
                                speaker="Speaker1",
                                voice_config=types.VoiceConfig(
                                    prebuilt_voice_config=types.PrebuiltVoiceConfig(
                                        voice_name=host_voice
                                    )
                                )
                            ),
                            types.SpeakerVoiceConfig(
                                speaker="Speaker2",
                                voice_config=types.VoiceConfig(
                                    prebuilt_voice_config=types.PrebuiltVoiceConfig(
                                        voice_name=guest_voice
                                    )
                                )
                            )
                        ]
                    )
                )
            )
        )
 
        audio_data = response.candidates[0].content.parts[0].inline_data.data
        audio_bytes = base64.b64decode(audio_data)
 
        with open(output_path, "wb") as f:
            f.write(audio_bytes)
 
        print(f"✅ Multi-speaker podcast generated: {output_path}")
 
    except Exception as e:
        print(f"❌ Error: {e}")
        raise
 
# Example script
sample_script = """
Speaker1: [enthusiasm] Welcome to the show. Today we're talking AI TTS, which I know sounds dry — but I promise it gets interesting.
 
Speaker2: [amusement] [laughs] High bar. Let's see if you can deliver.
 
Speaker1: [determination] Challenge accepted. So you've been running a podcast automation service for about six months now. What was the turning point?
 
Speaker2: [curiosity] Honestly? When I first heard output from Gemini 3.1 Flash TTS. [excitement] I thought I had accidentally played a human recording. The emotional range just wasn't what I expected from a machine.
 
Speaker1: [interest] What specifically stood out?
 
Speaker2: [whispers] The whisper tag. [normal voice] There's something about getting a model to actually whisper that feels like a boundary got crossed. Previous models would just lower the volume. This one actually changed the voice quality.
 
Speaker1: [curiosity] How much of your workflow is fully automated now?
 
Speaker2: [determination] Script generation, audio synthesis, chapter markers, and distribution to three platforms. [amusement] I touch it maybe an hour a week. The rest is the pipeline.
"""
 
generate_multi_speaker_podcast(
    script=sample_script,
    host_voice="Kore",
    guest_voice="Puck",
    output_path="interview.wav"
)

import re
from pathlib import Path
from pydub import AudioSegment
 
def split_script_by_exchanges(script: str, exchanges_per_chunk: int = 8) -> list[str]:
    """
    Split a script into chunks of N speaker exchanges each.
    Preserves the Speaker1:/Speaker2: prefix on each line.
    """
    lines = [l.strip() for l in script.strip().split("\n") if l.strip()]
    speaker_lines = [l for l in lines if re.match(r'^Speaker\d+:', l)]
 
    chunks = []
    for i in range(0, len(speaker_lines), exchanges_per_chunk):
        chunk = "\n\n".join(speaker_lines[i : i + exchanges_per_chunk])
        chunks.append(chunk)
 
    return chunks
 
def generate_long_podcast(
    full_script: str,
    output_path: str = "full_podcast.mp3",
    host_voice: str = "Kore",
    guest_voice: str = "Puck",
) -> None:
    """
    Process a full-length script by splitting into chunks,
    generating audio for each, then stitching into a single MP3.
    
    A 500ms silence is inserted between chunks to prevent
    abrupt transitions between API calls.
    """
    chunks = split_script_by_exchanges(full_script, exchanges_per_chunk=8)
    temp_files = []
 
    print(f"📦 Processing {len(chunks)} chunks...")
 
    for i, chunk in enumerate(chunks):
        temp_path = f"/tmp/chunk_{i:03d}.wav"
 
        try:
            generate_multi_speaker_podcast(
                script=chunk,
                host_voice=host_voice,
                guest_voice=guest_voice,
                output_path=temp_path,
            )
            temp_files.append(temp_path)
            print(f"  ✅ Chunk {i+1}/{len(chunks)} done")
 
        except Exception as e:
            print(f"  ⚠️  Chunk {i+1} failed (skipping): {e}")
            continue
 
    if not temp_files:
        raise RuntimeError("All chunks failed — nothing to stitch")
 
    combined = AudioSegment.empty()
    silence_gap = AudioSegment.silent(duration=500)
 
    for path in temp_files:
        segment = AudioSegment.from_wav(path)
        combined += segment + silence_gap
 
    combined.export(output_path, format="mp3", bitrate="192k")
    print(f"🎙️  Final output: {output_path} ({combined.duration_seconds:.1f}s)")
 
    for path in temp_files:
        Path(path).unlink(missing_ok=True)

def generate_podcast_script(
    topic: str,
    duration_minutes: int = 10,
    model: str = "gemini-2.5-pro-preview-05-06",
) -> str:
    """
    Generate a two-speaker podcast script on a given topic using Gemini.
    
    The generated script includes emotional tags throughout and follows
    a natural interview format with the host asking questions and the
    guest responding as a domain expert.
    
    Args:
        topic: The podcast episode topic
        duration_minutes: Target episode length in minutes
        model: Gemini model to use for script generation
    Returns:
        A formatted script string ready for TTS processing
    """
    word_count = duration_minutes * 130  # ~130 words/minute for podcast pacing
 
    prompt = f"""Write a {duration_minutes}-minute podcast script (~{word_count} words) on this topic: {topic}
 
Format requirements:
- Two speakers: Speaker1 (host) and Speaker2 (guest/expert)
- Include 1-2 emotional tags at the start of each line
- Use natural conversation — interruptions, follow-up questions, moments of humor
- Speaker1 asks questions, Speaker2 provides depth and insight
- Avoid jargon dumps; explain technical concepts through conversation
 
Available tags: [excitement], [enthusiasm], [curiosity], [determination], [amusement], [whispers], [nervousness], [interest], [awe], [hope], [laughs], [neutral], [positive]
 
Output the script only. No meta-commentary."""
 
    response = client.models.generate_content(
        model=model,
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.85,
            max_output_tokens=4096,
        ),
    )
 
    return response.text
 
def run_pipeline(topic: str, output_dir: str = "./output") -> dict:
    """
    Full topic → audio pipeline.
    
    Generates a script with Gemini 2.5 Pro, then synthesizes it with
    Gemini 3.1 Flash TTS, and saves both to disk.
    """
    import os
    from datetime import datetime
 
    os.makedirs(output_dir, exist_ok=True)
    ts = datetime.now().strftime("%Y%m%d_%H%M")
    label = topic[:30].replace(" ", "_")
 
    print(f"\n🎙️  Starting pipeline: {topic}")
 
    # Step 1: Script generation
    print("📝 Generating script...")
    script = generate_podcast_script(topic, duration_minutes=10)
 
    script_path = f"{output_dir}/{ts}_{label}_script.txt"
    with open(script_path, "w", encoding="utf-8") as f:
        f.write(script)
 
    # Step 2: Audio synthesis
    audio_path = f"{output_dir}/{ts}_{label}.mp3"
    print("🔊 Synthesizing audio...")
    generate_long_podcast(script, output_path=audio_path)
 
    return {"topic": topic, "script": script_path, "audio": audio_path}
 
# Batch example
topics = [
    "How indie developers are using Gemini to replace expensive contractors",
    "The real cost breakdown of running an AI-powered SaaS in 2026",
    "What Google I/O 2026 announcements mean for solo builders",
]
 
for topic in topics:
    try:
        result = run_pipeline(topic)
        print(f"  → {result['audio']}\n")
    except Exception as e:
        print(f"  ❌ Failed: {e}\n")

# ❌ Tag fights the formal register
bad = "[excitement] 本日のアップデートについてご説明申し上げます。"
 
# ✅ Tag and register work together
good = "[excitement] 今日のアップデート、本当にすごいんです！見てください！"

def generate_chunk_with_context(
    chunk: str,
    context_lines: str | None = None,
) -> str:
    """Returns TTS-ready text with optional context prefix stripped after generation."""
    if context_lines:
        prefixed = f"[Context - do not vocalize]\n{context_lines}\n\n[Script]\n{chunk}"
    else:
        prefixed = chunk
    # Pass prefixed to TTS API but note: the model will likely still read
    # the context. Better approach: use it in script generation, not TTS.
    return chunk  # Keep TTS call clean; use context only in script gen

import time
import random
from typing import Callable, TypeVar
 
T = TypeVar("T")
 
def with_backoff(
    fn: Callable[[], T],
    max_retries: int = 5,
    base_delay: float = 2.0,
) -> T:
    """
    Call fn() with exponential backoff on 429/503 errors.
    Raises on final failure or non-retryable errors.
    """
    for attempt in range(max_retries):
        try:
            return fn()
        except Exception as e:
            msg = str(e).lower()
            retryable = any(code in msg for code in ["429", "503", "rate limit", "overloaded"])
 
            if not retryable or attempt == max_retries - 1:
                raise
 
            delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), 60.0)
            print(f"  ⏳ Rate limit hit — waiting {delay:.1f}s (attempt {attempt+1}/{max_retries})")
            time.sleep(delay)
 
    raise RuntimeError("Exceeded max retries")
 
# Usage
def safe_generate(text: str, **kwargs) -> None:
    with_backoff(lambda: generate_speech(text, **kwargs))

import json
from google import genai
from google.genai import types
 
client = genai.Client(api_key=os.environ["GOOGLE_AI_API_KEY"])
 
def generate_show_notes(script: str, topic: str) -> dict:
    # Truncate for efficiency
    excerpt = script[:2500]
 
    prompt = (
        f'Given this podcast script on "{topic}", generate show notes.\n\n'
        f'Script excerpt:\n{excerpt}\n\n'
        'Return JSON with these keys:\n'
        '- title: engaging episode title under 70 characters\n'
        '- summary: 2-sentence RSS description\n'
        '- key_points: list of 4-6 main takeaways\n'
        '- estimated_duration_min: integer minutes\n'
        '- cta: one clear listener call-to-action'
    )
 
    response = client.models.generate_content(
        model="gemini-2.5-pro-preview-05-06",
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.4,
            response_mime_type="application/json",
        ),
    )
 
    return json.loads(response.text)
 
def run_full_pipeline(topic: str, output_dir: str = "./output") -> dict:
    import os
    from datetime import datetime
 
    os.makedirs(output_dir, exist_ok=True)
    ts = datetime.now().strftime("%Y%m%d_%H%M")
    label = topic[:30].replace(" ", "_")
 
    print(f"  1/3 Generating script...")
    script = generate_podcast_script(topic, duration_minutes=10)
 
    audio_path = f"{output_dir}/{ts}_{label}.mp3"
    print(f"  2/3 Synthesizing audio...")
    generate_long_podcast(script, output_path=audio_path)
 
    print(f"  3/3 Generating show notes...")
    notes = generate_show_notes(script, topic)
 
    metadata_path = f"{output_dir}/{ts}_{label}_metadata.json"
    with open(metadata_path, "w", encoding="utf-8") as f:
        json.dump({**notes, "audio_path": audio_path}, f, indent=2, ensure_ascii=False)
 
    return {"audio": audio_path, "metadata": metadata_path, "notes": notes}

def score_script_quality(script: str, topic: str) -> dict:
    excerpt = script[:2000]
 
    prompt = (
        f'Evaluate this podcast script on topic "{topic}".\n\n'
        f'Script excerpt:\n{excerpt}\n\n'
        'Score each dimension 1-10:\n'
        '1. emotional_variety: Are at least 4 different emotion tags used? Does tone shift naturally?\n'
        '2. information_density: Does each exchange introduce new information? (Penalize filler and repetition)\n'
        '3. conversation_naturalness: Does it sound like a real dialogue? (Penalize monologue blocks over 4 sentences)\n'
        '4. listener_value: Would a listener finish this feeling they learned something specific?\n\n'
        'Return JSON with keys: emotional_variety, information_density, conversation_naturalness, '
        'listener_value, overall (int), pass (bool, threshold=7), issues (list of strings)'
    )
 
    response = client.models.generate_content(
        model="gemini-2.5-pro-preview-05-06",
        contents=prompt,
        config=types.GenerateContentConfig(
            temperature=0.2,
            response_mime_type="application/json",
        ),
    )
 
    return json.loads(response.text)
 
def generate_script_with_quality_gate(
    topic: str,
    max_attempts: int = 3,
    min_score: int = 7,
) -> str:
    current_topic = topic
 
    for attempt in range(max_attempts):
        script = generate_podcast_script(current_topic)
        quality = score_script_quality(script, topic)
 
        print(f"  Quality attempt {attempt + 1}: {quality.get('overall', 0)}/10")
 
        if quality.get("pass") and quality.get("overall", 0) >= min_score:
            return script
 
        issues = quality.get("issues", ["low overall quality"])
        print(f"  Issues: {', '.join(issues)}")
 
        if attempt < max_attempts - 1:
            # Feed issues back into the next generation attempt
            issue_str = "; ".join(issues[:3])
            current_topic = f"{topic} (improve: {issue_str})"
 
    print(f"  Warning: returning best available script after {max_attempts} attempts")
    return script