Construire des agents vocaux IA pour l'immobilier

graph TD
    A[Phone Call] --> B[Speech Recognition]
    B --> C[Natural Language Processing]
    C --> D[Intent Classification]
    D --> E[Entity Extraction]
    E --> F[Conversation Management]
    F --> G[CRM Integration]
    G --> H[Appointment Scheduling]
    H --> I[Follow-up Actions]
    
    J[Text-to-Speech] --> K[Voice Response]
    F --> J

import speech_recognition as sr
import pyaudio
import wave
from pydub import AudioSegment
import openai
 
class VoiceProcessor:
    def __init__(self):
        self.recognizer = sr.Recognizer()
        self.microphone = sr.Microphone()
        self.openai_client = openai.OpenAI(api_key="your-api-key")
        
    def process_audio(self, audio_file_path):
        """Convert speech to text"""
        try:
            # Load and preprocess audio
            audio = AudioSegment.from_file(audio_file_path)
            
            # Normalize audio levels
            normalized_audio = audio.normalize()
            
            # Convert to format suitable for speech recognition
            wav_audio = normalized_audio.export(format="wav")
            
            # Use Google Speech Recognition
            with sr.AudioFile(wav_audio) as source:
                audio_data = self.recognizer.record(source)
                text = self.recognizer.recognize_google(audio_data)
                
            return {
                'success': True,
                'text': text,
                'confidence': 0.95  # illustrative placeholder value, not a measured score
            }
            
        except sr.UnknownValueError:
            return {
                'success': False,
                'error': 'Could not understand audio',
                'confidence': 0.0
            }
        except sr.RequestError as e:
            return {
                'success': False,
                'error': f'Speech recognition service error: {e}',
                'confidence': 0.0
            }
    
    def enhance_audio_quality(self, audio_file):
        """Improve audio quality for better recognition"""
        audio = AudioSegment.from_file(audio_file)
        
        # Remove background noise
        audio = audio.filter_silence(min_silence_len=1000, silence_thresh=-40)
        
        # Normalize volume
        audio = audio.normalize()
        
        # Apply noise reduction (simplified)
        audio = audio.low_pass_filter(3000)
        
        return audio

import spacy
from spacy import displacy
import re
from datetime import datetime, timedelta
 
class RealEstateNLP:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.intent_patterns = {
            'property_inquiry': [
                r'looking.*property', r'interested.*house', r'want.*buy',
                r'searching.*home', r'need.*place'
            ],
            'price_inquiry': [
                r'how.*much', r'price', r'cost', r'budget'
            ],
            'schedule_viewing': [
                r'schedule', r'viewing', r'appointment', r'visit',
                r'see.*property', r'tour'
            ],
            'location_inquiry': [
                r'where.*located', r'address', r'neighborhood',
                r'area', r'location'
            ],
            'property_details': [
                r'bedrooms', r'bathrooms', r'size', r'square.*feet',
                r'amenities', r'features'
            ]
        }
        
        self.entity_patterns = {
            'property_type': [r'house', r'apartment', r'condo', r'townhouse', r'villa'],
            'price_range': [r'\$[\d,]+', r'budget.*\$[\d,]+'],
            'location': [r'in\s+([A-Za-z\s]+)', r'near\s+([A-Za-z\s]+)'],
            'time_references': [r'tomorrow', r'next week', r'this weekend', r'asap']
        }
    
    def classify_intent(self, text):
        """Classify user intent from conversation"""
        doc = self.nlp(text.lower())
        intent_scores = {}
        
        for intent, patterns in self.intent_patterns.items():
            score = 0
            for pattern in patterns:
                matches = re.findall(pattern, text.lower())
                score += len(matches)
            intent_scores[intent] = score
        
        # Return the intent with highest score
        primary_intent = max(intent_scores, key=intent_scores.get)
        confidence = intent_scores[primary_intent] / sum(intent_scores.values())
        
        return {
            'intent': primary_intent,
            'confidence': confidence,
            'all_scores': intent_scores
        }
    
    def extract_entities(self, text):
        """Extract relevant entities from conversation"""
        doc = self.nlp(text)
        entities = {
            'property_type': None,
            'price_range': None,
            'location': None,
            'bedrooms': None,
            'bathrooms': None,
            'time_preference': None,
            'contact_info': {}
        }
        
        # Extract property type
        for pattern in self.entity_patterns['property_type']:
            match = re.search(pattern, text.lower())
            if match:
                entities['property_type'] = match.group()
                break
        
        # Extract price range
        price_match = re.search(r'\$([\d,]+)', text)
        if price_match:
            entities['price_range'] = int(price_match.group(1).replace(',', ''))
        
        # Extract location
        for pattern in self.entity_patterns['location']:
            match = re.search(pattern, text.lower())
            if match:
                entities['location'] = match.group(1).strip()
                break
        
        # Extract bedroom/bathroom count
        bedroom_match = re.search(r'(\d+)\s*bed', text.lower())
        if bedroom_match:
            entities['bedrooms'] = int(bedroom_match.group(1))
        
        bathroom_match = re.search(r'(\d+)\s*bath', text.lower())
        if bathroom_match:
            entities['bathrooms'] = int(bathroom_match.group(1))
        
        # Extract time preferences
        for pattern in self.entity_patterns['time_references']:
            match = re.search(pattern, text.lower())
            if match:
                entities['time_preference'] = match.group()
                break
        
        return entities

