에이전트 응답의 텍스트-음성 변환

에이전트에 TTS가 필요한 이유

OpenAI TTS 기초

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def text_to_speech(text, voice='alloy', output_path='response.mp3'):
    response = client.audio.speech.create(
        model='tts-1',          # tts-1 (fast) or tts-1-hd (higher quality)
        voice=voice,            # alloy, echo, fable, onyx, nova, shimmer
        input=text,
        response_format='mp3'   # mp3, opus, aac, flac
    )
    with open(output_path, 'wb') as f:
        f.write(response.content)
    print(f'Audio saved to: {output_path}')
    return output_path

# Available voices:
# alloy    - neutral, balanced
# echo     - warm, conversational
# fable    - expressive
# onyx     - deep, authoritative
# nova     - friendly, upbeat
# shimmer  - soft, clear

OpenAI를 사용한 스트리밍 TTS

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def stream_tts_to_file(text, output_path, voice='nova'):
    with client.audio.speech.with_streaming_response.create(
        model='tts-1',
        voice=voice,
        input=text
    ) as response:
        response.stream_to_file(output_path)
    return output_path

def stream_tts_to_bytes(text, voice='nova'):
    audio_chunks = []
    with client.audio.speech.with_streaming_response.create(
        model='tts-1',
        voice=voice,
        input=text
    ) as response:
        for chunk in response.iter_bytes(chunk_size=4096):
            audio_chunks.append(chunk)
    return b''.join(audio_chunks)

ElevenLabs SDK: 고품질 TTS

from elevenlabs import ElevenLabs, VoiceSettings
import os

client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY'))

def elevenlabs_tts(text, voice_id='pNInz6obpgDQGcFmaJgB', output_path='response.mp3'):
    # Common voice IDs:
    # Rachel:  21m00Tcm4TlvDq8ikWAM
    # Adam:    pNInz6obpgDQGcFmaJgB
    # Bella:   EXAVITQu4vr4xnSDxMaL

    audio = client.generate(
        text=text,
        voice=voice_id,
        model='eleven_multilingual_v2',  # supports 29 languages
        voice_settings=VoiceSettings(
            stability=0.5,        # 0.0-1.0: lower = more expressive
            similarity_boost=0.8, # 0.0-1.0: higher = closer to original voice
            style=0.3             # 0.0-1.0: style exaggeration
        )
    )

    with open(output_path, 'wb') as f:
        for chunk in audio:
            f.write(chunk)

    return output_path

ElevenLabs 스트리밍 TTS

from elevenlabs import ElevenLabs
import os

client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY'))

def streaming_tts(text, voice_id='pNInz6obpgDQGcFmaJgB'):
    audio_stream = client.generate(
        text=text,
        voice=voice_id,
        model='eleven_turbo_v2',  # fastest model, optimized for streaming
        stream=True
    )
    return audio_stream  # yields audio chunks as they are generated

def stream_to_file(text, output_path, voice_id='pNInz6obpgDQGcFmaJgB'):
    stream = streaming_tts(text, voice_id)
    with open(output_path, 'wb') as f:
        for chunk in stream:
            if chunk:
                f.write(chunk)
    print(f'Streaming TTS complete: {output_path}')

sounddevice를 사용한 오디오 재생

import sounddevice as sd
import soundfile as sf
import numpy as np
import tempfile
import os

def play_audio_file(audio_path):
    data, sample_rate = sf.read(audio_path)
    sd.play(data, sample_rate)
    sd.wait()  # block until playback finishes

def play_mp3_bytes(audio_bytes):
    # Write to temp file then play (soundfile needs a file)
    with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f:
        f.write(audio_bytes)
        tmp_path = f.name
    try:
        play_audio_file(tmp_path)
    finally:
        os.unlink(tmp_path)

def say(text, voice='nova'):
    audio_bytes = stream_tts_to_bytes(text, voice=voice)
    play_mp3_bytes(audio_bytes)

# Usage
say('Hello! How can I help you today?')

pygame을 사용한 오디오 재생

import pygame
import io
import tempfile
import os

pygame.mixer.init(frequency=44100, size=-16, channels=1, buffer=512)

def play_audio_pygame(audio_bytes):
    with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f:
        f.write(audio_bytes)
        tmp_path = f.name

    try:
        pygame.mixer.music.load(tmp_path)
        pygame.mixer.music.play()
        while pygame.mixer.music.get_busy():
            pygame.time.wait(50)  # check every 50ms
    finally:
        pygame.mixer.music.stop()
        os.unlink(tmp_path)

