ProyectoFinalDockerObservabilidad

Sleeping

ProyectoFinalDockerObservabilidad / app.py

Update app.py

d33da51 verified 7 months ago

3.51 kB

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import pipeline
import time
import logging
import torch

# 1. CONFIGURACIÓN DE LOGGING (Observabilidad: Logs)
# Esto permite que los mensajes aparezcan en la consola de Hugging Face
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# 2. MÉTRICAS EN MEMORIA (Observabilidad: Métricas)
# Diccionario para rastrear el uso de la API sin base de datos
METRICAS = {
    "total_predicciones": 0,
    "errores": 0,
    "inicio_sistema": time.time()
}

# Estructuras de Pydantic
class EntradaTexto(BaseModel):
    texto: str

class Prediccion(BaseModel):
    sentimiento: str
    confianza: str
    tiempo_ms: float
    texto: str
    contador: int # Nueva métrica para el cliente

app = FastAPI(title="MercaLatam Sentiment Analysis - Observable", version="2.0.0")

# 3. CARGA ROBUSTA DEL MODELO
logger.info("Iniciando carga del modelo DistilBERT de MercaLatam...")
try:
    classifier = pipeline(
        "sentiment-analysis", 
        model="./saved_model", 
        tokenizer="./saved_model",
        device=-1 # Forzamos CPU para el entorno Space
    )
    logger.info("✅ Modelo cargado exitosamente y listo para inferencia.")
except Exception as e:
    logger.error(f"❌ Error crítico al cargar el modelo: {e}")
    classifier = None

@app.get("/health")
def health():
    """Endpoint de salud mejorado con métricas de observabilidad"""
    uptime = round(time.time() - METRICAS["inicio_sistema"], 2)
    return {
        "status": "ok" if classifier else "error",
        "uptime_segundos": uptime,
        "predicciones_totales": METRICAS["total_predicciones"],
        "errores_registrados": METRICAS["errores"],
        "proyecto": "MercaLatam"
    }

@app.post("/predict", response_model=Prediccion)
def predecir(entrada: EntradaTexto):
    METRICAS["total_predicciones"] += 1
    
    # Validación con log de advertencia
    if not entrada.texto.strip():
        METRICAS["errores"] += 1
        logger.warning("Petición recibida con texto vacío.")
        raise HTTPException(status_code=400, detail="Texto vacío")
    
    if classifier is None:
        METRICAS["errores"] += 1
        logger.error("Intento de predicción sin modelo cargado.")
        raise HTTPException(status_code=503, detail="Modelo no disponible")

    try:
        inicio = time.time()
        
        # Inferencia
        res = classifier(entrada.texto)[0]
        
        tiempo_ms = round((time.time() - inicio) * 1000, 2)
        
        # Mapeo según tu entrenamiento
        label_map = {"LABEL_0": "NEGATIVO 🚩", "LABEL_1": "POSITIVO ✅"}
        sentimiento = label_map.get(res['label'], res['label'])
        
        # Log de éxito (Audit trail)
        logger.info(f"Predicción #{METRICAS['total_predicciones']} | Sentimiento: {sentimiento} | Latencia: {tiempo_ms}ms")
        
        return Prediccion(
            sentimiento=sentimiento,
            confianza=f"{round(res['score'] * 100, 2)}%",
            tiempo_ms=tiempo_ms,
            texto=entrada.texto,
            contador=METRICAS["total_predicciones"]
        )

    except Exception as e:
        METRICAS["errores"] += 1
        logger.error(f"Error durante la inferencia para el texto '{entrada.texto[:20]}...': {e}")
        raise HTTPException(status_code=500, detail="Error interno procesando la solicitud")

	from fastapi import FastAPI, HTTPException
	from pydantic import BaseModel
	from transformers import pipeline
	import time
	import logging
	import torch

	# 1. CONFIGURACIÓN DE LOGGING (Observabilidad: Logs)
	# Esto permite que los mensajes aparezcan en la consola de Hugging Face
	logging.basicConfig(
	level=logging.INFO,
	format="%(asctime)s [%(levelname)s] %(message)s",
	handlers=[logging.StreamHandler()]
	)
	logger = logging.getLogger(__name__)

	# 2. MÉTRICAS EN MEMORIA (Observabilidad: Métricas)
	# Diccionario para rastrear el uso de la API sin base de datos
	METRICAS = {
	"total_predicciones": 0,
	"errores": 0,
	"inicio_sistema": time.time()
	}

	# Estructuras de Pydantic
	class EntradaTexto(BaseModel):
	texto: str

	class Prediccion(BaseModel):
	sentimiento: str
	confianza: str
	tiempo_ms: float
	texto: str
	contador: int # Nueva métrica para el cliente

	app = FastAPI(title="MercaLatam Sentiment Analysis - Observable", version="2.0.0")

	# 3. CARGA ROBUSTA DEL MODELO
	logger.info("Iniciando carga del modelo DistilBERT de MercaLatam...")
	try:
	classifier = pipeline(
	"sentiment-analysis",
	model="./saved_model",
	tokenizer="./saved_model",
	device=-1 # Forzamos CPU para el entorno Space
	)
	logger.info("✅ Modelo cargado exitosamente y listo para inferencia.")
	except Exception as e:
	logger.error(f"❌ Error crítico al cargar el modelo: {e}")
	classifier = None

	@app.get("/health")
	def health():
	"""Endpoint de salud mejorado con métricas de observabilidad"""
	uptime = round(time.time() - METRICAS["inicio_sistema"], 2)
	return {
	"status": "ok" if classifier else "error",
	"uptime_segundos": uptime,
	"predicciones_totales": METRICAS["total_predicciones"],
	"errores_registrados": METRICAS["errores"],
	"proyecto": "MercaLatam"
	}

	@app.post("/predict", response_model=Prediccion)
	def predecir(entrada: EntradaTexto):
	METRICAS["total_predicciones"] += 1

	# Validación con log de advertencia
	if not entrada.texto.strip():
	METRICAS["errores"] += 1
	logger.warning("Petición recibida con texto vacío.")
	raise HTTPException(status_code=400, detail="Texto vacío")

	if classifier is None:
	METRICAS["errores"] += 1
	logger.error("Intento de predicción sin modelo cargado.")
	raise HTTPException(status_code=503, detail="Modelo no disponible")

	try:
	inicio = time.time()

	# Inferencia
	res = classifier(entrada.texto)[0]

	tiempo_ms = round((time.time() - inicio) * 1000, 2)

	# Mapeo según tu entrenamiento
	label_map = {"LABEL_0": "NEGATIVO 🚩", "LABEL_1": "POSITIVO ✅"}
	sentimiento = label_map.get(res['label'], res['label'])

	# Log de éxito (Audit trail)
	logger.info(f"Predicción #{METRICAS['total_predicciones']} \| Sentimiento: {sentimiento} \| Latencia: {tiempo_ms}ms")

	return Prediccion(
	sentimiento=sentimiento,
	confianza=f"{round(res['score'] * 100, 2)}%",
	tiempo_ms=tiempo_ms,
	texto=entrada.texto,
	contador=METRICAS["total_predicciones"]
	)

	except Exception as e:
	METRICAS["errores"] += 1
	logger.error(f"Error durante la inferencia para el texto '{entrada.texto[:20]}...': {e}")
	raise HTTPException(status_code=500, detail="Error interno procesando la solicitud")