Spaces:
Sleeping
Sleeping
File size: 3,699 Bytes
9632608 f08e8ab 6bcf7d9 9632608 1a451f5 3337b97 f08e8ab 9632608 f08e8ab 80a9980 f08e8ab 80a9980 f08e8ab 80a9980 5ecabb6 80a9980 f08e8ab 80a9980 f08e8ab 80a9980 f08e8ab 9632608 63e73e5 9632608 63e73e5 9632608 63e73e5 9632608 f08e8ab 9632608 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
import os
import requests
from huggingface_hub import hf_api
import time
# Cargar el token de Hugging Face desde el entorno
HF_TOKEN = os.getenv("HF_TOKEN")
MODEL_NAME = os.getenv("MODEL_NAME", "google/gemma-2b-it")
print("HF_TOKEN desde agent.py:", HF_TOKEN[:10] if HF_TOKEN else "NO DEFINIDO")
# Función para llamar al modelo y obtener respuesta
def call_model(prompt):
"""
Llama al modelo de Hugging Face y devuelve la respuesta generada.
"""
headers = {
"Authorization": f"Bearer {HF_TOKEN}",
"Content-Type": "application/json"
}
payload = {
"inputs": prompt,
"parameters": {"max_new_tokens": 512}
}
try:
response = requests.post(
f"https://api-inference.huggingface.co/models/{MODEL_NAME}",
headers=headers,
json=payload,
timeout=60
)
result = response.json()
if isinstance(result, dict) and result.get("error"):
raise ValueError(f"Error al procesar la solicitud: {result['error']}")
return result[0]["generated_text"] if isinstance(result, list) else result
except requests.exceptions.RequestException as e:
raise ValueError(f"Error de conexión con el modelo: {str(e)}")
except Exception as e:
raise ValueError(f"Error al obtener respuesta del modelo: {str(e)}")
# Función para esperar que el modelo esté listo
def wait_for_model():
"""
Espera hasta que el modelo esté completamente cargado y disponible para hacer solicitudes.
"""
headers = {
"Authorization": f"Bearer {HF_TOKEN}",
"Content-Type": "application/json"
}
print(f"Esperando que el modelo {MODEL_NAME} esté completamente cargado...")
while True:
try:
response = requests.get(
f"https://api-inference.huggingface.co/models/{MODEL_NAME}",
headers=headers,
timeout=30
)
if response.status_code == 200:
json_response = response.json()
if isinstance(json_response, dict) and json_response.get("status") == "loaded":
print(f"✅ Modelo {MODEL_NAME} completamente cargado.")
break
else:
print(f"⏳ Modelo aún cargando... Estado actual: {json_response.get('status', 'desconocido')}")
else:
print(f"❌ Estado inesperado: {response.status_code}")
except requests.exceptions.RequestException as e:
print(f"🌐 Error de red: {e}")
time.sleep(10) # Espera 10 segundos antes del siguiente intento
# Clase para representar la lógica del agente
class Agente:
def __init__(self):
self.name = "Agente Hugging Face"
self.version = "1.0"
def procesar_pregunta(self, pregunta):
"""
Procesa la pregunta y devuelve la respuesta generada por el modelo.
"""
print(f"Procesando pregunta: {pregunta}")
try:
respuesta = call_model(pregunta)
return respuesta
except ValueError as e:
print(f"Error al procesar la pregunta: {e}")
return f"ERROR: {e}"
# Función para ejecutar la evaluación de preguntas
def evaluar_preguntas(preguntas):
"""
Toma una lista de preguntas, las procesa y devuelve las respuestas generadas.
"""
agente = Agente()
respuestas = []
# Esperar a que el modelo esté listo
wait_for_model()
for pregunta in preguntas:
print(f"Evaluando: {pregunta}")
respuesta = agente.procesar_pregunta(pregunta)
respuestas.append(respuesta)
return respuestas
|