class ConversationManager:
    def __init__(self):
        self.nlp = RealEstateNLP()
        self.conversation_states = {}
        self.qualification_questions = [
            "What type of property are you looking for?",
            "What's your budget range?",
            "How many bedrooms do you need?",
            "What area are you interested in?",
            "When are you looking to move?",
            "Do you have a pre-approval for financing?"
        ]
    
    def process_conversation(self, user_id, text):
        """Main conversation processing logic"""
        # Get or create conversation state
        if user_id not in self.conversation_states:
            self.conversation_states[user_id] = {
                'stage': 'greeting',
                'collected_info': {},
                'question_index': 0,
                'conversation_history': []
            }
        
        state = self.conversation_states[user_id]
        state['conversation_history'].append({
            'user': text,
            'timestamp': datetime.now().isoformat()
        })
        
        # Classify intent and extract entities
        intent_result = self.nlp.classify_intent(text)
        entities = self.nlp.extract_entities(text)
        
        # Update collected information
        self._update_collected_info(state, entities)
        
        # Determine next action based on conversation stage
        response = self._generate_response(state, intent_result, entities)
        
        # Update conversation state
        state['conversation_history'].append({
            'agent': response['text'],
            'timestamp': datetime.now().isoformat()
        })
        
        return response
    
    def _update_collected_info(self, state, entities):
        """Update collected information from entities"""
        for key, value in entities.items():
            if value is not None:
                state['collected_info'][key] = value
    
    def _generate_response(self, state, intent_result, entities):
        """Generate appropriate response based on conversation state"""
        stage = state['stage']
        
        if stage == 'greeting':
            return self._handle_greeting(state)
        elif stage == 'qualification':
            return self._handle_qualification(state, intent_result, entities)
        elif stage == 'scheduling':
            return self._handle_scheduling(state, intent_result, entities)
        elif stage == 'confirmation':
            return self._handle_confirmation(state, intent_result, entities)
        else:
            return self._handle_fallback(state)
    
    def _handle_greeting(self, state):
        """Handle initial greeting and introduction"""
        state['stage'] = 'qualification'
        return {
            'text': "Hello! Thank you for calling. I'm here to help you find your perfect property. Let me ask you a few questions to better understand what you're looking for.",
            'action': 'continue',
            'next_question': self.qualification_questions[0]
        }
    
    def _handle_qualification(self, state, intent_result, entities):
        """Handle lead qualification questions"""
        question_index = state['question_index']
        
        # Check if we have enough information
        if self._is_qualification_complete(state):
            state['stage'] = 'scheduling'
            return self._handle_scheduling(state, intent_result, entities)
        
        # Ask next question
        if question_index < len(self.qualification_questions):
            question = self.qualification_questions[question_index]
            state['question_index'] += 1
            
            return {
                'text': question,
                'action': 'ask_question',
                'question_type': self._get_question_type(question_index)
            }
        
        return self._handle_scheduling(state, intent_result, entities)
    
    def _handle_scheduling(self, state, intent_result, entities):
        """Handle appointment scheduling"""
        if intent_result['intent'] == 'schedule_viewing':
            # Extract time preference
            time_pref = entities.get('time_preference', 'as soon as possible')
            
            # Generate available time slots
            available_slots = self._get_available_slots()
            
            return {
                'text': f"Great! I'd love to schedule a viewing for you. I have these times available: {', '.join(available_slots[:3])}. Which works best for you?",
                'action': 'schedule',
                'available_slots': available_slots
            }
        
        return {
            'text': "Would you like to schedule a viewing? I can show you some properties that match your criteria.",
            'action': 'offer_scheduling'
        }
    
    def _is_qualification_complete(self, state):
        """Check if we have enough information to proceed"""
        required_fields = ['property_type', 'price_range', 'location']
        collected = state['collected_info']
        
        return all(field in collected and collected[field] is not None 
                  for field in required_fields)

