Download src/huggingchat_client/client.py from FLAFUENTE/HUGCHAT: direct link, hf CLI and curl.
- Browser
- Download file 24.9 kB
-
https://huggingface.co/spaces/FLAFUENTE/HUGCHAT/resolve/main/src/huggingchat_client/client.py
- Command line
-
hf download hf://spaces/FLAFUENTE/HUGCHAT/src/huggingchat_client/client.py
-
curl -L -o client.py https://huggingface.co/spaces/FLAFUENTE/HUGCHAT/resolve/main/src/huggingchat_client/client.py
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() |