Ejemplo: Chatbot de Atención al Cliente con Python

import openai
import json
import datetime
from typing import Dict, List, Optional
import sqlite3
from dataclasses import dataclass
import logging

@dataclass
class CustomerProfile:
    """Perfil completo del cliente para personalización"""
    customer_id: str
    name: str
    email: str
    phone: str
    purchase_history: List[Dict]
    preferences: Dict
    interaction_history: List[Dict]
    sentiment_score: float
    priority_level: str

class IntelligentCustomerServiceBot:
    """Sistema completo de chatbot para atención al cliente"""
    
    def __init__(self, openai_api_key: str, db_path: str = "customer_service.db"):
        openai.api_key = openai_api_key
        self.db_path = db_path
        self.setup_database()
        self.setup_knowledge_base()
        self.setup_logging()
        
    def setup_database(self):
        """Configura base de datos para historial y perfiles de clientes"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Tabla de conversaciones
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS conversations (
                id INTEGER PRIMARY KEY,
                customer_id TEXT,
                timestamp DATETIME,
                channel TEXT,
                message TEXT,
                bot_response TEXT,
                sentiment REAL,
                resolved BOOLEAN,
                escalated BOOLEAN
            )
        ''')
        
        # Tabla de perfiles de cliente
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS customer_profiles (
                customer_id TEXT PRIMARY KEY,
                name TEXT,
                email TEXT,
                phone TEXT,
                profile_data TEXT,
                last_updated DATETIME
            )
        ''')
        
        # Tabla de knowledge base
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS knowledge_base (
                id INTEGER PRIMARY KEY,
                category TEXT,
                question TEXT,
                answer TEXT,
                keywords TEXT,
                usage_count INTEGER DEFAULT 0,
                last_updated DATETIME
            )
        ''')
        
        conn.commit()
        conn.close()
        
    def setup_knowledge_base(self):
        """Inicializa knowledge base con FAQs comunes"""
        common_faqs = [
            {
                'category': 'shipping',
                'question': 'tracking de pedido',
                'answer': 'Para consultar el estado de tu pedido, necesito tu número de orden. Una vez que lo proporciones, te daré información detallada sobre la ubicación actual y fecha estimada de entrega.',
                'keywords': 'pedido, tracking, seguimiento, envío, orden'
            },
            {
                'category': 'returns',
                'question': 'política de devoluciones',
                'answer': 'Aceptamos devoluciones dentro de los 30 días posteriores a la compra. El producto debe estar en condiciones originales. El proceso es simple: inicia la devolución online, imprime la etiqueta prepagada y envía el paquete.',
                'keywords': 'devolución, return, cambio, reembolso, política'
            },
            {
                'category': 'payment',
                'question': 'métodos de pago',
                'answer': 'Aceptamos tarjetas de crédito/débito (Visa, MasterCard, AmEx), PayPal, transferencias bancarias y pagos en efectivo contra entrega en áreas metropolitanas.',
                'keywords': 'pago, tarjeta, paypal, transferencia, efectivo'
            }
        ]
        
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        for faq in common_faqs:
            cursor.execute('''
                INSERT OR REPLACE INTO knowledge_base 
                (category, question, answer, keywords, last_updated)
                VALUES (?, ?, ?, ?, ?)
            ''', (faq['category'], faq['question'], faq['answer'], 
                  faq['keywords'], datetime.datetime.now()))
        
        conn.commit()
        conn.close()
        
    def setup_logging(self):
        """Configura logging para monitoreo y análisis"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('customer_service_bot.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
        
    def get_customer_profile(self, customer_id: str) -> Optional[CustomerProfile]:
        """Recupera perfil completo del cliente desde la base de datos"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT profile_data FROM customer_profiles WHERE customer_id = ?
        ''', (customer_id,))
        
        result = cursor.fetchone()
        conn.close()
        
        if result:
            profile_data = json.loads(result[0])
            return CustomerProfile(**profile_data)
        
        return None
    
    def analyze_sentiment(self, message: str) -> Dict[str, float]:
        """Analiza el sentimiento del mensaje del cliente"""
        try:
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[
                    {
                        "role": "system",
                        "content": """Analiza el sentimiento de este mensaje de cliente.
                        Devuelve solo un JSON con:
                        {
                            "sentiment": float(-1.0 a 1.0, donde -1=muy negativo, 0=neutral, 1=muy positivo),
                            "emotion": "string" (frustrated, happy, neutral, angry, confused, satisfied),
                            "urgency": float(0.0 a 1.0, donde 1.0=máxima urgencia),
                            "escalation_needed": boolean
                        }"""
                    },
                    {"role": "user", "content": message}
                ],
                max_tokens=150,
                temperature=0.3
            )
            
            return json.loads(response.choices[0].message.content)
            
        except Exception as e:
            self.logger.error(f"Error en análisis de sentimiento: {e}")
            return {
                "sentiment": 0.0,
                "emotion": "neutral", 
                "urgency": 0.5,
                "escalation_needed": False
            }
    
    def search_knowledge_base(self, query: str, limit: int = 3) -> List[Dict]:
        """Busca respuestas relevantes en la knowledge base"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Búsqueda por keywords (simulada - en producción usarías embeddings)
        query_words = query.lower().split()
        
        cursor.execute('''
            SELECT category, question, answer, usage_count 
            FROM knowledge_base 
            ORDER BY usage_count DESC
            LIMIT ?
        ''', (limit * 2,))  # Obtenemos más para filtrar
        
        results = cursor.fetchall()
        conn.close()
        
        # Filtrado básico por relevancia de keywords
        relevant_results = []
        for result in results:
            keywords = result[2] if len(result) > 3 else ""
            relevance_score = sum(1 for word in query_words if word in keywords.lower())
            
            if relevance_score > 0 or len(relevant_results) < limit:
                relevant_results.append({
                    'category': result[0],
                    'question': result[1],
                    'answer': result[2],
                    'usage_count': result[3],
                    'relevance': relevance_score
                })
        
        # Ordenar por relevancia y límite final
        relevant_results.sort(key=lambda x: (x['relevance'], x['usage_count']), reverse=True)
        return relevant_results[:limit]
    
    def generate_personalized_response(self, customer_message: str, 
                                     customer_profile: Optional[CustomerProfile],
                                     sentiment_analysis: Dict,
                                     knowledge_results: List[Dict]) -> str:
        """Genera respuesta personalizada usando IA"""
        
        # Construir contexto para el prompt
        context_parts = []
        
        if customer_profile:
            context_parts.append(f"Cliente: {customer_profile.name}")
            if customer_profile.purchase_history:
                last_purchase = customer_profile.purchase_history[-1]
                context_parts.append(f"Última compra: {last_purchase.get('product', 'N/A')}")
            
            if customer_profile.preferences:
                prefs = ", ".join(customer_profile.preferences.keys())
                context_parts.append(f"Preferencias: {prefs}")
        
        if sentiment_analysis['emotion'] != 'neutral':
            context_parts.append(f"Estado emocional detectado: {sentiment_analysis['emotion']}")
            if sentiment_analysis['urgency'] > 0.7:
                context_parts.append("NOTA: Cliente muestra alta urgencia")
        
        knowledge_context = ""
        if knowledge_results:
            knowledge_context = "\n".join([
                f"FAQ relevante: {kb['answer']}" for kb in knowledge_results[:2]
            ])
        
        context = "\n".join(context_parts)
        
        try:
            system_prompt = f"""Eres un asistente de atención al cliente experto y empático. 

CONTEXTO DEL CLIENTE:
{context}

INFORMACIÓN RELEVANTE DE LA BASE DE CONOCIMIENTOS:
{knowledge_context}

INSTRUCCIONES:
1. Responde de manera personalizada y empática
2. Si detectas frustración, reconócela y ofrece disculpas apropiadas
3. Sé específico y actionable en tus respuestas
4. Si no tienes información suficiente, pregunta de manera inteligente
5. Si es necesario escalamiento, indícalo claramente
6. Mantén un tono profesional pero cercano
7. Ofrece múltiples opciones cuando sea posible

Responde en español, de forma clara y concisa."""

            response = openai.ChatCompletion.create(
                model="gpt-4",
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": customer_message}
                ],
                max_tokens=500,
                temperature=0.7
            )
            
            return response.choices[0].message.content.strip()
            
        except Exception as e:
            self.logger.error(f"Error generando respuesta: {e}")
            return "Disculpa, estoy experimentando dificultades técnicas. Un agente humano te ayudará en breve."
    
    def should_escalate(self, sentiment_analysis: Dict, message: str, 
                       interaction_count: int) -> bool:
        """Determina si la conversación debe escalarse a un humano"""
        escalation_triggers = [
            sentiment_analysis['escalation_needed'],
            sentiment_analysis['urgency'] > 0.8,
            sentiment_analysis['sentiment'] < -0.7,
            interaction_count > 3,  # Más de 3 intercambios sin resolución
            any(word in message.lower() for word in [
                'supervisor', 'manager', 'complaint', 'legal', 'cancel',
                'supervisor', 'gerente', 'queja', 'legal', 'cancelar'
            ])
        ]
        
        return any(escalation_triggers)
    
    def log_interaction(self, customer_id: str, message: str, response: str,
                       channel: str, sentiment: Dict, resolved: bool = False,
                       escalated: bool = False):
        """Registra la interacción en la base de datos"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO conversations 
            (customer_id, timestamp, channel, message, bot_response, 
             sentiment, resolved, escalated)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ''', (customer_id, datetime.datetime.now(), channel, message, 
              response, sentiment['sentiment'], resolved, escalated))
        
        conn.commit()
        conn.close()
        
    def process_customer_message(self, customer_id: str, message: str, 
                               channel: str = "web") -> Dict:
        """Procesa mensaje completo del cliente y genera respuesta"""
        
        # 1. Obtener perfil del cliente
        customer_profile = self.get_customer_profile(customer_id)
        
        # 2. Analizar sentimiento
        sentiment_analysis = self.analyze_sentiment(message)
        
        # 3. Buscar en knowledge base
        knowledge_results = self.search_knowledge_base(message)
        
        # 4. Contar interacciones recientes
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            SELECT COUNT(*) FROM conversations 
            WHERE customer_id = ? AND timestamp > datetime('now', '-1 hour')
        ''', (customer_id,))
        interaction_count = cursor.fetchone()[0]
        conn.close()
        
        # 5. Determinar si necesita escalamiento
        needs_escalation = self.should_escalate(sentiment_analysis, message, interaction_count)
        
        if needs_escalation:
            response = """Entiendo que necesitas atención especializada. He notificado a nuestro equipo y un agente experto te contactará en los próximos 5 minutos. 
            
