Delate app.py

app.py

@@ -1,535 +0,0 @@
-"""
-BATUTO X ENTERPRISE - SISTEMA UNIFICADO
-- Orquestación inteligente de modelos
-- Generación multimodal
-- Interfaz profesional
-"""
-import os
-import time
-import json
-import asyncio
-from typing import Dict, List, Tuple, Any, Optional, Generator
-from datetime import datetime
-from pathlib import Path
-
-import gradio as gr
-from PIL import Image
-
-# Importar módulos optimizados
-from cognitive_kernel import (
-    IntentType,
-    analyze_intent_detailed,
-    select_model_with_context,
-    record_complete_interaction,
-    MEMORY,
-    ORCHESTRATOR
-)
-from core_llm import (
-    ENTERPRISE_MODELS,
-    create_sambanova_client,
-    stream_response_generator,
-    SambanovaClient
-)
-from image_tool_reve import (
-    generar_imagen_reve,
-    generar_imagen_con_metadatos,
-    AsyncReveGenerator
-)
-
-# ===================== CONFIGURACIÓN GLOBAL =====================
-
-import os
-
-API_KEY = os.getenv("SAMBANOVA_API_KEY_2")
-# Variable de estado en lugar de un error fatal
-SAMBANOVA_AVAILABLE = API_KEY is not None and API_KEY != ""
-
-# Si quieres, puedes mostrar una advertencia en los logs, pero no detener la app
-if not SAMBANOVA_AVAILABLE:
-    print("⚠️  ADVERTENCIA: SAMBANOVA_API_KEY_2 no está configurada.")
-    print("   La aplicación se iniciará, pero las funciones de texto no funcionarán.")
-    print("   Configúrala en: Settings -> Repository secrets")
-REVE_KEY = os.environ.get("REVE_API_KEY")
-if not REVE_KEY:
-    print("⚠️ ADVERTENCIA: REVE_API_KEY no configurada. Las imágenes no funcionarán.")
-
-OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "batuto_generations"))
-OUTPUT_DIR.mkdir(exist_ok=True)
-
-# ===================== CANDIDATOS POR INTENCIÓN =====================
-CANDIDATES = {
-    IntentType.IMAGE: {
-        "vision": ENTERPRISE_MODELS["Vision"]["id"]
-    },
-    IntentType.CODE: {
-        "coder": ENTERPRISE_MODELS["Coder"]["id"],
-        "heavy": ENTERPRISE_MODELS["Heavy"]["id"]
-    },
-    IntentType.REASONING: {
-        "reasoner": ENTERPRISE_MODELS["Reasoner"]["id"],
-        "heavy": ENTERPRISE_MODELS["Heavy"]["id"],
-        "scout": ENTERPRISE_MODELS["Scout"]["id"]
-    },
-    IntentType.ARCHITECTURE: {
-        "heavy": ENTERPRISE_MODELS["Heavy"]["id"],
-        "coder": ENTERPRISE_MODELS["Coder"]["id"]
-    },
-    IntentType.DEVOPS: {
-        "heavy": ENTERPRISE_MODELS["Heavy"]["id"],
-        "scout": ENTERPRISE_MODELS["Scout"]["id"]
-    },
-    IntentType.QA: {
-        "coder": ENTERPRISE_MODELS["Coder"]["id"],
-        "scout": ENTERPRISE_MODELS["Scout"]["id"]
-    },
-    IntentType.VISUAL: {
-        "vision": ENTERPRISE_MODELS["Vision"]["id"]
-    },
-    IntentType.GENERAL: {
-        "scout": ENTERPRISE_MODELS["Scout"]["id"],
-        "fast": ENTERPRISE_MODELS["Fast"]["id"],
-        "heavy": ENTERPRISE_MODELS["Heavy"]["id"]
-    }
-}
-
-# ===================== SYSTEM PROMPTS ESPECIALIZADOS =====================
-SYSTEM_PROMPTS = {
-    IntentType.IMAGE: """Eres especialista en ingeniería de prompts visuales.
-Analiza descripciones y sugiere mejoras para generación de imágenes.
-Sé específico sobre iluminación, composición y estilo.""",
-    
-    IntentType.CODE: """Eres ingeniero de software senior.
-Proporciona código limpio, eficiente y documentado.
-Explica tus decisiones técnicas.""",
-    
-    IntentType.REASONING: """Eres especialista en razonamiento analítico.
-Descompón problemas paso a paso, fundamenta tus conclusiones.""",
-    
-    IntentType.ARCHITECTURE: """Eres arquitecto de sistemas.
-Diseña soluciones escalables, mantenibles y seguras.
-Incluye diagramas conceptuales y decisiones técnicas.""",
-    
-    IntentType.DEVOPS: """Eres ingeniero DevOps senior.
-Automatiza, asegura y optimiza infraestructura.
-Prioriza prácticas cloud-native y GitOps.""",
-    
-    IntentType.QA: """Eres ingeniero de calidad automatizado.
-Diseña suites de pruebas robustas y mantenibles.
-Incluye métricas y estrategias de testing.""",
-    
-    IntentType.VISUAL: """Eres analista visual especializado.
-Describe imágenes con precisión técnica y artística.
-Proporciona contexto y sugerencias de mejora.""",
-    
-    IntentType.GENERAL: """Eres BATUTO X, asistente AI enterprise.
-Sé preciso, técnico y creativo cuando corresponda.
-Responde en español con claridad profesional."""
-}
-
-# ===================== ORQUESTADOR PRINCIPAL =====================
-class BatutoOrchestrator:
-    def __init__(self):
-        self.client = create_sambanova_client(API_KEY)
-        self.chat_history = []
-        self.session_id = f"session_{int(time.time())}"
-        
-    def prepare_messages(self, text: str, intent: IntentType, 
-                        context: List[Tuple] = None) -> List[Dict]:
-        """Prepara mensajes para el modelo"""
-        system_prompt = SYSTEM_PROMPTS.get(intent, SYSTEM_PROMPTS[IntentType.GENERAL])
-        
-        messages = [
-            {"role": "system", "content": system_prompt}
-        ]
-        
-        # Agregar contexto histórico (últimos 5 intercambios)
-        if context:
-            for user_msg, assistant_msg in context[-5:]:
-                if user_msg:
-                    messages.append({"role": "user", "content": user_msg})
-                if assistant_msg:
-                    messages.append({"role": "assistant", "content": assistant_msg})
-        
-        messages.append({"role": "user", "content": text})
-        return messages
-    
-    def generate_response(
-        self,
-        text: str,
-        history: List[Tuple],
-        stream: bool = True
-    ) -> Generator:
-        """Genera respuesta orquestada"""
-        start_time = time.time()
-        
-        # Analizar intención
-        intent, intent_metadata = analyze_intent_detailed(text, history)
-        
-        print(f"🎯 Intención detectada: {intent.value}")
-        print(f"📊 Metadatos: {json.dumps(intent_metadata, indent=2, ensure_ascii=False)}")
-        
-        # Caso especial: generación de imágenes
-        if intent == IntentType.IMAGE and REVE_KEY:
-            try:
-                # Generar imagen
-                result = generar_imagen_con_metadatos(text)
-                
-                if result["image"]:
-                    # Guardar imagen
-                    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-                    filename = f"image_{timestamp}.png"
-                    filepath = OUTPUT_DIR / filename
-                    result["image"].save(filepath)
-                    
-                    # Registrar interacción
-                    record_complete_interaction(
-                        prompt=text,
-                        intent=intent,
-                        model="REVE",
-                        latency=result["metadata"]["generation_time"],
-                        success=result["metadata"]["success"],
-                        response_quality=0.9 if result["metadata"]["success"] else 0.1,
-                        user_context=json.dumps({"session": self.session_id})
-                    )
-                    
-                    # Preparar respuesta con imagen
-                    response_text = f"🖼️ **Imagen generada exitosamente**\n\n"
-                    response_text += f"📁 Guardada en: `{filepath}`\n"
-                    response_text += f"⏱️ Tiempo: {result['metadata']['generation_time']:.2f}s\n"
-                    response_text += f"📐 Tamaño: {result['image'].size}\n\n"
-                    response_text += "La imagen incluye el sello BATUTO-ART y está lista para uso profesional."
-                    
-                    # Enviar imagen y texto
-                    yield {
-                        "type": "multimodal",
-                        "text": response_text,
-                        "image": result["image"],
-                        "filepath": str(filepath)
-                    }
-                else:
-                    error_msg = f"❌ Error generando imagen: {result['metadata'].get('error', 'Desconocido')}"
-                    yield error_msg
-                    
-            except Exception as e:
-                error_msg = f"❌ Error en generación de imagen: {str(e)}"
-                yield error_msg
-        
-        else:
-            # Seleccionar modelo óptimo
-            context_str = json.dumps({
-                "session": self.session_id,
-                "history_length": len(history),
-                "intent_history": [h.value for h in ORCHESTRATOR.intent_history]
-            })
-            
-            model, confidence, selection_data = select_model_with_context(
-                intent, CANDIDATES[intent], context_str
-            )
-            
-            print(f"🤖 Modelo seleccionado: {model} (confianza: {confidence:.2f})")
-            print(f"📝 Razón: {selection_data}")
-            
-            # Preparar mensajes
-            messages = self.prepare_messages(text, intent, history)
-            
-            # Generar respuesta
-            try:
-                if stream:
-                    full_response = ""
-                    
-                    for chunk in stream_response_generator(
-                        api_key=API_KEY,
-                        model=model,
-                        messages=messages,
-                        temperature=0.1,
-                        max_tokens=4096
-                    ):
-                        if isinstance(chunk, str):
-                            full_response += chunk
-                            yield full_response
-                        elif isinstance(chunk, dict):
-                            # Son las métricas finales
-                            metrics = chunk
-                            
-                            # Calcular calidad de respuesta
-                            response_quality = min(1.0, len(full_response) / 1000)  # Simple heuristic
-                            
-                            # Registrar interacción
-                            record_complete_interaction(
-                                prompt=text,
-                                intent=intent,
-                                model=model,
-                                latency=time.time() - start_time,
-                                success=metrics.get("success", False),
-                                response_quality=response_quality,
-                                user_context=context_str
-                            )
-                            
-                            # Log metrics
-                            print(f"📊 Métricas finales: {json.dumps(metrics, indent=2)}")
-                else:
-                    # Respuesta no streaming (para debugging)
-                    response = self.client.chat_completion(
-                        model=model,
-                        messages=messages,
-                        stream=False
-                    )
-                    
-                    full_response = response.choices[0].message.content
-                    
-                    # Registrar interacción
-                    record_complete_interaction(
-                        prompt=text,
-                        intent=intent,
-                        model=model,
-                        latency=time.time() - start_time,
-                        success=True,
-                        response_quality=0.8,
-                        user_context=context_str
-                    )
-                    
-                    yield full_response
-                    
-            except Exception as e:
-                error_msg = f"❌ Error en generación: {str(e)}"
-                yield error_msg
-                
-                # Registrar fallo
-                record_complete_interaction(
-                    prompt=text,
-                    intent=intent,
-                    model=model,
-                    latency=time.time() - start_time,
-                    success=False,
-                    response_quality=0.0,
-                    user_context=context_str
-                )
-
-# ===================== INTERFAZ GRADIO MEJORADA =====================
-def create_advanced_interface():
-    """Crea interfaz Gradio profesional"""
-    
-    orchestrator = BatutoOrchestrator()
-    
-    with gr.Blocks(
-        title="BATUTO X Enterprise",
-        theme=gr.themes.Soft(),
-        css="""
-        .gradio-container {
-            max-width: 1200px !important;
-            margin: 0 auto !important;
-        }
-        .model-card {
-            border: 1px solid #e0e0e0;
-            border-radius: 8px;
-            padding: 15px;
-            margin: 10px 0;
-            background: #f9f9f9;
-        }
-        .response-box {
-            border: 2px solid #4CAF50;
-            border-radius: 10px;
-            padding: 15px;
-            margin: 15px 0;
-        }
-        """
-    ) as app:
-        
-        # Header
-        gr.Markdown("""
-        # 🚀 BATUTO X Enterprise Orchestrator
-        ### **Plataforma Cognitiva Multimodal con Selección Inteligente de Modelos**
-        """)
-        
-        with gr.Row():
-            # Panel izquierdo - Información del sistema
-            with gr.Column(scale=1):
-                gr.Markdown("### 📊 Estado del Sistema")
-                
-                system_status = gr.Markdown("""
-                - **API Sambanova**: ✅ Conectado
-                - **API REVE**: {'✅' if REVE_KEY else '❌'}
-                - **Modelos disponibles**: 6
-                - **Intenciones soportadas**: 8
-                - **Memoria activa**: {len(MEMORY.nodes)} nodos
-                """)
-                
-                gr.Markdown("### 🎯 Intenciones Detectables")
-                
-                intents_list = gr.Markdown("""
-                - **🖼️ IMAGE**: Generación de imágenes
-                - **💻 CODE**: Programación y debugging
-                - **🧠 REASONING**: Razonamiento analítico
-                - **🏗️ ARCHITECTURE**: Arquitectura de sistemas
-                - **⚙️ DEVOPS**: Infraestructura y CI/CD
-                - **✅ QA**: Pruebas y calidad
-                - **👁️ VISUAL**: Análisis de imágenes
-                - **🌐 GENERAL**: Propósito general
-                """)
-                
-                # Botón para limpiar memoria
-                clear_btn = gr.Button("🧹 Limpiar Memoria de Contexto", variant="secondary")
-                
-                def clear_memory():
-                    MEMORY.nodes.clear()
-                    MEMORY.context_order.clear()
-                    MEMORY.intent_clusters.clear()
-                    return "✅ Memoria limpiada"
-                
-                clear_btn.click(clear_memory, outputs=system_status)
-            
-            # Panel principal - Chat
-            with gr.Column(scale=2):
-                # Chatbot con soporte multimodal
-        def create_advanced_interface():
-    """Crea interfaz Gradio profesional."""
-    
-    # Eliminamos theme y css del constructor Blocks
-    with gr.Blocks(
-        title="BATUTO X Enterprise Orchestrator"
-        # theme y css se pasan a launch() más adelante
-    ) as app:
-        
-        # ... HEADER y otros elementos ...
-        
-        with gr.Row():
-            # Panel izquierdo...
-            
-            # Panel principal - Chat
-            with gr.Column(scale=2):
-                # Chatbot con soporte multimodal
-                chatbot = gr.Chatbot(
-                    height=600,
-                    label="Conversación"
-                )
-                # ... resto del código del panel de chat ...
-        title="BATUTO X Enterprise Orchestrator"
-        # theme y css se pasan a launch() más adelante
-    ) as app:
-        
-        # ===== HEADER =====
-        gr.Markdown("""
-        # 🚀 BATUTO X — Enterprise AI Orchestrator
-        ### Orquestación Multimodal Inteligente con Detección Contextual
-        """)
-        
-        with gr.Row():
-            # ===== PANEL DE CONTROL IZQUIERDO =====
-            with gr.Column(scale=1):
-                gr.Markdown("### 🤖 Modelo")
-                modelo_selector = gr.Dropdown(
-                    choices=list(MODELS.keys()),
-                    value="🦙 Llama 4 Scout (17B)",
-                    label="Selecciona Modelo",
-                    interactive=True
-                )
-                info_modelo = gr.Markdown()
-                
-                with gr.Accordion("⚙️ Parámetros Avanzados", open=False):
-                    temperatura = gr.Slider(
-                        minimum=0.01, maximum=1.0, value=0.1, step=0.01,
-                        label="Temperatura (creatividad)"
-                    )
-                    max_tokens = gr.Slider(
-                        minimum=256, maximum=8192, value=4096, step=256,
-                        label="Tokens máximos"
-                    )
-                
-                with gr.Accordion("📈 Estado del Sistema", open=False):
-                    estado_sistema = gr.Markdown("Sistema inicializado...")
-            
-            # ===== PANEL DE CHAT PRINCIPAL =====
-            with gr.Column(scale=2):
-                # CHATBOT CORREGIDO: Eliminados show_copy_button y avatar_images
-                chatbot = gr.Chatbot(
-                    height=600,
-                    label="Conversación",
-                    # show_copy_button=True,     # <-- Eliminado
-                    # show_share_button=True,    # <-- Eliminado
-                    # avatar_images=("👤", "🤖") # <-- Eliminado
-                )
-                
-                modo_actual = gr.Markdown("**Modo activo**: Detectando...")
-                
-                entrada = gr.Textbox(
-                    label="Consulta",
-                    placeholder="Escribe tu consulta aquí...",
-                    lines=3
-                )
-                
-                with gr.Row():
-                    btn_enviar = gr.Button("🚀 Enviar", variant="primary", size="lg")
-                    btn_limpiar = gr.Button("🧹 Limpiar", variant="secondary")
-        
-        # ===== FUNCIONES DE INTERACCIÓN =====
-        def procesar_entrada_usuario(mensaje, historial):
-            if not mensaje.strip():
-                return "", historial, "**Modo**: Esperando entrada..."
-            
-            nuevo_historial = historial + [[mensaje, None]]
-            intent, _ = analyze_intent_detailed(mensaje, historial)
-            return "", nuevo_historial, f"**Modo activo**: {intent.value.replace('_', ' ').title()}"
-        
-        def generar_respuesta_con_parametros(historial, modelo, temp, tokens):
-            if not historial:
-                return historial, "**Modo activo**: Esperando entrada..."
-            
-            ultimo_mensaje = historial[-1][0]
-            
-            for respuesta in generar_respuesta_orquestada(
-                mensaje=ultimo_mensaje,
-                historia=historial[:-1],
-                modelo_seleccionado=modelo,
-                temperatura=temp,
-                max_tokens=tokens
-            ):
-                historial[-1][1] = respuesta
-                yield historial, "**Modo activo**: Generando..."
-            
-            intent, _ = analyze_intent_detailed(ultimo_mensaje, historial[:-1])
-            yield historial, f"**Modo activo**: {intent.value.replace('_', ' ').title()}"
-        
-        def limpiar_chat():
-            return [], "", "**Modo activo**: Esperando entrada..."
-        
-        # ===== CONEXIÓN DE EVENTOS =====
-        btn_enviar.click(
-            fn=procesar_entrada_usuario,
-            inputs=[entrada, chatbot],
-            outputs=[entrada, chatbot, modo_actual]
-        ).then(
-            fn=generar_respuesta_con_parametros,
-            inputs=[chatbot, modelo_selector, temperatura, max_tokens],
-            outputs=[chatbot, modo_actual]
-        )
-        
-        entrada.submit(
-            fn=procesar_entrada_usuario,
-            inputs=[entrada, chatbot],
-            outputs=[entrada, chatbot, modo_actual]
-        ).then(
-            fn=generar_respuesta_con_parametros,
-            inputs=[chatbot, modelo_selector, temperatura, max_tokens],
-            outputs=[chatbot, modo_actual]
-        )
-        
-        btn_limpiar.click(
-            fn=limpiar_chat,
-            outputs=[chatbot, entrada, modo_actual]
-        )
-    
-    return app
-# ===================== EJECUCIÓN PRINCIPAL =====================
-if __name__ == "__main__":
-    # ... (código de inicialización previo) ...
-    app = create_advanced_interface()
-    # Pasar theme y css al método launch()
-    app.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False,
-        show_error=True,
-        debug=False,
-        theme="soft",      # <-- Añadido aquí
-        css=""             # <-- Añadido aquí (puedes dejar tu CSS personalizado)
-    )