Building a Voice-Enabled AI Sales Agent

Environment Setup and Dependencies

This slide covers the essential environment configuration required for building a voice-enabled AI sales agent. The setup includes installing necessary Python packages, configuring environment variables, and initializing core dependencies for speech processing and language understanding.

import os
import torch
import whisper
import openai
import langchain
import chromadb
import gradio as gr
from typing import List, Dict

# Configure environment variables and API keys
openai.api_key = os.getenv("OPENAI_API_KEY")
CHROMA_DB_PATH = "knowledge_base"

# Initialize Whisper model for speech-to-text
model = whisper.load_model("base")

# Initialize ChromaDB client
chroma_client = chromadb.Client()
collection = chroma_client.create_collection(
    name="product_knowledge",
    metadata={"description": "Product information and specifications"}
)

print("Environment setup completed successfully")

Speech-to-Text Processing

Implementation of the speech recognition component using OpenAI's Whisper model. This module handles audio input processing, converts spoken language to text, and manages different audio formats and languages for robust voice recognition.

class AudioProcessor:
    def __init__(self, model_type: str = "base"):
        self.model = whisper.load_model(model_type)
        self.supported_formats = [".wav", ".mp3", ".m4a"]
    
    def process_audio(self, audio_path: str) -> Dict:
        """Process audio file and return transcription"""
        if not any(audio_path.endswith(fmt) for fmt in self.supported_formats):
            raise ValueError(f"Unsupported audio format. Use: {self.supported_formats}")
            
        # Transcribe audio using Whisper
        result = self.model.transcribe(audio_path)
        
        return {
            "text": result["text"],
            "language": result["language"],
            "segments": result["segments"]
        }

# Example usage
processor = AudioProcessor()
result = processor.process_audio("customer_query.wav")
print(f"Transcribed text: {result['text']}")

Language Model Integration

Advanced integration of GPT-4 through the OpenAI API, implementing context management and conversation handling. This component processes customer queries and generates appropriate responses based on the sales context.

class LanguageProcessor:
    def __init__(self, model: str = "gpt-4"):
        self.model = model
        self.context = []
        
    def process_query(self, query: str, context: List[str] = None) -> str:
        """Process customer query with context"""
        messages = [
            {"role": "system", "content": "You are a knowledgeable sales assistant."}
        ]
        
        # Add conversation context
        if context:
            for ctx in context:
                messages.append({"role": "assistant", "content": ctx})
        
        messages.append({"role": "user", "content": query})
        
        response = openai.ChatCompletion.create(
            model=self.model,
            messages=messages,
            temperature=0.7,
            max_tokens=150
        )
        
        return response.choices[0].message['content']

# Example usage
language_proc = LanguageProcessor()
response = language_proc.process_query("Tell me about your premium headphones")
print(f"AI Response: {response}")

Knowledge Base Implementation

The knowledge base implementation utilizes ChromaDB for efficient storage and retrieval of product information. This system enables semantic search capabilities and maintains product details in a vector database for quick access.