Mientras tanto, he documentado toda nuestra conversación para que el agente tenga contexto completo de tu situación."""
            
            # Notificar al sistema de escalamiento (implementar según infraestructura)
            self.logger.info(f"Escalamiento requerido para cliente {customer_id}")
            
        else:
            # 6. Generar respuesta personalizada
            response = self.generate_personalized_response(
                message, customer_profile, sentiment_analysis, knowledge_results
            )
        
        # 7. Registrar interacción
        self.log_interaction(
            customer_id, message, response, channel, 
            sentiment_analysis, resolved=not needs_escalation,
            escalated=needs_escalation
        )
        
        # 8. Actualizar usage_count en knowledge base si se usó
        if knowledge_results and not needs_escalation:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            for kb in knowledge_results[:1]:  # Solo el más relevante
                cursor.execute('''
                    UPDATE knowledge_base SET usage_count = usage_count + 1
                    WHERE question = ?
                ''', (kb['question'],))
            conn.commit()
            conn.close()
        
        return {
            'response': response,
            'sentiment_analysis': sentiment_analysis,
            'escalated': needs_escalation,
            'channel': channel,
            'timestamp': datetime.datetime.now().isoformat(),
            'customer_profile_available': customer_profile is not None
        }
    
    def generate_analytics_report(self, days: int = 7) -> Dict:
        """Genera reporte de análisis del chatbot"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        # Métricas básicas
        cursor.execute('''
            SELECT 
                COUNT(*) as total_conversations,
                AVG(sentiment) as avg_sentiment,
                COUNT(CASE WHEN resolved = 1 THEN 1 END) as resolved_count,
                COUNT(CASE WHEN escalated = 1 THEN 1 END) as escalated_count
            FROM conversations 
            WHERE timestamp > datetime('now', '-{} days')
        '''.format(days))
        
        metrics = cursor.fetchone()
        
        # Top categorías de consultas
        cursor.execute('''
            SELECT kb.category, COUNT(*) as usage_count
            FROM knowledge_base kb
            WHERE kb.usage_count > 0
            ORDER BY kb.usage_count DESC
            LIMIT 5
        ''')
        
        top_categories = cursor.fetchall()
        
        conn.close()
        
        total_conversations = metrics[0]
        resolution_rate = (metrics[2] / total_conversations * 100) if total_conversations > 0 else 0
        escalation_rate = (metrics[3] / total_conversations * 100) if total_conversations > 0 else 0
        
        return {
            'period_days': days,
            'total_conversations': total_conversations,
            'avg_sentiment': round(metrics[1] if metrics[1] else 0, 2),
            'resolution_rate': round(resolution_rate, 1),
            'escalation_rate': round(escalation_rate, 1),
            'customer_satisfaction_score': round((metrics[1] + 1) * 50, 1),  # Convertir a 0-100
            'top_inquiry_categories': [
                {'category': cat[0], 'count': cat[1]} for cat in top_categories
            ],
            'recommendations': self._generate_recommendations(resolution_rate, escalation_rate, metrics[1])
        }
    
    def _generate_recommendations(self, resolution_rate: float, 
                                escalation_rate: float, avg_sentiment: float) -> List[str]:
        """Genera recomendaciones basadas en métricas"""
        recommendations = []
        
        if resolution_rate < 80:
            recommendations.append("Expandir knowledge base con FAQs más específicas")
            
        if escalation_rate > 15:
            recommendations.append("Revisar triggers de escalamiento - posible sobre-escalación")
            
        if avg_sentiment < -0.2:
            recommendations.append("Mejorar detección emocional y respuestas empáticas")
            
        if not recommendations:
            recommendations.append("Rendimiento excelente - considerar nuevas funcionalidades")
            
        return recommendations

