Voice AI: STT, TTS, and Real-Time Pipelines

# implement transcribe_audio, transcribe_with_deepgram
from openai import OpenAI
from pathlib import Path
client = OpenAI()
def transcribe_audio(audio_path: str, language: str = None) -> dict:
    """Transcribe audio using OpenAI's Whisper API."""
    with open(audio_path, "rb") as audio_file:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            language=language, # ISO 639-1 code, e.g., "en"
            response_format="verbose_json",
            timestamp_granularities=["word", "segment"]
            )
        return {
            "text": transcript.text,
            "language": transcript.language,
            "duration": transcript.duration,
            "segments": [
            {
            "text": seg.text,
            "start": seg.start,
            "end": seg.end
            }
            for seg in (transcript.segments or [])
            ]
            }
        def transcribe_with_deepgram(audio_path: str) -> dict:
            """Transcribe audio using Deepgram's Nova-2 model."""
            from deepgram import DeepgramClient, PrerecordedOptions
            deepgram = DeepgramClient() # Uses DEEPGRAM_API_KEY env var
            with open(audio_path, "rb") as audio_file:
                payload = {"buffer": audio_file.read()}
                options = PrerecordedOptions(
                    model="nova-2",
                    smart_format=True, # Adds punctuation and formatting
                    utterances=True, # Detects speaker turns
                    diarize=True, # Speaker identification
                    language="en"
                    )
                response = deepgram.listen.rest.v("1").transcribe_file(
                    payload, options
                    )
                result = response.results
                return {
                    "transcript": result.channels[0].alternatives[0].transcript,
                    "confidence": result.channels[0].alternatives[0].confidence,
                    "words": [
                    {
                    "word": w.word,
                    "start": w.start,
                    "end": w.end,
                    "confidence": w.confidence,
                    "speaker": w.speaker
                    }
                    for w in result.channels[0].alternatives[0].words
                    ]
                    }
                # Example usage
                result = transcribe_audio("user_query.wav")
                print(f"Transcription: {result['text']}")
                print(f"Language: {result['language']}")
                print(f"Duration: {result['duration']:.1f}s")

# implement text_to_speech_openai
from openai import OpenAI
from pathlib import Path
client = OpenAI()
def text_to_speech_openai(text: str, voice: str = "nova",
    output_path: str = "response.mp3") -> str:
    """Generate speech from text using OpenAI's TTS API."""
    response = client.audio.speech.create(
        model="tts-1-hd", # Higher quality; use "tts-1" for lower latency
        voice=voice, # alloy, echo, fable, onyx, nova, shimmer
        input=text,
        speed=1.0 # 0.25 to 4.0
        )
    response.stream_to_file(output_path)
    return output_path
async def text_to_speech_elevenlabs(
    text: str,
    voice_id: str = "21m00Tcm4TlvDq8ikWAM", # Rachel
    output_path: str = "response.mp3"
    ) -> str:
    """Generate speech using ElevenLabs with streaming."""
    from elevenlabs import ElevenLabs
    eleven = ElevenLabs() # Uses ELEVEN_API_KEY env var
    audio_generator = eleven.text_to_speech.convert(
        voice_id=voice_id,
        text=text,
        model_id="eleven_turbo_v2_5",
        output_format="mp3_44100_128",
        voice_settings={
        "stability": 0.5,
        "similarity_boost": 0.75,
        "style": 0.3,
        "use_speaker_boost": True
        }
        )
    # Write streaming audio to file
    with open(output_path, "wb") as f:
        for chunk in audio_generator:
            f.write(chunk)
            return output_path
            # Simple usage
            output = text_to_speech_openai(
                "Hello! I'd be happy to help you check your order status. "
                "Could you provide me with your order number?",
                voice="nova"
                )
            print(f"Audio saved to: {output}")

# Implementation example
import asyncio
from pipecat.frames.frames import TextFrame, AudioRawFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.services.openai import OpenAILLMService
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.transports.services.daily import DailyTransport
async def create_voice_bot():
    """Create a real-time voice AI bot using Pipecat."""
    # Transport layer (handles WebRTC audio/video)
    transport = DailyTransport(
        room_url="https://your-domain.daily.co/room-name",
        token="your-daily-token",
        bot_name="Assistant"
    )
    # Speech-to-text
    stt = DeepgramSTTService(
        api_key="your-deepgram-key",
        model="nova-2",
        language="en"
    )
    # Language model
    llm = OpenAILLMService(
        api_key="your-openai-key",
        model="gpt-4o",
        system_prompt=(
        "You are a helpful voice assistant. Keep responses "
        "concise (1-2 sentences) since this is a voice conversation. "
        "Be natural and conversational."
        )
    )
    # Text-to-speech
    tts = CartesiaTTSService(
        api_key="your-cartesia-key",
        voice_id="a0e99841-438c-4a64-b679-ae501e7d6091",
        model_id="sonic-english",
        sample_rate=16000
    )
    # Build the pipeline: audio in -> STT -> LLM -> TTS -> audio out
    pipeline = Pipeline([
        transport.input(), # Receive user audio
        stt, # Transcribe to text
        llm, # Generate response
        tts, # Synthesize speech
        transport.output() # Send audio to user
    ])
    task = PipelineTask(pipeline)
    await task.run()
# Run the voice bot
asyncio.run(create_voice_bot())

class InterruptionHandler:
    """Manages user interruptions during system speech."""
    def __init__(self):
        self.is_speaking = False
        self.current_utterance: str = ""
        self.spoken_so_far: str = ""
    async def on_speech_started(self, text: str):
        """Called when the system starts speaking."""
        self.is_speaking = True
        self.current_utterance = text
        self.spoken_so_far = ""
    async def on_speech_chunk_played(self, chunk_text: str):
        """Track how much of the response has been spoken."""
        self.spoken_so_far += chunk_text
    async def on_user_interruption(self, user_audio_detected: bool):
        """Handle user interrupting system speech."""
        if not self.is_speaking or not user_audio_detected:
            return None
            self.is_speaking = False
            # Calculate what was and was not heard
            unspoken = self.current_utterance[len(self.spoken_so_far):]
            return {
                "action": "interrupted",
                "spoken_portion": self.spoken_so_far.strip(),
                "unspoken_portion": unspoken.strip(),
                "context_note": (
                f"System was saying: '{self.spoken_so_far.strip()}' "
                f"but was interrupted. The rest ('{unspoken.strip()[:50]}...') "
                "was not heard by the user."
                )
                }
            async def on_speech_completed(self):
                """Called when system finishes speaking without interruption."""
                self.is_speaking = False
                self.spoken_so_far = ""
                self.current_utterance = ""