""" 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()