Voice AI Agents

PDF Documents
    ↓
Document Processing
├─ RecursiveCharacterTextSplitter
├─ FastEmbed embeddings
└─ Qdrant vector storage
    ↓
User Question (Text/Voice)
    ↓
Retrieval Pipeline
├─ Query embedding
├─ Vector similarity search
└─ Context retrieval
    ↓
Agent Processing
├─ Processing Agent (answer generation)
└─ TTS Agent (speech optimization)
    ↓
Voice Generation
├─ OpenAI TTS API
├─ Voice selection
└─ MP3 output
    ↓
User Response (Text + Audio)

cd voice_ai_agents/voice_rag_openaisdk
pip install -r requirements.txt
streamlit run rag_voice.py

from agno import Agent, OpenAI
from agno.storage import QdrantStorage
from agno.embeddings import FastEmbed

# Create processing agent
processing_agent = Agent(
    name="Document Processor",
    model=OpenAI(id="gpt-4o"),
    instructions="Generate clear, spoken-word friendly responses",
    storage=QdrantStorage(...)
)

# Create TTS agent
tts_agent = Agent(
    name="TTS Optimizer",
    model=OpenAI(id="gpt-4o"),
    instructions="Optimize text for natural speech synthesis"
)

# Generate and synthesize
response = processing_agent.run(query)
optimized = tts_agent.run(response)
audio = openai.audio.speech.create(
    model="tts-1",
    voice="nova",
    input=optimized
)

Documentation Website
    ↓
Firecrawl API (Web Crawling)
    ↓
Content Extraction & Processing
    ↓
Qdrant Vector Database
├─ FastEmbed embeddings
├─ Semantic indexing
└─ Efficient retrieval
    ↓
User Query
    ↓
AI Agent Team
├─ Documentation Processor Agent
│   └─ Analyzes docs and generates answers
├─ TTS Agent
│   └─ Optimizes for natural speech
└─ Voice synthesis
    ↓
Text + Voice Response

cd voice_ai_agents/customer_support_voice_agent
pip install -r requirements.txt
streamlit run ai_voice_agent_docs.py

User Input
├─ Location (e.g., "Paris, France")
├─ Interests (History, Architecture, Culture, Food)
├─ Duration (15, 30, or 60 minutes)
└─ Custom preferences
    ↓
Orchestrator Agent Planning
├─ Web search for location information
├─ Time allocation based on interests
├─ Content distribution planning
└─ Transition coordination
    ↓
Expert Agents Generate Content (Parallel)
├─ History Agent → Historical narratives
├─ Architecture Agent → Building descriptions
├─ Culture Agent → Cultural insights
└─ Culinary Agent → Food experiences
    ↓
Orchestrator Assembles Tour
├─ Weaves narratives together
├─ Adds transitions
├─ Ensures proper pacing
└─ Balances content by interest weights
    ↓
Voice Synthesis (GPT-4o Mini Audio)
├─ Different voices for each agent
├─ Natural transitions
├─ Expressive delivery
└─ Appropriate tone per topic
    ↓
Complete Audio Tour

cd voice_ai_agents/ai_audio_tour_agent
pip install -r requirements.txt
streamlit run ai_audio_tour_agent.py

Introduction (2 min)
└─ Orchestrator sets scene

History Section (8 min)
├─ Ancient origins
├─ Key historical events
└─ Notable figures

Architecture Section (7 min)
├─ Iconic buildings
├─ Architectural styles
└─ Design elements

Culture Section (7 min)
├─ Local traditions
├─ Arts and music
└─ Cultural practices

Culinary Section (5 min)
├─ Signature dishes
├─ Food culture
└─ Dining experiences

Conclusion (1 min)
└─ Orchestrator wraps up

from agno import Agent, OpenAI
import openai

# Create agent with TTS capability
agent = Agent(
    name="Voice Assistant",
    model=OpenAI(id="gpt-4o"),
    instructions="Generate clear, conversational responses"
)

# Generate response
response = agent.run(user_query)

# Convert to speech
audio = openai.audio.speech.create(
    model="tts-1",
    voice="nova",
    input=response.content
)

# Save or stream audio
audio.stream_to_file("response.mp3")

from agno import Agent
from agno.storage import QdrantStorage
from agno.embeddings import FastEmbed

# RAG agent with voice output
rag_agent = Agent(
    name="Voice RAG",
    model=OpenAI(id="gpt-4o"),
    storage=QdrantStorage(
        collection="docs",
        embedder=FastEmbed()
    ),
    instructions="""
    Answer questions using the knowledge base.
    Format responses for natural speech:
    - Use conversational language
    - Add appropriate pauses with commas
    - Spell out acronyms on first use
    - Keep sentences clear and concise
    """
)

# Query with voice output
response = rag_agent.run(query)
audio = synthesize_speech(response.content)

# Specialized agents with different voices
history_agent = Agent(
    name="Historian",
    voice="onyx",  # Deep, authoritative
    instructions="Deliver historical narratives"
)

culture_agent = Agent(
    name="Culture Guide",
    voice="nova",  # Friendly, energetic
    instructions="Explore cultural experiences"
)

orchestrator = Agent(
    name="Tour Guide",
    team=[history_agent, culture_agent],
    voice="alloy",  # Neutral, balanced
    instructions="Coordinate tour narrative"
)

# Generate multi-voice tour
tour = orchestrator.run(tour_request)

import asyncio

async def stream_voice_response(query):
    # Stream text response
    async for chunk in agent.run_stream(query):
        print(chunk.content, end="", flush=True)
        text_buffer += chunk.content
        
        # Generate audio for complete sentences
        if chunk.content.endswith(('.', '!', '?')):
            audio = await synthesize_async(text_buffer)
            await play_audio(audio)
            text_buffer = ""

def optimize_for_speech(text: str) -> str:
    """
    Optimize text for natural-sounding speech.
    """
    # Spell out acronyms
    text = text.replace("AI", "A.I.")
    text = text.replace("API", "A.P.I.")
    
    # Add pauses for readability
    text = text.replace(". ", ". ... ")
    
    # Remove markdown formatting
    text = re.sub(r'\*\*|\*|_', '', text)
    
    # Convert numbers to words for clarity
    text = text.replace("1st", "first")
    text = text.replace("2nd", "second")
    
    return text

# High-quality audio
audio = openai.audio.speech.create(
    model="tts-1-hd",  # HD quality
    voice="nova",
    input=text,
    response_format="mp3",
    speed=1.0  # Normal speed (0.25-4.0)
)

# Optimize for streaming
audio = openai.audio.speech.create(
    model="tts-1",  # Standard quality, faster
    voice="nova",
    input=text,
    response_format="opus"  # Better for streaming
)