FLAFUENTE's picture
Initial clean deploy to Hugging Face Space
6f22d1c
Raw History Blame Contribute Delete
24.9 kB
"""
HuggingChat Client - Implementacion principal del cliente de API
FILOSOFIA DE DISEÑO:
===================
Este modulo implementa un cliente robusto y educativo para la API oficial
de HuggingChat. El diseño prioriza:
1. CLARIDAD PEDAGOGICA: Cada decision de diseño esta documentada con su
contexto historico y justificacion tecnica
2. ROBUSTEZ EMPRESARIAL: Manejo comprehensivo de errores, timeouts y
casos edge comunes en entornos de produccion
3. MODULARIDAD: Separacion clara de responsabilidades siguiendo principios
SOLID y patrones de diseño modernos
4. SEGURIDAD: Implementacion de mejores practicas para el manejo de
credenciales y comunicacion con APIs externas
"""
import os
import json
import time
from typing import List, Dict, Optional, Any
from pathlib import Path
# Importamos requests para HTTP, considerado el estandar de facto en Python
# para comunicacion web por su simplicidad y robustez comparado con urllib
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# Importamos python-dotenv para manejo seguro de variables de entorno
# Esto evita hardcodear credenciales en el codigo fuente
from dotenv import load_dotenv
class HuggingChatClient:
"""
Cliente oficial para la API de HuggingChat de Hugging Face
EVOLUCION DEL ACCESO A HUGGINGCHAT:
===================================
EPOCA 1 (2023-early 2024): Web Scraping Hacky
----------------------------------------------
Los desarrolladores usaban bibliotecas como 'hugchat' de Soulter, que:
- Simulaba un navegador web con requests/selenium
- Parseaba HTML del chat web de HuggingFace
- Manejaba cookies y tokens CSRF manualmente
- Era fragil ante cambios en la interfaz web
- Violaba implicitamente los ToS de HuggingFace
Ejemplo del metodo obsoleto:
```python
# OBSOLETO - NO USAR
from hugchat import hugchat
chatbot = hugchat.ChatBot(cookie_dict_from_browser)
```
EPOCA 2 (2024-presente): API Oficial REST
-----------------------------------------
HuggingFace introdujo una API oficial que:
- Proporciona endpoint estandarizado /v1/chat/completions
- Usa autenticacion Bearer Token limpia y segura
- Mantiene compatibilidad con formato OpenAI
- Garantiza estabilidad y soporte oficial
- Permite uso comercial bajo terminos claros
DISEÑO DE ESTA IMPLEMENTACION:
=============================
1. PATRON SINGLETON-LIKE: Una instancia por token/configuracion
2. SESSION REUSE: Reutilizamos conexiones HTTP para eficiencia
3. RETRY STRATEGY: Implementamos reintentos automaticos con backoff
4. GRACEFUL DEGRADATION: Manejo elegante de errores de red y API
5. LOGGING COMPREHENSIVO: Trazabilidad completa para debugging
"""
def __init__(self, token: Optional[str] = None, model: str = "meta-llama/Llama-3.2-3B-Instruct"):
"""
Inicializa el cliente de HuggingChat con configuracion robusta
FILOSOFIA DE INICIALIZACION:
===========================
La inicializacion sigue el principio "fail fast" - detectamos problemas
de configuracion inmediatamente en lugar de fallar silenciosamente
durante la ejecucion.
CARGA DE CREDENCIALES:
=====================
Implementamos una cascada de fuentes de credenciales:
1. Parametro explicito (para testing y casos especiales)
2. Variable de entorno HF_TOKEN (metodo recomendado)
3. Archivo .env en el directorio actual (para desarrollo)
4. Archivo .env en el directorio padre (estructura de proyecto)
Esta jerarquia permite flexibilidad sin comprometer seguridad.
Args:
token (str, optional): Token de Hugging Face. Si no se proporciona,
se buscara en variables de entorno o .env
model (str): Modelo a usar. Por defecto Llama-3.2-3B-Instruct
que ofrece buen balance calidad/velocidad
Raises:
ValueError: Si no se encuentra un token valido
ConnectionError: Si no se puede validar la conexion con la API
"""
# FASE 1: Carga de configuracion desde multiples fuentes
# =====================================================
# Cargamos variables de entorno desde .env si existe
# Esto permite desarrollo local sin exponer credenciales en git
load_dotenv()
# Establecemos el token siguiendo la cascada de prioridades
self.token = token or os.getenv('HF_TOKEN')
if not self.token:
raise ValueError(
"Token de Hugging Face requerido. "
"Proporciona el token via parametro, variable de entorno HF_TOKEN, "
"o archivo .env con la linea: HF_TOKEN=tu_token_aqui"
)
# FASE 2: Configuracion del modelo y endpoints
# ===========================================
self.model = model
# Usamos el endpoint oficial de Hugging Face para chat completions
# Este endpoint es compatible con la especificacion OpenAI, lo que
# facilita la migracion entre proveedores de LLM
# Nota: El endpoint se actualizo en 2024 de api-inference a router
self.base_url = "https://router.huggingface.co"
self.chat_endpoint = f"{self.base_url}/v1/chat/completions"
# FASE 3: Configuracion de la sesion HTTP
# ======================================
# Creamos una sesion persistente para reutilizar conexiones TCP
# Esto mejora significativamente el rendimiento en conversaciones largas
self.session = requests.Session()
# Configuramos headers estandar que se enviaran con cada request
self.session.headers.update({
'Authorization': f'Bearer {self.token}',
'Content-Type': 'application/json',
'User-Agent': 'HuggingChatClient/1.0.0 (Python/3.10+)'
})
# FASE 4: Estrategia de reintentos robusta
# =======================================
# Configuramos reintentos automaticos para manejar fallos transitorios
# comunes en APIs de ML (timeouts, rate limiting, errores 5xx)
retry_strategy = Retry(
total=3, # Maximo 3 reintentos
backoff_factor=1, # Delay exponencial: 1s, 2s, 4s
status_forcelist=[429, 500, 502, 503, 504], # Codigos HTTP a reintentar
allowed_methods=["POST"] # Solo reintentamos operaciones seguras
)
# Montamos el adaptador con la estrategia de reintentos
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("https://", adapter)
self.session.mount("http://", adapter)
# FASE 5: Inicializacion del historial de conversacion
# ===================================================
# Mantenemos el historial para contexto en conversaciones multi-turno
# Esto permite al modelo mantener coherencia a lo largo de la sesion
self.conversation_history: List[Dict[str, str]] = []
# FASE 6: Validacion inicial de conectividad
# =========================================
print("🔧 Inicializando cliente HuggingChat...")
print(f"📡 Endpoint: {self.chat_endpoint}")
print(f"🤖 Modelo: {self.model}")
print(f"🔑 Token: {'*' * (len(self.token) - 8) + self.token[-8:]}")
# Validamos que el token funciona con una llamada de prueba
try:
self._validate_connection()
print("✅ Conexion validada exitosamente")
except Exception as e:
print(f"❌ Error validando conexion: {e}")
raise
def _validate_connection(self) -> None:
"""
Valida que el token y la conexion funcionan correctamente
PROPOSITO DE LA VALIDACION:
==========================
Esta validacion temprana evita que el usuario descubra problemas
de autenticacion despues de configurar toda su aplicacion.
Enviamos un mensaje minimo para verificar:
1. Token valido y con permisos suficientes
2. Conectividad de red estable
3. Modelo especificado disponible
4. Formato de respuesta correcto
Raises:
ConnectionError: Si hay problemas de conectividad
ValueError: Si el token o modelo son invalidos
"""
test_payload = {
"model": self.model,
"messages": [{"role": "user", "content": "test"}],
"max_tokens": 1,
"temperature": 0.1
}
try:
response = self.session.post(
self.chat_endpoint,
json=test_payload,
timeout=10 # Timeout corto para validacion rapida
)
if response.status_code == 401:
raise ValueError("Token de Hugging Face invalido o expirado")
elif response.status_code == 403:
raise ValueError("Token sin permisos para acceder a la API de Inferencia")
elif response.status_code == 404:
raise ValueError(f"Modelo '{self.model}' no encontrado o no disponible")
elif response.status_code != 200:
raise ConnectionError(f"Error de API: {response.status_code} - {response.text}")
except requests.exceptions.Timeout:
raise ConnectionError("Timeout conectando con la API de Hugging Face")
except requests.exceptions.ConnectionError:
raise ConnectionError("No se pudo conectar con la API de Hugging Face")
def send_message(self, message: str, use_history: bool = True,
max_tokens: int = 512, temperature: float = 0.7) -> str:
"""
Envia un mensaje al modelo y retorna la respuesta
DISEÑO DE LA COMUNICACION:
=========================
Esta funcion implementa el core de la comunicacion con HuggingChat.
El diseño considera:
1. CONTEXT MANAGEMENT: Manejo optimo del historial de conversacion
para mantener coherencia sin exceder limites de tokens
2. PARAMETER TUNING: Parametros configurables para diferentes use cases:
- temperature: Control de creatividad/determinismo
- max_tokens: Control de longitud de respuesta
- use_history: Toggle para conversaciones vs consultas aisladas
3. ERROR HANDLING: Manejo granular de diferentes tipos de errores
con mensajes informativos en español
4. PERFORMANCE: Optimizaciones para minimizar latencia y costo
COMPARACION CON METODOS ANTIGUOS:
================================
METODO OBSOLETO (Web Scraping):
```python
# OBSOLETO - Codigo fragil y complejo
browser.find_element(By.ID, "chat-input").send_keys(message)
browser.find_element(By.CLASS_NAME, "send-button").click()
time.sleep(2) # Espera arbitraria
response = browser.find_element(By.CLASS_NAME, "response").text
```
METODO ACTUAL (API REST):
```python
# MODERNO - Codigo limpio y robusto
response = client.send_message("Tu pregunta aqui")
```
Args:
message (str): Mensaje a enviar al modelo
use_history (bool): Si incluir historial de conversacion previa
max_tokens (int): Limite de tokens en la respuesta
temperature (float): Control de aleatoriedad (0.0 = determinista, 1.0 = creativo)
Returns:
str: Respuesta del modelo
Raises:
ValueError: Si el mensaje esta vacio o parametros invalidos
ConnectionError: Si hay problemas de conectividad
RuntimeError: Si la API retorna un error inesperado
"""
# VALIDACION DE ENTRADA
# ====================
if not message or not message.strip():
raise ValueError("El mensaje no puede estar vacio")
if not 0.0 <= temperature <= 2.0:
raise ValueError("Temperature debe estar entre 0.0 y 2.0")
if max_tokens < 1 or max_tokens > 4096:
raise ValueError("max_tokens debe estar entre 1 y 4096")
# CONSTRUCCION DEL CONTEXTO
# ========================
messages = []
if use_history and self.conversation_history:
# Incluimos el historial para mantener contexto
# Limitamos el historial para evitar exceder el limite de tokens
messages.extend(self.conversation_history[-10:]) # Ultimos 10 intercambios
# Agregamos el mensaje actual del usuario
user_message = {"role": "user", "content": message.strip()}
messages.append(user_message)
# CONSTRUCCION DEL PAYLOAD
# =======================
payload = {
"model": self.model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
"stream": False # Respuesta completa, no streaming
}
# EJECUCION DE LA REQUEST
# =====================
print(f"🤔 Enviando mensaje al modelo {self.model}...")
start_time = time.time()
try:
response = self.session.post(
self.chat_endpoint,
json=payload,
timeout=30 # Timeout generoso para generacion
)
elapsed_time = time.time() - start_time
print(f"⏱️ Respuesta recibida en {elapsed_time:.2f} segundos")
# MANEJO DE ERRORES HTTP
# =====================
if response.status_code == 400:
error_detail = response.json().get('error', 'Error de formato en la request')
raise ValueError(f"Request malformada: {error_detail}")
elif response.status_code == 401:
raise ValueError(
"Token de autenticacion invalido o expirado. "
"Verifica tu token en https://huggingface.co/settings/tokens"
)
elif response.status_code == 403:
raise ValueError(
"Sin permisos para acceder a este modelo. "
"Verifica que tu token tenga permisos de 'Inference API'"
)
elif response.status_code == 429:
raise RuntimeError(
"Rate limit excedido. "
"La API de Hugging Face esta limitando las requests. "
"Espera unos minutos antes de intentar nuevamente."
)
elif response.status_code == 503:
raise RuntimeError(
f"Modelo {self.model} temporalmente no disponible. "
"El modelo podria estar cargandose o en mantenimiento."
)
elif response.status_code != 200:
raise RuntimeError(f"Error inesperado de API: {response.status_code} - {response.text}")
# PROCESAMIENTO DE LA RESPUESTA
# ============================
try:
response_data = response.json()
except json.JSONDecodeError:
raise RuntimeError("Respuesta de API malformada (JSON invalido)")
# Extraemos el contenido de la respuesta siguiendo el formato OpenAI
try:
assistant_response = response_data['choices'][0]['message']['content']
except (KeyError, IndexError) as e:
raise RuntimeError(f"Formato de respuesta inesperado: {e}")
# ACTUALIZACION DEL HISTORIAL
# ==========================
if use_history:
# Agregamos tanto el mensaje del usuario como la respuesta al historial
self.conversation_history.append(user_message)
self.conversation_history.append({
"role": "assistant",
"content": assistant_response
})
# Limitamos el historial para evitar crecimiento excesivo
if len(self.conversation_history) > 20: # Maximo 10 intercambios
self.conversation_history = self.conversation_history[-20:]
# LOGGING Y RETORNO
# ================
print(f"✅ Respuesta generada exitosamente ({len(assistant_response)} caracteres)")
return assistant_response
except requests.exceptions.Timeout:
raise ConnectionError(
"Timeout esperando respuesta del modelo. "
"El modelo podria estar procesando una request compleja."
)
except requests.exceptions.ConnectionError:
raise ConnectionError(
"Error de conectividad con la API de Hugging Face. "
"Verifica tu conexion a internet."
)
def interactive_chat(self) -> None:
"""
Inicia un chat interactivo en la terminal
DISEÑO DE LA INTERFAZ INTERACTIVA:
=================================
Esta funcion implementa una interfaz de chat amigable que demuestra
las capacidades del cliente. Incluye:
1. LOOP PRINCIPAL: Ciclo de entrada/respuesta con manejo de errores
2. COMANDOS ESPECIALES: Shortcuts para funcionalidades comunes
3. GRACEFUL EXIT: Salida limpia con confirmacion
4. ERROR RECOVERY: Continuacion tras errores no fatales
EVOLUCION DE LAS INTERFACES:
===========================
EPOCA ANTERIOR (Web Scraping):
- Interfaces complejas que simulaban navegadores
- Manejo manual de cookies y sesiones
- Interfaces fragiles ante cambios del DOM
EPOCA ACTUAL (API REST):
- Interfaces simples y robustas
- Estado manejado por la propia API
- Enfoque en UX y manejo de errores
COMANDOS DISPONIBLES:
====================
- /help: Muestra ayuda
- /clear: Limpia el historial de conversacion
- /model: Muestra el modelo actual
- /quit, /exit: Sale del chat
"""
print("\n" + "="*60)
print("🤖 HUGGINGCHAT CLIENT - CHAT INTERACTIVO")
print("="*60)
print(f"Modelo activo: {self.model}")
print("Comandos disponibles:")
print(" /help - Muestra esta ayuda")
print(" /clear - Limpia el historial de conversacion")
print(" /model - Muestra informacion del modelo")
print(" /quit - Sale del chat")
print("="*60)
print("Escribe tu mensaje y presiona Enter. Ctrl+C para salir.")
print()
try:
while True:
try:
# Obtenemos input del usuario con prompt personalizado
user_input = input("👤 Tu: ").strip()
# Manejamos comandos especiales
if user_input.lower() in ['/quit', '/exit', '/q']:
print("👋 Hasta luego!")
break
elif user_input.lower() in ['/help', '/h']:
print("\n📖 AYUDA:")
print(" /help - Muestra esta ayuda")
print(" /clear - Limpia el historial de conversacion")
print(" /model - Muestra informacion del modelo")
print(" /quit - Sale del chat")
print()
continue
elif user_input.lower() in ['/clear', '/c']:
self.conversation_history = []
print("🧹 Historial de conversacion limpiado")
print()
continue
elif user_input.lower() in ['/model', '/m']:
print(f"\n🤖 INFORMACION DEL MODELO:")
print(f" Modelo: {self.model}")
print(f" Endpoint: {self.chat_endpoint}")
print(f" Historial: {len(self.conversation_history)} mensajes")
print()
continue
# Ignoramos entradas vacias
if not user_input:
continue
# Enviamos el mensaje al modelo
print("🤖 Asistente: ", end="", flush=True)
response = self.send_message(user_input)
print(response)
print()
except ValueError as e:
print(f"❌ Error de validacion: {e}")
print()
except ConnectionError as e:
print(f"🌐 Error de conexion: {e}")
print("Intentalo nuevamente en unos momentos.")
print()
except RuntimeError as e:
print(f"⚠️ Error del servicio: {e}")
print()
except Exception as e:
print(f"💥 Error inesperado: {e}")
print("El chat continuara funcionando.")
print()
except KeyboardInterrupt:
print("\n\n👋 Chat interrumpido por el usuario. Hasta luego!")
except Exception as e:
print(f"\n💥 Error critico: {e}")
print("El chat debe cerrarse.")
def clear_history(self) -> None:
"""
Limpia el historial de conversacion
UTILIDAD DEL CLEAR:
==================
Limpiar el historial es util para:
1. Iniciar una nueva conversacion sin contexto previo
2. Liberar memoria en conversaciones muy largas
3. Evitar que contexto irrelevante afecte nuevas respuestas
4. Reset del estado para testing o debugging
"""
self.conversation_history = []
print("🧹 Historial de conversacion limpiado")
def get_conversation_history(self) -> List[Dict[str, str]]:
"""
Retorna el historial de conversacion actual
Returns:
List[Dict[str, str]]: Lista de mensajes con roles y contenido
"""
return self.conversation_history.copy()
def set_model(self, model: str) -> None:
"""
Cambia el modelo de lenguaje a usar
MODELOS POPULARES EN HUGGING FACE:
==================================
- meta-llama/Llama-3.2-3B-Instruct: Balanceado, rapido
- meta-llama/Llama-3.2-1B-Instruct: Muy rapido, menos capaz
- microsoft/DialoGPT-large: Especializado en conversacion
- google/flan-t5-xxl: Bueno para tareas especificas
Args:
model (str): Nombre del modelo en formato "org/model-name"
"""
old_model = self.model
self.model = model
try:
# Validamos que el nuevo modelo funciona
self._validate_connection()
print(f"✅ Modelo cambiado de '{old_model}' a '{model}'")
except Exception as e:
# Si falla, revertimos al modelo anterior
self.model = old_model
raise ValueError(f"No se pudo cambiar al modelo '{model}': {e}")
def __del__(self):
"""
Destructor que cierra la sesion HTTP limpiamente
IMPORTANCIA DE LA LIMPIEZA:
==========================
Cerrar las conexiones HTTP apropiadamente:
1. Libera recursos del sistema operativo
2. Evita warnings de Python sobre conexiones abiertas
3. Permite reutilizar puertos de red
4. Es una buena practica de programacion defensiva
"""
if hasattr(self, 'session'):
self.session.close()