Update app.py

app.py

@@ -6,285 +6,115 @@ from PIL import Image
 import base64
 import io
 import requests
-from functools import lru_cache
-import torch
-# Importar la clase de excepción de la API de SambaNova si está disponible, si no, usar la base
-try:
-    from sambanova.exceptions import APIError as SambaNovaAPIError
-except ImportError:
-    # Usar una excepción genérica si la librería no está instalada o no la expone
-    SambaNovaAPIError = Exception
 
-print("🚀 Iniciando aplicación...")
-
-# ------------ CONFIGURACIÓN SIMPLIFICADA ------------
+# --- Config ---
 class Config:
     def __init__(self):
-        # Buscar API keys en environment variables (secrets de Hugging Face)
         self.SAMBANOVA_API_KEY = os.getenv("SAMBANOVA_API_KEY")
         self.BRIA_API_TOKEN = os.getenv("BRIA_API_TOKEN")
-        self.HUGGINGFACE_TOKEN = os.getenv("HUGGINGFACE_TOKEN")
-        
-        self.validate_keys()
+        self.validate()
 
-    def validate_keys(self):
-        if self.SAMBANOVA_API_KEY:
-            print(f"✅ SAMBANOVA_API_KEY configurado (primeros 10 chars): {self.SAMBANOVA_API_KEY[:10]}...")
-        else:
-            print("❌ SAMBANOVA_API_KEY no encontrado")
-            print("💡 Agrega el secret en: Settings → Repository secrets → SAMBANOVA_API_KEY")
-            
-        if self.BRIA_API_TOKEN:
-            print("✅ BRIA_API_TOKEN configurado")
-        else:
-            print("⚠️ BRIA_API_TOKEN no configurado - Imagen deshabilitada")
+    def validate(self):
+        ok_samba = bool(self.SAMBANOVA_API_KEY)
+        ok_bria = bool(self.BRIA_API_TOKEN)
+        print(f"✅ SambaNova: {'Configurado' if ok_samba else 'Falta SAMBANOVA_API_KEY'}")
+        print(f"✅ Bria: {'Configurado' if ok_bria else 'Falta BRIA_API_TOKEN'}")
 
 config = Config()
 
-# ------------ IMPORTACION CONDICIONAL CON MEJOR MANEJO DE ERRORES ------------
-SambaNova = None
-try:
-    # Intenta importar la clase SambaNova
-    from sambanova import SambaNova
-    print("✅ SambaNova importado correctamente")
-except ImportError:
-    print("⚠️ SambaNova no disponible - usando solo modelos locales")
-    print("💡 Para instalar: pip install sambanova")
-
-# ------------ MODELOS INTEGRADOS ------------
-MODELS = {
-    "chat": {
-        "llama3": "NousResearch/Meta-Llama-3-8B-Instruct",
-        "qwen": "Qwen/Qwen1.5-7B-Chat",
-        "deepseek": "deepseek-ai/deepseek-llm-7b-chat",
-    },
-    "code": {
-        "starcoder2": "bigcode/starcoder2-3b",
-        "deepseek-coder": "deepseek-ai/deepseek-coder-1.3b-instruct"
-    },
-    "sambanova": {
-        # 💡 Reemplaza con el nombre de modelo de SambaNova correcto.
-        "chat_model": "Llama-3.3-Swallow-70B-Instruct-v0.4" 
-    }
-}
-
-# ------------ CARGA DE MODELOS SIMPLIFICADA ------------
-class ModelLoader:
-    @lru_cache(maxsize=2)
-    def load_hf_model(self, model_name: str):
+# --- Cliente SambaNova ---
+class SambaNovaClient:
+    BASE_URL = "https://api.sambanova.ai/v1/chat/completions"
+    
+    def __init__(self, api_key: str):
+        self.api_key = api_key
+
+    def generate(self, prompt: str, model: str = "Maverick") -> str:
+        headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json"
+        }
+        data = {
+            "model": model,
+            "messages": [{"role": "user", "content": prompt}],
+            "temperature": 0.7,
+            "max_tokens": 500
+        }
         try:
