import accelerate import gradio as gr import joblib import numpy as np import requests import torch import os from transformers import AutoModelForCausalLM, AutoTokenizer from langchain.memory import ConversationBufferMemory # Configuración del modelo de lenguaje MODEL_NAME = "PlanTL-GOB-ES/roberta-base-bne" device = "cuda" if torch.cuda.is_available() else "cpu" HF_TOKEN = os.getenv("HF_TOKEN") # Token de Hugging Face if not HF_TOKEN: raise ValueError("❌ ERROR: No se encontró HF_TOKEN. Asegúrate de definirlo en las variables de entorno.") print("🔄 Cargando modelo de lenguaje...") tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, token=HF_TOKEN) model = AutoModelForCausalLM.from_pretrained( MODEL_NAME, torch_dtype=torch.float32, device_map="auto", token=HF_TOKEN ) # Memoria conversacional memory = ConversationBufferMemory() # API de Node-RED NODE_RED_URL = "https://appairedecolmena.es/colmena1/datos" USERNAME = "user" PASSWORD = "velutina" def obtener_datos_colmena(): """Obtiene los datos más recientes de Node-RED con autenticación.""" try: respuesta = requests.get(NODE_RED_URL, auth=(USERNAME, PASSWORD), timeout=5) if respuesta.status_code == 200: datos = respuesta.json() if "data" in datos and isinstance(datos["data"], list) and datos["data"]: return datos["data"][-1] # Último registro return {"error": "No hay datos recientes en Node-RED."} else: return {"error": f"Error en la API: {respuesta.status_code}"} except Exception as e: return {"error": str(e)} def generar_recomendacion(datos): """Genera recomendaciones basadas en el estado de la colmena.""" # ✅ Convertimos a float para evitar errores temperaturaInterior = float(datos.get("temperaturaInterior", 0)) humedadInterior = float(datos.get("humedadInterior", 0)) co2 = float(datos.get("co2", 0)) recomendaciones = [] if temperaturaInterior > 35: recomendaciones.append("🌡 La temperatura interior es alta. Podrías mejorar la ventilación para evitar el sobrecalentamiento de la colmena.") elif temperaturaInterior < 20: recomendaciones.append("❄ La temperatura interior es baja. Asegúrate de que la calefacción funcione correctamente para evitar estrés en las abejas.") if humedadInterior > 80: recomendaciones.append("💧 La humedad es alta. Esto puede favorecer hongos y enfermedades. Asegúrate de que la ventilación sea adecuada.") elif humedadInterior < 40: recomendaciones.append("🔥 La humedad es baja. Las abejas necesitan un nivel de humedad adecuado para la crianza, podrías aumentar la humedad si es necesario.") if co2 > 1000: recomendaciones.append("🌿 El nivel de CO₂ es alto. Esto puede indicar problemas de ventilación. Verifica que el sistema de ventilación esté activo.") if not recomendaciones: recomendaciones.append("✅ Todo parece estar en orden. Sigue monitoreando regularmente para asegurar el bienestar de la colmena.") return "\n".join(recomendaciones) def conversar_con_colmena(mensaje): """Genera una respuesta combinando el modelo de lenguaje con recomendaciones sobre la colmena.""" datos = obtener_datos_colmena() if "error" in datos: return datos["error"] datos_relevantes = generar_recomendacion(datos) contexto = f"Datos actuales de la colmena: {datos_relevantes}\nUsuario: {mensaje}\nColmena:" inputs = tokenizer(contexto, return_tensors="pt").to(device) print(f"🔄 Enviando entrada al modelo: {contexto}") with torch.no_grad(): output = model.generate( **inputs, max_length=100, do_sample=True, top_k=50, temperature=0.7, pad_token_id=tokenizer.eos_token_id ) respuesta = tokenizer.decode(output[0], skip_special_tokens=True).strip() print(f"✅ Respuesta generada por la IA: '{respuesta}'") if not respuesta: return "🤖 No pude generar una respuesta. Inténtalo de nuevo con otra pregunta." return f"{datos_relevantes}\n\n🤖 Respuesta de la IA: {respuesta}" iface = gr.Interface( fn=conversar_con_colmena, inputs="text", outputs="text", title="🐝 Chat con la Colmena", description="Consulta el estado de la colmena y recibe recomendaciones." ) iface.launch()