import requests
import json
from datetime import datetime
 
class CRMIntegration:
    def __init__(self, crm_api_key, crm_base_url):
        self.api_key = crm_api_key
        self.base_url = crm_base_url
        self.headers = {
            'Authorization': f'Bearer {crm_api_key}',
            'Content-Type': 'application/json'
        }
    
    def create_lead(self, lead_data):
        """Create a new lead in CRM"""
        lead_payload = {
            'first_name': lead_data.get('first_name', ''),
            'last_name': lead_data.get('last_name', ''),
            'email': lead_data.get('email', ''),
            'phone': lead_data.get('phone', ''),
            'property_type': lead_data.get('property_type', ''),
            'price_range': lead_data.get('price_range', 0),
            'location': lead_data.get('location', ''),
            'bedrooms': lead_data.get('bedrooms', 0),
            'bathrooms': lead_data.get('bathrooms', 0),
            'move_date': lead_data.get('move_date', ''),
            'financing_status': lead_data.get('financing_status', 'unknown'),
            'lead_source': 'voice_agent',
            'qualification_score': self._calculate_lead_score(lead_data),
            'created_at': datetime.now().isoformat()
        }
        
        response = requests.post(
            f"{self.base_url}/leads",
            headers=self.headers,
            json=lead_payload
        )
        
        if response.status_code == 201:
            return response.json()
        else:
            raise Exception(f"Failed to create lead: {response.text}")
    
    def schedule_appointment(self, lead_id, appointment_data):
        """Schedule appointment in CRM"""
        appointment_payload = {
            'lead_id': lead_id,
            'appointment_date': appointment_data['date'],
            'appointment_time': appointment_data['time'],
            'property_address': appointment_data.get('property_address', ''),
            'agent_id': appointment_data.get('agent_id', ''),
            'notes': appointment_data.get('notes', ''),
            'status': 'scheduled'
        }
        
        response = requests.post(
            f"{self.base_url}/appointments",
            headers=self.headers,
            json=appointment_payload
        )
        
        if response.status_code == 201:
            return response.json()
        else:
            raise Exception(f"Failed to schedule appointment: {response.text}")
    
    def _calculate_lead_score(self, lead_data):
        """Calculate lead qualification score"""
        score = 0
        
        # Basic information (40 points)
        if lead_data.get('first_name'): score += 10
        if lead_data.get('last_name'): score += 10
        if lead_data.get('email'): score += 10
        if lead_data.get('phone'): score += 10
        
        # Property preferences (30 points)
        if lead_data.get('property_type'): score += 10
        if lead_data.get('price_range'): score += 10
        if lead_data.get('location'): score += 10
        
        # Urgency indicators (30 points)
        if lead_data.get('move_date'):
            move_date = datetime.strptime(lead_data['move_date'], '%Y-%m-%d')
            days_until_move = (move_date - datetime.now()).days
            if days_until_move <= 30:
                score += 20
            elif days_until_move <= 90:
                score += 10
        
        if lead_data.get('financing_status') == 'pre_approved':
            score += 10
        
        return min(score, 100)  # Cap at 100

import pyttsx3
import pydub
from pydub.playback import play
import io
 
class VoiceResponse:
    def __init__(self):
        self.tts_engine = pyttsx3.init()
        self._configure_voice()
    
    def _configure_voice(self):
        """Configure voice settings for natural speech"""
        voices = self.tts_engine.getProperty('voices')
        
        # Select a natural-sounding voice
        for voice in voices:
            if 'english' in voice.name.lower():
                self.tts_engine.setProperty('voice', voice.id)
                break
        
        # Set speech rate and volume
        self.tts_engine.setProperty('rate', 180)  # Words per minute
        self.tts_engine.setProperty('volume', 0.9)
    
    def generate_speech(self, text):
        """Convert text to speech"""
        # Add natural pauses and emphasis
        enhanced_text = self._enhance_speech_text(text)
        
        # Generate audio
        audio_buffer = io.BytesIO()
        self.tts_engine.save_to_file(enhanced_text, 'temp_audio.wav')
        self.tts_engine.runAndWait()
        
        # Load and return audio
        audio = pydub.AudioSegment.from_wav('temp_audio.wav')
        return audio
    
    def _enhance_speech_text(self, text):
        """Add natural speech patterns to text"""
        # Add pauses for better comprehension
        text = text.replace('.', '. ')
        text = text.replace(',', ', ')
        text = text.replace('?', '? ')
        text = text.replace('!', '! ')
        
        # Add emphasis to important words
        emphasis_words = ['important', 'available', 'schedule', 'viewing']
        for word in emphasis_words:
            text = text.replace(word, f"<emphasis>{word}</emphasis>")
        
        return text
    
    def play_response(self, audio):
        """Play the generated audio response"""
        play(audio)