-            from transformers import AutoModelForCausalLM, AutoTokenizer
-            
-            print(f"📥 Cargando modelo: {model_name}")
-            
-            # Configuración simplificada para CPU
-            tokenizer = AutoTokenizer.from_pretrained(model_name)
-            
-            # Nota: Si el modelo es muy grande para tu entorno, esta carga puede ser lenta o fallar.
-            model = AutoModelForCausalLM.from_pretrained(
-                model_name,
-                device_map="auto",
-                torch_dtype=torch.float32,
-                low_cpu_mem_usage=True
-            )
-            model.eval()
-            print(f"✅ Modelo {model_name} cargado exitosamente")
-            return model, tokenizer
-            
+            response = requests.post(self.BASE_URL, headers=headers, json=data, timeout=30)
+            response.raise_for_status()
+            return response.json()["choices"][0]["message"]["content"].strip()
         except Exception as e:
-            print(f"❌ Error cargando modelo {model_name}: {e}")
-            return None, None
+            return f"❌ Error SambaNova: {str(e)}"
 
-model_loader = ModelLoader()
-
-# ------------ HERRAMIENTAS SIMPLIFICADAS ------------
+# --- Herramientas ---
 class AI_Tools:
     def __init__(self):
-        self.sn_client = None
-        if SambaNova is not None and config.SAMBANOVA_API_KEY:
-            try:
-                # 💡 VERIFICACIÓN CLAVE: Inicialización del cliente
-                self.sn_client = SambaNova(api_key=config.SAMBANOVA_API_KEY)
-                print("✅ Cliente SambaNova inicializado")
-            except Exception as e:
-                # Se imprime el error específico de la inicialización
-                print(f"❌ Error inicializando SambaNova. Asegúrate que la clave es válida: {e}")
-
-    async def generate_text(self, model_type: str, model_name: str, prompt: str) -> str:
-        try:
-            if model_type == "sambanova" and self.sn_client:
-                return await self._generate_sambanova(model_name, prompt)
-            else:
-                return await self._generate_hf(model_name, prompt)
-        except Exception as e:
-            # Error de alto nivel
-            return f"❌ Error general en la generación: {str(e)}"
+        self.samba = SambaNovaClient(config.SAMBANOVA_API_KEY) if config.SAMBANOVA_API_KEY else None
 
-    async def _generate_sambanova(self, model_name: str, prompt: str) -> str:
-        # 💡 GUARDIA: Si por alguna razón el cliente falla después de la inicialización
-        if not self.sn_client:
-            return "❌ Cliente SambaNova no disponible (inicialización fallida)."
-            
-        try:
-            # Usar run_in_executor para llamadas síncronas de la librería sambanova.
-            response = await asyncio.get_event_loop().run_in_executor(
-                None,
-                # La llamada síncrona real
-                lambda: self.sn_client.chat.completions.create(
-                    model=model_name,
-                    messages=[{"role": "user", "content": prompt}],
-                    temperature=0.7,
-                    max_tokens=500
-                )
-            )
-            return response.choices[0].message.content
-        
-        # Captura de errores de la API de SambaNova (si es posible) o errores generales de conexión
-        except SambaNovaAPIError as e:
-            # 💡 MENSAJE CLARO DE FALLO: Si falla, el error aparecerá en el chat.
-            return f"❌ Error de API de SambaNova. Verifica el nombre del modelo ({model_name}) y la clave API. Error: {str(e)}"
-        except Exception as e:
-            return f"❌ Error de conexión/ejecución SambaNova: {type(e).__name__} - {str(e)}"
+    def generate_text(self, prompt: str) -> str:
+        if not self.samba:
+            return "❌ SAMBANOVA_API_KEY no configurada."
+        return self.samba.generate(prompt, model="Maverick")
 
