Spaces:
Sleeping
Sleeping
File size: 4,491 Bytes
9632608 f08e8ab 6bcf7d9 9632608 1a451f5 3337b97 f08e8ab 9632608 5bafa85 9632608 5bafa85 9632608 5bafa85 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 |
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, max_retries=5, wait_seconds=10):
"""
Llama al modelo de Hugging Face y devuelve la respuesta generada.
Reintenta si el modelo aún se está cargando.
"""
headers = {
"Authorization": f"Bearer {HF_TOKEN}",
"Content-Type": "application/json"
}
payload = {
"inputs": prompt,
"parameters": {"max_new_tokens": 512}
}
for attempt in range(1, max_retries + 1):
try:
response = requests.post(
f"https://api-inference.huggingface.co/models/{MODEL_NAME}",
headers=headers,
json=payload,
timeout=60
)
result = response.json()
# Verificar si hay un error
if isinstance(result, dict) and result.get("error"):
error_msg = result["error"]
print(f"[Intento {attempt}] Error del modelo: {error_msg}")
if "is currently loading" in error_msg and attempt < max_retries:
print(f"🔁 Esperando {wait_seconds} segundos antes de reintentar...")
time.sleep(wait_seconds)
continue
else:
raise ValueError(f"Error al procesar la solicitud: {error_msg}")
# Respuesta válida
return result[0]["generated_text"] if isinstance(result, list) else result
except requests.exceptions.RequestException as e:
print(f"[Intento {attempt}] Error de red: {e}")
if attempt < max_retries:
time.sleep(wait_seconds)
continue
else:
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)}")
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
|