class RealEstateVoiceAgent:
    def __init__(self, crm_api_key, crm_base_url):
        self.voice_processor = VoiceProcessor()
        self.conversation_manager = ConversationManager()
        self.crm_integration = CRMIntegration(crm_api_key, crm_base_url)
        self.voice_response = VoiceResponse()
        
    def handle_incoming_call(self, audio_file_path):
        """Main entry point for handling incoming calls"""
        try:
            # Step 1: Convert speech to text
            speech_result = self.voice_processor.process_audio(audio_file_path)
            
            if not speech_result['success']:
                return self._handle_speech_error(speech_result['error'])
            
            # Step 2: Process conversation
            user_id = self._generate_user_id()  # Generate unique user ID
            response = self.conversation_manager.process_conversation(
                user_id, speech_result['text']
            )
            
            # Step 3: Generate voice response
            audio_response = self.voice_response.generate_speech(response['text'])
            
            # Step 4: Handle actions (CRM integration, scheduling, etc.)
            if response.get('action') == 'create_lead':
                lead_data = self.conversation_manager.get_collected_info(user_id)
                crm_lead = self.crm_integration.create_lead(lead_data)
                response['crm_lead_id'] = crm_lead['id']
            
            elif response.get('action') == 'schedule':
                appointment_data = response.get('appointment_data', {})
                appointment = self.crm_integration.schedule_appointment(
                    response['crm_lead_id'], appointment_data
                )
                response['appointment_id'] = appointment['id']
            
            return {
                'success': True,
                'audio_response': audio_response,
                'response_text': response['text'],
                'next_action': response.get('action'),
                'conversation_state': self.conversation_manager.get_state(user_id)
            }
            
        except Exception as e:
            return self._handle_system_error(str(e))
    
    def _handle_speech_error(self, error):
        """Handle speech recognition errors"""
        error_responses = {
            'Could not understand audio': "I'm sorry, I didn't catch that. Could you please repeat?",
            'Speech recognition service error': "I'm having trouble hearing you. Please speak a bit louder.",
        }
        
        response_text = error_responses.get(error, "I'm sorry, there was a technical issue. Please try again.")
        audio_response = self.voice_response.generate_speech(response_text)
        
        return {
            'success': False,
            'error': error,
            'audio_response': audio_response,
            'response_text': response_text
        }
    
    def _handle_system_error(self, error):
        """Handle system errors"""
        response_text = "I'm sorry, there was a technical issue. Please call back in a few minutes."
        audio_response = self.voice_response.generate_speech(response_text)
        
        return {
            'success': False,
            'error': error,
            'audio_response': audio_response,
            'response_text': response_text
        }

# Docker configuration for scalable deployment
version: '3.8'
services:
  voice-agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - CRM_API_KEY=${CRM_API_KEY}
      - CRM_BASE_URL=${CRM_BASE_URL}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./audio_files:/app/audio_files
    depends_on:
      - redis
      - postgres
  
  redis:
    image: redis:alpine
    ports:
      - "6379:6379"
  
  postgres:
    image: postgres:13
    environment:
      - POSTGRES_DB=voice_agent
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data
 
volumes:
  postgres_data:

import logging
import time
from functools import wraps
 
class PerformanceMonitor:
    def __init__(self):
        self.logger = logging.getLogger('voice_agent')
        self.metrics = {}
    
    def track_performance(self, operation_name):
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    execution_time = time.time() - start_time
                    
                    self.logger.info(f"{operation_name} completed in {execution_time:.2f}s")
                    self._update_metrics(operation_name, execution_time, success=True)
                    
                    return result
                except Exception as e:
                    execution_time = time.time() - start_time
                    self.logger.error(f"{operation_name} failed after {execution_time:.2f}s: {e}")
                    self._update_metrics(operation_name, execution_time, success=False)
                    raise
            return wrapper
        return decorator
    
    def _update_metrics(self, operation, execution_time, success):
        if operation not in self.metrics:
            self.metrics[operation] = {
                'total_calls': 0,
                'successful_calls': 0,
                'total_time': 0,
                'average_time': 0
            }
        
        self.metrics[operation]['total_calls'] += 1
        self.metrics[operation]['total_time'] += execution_time
        self.metrics[operation]['average_time'] = (
            self.metrics[operation]['total_time'] / 
            self.metrics[operation]['total_calls']
        )
        
        if success:
            self.metrics[operation]['successful_calls'] += 1