Voice & Multimodal Interfaces

# implement transcribe_audio, transcribe_with_deepgram
# Key operations: results display, API interaction
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
# Key operations: results display, chunking strategy, API interaction
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
# Key operations: prompt construction, API interaction
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())

# Define InterruptionHandler; implement __init__
# Key operations: chunking strategy
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 = ""

# Define MultimodalConversation; implement encode_image_to_base64, __init__, send_text
# Key operations: results display, prompt construction, API interaction
import base64
from openai import OpenAI

client = OpenAI()

def encode_image_to_base64(image_path: str) -> str:
 """Read an image file and encode it as base64."""
 with open(image_path, "rb") as f:
 return base64.b64encode(f.read()).decode("utf-8")

class MultimodalConversation:
 """Conversational AI with vision capabilities."""

 def __init__(self, system_prompt: str):
 self.system_prompt = system_prompt
 self.history: list[dict] = []

 def send_text(self, user_message: str) -> str:
 """Send a text-only message."""
 self.history.append({
 "role": "user",
 "content": user_message
 })
 return self._get_response()

 def send_image(self, image_path: str,
 question: str = "What do you see?") -> str:
 """Send an image with an optional question."""
 b64_image = encode_image_to_base64(image_path)

 # Determine MIME type from extension
 ext = image_path.rsplit(".", 1)[-1].lower()
 mime_map = {"jpg": "jpeg", "jpeg": "jpeg",
 "png": "png", "gif": "gif", "webp": "webp"}
 mime_type = f"image/{mime_map.get(ext, 'jpeg')}"

 self.history.append({
 "role": "user",
 "content": [
 {"type": "text", "text": question},
 {
 "type": "image_url",
 "image_url": {
 "url": f"data:{mime_type};base64,{b64_image}",
 "detail": "high"
 }
 }
 ]
 })
 return self._get_response()

 def send_image_url(self, url: str, question: str) -> str:
 """Send an image via URL with a question."""
 self.history.append({
 "role": "user",
 "content": [
 {"type": "text", "text": question},
 {
 "type": "image_url",
 "image_url": {"url": url, "detail": "auto"}
 }
 ]
 })
 return self._get_response()

 def _get_response(self) -> str:
 """Get a response from the multimodal LLM."""
 messages = [
 {"role": "system", "content": self.system_prompt},
 *self.history
 ]
 response = client.chat.completions.create(
 model="gpt-4o",
 messages=messages,
 max_tokens=1000
 )
 assistant_msg = response.choices[0].message.content
 self.history.append({
 "role": "assistant", "content": assistant_msg
 })
 return assistant_msg

# Example: Visual troubleshooting assistant
troubleshooter = MultimodalConversation(
 system_prompt=(
 "You are a technical support assistant that can analyze "
 "screenshots and photos to help users troubleshoot issues. "
 "When shown an image, describe what you see and provide "
 "specific, actionable solutions."
 )
)

# User shares a screenshot of an error
response = troubleshooter.send_image(
 "error_screenshot.png",
 "I keep getting this error when I try to start the application. "
 "What should I do?"
)
print(response)