class KnowledgeBase:
    def __init__(self, collection_name: str = "products"):
        self.client = chromadb.Client()
        self.collection = self.client.get_or_create_collection(collection_name)
    
    def add_product(self, 
                   product_id: str,
                   name: str,
                   description: str,
                   specifications: Dict):
        """Add product information to knowledge base"""
        self.collection.add(
            documents=[description],
            metadatas=[{
                "name": name,
                "specifications": str(specifications)
            }],
            ids=[product_id]
        )
    
    def search_products(self, query: str, n_results: int = 3):
        """Search products using semantic similarity"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        return results

# Example usage
kb = KnowledgeBase()
kb.add_product(
    "HDPH001",
    "Premium Wireless Headphones",
    "High-fidelity wireless headphones with noise cancellation",
    {"battery": "20h", "type": "over-ear"}
)

Conversational Memory Implementation

This component maintains the conversation history and context using LangChain's memory management system. It enables the AI agent to reference previous interactions and maintain coherent, contextual conversations with customers.

from langchain.memory import ConversationBufferMemory
from langchain.schema import HumanMessage, AIMessage

class ConversationManager:
    def __init__(self, max_history: int = 5):
        self.memory = ConversationBufferMemory(k=max_history)
        self.current_conversation = []
    
    def add_interaction(self, user_message: str, ai_response: str):
        """Add a new interaction to conversation history"""
        self.memory.chat_memory.add_message(HumanMessage(content=user_message))
        self.memory.chat_memory.add_message(AIMessage(content=ai_response))
        
    def get_context(self) -> List[str]:
        """Retrieve conversation context"""
        return self.memory.load_memory_variables({})["history"]
    
    def clear_memory(self):
        """Clear conversation history"""
        self.memory.clear()

# Example usage
conv_manager = ConversationManager()
conv_manager.add_interaction(
    "What's the battery life of your wireless headphones?",
    "Our premium wireless headphones offer 20 hours of battery life."
)
print(f"Conversation context: {conv_manager.get_context()}")

Sentiment Analysis Module

Advanced sentiment analysis implementation for real-time customer emotion tracking. This module processes customer interactions to adapt the AI's responses based on detected emotional states and satisfaction levels.

import numpy as np
from transformers import pipeline

class SentimentAnalyzer:
    def __init__(self):
        self.analyzer = pipeline('sentiment-analysis')
        self.emotion_scores = []
    
    def analyze_sentiment(self, text: str) -> Dict:
        """Analyze sentiment of customer message"""
        result = self.analyzer(text)[0]
        score = float(result['score'])
        label = result['label']
        
        self.emotion_scores.append(score)
        
        return {
            'sentiment': label,
            'confidence': score,
            'trend': self._calculate_trend()
        }
    
    def _calculate_trend(self) -> str:
        """Calculate sentiment trend over conversation"""
        if len(self.emotion_scores) < 2:
            return "neutral"
        
        trend = np.mean(self.emotion_scores[-2:])
        if trend > 0.6:
            return "improving"
        elif trend < 0.4:
            return "declining"
        return "stable"

# Example usage
sentiment_analyzer = SentimentAnalyzer()
result = sentiment_analyzer.analyze_sentiment(
    "I'm really impressed with the sound quality!"
)
print(f"Sentiment analysis result: {result}")

Voice Response Generation

Implementation of text-to-speech capabilities for dynamic voice responses. This module converts AI-generated text responses into natural-sounding speech using advanced TTS models and audio processing.

from TTS.api import TTS
import soundfile as sf
import numpy as np

class VoiceGenerator:
    def __init__(self, model_name: str = "tts_models/en/ljspeech/tacotron2-DDC"):
        self.tts = TTS(model_name)
        self.sample_rate = 22050
        
    def generate_response(self, text: str, output_path: str) -> str:
        """Generate voice response from text"""
        try:
            # Generate speech waveform
            wav = self.tts.tts(text)
            
            # Apply audio enhancements
            wav = self._enhance_audio(wav)
            
            # Save to file
            sf.write(output_path, wav, self.sample_rate)
            
            return output_path
            
    def _enhance_audio(self, wav: np.ndarray) -> np.ndarray:
        """Apply audio enhancements for better quality"""
        # Normalize audio
        wav = wav / np.max(np.abs(wav))
        
        # Apply subtle compression
        threshold = 0.3
        wav = np.where(
            np.abs(wav) > threshold,
            threshold + (np.abs(wav) - threshold) * 0.6,
            wav
        )
        
        return wav

# Example usage
voice_gen = VoiceGenerator()
audio_file = voice_gen.generate_response(
    "Thank you for your interest in our products!",
    "response.wav"
)

Interactive Interface Implementation

Creation of a user-friendly interface using Gradio, enabling both voice and text interactions. This component provides real-time transcription, response generation, and audio playback capabilities.

import gradio as gr

class SalesAgentInterface:
    def __init__(self):
        self.audio_processor = AudioProcessor()
        self.language_proc = LanguageProcessor()
        self.voice_gen = VoiceGenerator()
        
    def create_interface(self):
        """Create and launch the interactive interface"""
        with gr.Blocks() as interface:
            gr.Markdown("## AI Sales Assistant")
            
            with gr.Row():
                # Input methods
                text_input = gr.Textbox(label="Type your question")
                audio_input = gr.Audio(
                    source="microphone",
                    type="filepath",
                    label="Or speak your question"
                )
            
            with gr.Row():
                # Response display
                text_output = gr.Textbox(label="AI Response")
                audio_output = gr.Audio(label="Voice Response")
            
            # Handle text input
            text_input.submit(
                fn=self._process_text_query,
                inputs=text_input,
                outputs=[text_output, audio_output]
            )
            
            # Handle voice input
            audio_input.change(
                fn=self._process_voice_query,
                inputs=audio_input,
                outputs=[text_output, audio_output]
            )
        
        return interface
    
    def _process_text_query(self, text: str):
        """Process text input and generate response"""
        response = self.language_proc.process_query(text)
        audio_file = self.voice_gen.generate_response(
            response,
            "response.wav"
        )
        return response, audio_file
    
    def _process_voice_query(self, audio_path: str):
        """Process voice input and generate response"""
        text = self.audio_processor.process_audio(audio_path)["text"]
        return self._process_text_query(text)

# Launch interface
agent_interface = SalesAgentInterface()
interface = agent_interface.create_interface()
interface.launch()

Real-time Analytics Integration

Implementation of real-time analytics tracking system to monitor customer interactions, measure conversion rates, and analyze user engagement patterns. This module provides valuable insights for continuous improvement of the AI sales agent.

from datetime import datetime
import pandas as pd
from typing import Optional

class InteractionAnalytics:
    def __init__(self):
        self.interactions_df = pd.DataFrame(
            columns=['timestamp', 'query_type', 'query', 'response', 
                    'sentiment', 'duration', 'conversion']
        )
        
    def log_interaction(self, 
                       query_type: str,
                       query: str,
                       response: str,
                       sentiment: Optional[float] = None,
                       duration: Optional[float] = None,
                       conversion: bool = False):
        """Log customer interaction details"""
        interaction = {
            'timestamp': datetime.now(),
            'query_type': query_type,
            'query': query,
            'response': response,
            'sentiment': sentiment,
            'duration': duration,
            'conversion': conversion
        }
        
        self.interactions_df = pd.concat([
            self.interactions_df,
            pd.DataFrame([interaction])
        ], ignore_index=True)
        
    def generate_metrics(self) -> Dict:
        """Calculate key performance metrics"""
        metrics = {
            'total_interactions': len(self.interactions_df),
            'avg_sentiment': self.interactions_df['sentiment'].mean(),
            'conversion_rate': (
                self.interactions_df['conversion'].sum() / 
                len(self.interactions_df)
            ),
            'avg_duration': self.interactions_df['duration'].mean()
        }
        
        return metrics

# Example usage
analytics = InteractionAnalytics()
analytics.log_interaction(
    query_type='voice',
    query="Tell me about your headphones",
    response="Our premium headphones feature...",
    sentiment=0.8,
    duration=45.5,
    conversion=True
)
print(f"Performance metrics: {analytics.generate_metrics()}")

Dynamic Product Knowledge Management

Advanced implementation of a dynamic product knowledge management system that automatically updates product information and maintains real-time inventory status through API integrations.

import aiohttp
import asyncio
from datetime import datetime

class ProductManager:
    def __init__(self, api_endpoint: str):
        self.api_endpoint = api_endpoint
        self.knowledge_base = KnowledgeBase()
        self.last_update = None
        
    async def update_product_info(self):
        """Update product information from external API"""
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{self.api_endpoint}/products") as response:
                products = await response.json()
                
                for product in products:
                    # Update knowledge base
                    self.knowledge_base.add_product(
                        product_id=product['id'],
                        name=product['name'],
                        description=product['description'],
                        specifications=product['specs']
                    )
                    
                self.last_update = datetime.now()
                
    async def get_inventory_status(self, product_id: str) -> Dict:
        """Get real-time inventory status"""
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.api_endpoint}/inventory/{product_id}"
            ) as response:
                return await response.json()
    
    def format_product_response(self, product_id: str, 
                              include_inventory: bool = True) -> str:
        """Format product information for customer response"""
        product = self.knowledge_base.get_product(product_id)
        
        if include_inventory:
            inventory = asyncio.run(self.get_inventory_status(product_id))
            product.update({'inventory': inventory})
            
        return self._generate_response_text(product)
    
    def _generate_response_text(self, product_data: Dict) -> str:
        """Generate formatted response text"""
        return f"{product_data['name']}: {product_data['description']} " \
               f"Available: {product_data.get('inventory', {}).get('in_stock', 'N/A')}"

# Example usage
product_manager = ProductManager("https://api.example.com")
asyncio.run(product_manager.update_product_info())
response = product_manager.format_product_response("HDPH001")
print(f"Product response: {response}")

Enhanced Security Implementation

Implementation of robust security measures including input validation, rate limiting, and authentication to protect the AI sales agent from potential misuse and ensure secure customer interactions.

import hashlib
import time
from functools import wraps
from typing import Callable, Dict

class SecurityManager:
    def __init__(self):
        self.rate_limits = {}
        self.blocked_ips = set()
        self.request_logs = {}
        
    def rate_limiter(self, 
                     max_requests: int = 10,
                     window_seconds: int = 60) -> Callable:
        """Rate limiting decorator"""
        def decorator(func):
            @wraps(func)
            def wrapper(self, client_id: str, *args, **kwargs):
                current_time = time.time()
                
                # Clean old requests
                self._clean_old_requests(current_time, window_seconds)
                
                # Check rate limit
                if self._is_rate_limited(client_id, current_time,
                                       max_requests, window_seconds):
                    raise Exception("Rate limit exceeded")
                
                # Log request
                self._log_request(client_id, current_time)
                
                return func(self, *args, **kwargs)
            return wrapper
        return decorator
    
    def validate_input(self, text: str) -> bool:
        """Validate input for potential security risks"""
        # Check input length
        if len(text) > 1000:
            return False
            
        # Check for suspicious patterns
        suspicious_patterns = [
            "script",
            "exec(",
            "eval(",
            "<script>",
            "DROP TABLE"
        ]
        
        return not any(pattern.lower() in text.lower() 
                      for pattern in suspicious_patterns)
    
    def _clean_old_requests(self, current_time: float, window: int):
        """Remove old requests from tracking"""
        cutoff = current_time - window
        self.request_logs = {
            client: [req for req in requests if req > cutoff]
            for client, requests in self.request_logs.items()
        }
    
    def _is_rate_limited(self, client_id: str, current_time: float,
                        max_requests: int, window: int) -> bool:
        """Check if client has exceeded rate limit"""
        if client_id in self.blocked_ips:
            return True
            
        requests = self.request_logs.get(client_id, [])
        return len(requests) >= max_requests
    
    def _log_request(self, client_id: str, timestamp: float):
        """Log new request"""
        if client_id not in self.request_logs:
            self.request_logs[client_id] = []
        self.request_logs[client_id].append(timestamp)

# Example usage
security = SecurityManager()

@security.rate_limiter(max_requests=5, window_seconds=60)
def process_request(query: str):
    if not security.validate_input(query):
        raise ValueError("Invalid input detected")
    return "Processed query: " + query

try:
    result = process_request("client123", "Tell me about your products")
    print(result)
except Exception as e:
    print(f"Error: {str(e)}")

Performance Testing Suite

Implementation of comprehensive testing framework to evaluate the AI sales agent's performance across multiple dimensions including response accuracy, latency, and conversation coherence using automated test scenarios.

import unittest
import time
from typing import List, Tuple

class SalesAgentTester:
    def __init__(self, agent):
        self.agent = agent
        self.test_cases = []
        self.results = {}
        
    def add_test_case(self, 
                      query: str,
                      expected_response: str,
                      test_type: str = 'accuracy'):
        """Add test case to suite"""
        self.test_cases.append({
            'query': query,
            'expected': expected_response,
            'type': test_type
        })
    
    def run_performance_tests(self) -> Dict:
        """Execute all test cases and measure performance"""
        results = {
            'accuracy': [],
            'latency': [],
            'coherence': []
        }
        
        for case in self.test_cases:
            # Measure response time
            start_time = time.time()
            response = self.agent.process_query(case['query'])
            latency = time.time() - start_time
            
            # Calculate metrics
            accuracy = self._calculate_accuracy(
                response,
                case['expected']
            )
            coherence = self._measure_coherence(response)
            
            # Store results
            results['accuracy'].append(accuracy)
            results['latency'].append(latency)
            results['coherence'].append(coherence)
        
        return self._aggregate_results(results)
    
    def _calculate_accuracy(self, 
                          response: str,
                          expected: str) -> float:
        """Calculate response accuracy score"""
        response_tokens = set(response.lower().split())
        expected_tokens = set(expected.lower().split())
        
        intersection = response_tokens.intersection(expected_tokens)
        union = response_tokens.union(expected_tokens)
        
        return len(intersection) / len(union)
    
    def _measure_coherence(self, response: str) -> float:
        """Measure response coherence"""
        sentences = response.split('.')
        coherence_scores = []
        
        for i in range(len(sentences) - 1):
            score = self._sentence_similarity(
                sentences[i],
                sentences[i + 1]
            )
            coherence_scores.append(score)
            
        return sum(coherence_scores) / len(coherence_scores) if coherence_scores else 0
    
    def _sentence_similarity(self, s1: str, s2: str) -> float:
        """Calculate similarity between two sentences"""
        s1_tokens = set(s1.lower().split())
        s2_tokens = set(s2.lower().split())
        
        intersection = s1_tokens.intersection(s2_tokens)
        union = s1_tokens.union(s2_tokens)
        
        return len(intersection) / len(union) if union else 0
    
    def _aggregate_results(self, results: Dict) -> Dict:
        """Aggregate test results"""
        return {
            'avg_accuracy': sum(results['accuracy']) / len(results['accuracy']),
            'avg_latency': sum(results['latency']) / len(results['latency']),
            'avg_coherence': sum(results['coherence']) / len(results['coherence']),
            'total_tests': len(self.test_cases)
        }

# Example usage
tester = SalesAgentTester(language_processor)
tester.add_test_case(
    "What are the features of your wireless headphones?",
    "Our wireless headphones feature active noise cancellation, " \
    "20-hour battery life, and premium sound quality."
)
results = tester.run_performance_tests()
print(f"Performance test results: {results}")

Deployment and Scaling Architecture

Implementation of a scalable deployment architecture using containerization and load balancing to handle multiple concurrent customer interactions while maintaining performance and reliability.

import docker
from kubernetes import client, config
from typing import List, Dict

class DeploymentManager:
    def __init__(self, 
                 docker_image: str,
                 namespace: str = "sales-agent"):
        self.docker_image = docker_image
        self.namespace = namespace
        self.docker_client = docker.from_env()
        
        # Initialize Kubernetes config
        config.load_kube_config()
        self.k8s_apps_v1 = client.AppsV1Api()
        self.k8s_core_v1 = client.CoreV1Api()
        
    def deploy_agent(self, 
                    replicas: int = 3,
                    resources: Dict = None):
        """Deploy AI sales agent to Kubernetes cluster"""
        if resources is None:
            resources = {
                'requests': {
                    'cpu': '500m',
                    'memory': '512Mi'
                },
                'limits': {
                    'cpu': '1000m',
                    'memory': '1Gi'
                }
            }
            
        # Create deployment
        deployment = client.V1Deployment(
            metadata=client.V1ObjectMeta(
                name="sales-agent",
                namespace=self.namespace
            ),
            spec=client.V1DeploymentSpec(
                replicas=replicas,
                selector=client.V1LabelSelector(
                    match_labels={"app": "sales-agent"}
                ),
                template=client.V1PodTemplateSpec(
                    metadata=client.V1ObjectMeta(
                        labels={"app": "sales-agent"}
                    ),
                    spec=client.V1PodSpec(
                        containers=[
                            client.V1Container(
                                name="sales-agent",
                                image=self.docker_image,
                                resources=resources,
                                ports=[
                                    client.V1ContainerPort(
                                        container_port=8080
                                    )
                                ]
                            )
                        ]
                    )
                )
            )
        )
        
        # Create the deployment
        self.k8s_apps_v1.create_namespaced_deployment(
            namespace=self.namespace,
            body=deployment
        )
        
        # Create service
        service = client.V1Service(
            metadata=client.V1ObjectMeta(
                name="sales-agent-service"
            ),
            spec=client.V1ServiceSpec(
                selector={"app": "sales-agent"},
                ports=[
                    client.V1ServicePort(
                        port=80,
                        target_port=8080
                    )
                ],
                type="LoadBalancer"
            )
        )
        
        self.k8s_core_v1.create_namespaced_service(
            namespace=self.namespace,
            body=service
        )
        
    def scale_deployment(self, replicas: int):
        """Scale the deployment"""
        self.k8s_apps_v1.patch_namespaced_deployment_scale(
            name="sales-agent",
            namespace=self.namespace,
            body={"spec": {"replicas": replicas}}
        )

# Example usage
deployment_manager = DeploymentManager("sales-agent:latest")
deployment_manager.deploy_agent(replicas=3)

Additional Resources

• ArXiv: "Attention Is All You Need" - https://arxiv.org/abs/1706.03762
• ArXiv: "Neural Machine Translation by Jointly Learning to Align and Translate" - https://arxiv.org/abs/1409.0473
• ArXiv: "BERT: Pre-training of Deep Bidirectional Transformers for Language Understanding" - https://arxiv.org/abs/1810.04805
• Search for "Conversational AI architectures" on Google Scholar
• Visit OpenAI documentation for GPT-4 implementation details
• Explore LangChain documentation for advanced conversation management patterns