-    async def _generate_hf(self, model_name: str, prompt: str) -> str:
-        try:
-            model, tokenizer = model_loader.load_hf_model(model_name)
-            if model is None:
-                return "❌ Modelo local no disponible"
-                
-            inputs = tokenizer(prompt, return_tensors="pt")
-            
-            with torch.no_grad():
-                outputs = model.generate(
-                    **inputs, 
-                    max_new_tokens=256,
-                    temperature=0.7,
-                    do_sample=True,
-                    pad_token_id=tokenizer.eos_token_id
-                )
-            
-            response = tokenizer.decode(outputs[0], skip_special_tokens=True)
-            # Remover el prompt del response
-            return response[len(prompt):].strip()
-            
-        except Exception as e:
-            return f"❌ Error modelo local: {str(e)}"
-
-    async def generate_image(self, prompt: str) -> Optional[Image.Image]:
+    def generate_image(self, prompt: str) -> Optional[Image.Image]:
         if not config.BRIA_API_TOKEN:
             return None
-        
         try:
-            url = "https://api.bria.ai/v1/generate"
-            headers = {
-                "Authorization": f"Bearer {config.BRIA_API_TOKEN}",
-                "Content-Type": "application/json"
-            }
-            json_data = {
-                "prompt": prompt,
-                "options": {"resolution": "512x512"}
-            }
-
-            response = await asyncio.get_event_loop().run_in_executor(
-                None,
-                lambda: requests.post(url, headers=headers, json=json_data, timeout=30)
-            )
-            
+            url = "https://api.bria.ai/v1/generate"  # ✅ sin espacio
+            headers = {"Authorization": f"Bearer {config.BRIA_API_TOKEN}"}
+            json_data = {"prompt": prompt, "options": {"resolution": "512x512"}}
+            response = requests.post(url, headers=headers, json=json_data, timeout=30)
             if response.status_code == 200:
-                data = response.json()
-                if "image_base64" in data:
-                    img_bytes = base64.b64decode(data["image_base64"])
+                img_b64 = response.json().get("image_base64")
+                if img_b64:
+                    img_bytes = base64.b64decode(img_b64)
                     return Image.open(io.BytesIO(img_bytes))
-            
-            return None
         except Exception as e:
-            print(f"⚠️ Error imagen: {e}")
-            return None
-            # ------------ INTERFAZ Y LÓGICA DE PROCESAMIENTO ------------
-tools = AI_Tools()
+            print(f"⚠️ Error en Bria: {e}")
+        return None
 
-async def process_input(message: str, image: Optional[Image.Image], history: list) -> list:
-    try:
-        msg_lower = message.lower()
-        
-        # Generación de imágenes
-        if image or any(k in msg_lower for k in ["imagen", "genera imagen", "foto"]):
-            img = await tools.generate_image(message)
-            if img:
-                return history + [(message, ("Imagen generada:", img))]
-            return history + [(message, "❌ Imagen no disponible - BRIA no configurado")]
-        
-        # Generación de código
-        elif any(k in msg_lower for k in ["código", "code", "programa"]):
-            response = await tools.generate_text("hf", MODELS["code"]["deepseek-coder"], message)
-            return history + [(message, f"```python\n{response}\n```")]
-        
-        # Chat normal
-        else:
-            # Intentar SambaNova primero
-            if tools.sn_client:
-                # 💡 USO DEL MODELO SAMBANOVA CONFIGURADO
-                sn_model_name = MODELS["sambanova"]["chat_model"]
-                response = await tools.generate_text("sambanova", sn_model_name, message)
-            else:
-                # Fallback a modelo local
-                response = await tools.generate_text("hf", MODELS["chat"]["llama3"], message)
-            
-            return history + [(message, response)]
-            
-    except Exception as e:
-        return history + [(message, f"❌ Error inesperado: {str(e)}")]
+# --- Lógica de procesamiento (síncrona para Gradio) ---
+tools = AI_Tools()
 
