Upload app.py

app.py

@@ -0,0 +1,702 @@
+import gradio as gr
+import asyncio
+import json
+from typing import List, Dict, Any, Optional, Tuple, Generator
+from datetime import datetime
+from enum import Enum
+import hashlib
+import logging
+from dataclasses import dataclass
+from typing import Literal
+
+# ==================== CONFIGURACIÓN DE LOGGING ESTRUCTURADO ====================
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler('chat_interface.log'),
+        logging.StreamHandler()
+    ]
+)
+logger = logging.getLogger(__name__)
+
+# ==================== TIPADO Y ESTRUCTURAS DE DATOS ====================
+@dataclass
+class ChatMessage:
+    """Estructura type-safe para mensajes de chat"""
+    role: Literal["user", "assistant", "system"]
+    content: str
+    timestamp: str
+    message_id: str
+    
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "role": self.role,
+            "content": self.content,
+            "timestamp": self.timestamp,
+            "id": self.message_id
+        }
+
+class IntentType(Enum):
+    """Tipos de intención detectados"""
+    QUERY = "consulta"
+    CODE_GENERATION = "generacion_codigo"
+    SYSTEM_DIAGNOSTIC = "diagnostico_sistema"
+    OPTIMIZATION = "optimizacion"
+    UNKNOWN = "desconocido"
+
+@dataclass
+class IntentAnalysis:
+    """Resultado del análisis de intención"""
+    intent: IntentType
+    confidence: float
+    entities: Dict[str, Any]
+    requires_streaming: bool
+
+# ==================== SERVICIO DE ANÁLISIS DE INTENCIÓN ====================
+class IntentAnalyzer:
+    """Servicio para análisis de intención con validación robusta"""
+    
+    @staticmethod
+    def analyze_intent_detailed(
+        message: str, 
+        history: List[Dict[str, Any]]
+    ) -> Tuple[IntentType, Dict[str, Any]]:
+        """
+        Analiza la intención del mensaje con validación de inputs.
+        
+        Args:
+            message: Mensaje del usuario (validado y sanitizado)
+            history: Historial de conversación en formato Gradio 4.x
+            
+        Returns:
+            Tuple[IntentType, Dict]: Tipo de intención y metadatos
+            
+        Raises:
+            ValueError: Si el mensaje está vacío o mal formado
+        """
+        # Validación de inputs
+        if not isinstance(message, str):
+            raise TypeError(f"Mensaje debe ser string, recibido: {type(message)}")
+        
+        if not message.strip():
+            raise ValueError("Mensaje no puede estar vacío")
+        
+        # Sanitización básica
+        sanitized_message = message.strip()[:2000]  # Limitación de longitud
+        
+        # Patrones de detección de intención
+        message_lower = sanitized_message.lower()
+        
+        # Diccionario de palabras clave para cada intención
+        intent_patterns = {
+            IntentType.CODE_GENERATION: [
+                "código", "programa", "función", "clase", "implementa",
+                "desarrolla", "escribe un", "crea un", "genera código",
+                "python", "javascript", "java", "typescript", "html", "css"
+            ],
+            IntentType.SYSTEM_DIAGNOSTIC: [
+                "error", "falla", "problema", "debug", "depurar",
+                "solucionar", "arreglar", "no funciona", "bug",
+                "excepción", "traceback", "stack trace"
+            ],
+            IntentType.OPTIMIZATION: [
+                "optimizar", "mejorar", "rendimiento", "eficiencia",
+                "velocidad", "lento", "bottleneck", "cuello de botella",
+                "escalar", "scaling", "performance", "memoria", "cpu"
+            ],
+            IntentType.QUERY: [
+                "qué", "cómo", "cuándo", "dónde", "por qué",
+                "explica", "describe", "información", "ayuda",
+                "tutorial", "guía", "ejemplo"
+            ]
+        }
+        
+        # Análisis de frecuencia y confianza
+        intent_scores = {intent: 0.0 for intent in IntentType}
+        
+        for intent, patterns in intent_patterns.items():
+            for pattern in patterns:
+                if pattern in message_lower:
+                    intent_scores[intent] += 1.0
+        
+        # Ponderación por longitud de palabra clave
+        word_count = len(sanitized_message.split())
+        for intent in intent_scores:
+            if intent_scores[intent] > 0:
+                intent_scores[intent] = min(1.0, intent_scores[intent] / (word_count * 0.5))
+        
+        # Determinación de intención principal
+        if max(intent_scores.values()) > 0:
+            main_intent = max(intent_scores, key=intent_scores.get)
+            confidence = intent_scores[main_intent]
+        else:
+            main_intent = IntentType.UNKNOWN
+            confidence = 0.0
+        
+        # Metadatos adicionales
+        metadata = {
+            "confidence": round(confidence, 2),
+            "message_length": len(sanitized_message),
+            "word_count": word_count,
+            "has_code_snippet": "```" in sanitized_message,
+            "timestamp": datetime.now().isoformat(),
+            "intent_scores": {k.name: v for k, v in intent_scores.items()}
+        }
+        
+        logger.info(f"Intención detectada: {main_intent.value} (confianza: {confidence:.2f})")
+        
+        return main_intent, metadata
+
+# ==================== SERVICIO DE GENERACIÓN DE RESPUESTAS ====================
+class ResponseGenerator:
+    """Servicio para generación de respuestas con streaming"""
+    
+    def __init__(self):
+        self.intent_analyzer = IntentAnalyzer()
+        self.response_templates = self._load_templates()
+    
+    def _load_templates(self) -> Dict[IntentType, str]:
+        """Carga plantillas de respuesta según intención"""
+        return {
+            IntentType.CODE_GENERATION: """
+            **🔧 Generando solución de código...**
+            
+            He analizado tu solicitud y procedo a generar una implementación robusta y lista para producción.
+            
+            **Consideraciones técnicas aplicadas:**
+            • Type hints completos para mypy
+            • Logging estructurado con niveles apropiados
+            • Manejo de errores con excepciones específicas
+            • Principios SOLID y Clean Code
+            • Validación de inputs y sanitización
+            
+            **Código generado:**
+            ```python
+            # Implementación específica basada en tu solicitud
+            """
+            ,
+            IntentType.SYSTEM_DIAGNOSTIC: """
+            **🔍 Analizando diagnóstico del sistema...**
+            
+            Procedo a analizar el problema reportado con metodología sistemática:
+            
+            1. **Identificación del root cause**
+            2. **Análisis de logs y métricas**
+            3. **Propuesta de solución escalable**
+            4. **Prevención de recurrencia**
+            
+            """
+            ,
+            IntentType.OPTIMIZATION: """
+            **⚡ Iniciando proceso de optimización...**
+            
+            Optimización aplicada con enfoque en:
+            • Complejidad algorítmica (Big O)
+            • Uso eficiente de memoria
+            • Paralelización y concurrencia
+            • Caching estratégico
+            • Minimización de I/O
+            
+            """
+            ,
+            IntentType.QUERY: """
+            **📚 Proporcionando respuesta detallada...**
+            
+            Basado en tu consulta, te proporciono información completa y verificada:
+            
+            """
+            ,
+            IntentType.UNKNOWN: """
+            **🤖 Procesando tu solicitud...**
+            
+            He recibido tu mensaje. Para brindarte la mejor asistencia:
+            
+            • ¿Podrías especificar si necesitas generación de código?
+            • ¿O prefieres una explicación técnica detallada?
+            • ¿Estás enfrentando un problema específico del sistema?
+            
+            Estoy listo para ayudarte en cualquiera de estas áreas.
+            """
+        }
+    
+    async def generate_streaming_response(
+        self,
+        message: str,
+        history: List[Dict[str, Any]]
+    ) -> Generator[Tuple[List[Dict[str, Any]], str], None, None]:
+        """
+        Genera respuesta con streaming en tiempo real.
+        
+        Args:
+            message: Mensaje del usuario
+            history: Historial de conversación
+            
+        Yields:
+            Tupla[List[Dict], str]: Historial actualizado y texto de modo
+        """
+        try:
+            # 1. Análisis de intención
+            intent, metadata = self.intent_analyzer.analyze_intent_detailed(message, history)
+            mode_text = f"**Modo activo**: {intent.value.replace('_', ' ').title()}"
+            
+            # 2. Crear historial temporal para streaming
+            temp_history = history.copy()
+            
+            # 3. Añadir mensaje del usuario al historial
+            user_message = {
+                "role": "user",
+                "content": message,
+                "timestamp": datetime.now().isoformat()
+            }
+            temp_history.append(user_message)
+            
+            # 4. Añadir placeholder de asistente
+            assistant_placeholder = {
+                "role": "assistant",
+                "content": "",
+                "timestamp": datetime.now().isoformat()
+            }
+            temp_history.append(assistant_placeholder)
+            
+            yield temp_history, mode_text
+            
+            # 5. Generar respuesta con streaming simulado
+            template = self.response_templates.get(intent, self.response_templates[IntentType.UNKNOWN])
+            
+            # Streaming chunk por chunk
+            chunks = self._split_into_chunks(template)
+            accumulated_response = ""
+            
+            for chunk in chunks:
+                await asyncio.sleep(0.05)  # Simular procesamiento
+                accumulated_response += chunk
+                
+                # Actualizar contenido del asistente en el historial
+                temp_history[-1]["content"] = accumulated_response
+                yield temp_history, mode_text
+            
+            # 6. Añadir sección específica basada en el mensaje
+            specific_response = f"\n\n**📝 Detalles específicos para tu solicitud:**\n\n"
+            specific_response += f"• **Mensaje analizado:** `{message[:100]}{'...' if len(message) > 100 else ''}`\n"
+            specific_response += f"• **Confianza de intención:** {metadata['confidence'] * 100:.1f}%\n"
+            specific_response += f"• **Longitud del mensaje:** {metadata['message_length']} caracteres\n"
+            
+            chunks_specific = self._split_into_chunks(specific_response)
+            for chunk in chunks_specific:
+                await asyncio.sleep(0.03)
+                accumulated_response += chunk
+                temp_history[-1]["content"] = accumulated_response
+                yield temp_history, mode_text
+            
+            # 7. Añadir marca de finalización
+            completion_marker = "\n\n---\n*Respuesta generada con SISTEMA_PROTOCOLO_POTENCIACION_ACTIVO v2.0*"
+            temp_history[-1]["content"] = accumulated_response + completion_marker
+            
+            yield temp_history, mode_text
+            
+        except Exception as e:
+            logger.error(f"Error en generación de respuesta: {str(e)}", exc_info=True)
+            
+            # Respuesta de error controlada
+            error_history = history.copy()
+            error_history.append({
+                "role": "user",
+                "content": message,
+                "timestamp": datetime.now().isoformat()
+            })
+            error_history.append({
+                "role": "assistant",
+                "content": f"**⚠️ Error en el procesamiento:**\n\n{str(e)}\n\nPor favor, intenta nuevamente o reformula tu solicitud.",
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            yield error_history, "**Modo activo**: Manejo de errores"
+    
+    def _split_into_chunks(self, text: str, chunk_size: int = 20) -> List[str]:
+        """Divide texto en chunks para streaming"""
+        words = text.split()
+        chunks = []
+        current_chunk = []
+        current_length = 0
+        
+        for word in words:
+            if current_length + len(word) + 1 > chunk_size and current_chunk:
+                chunks.append(" ".join(current_chunk) + " ")
+                current_chunk = [word]
+                current_length = len(word)
+            else:
+                current_chunk.append(word)
+                current_length += len(word) + 1
+        
+        if current_chunk:
+            chunks.append(" ".join(current_chunk))
+        
+        return chunks
+
+# ==================== INTERFAZ GRADIO 4.x COMPATIBLE ====================
+def create_advanced_interface():
+    """
+    Crea interfaz avanzada de chat compatible con Gradio 4.x.
+    Implementa formato correcto de mensajes: {"role": "rol", "content": "texto"}
+    """
+    
+    # Instanciar servicios
+    intent_analyzer = IntentAnalyzer()
+    response_generator = ResponseGenerator()
+    
+    # Cache para historiales de sesión
+    session_histories = {}
+    
+    def get_session_history(session_id: str) -> List[Dict[str, Any]]:
+        """Obtiene o crea historial para sesión"""
+        if session_id not in session_histories:
+            # Mensaje inicial del sistema
+            session_histories[session_id] = [{
+                "role": "system",
+                "content": "Eres un asistente especializado en ingeniería de software, DevOps y arquitectura de sistemas distribuidos. Proporcionas soluciones técnicas precisas y listas para producción.",
+                "timestamp": datetime.now().isoformat()
+            }]
+        return session_histories[session_id]
+    
+    def generate_session_id() -> str:
+        """Genera ID único de sesión"""
+        timestamp = datetime.now().isoformat()
+        hash_obj = hashlib.md5(timestamp.encode())
+        return hash_obj.hexdigest()[:12]
+    
+    # ==================== FUNCIONES DE INTERACCIÓN ====================
+    def process_message(
+        message: str, 
+        history: List[Dict[str, Any]], 
+        session_id: str
+    ) -> Tuple[str, List[Dict[str, Any]], str]:
+        """
+        Procesa mensaje del usuario - FORMATO GRADIO 4.x CORRECTO
+        
+        Args:
+            message: Mensaje del usuario
+            history: Historial en formato {"role": "rol", "content": "texto"}
+            session_id: ID de sesión para persistencia
+            
+        Returns:
+            Tuple[str, List[Dict], str]: Mensaje vacío, historial actualizado, texto de modo
+        """
+        # Validación robusta
+        if not message or not isinstance(message, str):
+            return "", history, "**Modo**: Error - Mensaje inválido"
+        
+        sanitized_message = message.strip()
+        if not sanitized_message:
+            return "", history, "**Modo**: Esperando entrada válida..."
+        
+        # Obtener historial de sesión
+        session_history = get_session_history(session_id)
+        
+        # Crear mensaje de usuario en formato Gradio 4.x
+        user_message = {
+            "role": "user",
+            "content": sanitized_message,
+            "timestamp": datetime.now().isoformat()
+        }
+        
+        # Actualizar historiales
+        new_history = history + [user_message] if history else [user_message]
+        session_history.append(user_message)
+        
+        try:
+            # Análisis de intención
+            intent, metadata = intent_analyzer.analyze_intent_detailed(
+                sanitized_message, 
+                new_history
+            )
+            
+            mode_text = (
+                f"**Modo activo**: {intent.value.replace('_', ' ').title()}\n"
+                f"**Confianza**: {metadata['confidence'] * 100:.1f}%\n"
+                f"**Tokens**: {metadata['word_count']}"
+            )
+            
+        except Exception as e:
+            logger.warning(f"Error en análisis de intención: {e}")
+            mode_text = "**Modo activo**: Procesamiento directo"
+        
+        # Limpiar input
+        return "", new_history, mode_text
+    
+    async def generate_response_stream(
+        message: str,
+        history: List[Dict[str, Any]],
+        session_id: str
+    ):
+        """
+        Genera respuesta con streaming - FORMATO GRADIO 4.x CORRECTO
+        
+        Args:
+            message: Mensaje del usuario
+            history: Historial en formato correcto
+            session_id: ID de sesión
+            
+        Yields:
+            Historial actualizado y texto de modo
+        """
+        async for updated_history, mode_text in response_generator.generate_streaming_response(
+            message, 
+            history
+        ):
+            # Actualizar historial de sesión
+            session_histories[session_id] = updated_history.copy()
+            yield updated_history, mode_text
+    
+    # ==================== INTERFAZ GRADIO ====================
+    with gr.Blocks(
+        title="🧠 SISTEMA_PROTOCOLO_POTENCIACION_ACTIVO v2.0",
+        theme=gr.themes.Soft(),
+        css="""
+        .gradio-container {
+            max-width: 1200px !important;
+            margin: 0 auto !important;
+        }
+        .chatbot {
+            height: 600px !important;
+            overflow-y: auto !important;
+            border: 1px solid #e0e0e0 !important;
+            border-radius: 10px !important;
+            padding: 20px !important;
+        }
+        .status-panel {
+            background: linear-gradient(135deg, #667eea 0%, #764ba2 100%) !important;
+            color: white !important;
+            padding: 15px !important;
+            border-radius: 10px !important;
+            margin-bottom: 20px !important;
+        }
+        """
+    ) as interface:
+        
+        # Estado de sesión
+        session_state = gr.State(value=generate_session_id())
+        
+        gr.Markdown("""
+        # 🧠 SISTEMA_PROTOCOLO_POTENCIACION_ACTIVO v2.0
+        ### *Asistente de IA Senior en Ingeniería de Software y DevOps*
+        
+        ---
+        
+        **Especialidades:** 
+        • Arquitectura de Sistemas Distribuidos • Microservicios • DevOps • Cloud Security
+        • Optimización de Rendimiento • Code Review Profesional • Soluciones Production-Ready
+        
+        **Modo actual:** `OPTIMIZACION_ASCENDENTE_EXCLUSIVA`
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=3):
+                # Chatbot con formato correcto para Gradio 4.x
+                chatbot = gr.Chatbot(
+                    label="Conversación",
+                    height=600,
+                    show_label=False,
+                    bubble_full_width=False,
+                    avatar_images=(
+                        "https://cdn-icons-png.flaticon.com/512/3135/3135715.png",  # User
+                        "https://cdn-icons-png.flaticon.com/512/4712/4712035.png"   # Assistant
+                    ),
+                    sanitize_html=False,
+                    show_copy_button=True
+                )
+                
+                # Panel de estado
+                status_display = gr.Markdown(
+                    "**Modo**: Inicializando sistema...",
+                    elem_classes="status-panel"
+                )
+                
+                # Input con validación
+                msg_input = gr.Textbox(
+                    label="Escribe tu mensaje técnico",
+                    placeholder="Ejemplo: 'Optimiza este microservicio Python para alta concurrencia...'",
+                    lines=3,
+                    max_lines=6,
+                    show_label=False
+                )
+                
+                with gr.Row():
+                    submit_btn = gr.Button(
+                        "🚀 Enviar Consulta Técnica",
+                        variant="primary",
+                        scale=2
+                    )
+                    clear_btn = gr.Button(
+                        "🧹 Limpiar Chat",
+                        variant="secondary",
+                        scale=1
+                    )
+            
+            with gr.Column(scale=1):
+                # Panel de información técnica
+                gr.Markdown("### 📊 Métricas del Sistema")
+                
+                session_id_display = gr.Textbox(
+                    label="ID de Sesión",
+                    value=session_state.value,
+                    interactive=False
+                )
+                
+                message_count = gr.Number(
+                    label="Mensajes en sesión",
+                    value=0,
+                    interactive=False
+                )
+                
+                gr.Markdown("""
+                ### ⚙️ Configuración
+                
+                - **Formato:** Gradio 4.x Compatible
+                - **Streaming:** Activado
+                - **Persistencia:** Por sesión
+                - **Logging:** Estructurado
+                - **Seguridad:** Input Sanitization
+                """)
+        
+        # ==================== EVENT HANDLERS ====================
+        
+        def update_message_count(history):
+            """Actualiza contador de mensajes"""
+            user_messages = [m for m in history if m.get("role") == "user"]
+            return len(user_messages)
+        
+        # Submit handler
+        submit_event = msg_input.submit(
+            fn=process_message,
+            inputs=[msg_input, chatbot, session_state],
+            outputs=[msg_input, chatbot, status_display],
+            queue=True
+        ).then(
+            fn=update_message_count,
+            inputs=[chatbot],
+            outputs=[message_count]
+        ).then(
+            fn=generate_response_stream,
+            inputs=[msg_input, chatbot, session_state],
+            outputs=[chatbot, status_display]
+        )
+        
+        # Button click handler
+        submit_btn.click(
+            fn=process_message,
+            inputs=[msg_input, chatbot, session_state],
+            outputs=[msg_input, chatbot, status_display],
+            queue=True
+        ).then(
+            fn=update_message_count,
+            inputs=[chatbot],
+            outputs=[message_count]
+        ).then(
+            fn=generate_response_stream,
+            inputs=[msg_input, chatbot, session_state],
+            outputs=[chatbot, status_display]
+        )
+        
+        # Clear handler
+        def clear_chat():
+            """Limpia el chat y genera nueva sesión"""
+            new_session_id = generate_session_id()
+            initial_message = [{
+                "role": "system",
+                "content": "Sesión reiniciada. Listo para asistencia técnica.",
+                "timestamp": datetime.now().isoformat()
+            }]
+            return (
+                initial_message,  # chatbot
+                "**Modo**: Sistema reiniciado - Esperando consulta técnica",  # status
+                new_session_id,  # session_state
+                0,  # message_count
+                new_session_id  # session_id_display
+            )
+        
+        clear_btn.click(
+            fn=clear_chat,
+            outputs=[chatbot, status_display, session_state, message_count, session_id_display]
+        )
+        
+        # ==================== CONFIGURACIÓN ADICIONAL ====================
+        
+        # Ejemplos de consultas
+        gr.Examples(
+            examples=[
+                ["Optimiza este endpoint REST para manejar 10K RPS con Python async/await"],
+                ["Diseña una arquitectura de microservicios con auto-scaling y circuit breaker"],
+                ["Implementa un sistema de logging distribuido con OpenTelemetry y Grafana"],
+                ["Genera un Dockerfile production-ready para aplicación Python con seguridad hardening"],
+                ["Crea un sistema de CI/CD con GitHub Actions para despliegue blue-green en Kubernetes"]
+            ],
+            inputs=msg_input,
+            label="📋 Ejemplos de consultas técnicas"
+        )
+        
+        # Footer
+        gr.Markdown("""
+        ---
+        
+        **🔒 Seguridad activa:** Todas las consultas son procesadas con sanitización de inputs y validación de tipos.
+        **📈 Logging:** Todas las interacciones son registradas para mejora continua del sistema.
+        **⚡ Performance:** Respuestas optimizadas con streaming en tiempo real.
+        
+        *Sistema compatible con Gradio 4.x - Formato de mensajes: `{"role": "rol", "content": "texto"}`*
+        """)
+    
+    return interface
+
+# ==================== PUNTO DE ENTRADA ====================
+if __name__ == "__main__":
+    # Validación de entorno
+    try:
+        import gradio as gr
+        GRADIO_VERSION = gr.__version__
+        logger.info(f"Gradio versión detectada: {GRADIO_VERSION}")
+        
+        if not GRADIO_VERSION.startswith("4."):
+            logger.warning(
+                f"Esta interfaz está optimizada para Gradio 4.x. "
+                f"Versión actual: {GRADIO_VERSION}. "
+                f"Puede haber incompatibilidades."
+            )
+        
+        # Crear e iniciar interfaz
+        interface = create_advanced_interface()
+        
+        # Configuración para Hugging Face Spaces
+        interface.queue(
+            max_size=20,
+            default_concurrency_limit=10,
+            api_open=True
+        ).launch(
+            server_name="0.0.0.0" if "SPACE_ID" in os.environ else None,
+            server_port=7860,
+            share=False,
+            debug=False,
+            show_error=True,
+            favicon_path="https://cdn-icons-png.flaticon.com/512/4712/4712035.png"
+        )
+        
+    except Exception as e:
+        logger.critical(f"Error crítico al iniciar la aplicación: {str(e)}", exc_info=True)
+        
+        # Interfaz de fallback mínima
+        with gr.Blocks() as fallback:
+            gr.Markdown(f"""
+            # ⚠️ Error en la inicialización del sistema
+            
+            **Detalles técnicos:**
+            ```python
+            {str(e)}
+            ```
+            
+            Por favor, verifica:
+            1. Dependencias de Gradio 4.x instaladas
+            2. Formato de mensajes correcto
+            3. Logs del sistema para diagnóstico
+            """)
+        
+        fallback.launch()