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()