-# ------------ APLICACIÓN GRADIO ------------
-with gr.Blocks(title="AI Assistant", theme=gr.themes.Soft()) as app:
-    gr.Markdown("# 🤖 AI Assistant")
-    
-    # El estado del cliente SN ahora reflejará si la inicialización fue exitosa o no
-    status_text = "SambaNova: ✅ Conectado" if tools.sn_client else "SambaNova: ❌ No disponible (Verifica logs de inicialización)"
-    gr.Markdown(f"**{status_text}** | **Modelos locales:** ✅ Disponibles")
-    
-    chatbot = gr.Chatbot(height=400)
+def process_input(message: str, image: Optional[Image.Image], history: list) -> list:
+    msg_lower = message.lower()
     
-    with gr.Row():
-        msg = gr.Textbox(
-            label="Mensaje",
-            placeholder="Escribe tu pregunta...",
-            scale=4
-        )
-        submit_btn = gr.Button("Enviar", variant="primary", scale=1)
+    # Imagen
+    if any(kw in msg_lower for kw in ["imagen", "foto", "genera imagen"]):
+        img = tools.generate_image(message)
+        if img:
+            return history + [(message, ("", img))]
+        return history + [(message, "❌ Imagen no generada (Bria no disponible o error).")]
     
-    img_input = gr.Image(type="pil", label="Imagen (opcional)", height=150)
-    
-    # Ejemplos
-    gr.Examples(
-        examples=[
-            "Explique la teoría de la relatividad",
-            "Escribe una función Python para ordenar una lista",
-            "Genera una imagen de un paisaje montañoso"
-        ],
-        inputs=msg
-    )
+    # Texto (usa SambaNova/Maverick)
+    response = tools.generate_text(message)
+    return history + [(message, response)]
 
-    def clear_chat():
-        return []
+# --- Interfaz Gradio ---
+with gr.Blocks(title="AI Assistant", theme=gr.themes.Soft()) as app:
+    gr.Markdown("# 🤖 AI Assistant (Maverick + Bria)")
+    status = "✅ SambaNova: Activo" if tools.samba else "❌ SambaNova: Inactivo"
+    gr.Markdown(f"**{status}** | **Bria:** {'✅ Activo' if config.BRIA_API_TOKEN else '❌ Inactivo'}")
     
+    chatbot = gr.Chatbot(height=450)
+    msg = gr.Textbox(label="Mensaje", placeholder="Ej: 'Genera una imagen de un gato en la luna' o 'Explica la fusión nuclear'")
+    img_input = gr.Image(type="pil", label="Imagen (opcional)", visible=False)  # No usado realmente
+
+    gr.Examples([
+        "Escribe un poema sobre el otoño",
+        "Genera una imagen de un robot andando en bicicleta",
+        "¿Qué es la computación cuántica?"
+    ], inputs=msg)
+
+    def clear(): return []
+
     clear_btn = gr.Button("Limpiar chat")
-    clear_btn.click(clear_chat, outputs=chatbot)
+    clear_btn.click(clear, outputs=chatbot)
 
-    submit_btn.click(
-        process_input,
-        inputs=[msg, img_input, chatbot],
-        outputs=chatbot
-    )
-    
-    msg.submit(
-        process_input,
-        inputs=[msg, img_input, chatbot],
-        outputs=chatbot
-    )
+    # ✅ Asíncrono en el fondo, pero interfaz síncrona (compatible con Gradio)
+    msg.submit(process_input, [msg, img_input, chatbot], chatbot, queue=True)
+    gr.Button("Enviar").click(process_input, [msg, img_input, chatbot], chatbot, queue=True)
 
 if __name__ == "__main__":
-    # Asegúrate de ejecutarlo con 'python -m uvicorn tu_archivo:app --reload' 
-    # si estás usando Gradio en un entorno de servidor.
-    app.launch(server_name="0.0.0.0", server_port=7860)
-    
+    app.queue().launch(server_name="0.0.0.0", server_port=7860)