def stop_audio():
    if pygame.mixer.music.get_busy():
        pygame.mixer.music.stop()
        print('Audio stopped (interrupt)')

TTS 응답 캐싱

import hashlib
import os

TTS_CACHE_DIR = '/tmp/tts_cache'
os.makedirs(TTS_CACHE_DIR, exist_ok=True)

def cached_tts(text, voice='nova'):
    cache_key = hashlib.md5(f'{voice}:{text}'.encode()).hexdigest()
    cache_path = os.path.join(TTS_CACHE_DIR, f'{cache_key}.mp3')

    if os.path.exists(cache_path):
        print('TTS cache hit')
        return cache_path

    # Generate and save
    audio_bytes = stream_tts_to_bytes(text, voice=voice)
    with open(cache_path, 'wb') as f:
        f.write(audio_bytes)

    print(f'TTS cached: {cache_path}')
    return cache_path

# Pre-warm cache for common phrases on startup
COMMON_PHRASES = [
    'Hello! How can I help you today?',
    'I did not catch that. Could you repeat?',
    'Let me look that up for you.',
    'Thank you, goodbye!'
]

def prewarm_tts_cache():
    for phrase in COMMON_PHRASES:
        cached_tts(phrase)
    print(f'Pre-warmed {len(COMMON_PHRASES)} phrases')

스트리밍을 위한 문장별 TTS

import re

def split_into_sentences(text):
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    return [s.strip() for s in sentences if s.strip()]

def stream_response_as_voice(llm_response_text, voice='nova'):
    sentences = split_into_sentences(llm_response_text)
    print(f'Speaking {len(sentences)} sentences')

    for i, sentence in enumerate(sentences):
        if not sentence:
            continue
        print(f'Speaking ({i+1}/{len(sentences)}): {sentence[:60]}')
        cache_path = cached_tts(sentence, voice=voice)
        play_audio_file(cache_path)

# Example
response = 'The weather today is sunny and 22 degrees. Perfect for outdoor activities. Bring sunscreen if you plan to be outside for more than an hour.'
stream_response_as_voice(response)

음성과 모델 선택

VOICE_PROFILES = {
    'assistant':    {'service': 'openai',     'voice': 'nova',   'model': 'tts-1'},
    'professional': {'service': 'openai',     'voice': 'onyx',   'model': 'tts-1-hd'},
    'narrator':     {'service': 'openai',     'voice': 'fable',  'model': 'tts-1-hd'},
    'premium':      {'service': 'elevenlabs', 'voice_id': '21m00Tcm4TlvDq8ikWAM', 'model': 'eleven_multilingual_v2'}
}

def speak(text, persona='assistant'):
    profile = VOICE_PROFILES.get(persona, VOICE_PROFILES['assistant'])

    if profile['service'] == 'openai':
        audio_bytes = stream_tts_to_bytes(text, voice=profile['voice'])
        play_mp3_bytes(audio_bytes)
    elif profile['service'] == 'elevenlabs':
        output_path = elevenlabs_tts(text, voice_id=profile['voice_id'])
        play_audio_file(output_path)

speak('Your meeting has been rescheduled.', persona='professional')

말하기 속도와 음높이 조정

from pydub import AudioSegment
from pydub.effects import speedup
import os

def adjust_speech_speed(audio_path, speed_factor=1.0, output_path=None):
    if speed_factor == 1.0:
        return audio_path

    audio = AudioSegment.from_file(audio_path)

    if speed_factor > 1.0:
        # Speed up without changing pitch
        audio = speedup(audio, playback_speed=speed_factor)
    else:
        # Slow down: overlay with silence (simple method)
        # For production: use ffmpeg with atempo filter
        slow_factor = 1.0 / speed_factor
        audio = audio._spawn(
            audio.raw_data,
            overrides={'frame_rate': int(audio.frame_rate / slow_factor)}
        ).set_frame_rate(audio.frame_rate)

    if output_path is None:
        base = os.path.splitext(audio_path)[0]
        output_path = f'{base}_speed{speed_factor}.mp3'

    audio.export(output_path, format='mp3')
    return output_path

# Usage
path = text_to_speech('Your meeting starts in 5 minutes.')
fast_path = adjust_speech_speed(path, speed_factor=1.2)  # 20% faster for alerts

지식 확인

복습: 에이전트 응답의 텍스트 음성 변환