# Ejemplo de uso
def main():
    # Configuración
    OPENAI_API_KEY = "tu_openai_api_key"
    
    # Inicializar bot
    bot = IntelligentCustomerServiceBot(OPENAI_API_KEY)
    
    # Simular conversación
    customer_id = "customer_12345"
    
    # Ejemplo 1: Consulta sobre pedido
    message1 = "Hola, no he recibido mi pedido y estoy muy preocupado"
    result1 = bot.process_customer_message(customer_id, message1, "web")
    
    print("="*50)
    print("CONVERSACIÓN CON CHATBOT INTELIGENTE")
    print("="*50)
    print(f"Cliente: {message1}")
    print(f"Bot: {result1['response']}")
    print(f"Sentimiento: {result1['sentiment_analysis']['emotion']}")
    print(f"Escalado: {'Sí' if result1['escalated'] else 'No'}")
    
    # Ejemplo 2: Seguimiento
    if not result1['escalated']:
        message2 = "Mi número de pedido es ORD-789456"
        result2 = bot.process_customer_message(customer_id, message2, "web")
        
        print(f"\nCliente: {message2}")
        print(f"Bot: {result2['response']}")
    
    # Generar reporte de análisis
    analytics = bot.generate_analytics_report(7)
    
    print("\n" + "="*50)
    print("REPORTE DE ANÁLISIS (7 DÍAS)")
    print("="*50)
    print(f"Total conversaciones: {analytics['total_conversations']}")
    print(f"Tasa de resolución: {analytics['resolution_rate']}%")
    print(f"Tasa de escalamiento: {analytics['escalation_rate']}%")
    print(f"Satisfacción del cliente: {analytics['customer_satisfaction_score']}/100")
    
    print("\nRecomendaciones:")
    for rec in analytics['recommendations']:
        print(f"• {rec}")

if __name__ == "__main__":
    main()