Implementación de ingeniería de IA

Has ejecutado un modelo en tu computadora portátil que genera respuestas rápida y correctamente, pero cuando lo pones en un servidor y cien usuarios lo llaman al mismo tiempo, la situación cambia:Algunos usuarios esperan diez segundos para recibir una respuesta, algunas solicitudes simplemente superan el tiempo de espera y arrojan errores, la memoria de la GPU se llena en un momento, y el importe de la factura sube de forma que te duele.

Este es el problema que debe resolver el despliegue de IA:Convertir un modelo que funciona en un servicio útil.。

El cuello de botella de los servicios web tradicionales suele estar en la CPU y la base de datos.

El cuello de botella de los servicios de IA se encuentra principalmente en la GPU: si hay suficiente memoria de video, si el cálculo es rápido, y cómo se ponen en cola las solicitudes concurrentes.

Los desafíos centrales del despliegue de IA se pueden resumir en tres palabras: latencia, rendimiento (throughput) y costo.

Métrica	Significado	Experiencia del usuario
Latencia (Latency)	Tiempo desde que el usuario envía la solicitud hasta que recibe la primera palabra	¿Es rápido?
Rendimiento (Throughput)	Cuántas solicitudes puede procesar por segundo	Si puede atender a muchas personas al mismo tiempo
Costo (Cost)	Costo de GPUs/servidores para ejecutar el servicio	¿Es caro?

Un buen despliegue de IA consiste en encontrar el equilibrio entre estos tres: latencia suficientemente baja, rendimiento suficientemente alto y costo controlable.

Marco de servicio de modelos

Para envolver un modelo entrenado como un servicio API, se necesita un marco especializado. Esta sección presenta tres opciones principales: vLLM, TGI y Ollama.

vLLM: Inferencia de alto rendimiento impulsada por PagedAttention

vLLM es un motor de inferencia desarrollado por el equipo de UC Berkeley, y su mayor característica es la velocidad.

La innovación principal de vLLM esPagedAttention—una técnica para gestionar eficientemente la memoria de video, inspirada en la paginación de memoria virtual de los sistemas operativos.

En los marcos de inferencia tradicionales, la caché KV (cache de claves y valores) de cada solicitud necesita ocupar un espacio contiguo de memoria de video.

Cuando las solicitudes tienen longitudes diferentes, la memoria de video se fragmenta y la utilización es baja.

PagedAttention divide la memoria de video en páginas de tamaño fijo; la caché KV de cada solicitud puede almacenarse de forma dispersa en diferentes páginas, y las posiciones se registran mediante una tabla de páginas.

De esta manera, la utilización de la memoria de video mejora significativamente y se pueden atender más solicitudes al mismo tiempo.

Ejemplo

# Instalar vLLM
pip install vllm

# Iniciar un servicio API compatible con OpenAI usando vLLM
# --model especifica el modelo, --host y --port especifican la dirección de escucha
# --tensor-parallel-size para usar varias GPUs en inferencia paralela
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--host 0.0.0.0 \
--port 8000 \
--tensor-parallel-size 1 \
--gpu-memory-utilization 0.9 \
--max-model-len 8192

Una vez iniciado el servicio, se puede llamar directamente con el SDK de OpenAI:

Ejemplo

# Ruta del archivo: test_vllm_client.py
from openai import OpenAI

# Conectar al servicio vLLM local
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="ejemplo-demo-key" # vLLM no requiere una clave API real por defecto
)

# Llamar a chat completions (completado de chat)
response = client.chat.completions.create(
model="Qwen/Qwen2.5-7B-Instruct",
messages=[
{"role": "system", "content": "Eres un asistente de IA útil."},
{"role": "user", "content": "Presenta el tutorial de ejemplo en una oración."}
],
temperature=0.7,
max_tokens=500,
stream=True # Salida en streaming
)

print("Respuesta de IA:", end="", flush=True)
for chunk in response:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

vLLM también admite la escritura de servicios personalizados directamente con código Python:

Ejemplo

# Ruta del archivo: custom_vllm_server.py
from vllm import LLM, SamplingParams

# Inicializar el modelo
llm = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
gpu_memory_utilization=0.9, # Proporción de uso de memoria de video GPU
tensor_parallel_size=1, # Grado de paralelismo tensorial (número de GPUs)
max_model_len=8192, # Longitud máxima de contexto
)

# Configurar parámetros de muestreo
sampling_params = SamplingParams(
temperature=0.7,
top_p=0.9,
max_tokens=500,
stop=["</s>"],
)

# Inferencia por lotes
prompts = [
"Presenta ejemplo",
"¿Qué es la IA?",
"¿Cómo aprender a programar?",
]

# Generar respuesta
outputs = llm.generate(prompts, sampling_params)

# Imprimir resultado
for output in outputs:
prompt = output.prompt
generated_text = output.outputs[0].text
print(f"Prompt: {prompt}")
print(f"Generated: {generated_text}")
print("-" * 50)

TGI: Motor de inferencia de Hugging Face

TGI (Text Generation Inference) es un marco de servicio de inferencia lanzado por Hugging Face.

Su característica es un ecosistema completo, con integración profunda con Hugging Face Hub, y soporte para optimizaciones como Flash Attention y procesamiento por lotes dinámico.

Ejemplo

# Ejecutar TGI con Docker (la forma más sencilla)
# Nota: primero debe instalar Docker y configurar el soporte de GPU
docker run -d \
--gpus all \
-p 8080:80 \
-v $PWD/data:/data \
--name tgi-server \
ghcr.io/huggingface/text-generation-inference:latest \
--model-id Qwen/Qwen2.5-7B-Instruct \
--max-input-length 4096 \
--max-total-tokens 8192 \
--max-batch-prefill-tokens 4096

# O instalar directamente con Python (entorno de desarrollo)
# pip install text-generation

Llamar al servicio TGI:

Ejemplo

# Ruta del archivo: test_tgi_client.py
from text_generation import Client

# Conectar al servicio TGI
client = Client("http://localhost:8080", timeout=60)

# Llamada sin streaming
response = client.generate(
"Presenta el tutorial de ejemplo",
max_new_tokens=500,
temperature=0.7,
top_p=0.9,
)
print(response.generated_text)

print("-" * 50)

# Llamada con streaming
print("Salida en streaming:")
for chunk in client.generate_stream(
"Escribe un Hello World en Python",
max_new_tokens=200,
):
if not chunk.token.special:
print(chunk.token.text, end="", flush=True)
print()

Ollama: De la demostración local al despliegue en producción

Ollama es conocido por "ejecutar modelos locales con un clic", y muchos lo usan para desarrollo y demostraciones.

Pero en realidad Ollama también es adecuado para despliegues de producción a pequeña escala: es simple, estable y con uso de recursos controlable.

Ejemplo

# Instalar Ollama (servidor Linux)
curl -fsSL https://ollama.com/install.sh | sh

# Configurar Ollama para escuchar en todas las interfaces de red (preste atención al firewall en producción)
# Editar el archivo de servicio systemd
sudo systemctl edit ollama.service

# Agregar el siguiente contenido:
# [Service]
# Environment="OLLAMA_HOST=0.0.0.0:11434"
# Environment="OLLAMA_MODELS=/var/lib/ollama/models"

# Reiniciar el servicio
sudo systemctl restart ollama

# Descargar y ejecutar el modelo
ollama pull qwen2.5:7b

Llamar a Ollama mediante su API:

Ejemplo

# Ruta del archivo: test_ollama_client.py
import requests
import json

# Dirección de la API de Ollama
base_url = "http://localhost:11434/api"

# Interfaz de chat (streaming)
def chat_stream(model, messages):
url = f"{base_url}/chat"
data = {
"model": model,
"messages": messages,
"stream": True,
"options": {
"temperature": 0.7,
"num_ctx": 8192,
}
}

response = requests.post(url, json=data, stream=True)
full_response = ""

for line in response.iter_lines():
if line:
chunk = json.loads(line)
if "message" in chunk:
content = chunk["message"].get("content", "")
full_response += content
print(content, end="", flush=True)
if chunk.get("done", False):
break

print()
return full_response

# Ejemplo de llamada
messages = [
{"role": "system", "content": "Eres un asistente de IA útil."},
{"role": "user", "content": Presenta el tutorial de ejemplo}
]

print(Respuesta de la IA:, end="")
result = chat_stream("qwen2.5:7b", messages)

Comparativa de marcos de trabajo

Los tres frameworks tienen sus propias características. ¿Cómo elegir? Depende de tus necesidades.

Framework	Ventajas	Desventajas	Escenarios adecuados
vLLM	El más potente, con el mayor throughput	Configuración relativamente compleja	Entornos de producción de alta concurrencia, buscando máximo rendimiento
TGI	Ecosistema Hugging Face, despliegue sencillo	Rendimiento ligeramente inferior a vLLM	Si ya usas el ecosistema de herramientas de Hugging Face
Ollama	Despliegue minimalista, bajo costo de mantenimiento	No apto para concurrencia muy alta	Servicios de pequeña y mediana escala, lanzamiento rápido

Etapa inicial: usa primero Ollama, simple y confiable. Cuando aumente el tráfico: migra a vLLM, priorizando el rendimiento.

Técnicas de optimización de inferencia

El mismo modelo, ejecutado de diferentes maneras, puede tener diferencias de varias veces en velocidad y costo.

Esta sección presenta técnicas de optimización de inferencia comunes en entornos de producción.

Procesamiento por lotes continuo (Continuous Batching)

El procesamiento por lotes tradicional es estático: se acumula un lote de solicitudes, se procesan juntas, y solo después de completarlas todas se procesa el siguiente lote.

El problema es: algunas solicitudes son muy cortas (por ejemplo, generar solo 10 caracteres), y otras muy largas (generar 500 caracteres).

Las solicitudes cortas deben esperar a que las largas terminen para liberar recursos juntas, y la utilización de la GPU no aumenta.

Procesamiento por lotes continuoes dinámico: cuando una solicitud se completa, inmediatamente se agrega una nueva solicitud de la cola, sin esperar a que todo el lote termine.

Así la GPU trabaja casi siempre a plena carga, y el throughput aumenta significativamente.

Tanto vLLM como TGI incluyen procesamiento por lotes continuo integrado; no necesitas implementarlo manualmente, solo activarlo.

Muestreo especulativo (Speculative Decoding)

Un modelo grande, para generar un carácter, necesita ejecutar una propagación hacia adelante completa, lo cual es lento.

La idea del muestreo especulativo es:usar un modelo pequeño para "adivinar" los siguientes caracteres, y luego hacer que el modelo grande los verifique de una sola vez.。

El modelo pequeño adivina rápido; aunque no siempre acierta, la parte acertada puede emitirse en lote.

La parte incorrecta se descarta y se vuelve a adivinar.

De esta manera, la velocidad general puede mejorar de 2 a 3 veces, y la calidad de salida no cambia.

Ejemplo

# Habilitar muestreo especulativo en vLLM
# --speculative-model especifica el modelo pequeño (modelo borrador)
# --num-speculative-tokens especifica cuántos caracteres adivinar cada vez
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen2.5-7B-Instruct \
--speculative-model Qwen/Qwen2.5-0.5B-Instruct \
--num-speculative-tokens 5 \
--host 0.0.0.0 \
--port 8000

Inferencia cuantizada: AWQ y GPTQ

La cuantización de modelos consiste en comprimir los pesos del modelo desde números de punto flotante de 16 bits (FP16) a enteros de 8 o 4 bits.

La ocupación de memoria de video se reduce a la mitad o incluso menos, y la velocidad de inferencia puede ser aún mayor.

Los dos esquemas de cuantización más utilizados son AWQ y GPTQ.

Esquema de cuantización	Características	Pérdida de precisión	Velocidad de inferencia
AWQ	Cuantización de pesos sensible a activaciones	Mínima	Rápido
GPTQ	Cuantización optimizada por capas	Pequeño	Rápido
GGUF	Adecuada para ejecución local	Aceptable	Rápido

Ejemplo

# Ejecutar directamente un modelo cuantizado con AWQ usando vLLM
# Muchos modelos cuantizados se pueden descargar directamente desde Hugging Face
python -m vllm.entrypoints.openai.api_server \
--model TheBloke/Qwen2.5-7B-Instruct-AWQ \
--quantization awq \
--host 0.0.0.0 \
--port 8000

# O usar Ollama para ejecutar modelos cuantizados GGUF
ollama run qwen2.5:7b-instruct-q4_0

Cargar un modelo cuantizado AWQ con Python:

Ejemplo

# Ruta del archivo: awq_inference.py
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# Cargar modelo cuantizado AWQ
model_name = "TheBloke/Qwen2.5-7B-Instruct-AWQ"

# Nota: para ejecutarlo realmente se necesita instalar la librería autoawq
# pip install autoawq

tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)

# Inferencia
messages = [
{"role": "system", "content": "Eres un asistente de IA servicial."},
{"role": "user", "content": "Presenta ejemplo"}
]

text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)

inputs = tokenizer([text], return_tensors="pt").to(model.device)

with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=500,
temperature=0.7,
)

response = tokenizer.decode(
outputs[0][inputs.input_ids.shape[1]:],
skip_special_tokens=True
)
print(response)

Destilación de modelos

La destilación consiste en usar un modelo grande (maestro) para entrenar un modelo pequeño (estudiante), de modo que el modelo pequeño aprenda las capacidades del modelo grande.

El modelo pequeño es más rápido y más económico; aunque su rendimiento es ligeramente inferior, es suficiente en muchos escenarios.

Por ejemplo, usando Qwen2.5-72B como maestro para entrenar un estudiante Qwen2.5-7B, el resultado puede acercarse al 72B, pero con la velocidad y el costo de un 7B.

Los beneficios de la destilación suelen ser:El tamaño del modelo se reduce a 1/10, la velocidad de inferencia mejora de 5 a 10 veces, y el rendimiento se conserva en más del 90%.。

Diseño de servicios API

Envolver el modelo es solo el primer paso; una API en producción necesita muchas más funciones: autenticación, limitación de velocidad, respuestas en streaming y manejo de errores.

FastAPI es una excelente opción para esto.

Encapsulación de servicios LLM con FastAPI

Vamos a escribir un servicio API completo que incluya autenticación, limitación de velocidad y respuestas en streaming.

Ejemplo

# Ruta del archivo: llm_api_server.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import List, Optional, AsyncGenerator
from enum import Enum
import time
import asyncio
import uuid
from datetime import datetime, timedelta
from collections import defaultdict

# ============================================
# 1. Inicialización de la aplicación y configuración
# ============================================

app = FastAPI(title="EJEMPLO LLM API", version="1.0.0")

# Autenticación simple por API Key (en producción se recomienda un esquema más robusto)
VALID_API_KEYS = {
"sk-ejemplo-123456": {"user": "demo_user", "quota": 1000},
"sk-ejemplo-789012": {"user": "pro_user", "quota": 10000},
}

# Limitador de velocidad simple (implementación en memoria, en producción se recomienda Redis)
class RateLimiter:
def __init__(self):
self.requests = defaultdict(list) # key -> [timestamp, ...]

async def is_allowed(self, key: str, max_requests: int, window_seconds: int) -> bool:
"""Comprueba si se ha superado el límite de velocidad"""
now = time.time()
# Limpiar registros expirados
self.requests[key] = [t for t in self.requests[key] if now - t < window_seconds]
# Comprobar si se superó el límite
if len(self.requests[key]) >= max_requests:
return False
# Registrar esta solicitud
self.requests[key].append(now)
return True

rate_limiter = RateLimiter()

# ============================================
# 2. Modelo de datos
# ============================================

class Role(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"

class Message(BaseModel):
role: Role = Field(..., description="Rol del mensaje")
content: str = Field(..., description="Contenido del mensaje")

class ChatCompletionRequest(BaseModel):
model: str = Field(..., description="Nombre del modelo")
messages: List[Message] = Field(..., description="Lista de mensajes de la conversación")
temperature: Optional[float] = Field(default=0.7, ge=0, le=2, description="Temperatura de muestreo")
top_p: Optional[float] = Field(default=0.9, ge=0, le=1, description="Muestreo Top-P")
max_tokens: Optional[int] = Field(default=500, ge=1, le=4096, description="Número máximo de tokens a generar")
stream: Optional[bool] = Field(default=False, description="Si se debe emitir en streaming")

class ChatCompletionResponse(BaseModel):
id: str
object: str = "chat.completion"
created: int
model: str
choices: List[dict]
usage: dict

# ============================================
# 3. Dependencia de autenticación
# ============================================

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""Verifica la API Key y devuelve la información del usuario"""
api_key = credentials.credentials
if api_key not in VALID_API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API Key"
)
return VALID_API_KEYS[api_key]

# ============================================
# 4. Simulación de generación del LLM (en proyectos reales, reemplazar por una llamada a un modelo real)
# ============================================

async def mock_llm_generate(messages: List[Message], max_tokens: int) -> str:
"""Simula el proceso de generación del LLM"""
# Esto es solo una simulación; en un proyecto real deberías llamar al modelo real
response_text = (
"¡Hola! Soy el asistente de IA de EJEMPLO."\n\n"
"EJEMPLO ofrece una gran variedad de tutoriales de programación, incluyendo Python, Java, C++, desarrollo frontend, etc."\n"
"¡Puedes visitar https://www.example.com para aprender más contenido!"\n\n"
"¿En qué puedo ayudarte?"
)
# Simular latencia de generación
await asyncio.sleep(0.5)
return response_text

async def mock_llm_generate_stream(messages: List[Message], max_tokens: int) -> AsyncGenerator[str, None]:
"""Simula la generación en streaming del LLM"""
response_text = await mock_llm_generate(messages, max_tokens)
# Salida carácter por carácter para simular el efecto de streaming
for char in response_text:
yield char
await asyncio.sleep(0.03) # Retraso de 30 ms por carácter

# ============================================
# 5. Endpoints de la API
# ============================================

@app.post("/v1/chat/completions")
async def chat_completions(
request: ChatCompletionRequest,
user: dict = Depends(get_current_user)
):
"""Interfaz de chat completion, compatible con el formato de OpenAI"""

# Verificación de límite de velocidad: máximo 60 solicitudes por minuto por usuario
user_key = user["user"]
if not await rate_limiter.is_allowed(user_key, max_requests=60, window_seconds=60):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Rate limit exceeded"
)

# Generar ID de solicitud
request_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
created = int(datetime.now().timestamp())

if request.stream:
# Respuesta en streaming
async def stream_generator():
async for token in mock_llm_generate_stream(request.messages, request.max_tokens):
# Construir chunk en formato SSE
chunk = {
"id": request_id,
"object": "chat.completion.chunk",
"created": created,
"model": request.model,
"choices": [{
"index": 0,
"delta": {"content": token},
"finish_reason": None
}]
}
yield f"data: {__import__('json').dumps(chunk)}\n\n"
# Enviar marcador de finalización
yield "data: [DONE]\n\n"

return StreamingResponse(
stream_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"}
)

else:
# Respuesta no streaming
response_text = await mock_llm_generate(request.messages, request.max_tokens)

return ChatCompletionResponse(
id=request_id,
created=created,
model=request.model,
choices=[{
"index": 0,
"message": {
"role": "assistant",
"content": response_text
},
"finish_reason": "stop"
}],
usage={
"prompt_tokens": sum(len(m.content) for m in request.messages),
"completion_tokens": len(response_text),
"total_tokens": sum(len(m.content) for m in request.messages) + len(response_text)
}
)

@app.get("/health")
async def health_check():
"""Endpoint de verificación de salud"""
return {"status": "ok", "timestamp": datetime.now().isoformat()}

@app.get("/v1/models")
async def list_models(user: dict = Depends(get_current_user)):
"""Listar los modelos disponibles"""
return {
"object": "list",
"data": [
{
"id": "qwen2.5-7b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "ejemplo"
},
{
"id": "llama-3.1-8b-instruct",
"object": "model",
"created": 1699000000,
"owned_by": "ejemplo"
}
]
}

if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

Iniciar el servicio:

Ejemplo

# Instalar dependencias
pip install fastapi uvicorn python-multipart pydantic

# Iniciar el servicio
python llm_api_server.py

# Probar en otra terminal
curl -X POST http://localhost:8000/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-ejemplo-123456" \
-d '{
"model": "qwen2.5-7b-instruct",
"messages": [
{"role": "user", "content": "Hola"}
],
"stream": false
}'

Llamar con un cliente Python:

Ejemplo

# Ruta del archivo: client_example.py
from openai import OpenAI

# Conectar a nuestro propio servicio API
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="sk-ejemplo-123456"
)

# Probar el modo no streaming
print("Llamada no streaming:")
response = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Presenta ejemplo"}],
stream=False
)
print(response.choices[0].message.content)

print("\n" + "="*50 + "\n")

# Probar el modo streaming
print("Llamada en streaming:")
stream = client.chat.completions.create(
model="qwen2.5-7b-instruct",
messages=[{"role": "user", "content": "Presenta ejemplo"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

Ventajas de interfaces compatibles con OpenAI

Hacer que tu API sea compatible con el formato de OpenAI tiene varias ventajas claras:

Los usuarios no necesitan cambiar su código, solo modificar base_url y api_key.

Ecosistema rico: cualquier SDK, herramienta o aplicación que admita OpenAI puede usar directamente tu servicio.

No necesitas diseñar tu propio formato de API; el diseño de OpenAI ya está muy bien pensado.

Autenticación y limitación de velocidad en producción

El ejemplo anterior usó limitación de velocidad simple en memoria; en producción se recomienda usar Redis para limitación distribuida.

Ejemplo

# Ruta del archivo: redis_rate_limiter.py
import redis.asyncio as redis
import time

class RedisRateLimiter:
"""Limitador de velocidad distribuido basado en Redis"""

def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)

async def is_allowed(
self,
key: str,
max_requests: int,
window_seconds: int
) -> bool:
"""
Algoritmo de limitación de velocidad por ventana deslizante
key: clave de limitación (por ejemplo, user_id o api_key)
max_requests: número máximo de solicitudes en la ventana
window_seconds: tamaño de la ventana (segundos)
"""
now = time.time()
window_start = now - window_seconds

# Claves de Redis
redis_key = f"rate_limit:{key}"

# Usar operación atómica con pipeline
async with self.redis.pipeline() as pipe:
# 1. Eliminar registros fuera de la ventana
pipe.zremrangebyscore(redis_key, 0, window_start)
# 2. Contar las solicitudes dentro de la ventana actual
pipe.zcard(redis_key)
# 3. Agregar la solicitud actual
pipe.zadd(redis_key, {str(now): now})
# 4. Establecer tiempo de expiración
pipe.expire(redis_key, window_seconds)
# Ejecutar
_, current_count, _, _ = await pipe.execute()

return current_count < max_requests

async def close(self):
await self.redis.close()

# Ejemplo de uso
async def main():
limiter = RedisRateLimiter("redis://localhost:6379")

# Prueba: mismo usuario, máximo 10 solicitudes por minuto
user_id = "user_123"
for i in range(15):
allowed = await limiter.is_allowed(user_id, max_requests=10, window_seconds=60)
print(f"Solicitud {i+1}: {'Permitido' if allowed else 'Denegado'}")

await limiter.close()

if __name__ == "__main__":
import asyncio
asyncio.run(main())

La autenticación y la limitación de velocidad son la infraestructura de una API; es mejor hacerlas un poco más complejas que arrepentirse cuando te lo saturan.

Balanceo de carga y escalado

Cuando hay cada vez más usuarios y un solo servidor no da abasto, necesitas añadir máquinas y distribuir el tráfico entre varias.

Esto es lo que resuelven el balanceo de carga y el escalado automático.

Diseño de servicios sin estado

Primero, recuerda:El servicio de IA debe ser sin estado (stateless)。

Sin estado significa que las solicitudes no comparten memoria; cualquier servidor puede procesar cualquier solicitud.

Así el escalado es simple: solo añade máquinas, sin preocuparte por la sincronización de datos.

El estado de la sesión (por ejemplo, el historial de la conversación) puede almacenarse en el cliente (para que el usuario lo envíe cada vez) o en Redis.

Proxy inverso con Nginx

Nginx es el balanceador de carga más utilizado; puede distribuir el tráfico entre múltiples servidores backend.

Ejemplo

# Ruta del archivo: nginx.conf
# Configuración de Nginx: balanceo de carga para múltiples servicios LLM

upstream llm_backends {
# Algoritmo de menos conexiones: envía la solicitud al servidor con menos conexiones actuales
least_conn;

# Lista de servidores backend
server 192.168.1.101:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.102:8000 weight=1 max_fails=3 fail_timeout=30s;
server 192.168.1.103:8000 weight=1 max_fails=3 fail_timeout=30s;

# Si hay varios modelos de GPU, puedes asignar diferentes proporciones de tráfico según el peso (weight)
# weight=2 significa que este servidor recibe el doble de tráfico que los demás
}

server {
listen 80;
server_name api.example.com;

# Reenvío de solicitudes API
location /v1/ {
# Reenviar al backend
proxy_pass http://llm_backends;

# Configuración de tiempo de espera (la generación de IA puede ser lenta)
proxy_connect_timeout 10s;
proxy_send_timeout 600s;
proxy_read_timeout 600s;

# Pasar la IP real del cliente
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

# Configuración necesaria para respuestas en streaming
proxy_buffering off;
proxy_cache off;
proxy_set_header Connection '';
proxy_http_version 1.1;
}

# Endpoint de comprobación de salud
location /health {
proxy_pass http://llm_backends/health;
proxy_connect_timeout 5s;
proxy_read_timeout 5s;
}

# Límite de velocidad: máximo 10 solicitudes por segundo por IP
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
limit_req zone=api_limit burst=20 nodelay;
}

Escalado automático con Kubernetes HPA

Si despliegas con Kubernetes, puedes usar HPA (Horizontal Pod Autoscaler) para escalar automáticamente según la carga.

Ejemplo

# Ruta del archivo: llm-deployment.yaml
# Configuración de despliegue en Kubernetes para el servicio LLM

# 1. Deployment: desplegar Pods de LLM
apiVersion: apps/v1
kind: Deployment
metadata:
name: llm-server
namespace: ejemplo
spec:
replicas: 3 # Número inicial de réplicas
selector:
matchLabels:
app: llm-server
template:
metadata:
labels:
app: llm-server
spec:
containers:
- name: llm-server
image: ejemplo/llm-server:latest
ports:
- containerPort: 8000
resources:
requests:
cpu: "4"
memory: "16Gi"
nvidia.com/gpu: 1 # Solicitar 1 GPU
limits:
cpu: "8"
memory: "32Gi"
nvidia.com/gpu: 1 # Limitar a 1 GPU
env:
- name: MODEL_NAME
value: "Qwen/Qwen2.5-7B-Instruct"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300 # La carga del modelo requiere tiempo
periodSeconds: 30
readinessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 300
periodSeconds: 10
tolerations:
- key: "nvidia.com/gpu"
operator: "Exists"
effect: "NoSchedule"

---
# 2. Service: Service de balanceo de carga
apiVersion: v1
kind: Service
metadata:
name: llm-service
namespace: ejemplo
spec:
selector:
app: llm-server
ports:
- port: 80
targetPort: 8000
type: LoadBalancer

---
# 3. HPA: escalado automático
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: llm-hpa
namespace: ejemplo
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: llm-server
minReplicas: 3 # Mínimo 3 Pods
maxReplicas: 20 # Máximo 20 Pods
metrics:
# Escalar según el uso de CPU
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
# Escalar según el uso de GPU (requiere métricas personalizadas)
- type: Pods
pods:
metric:
name: gpu_utilization
target:
type: AverageValue
averageValue: 80
# Escalar según la longitud de la cola (métrica personalizada)
- type: Pods
pods:
metric:
name: pending_requests
target:
type: AverageValue
averageValue: 10
behavior:
scaleUp:
stabilizationWindowSeconds: 60 # Observar 60 segundos antes de escalar
policies:
- type: Percent
value: 50
periodSeconds: 60
scaleDown:
stabilizationWindowSeconds: 300 # Observar 5 minutos antes de reducir
policies:
- type: Percent
value: 20
periodSeconds: 120

Gestión del grupo de recursos GPU

Las GPU son caras, hay que maximizar su uso.

Una estrategia común es agrupar por tipo de modelo: colocar los Pods del mismo modelo juntos.

Por ejemplo:

Pool de recursos	Modelo de GPU	Modelo en ejecución	Uso
small-pool	RTX 3090 / L4	Modelo 7B	Tráfico diario
medium-pool	A10 / L40	Modelo 13B-70B	Tareas complejas
large-pool	A100 / H100	Modelo 70B+	Usuarios de alto valor
spot-pool	Varios tipos de máquinas	Procesamiento por lotes offline	Usar instancias Spot de bajo costo

Pruebas A/B de modelos de IA

Has lanzado una nueva versión del modelo, ¿cómo saber si es mejor que la anterior? Con pruebas A/B.

El núcleo de las pruebas A/B es:Dividir el tráfico en varios grupos, cada grupo usa un modelo diferente y comparar las métricas.。

Estrategia de división de tráfico

Hay varias formas comunes de dividir el tráfico:

Ejemplo

# Ruta del archivo: ab_test_router.py
import hashlib
import random
from typing import Dict, List, Tuple
from dataclasses import dataclass

@dataclass
class Variant:
"""Una variante de las pruebas A/B"""
name: str # Nombre de la variante
model_name: str # Nombre del modelo
weight: int # Peso del tráfico
config: dict # Otra configuración

class ABTestRouter:
"""Enrutador de tráfico para pruebas A/B"""

def __init__(self, test_name: str, variants: List[Variant]):
self.test_name = test_name
self.variants = variants
# Calcular el peso total
total_weight = sum(v.weight for v in variants)
# Construir rangos segmentados
self.ranges: List[Tuple[int, int, Variant]] = []
current = 0
for v in variants:
self.ranges.append((current, current + v.weight, v))
current += v.weight
self.total_weight = total_weight

def get_variant_by_user(self, user_id: str) -> Variant:
"""
División estable basada en el ID de usuario
El mismo usuario siempre se asigna al mismo grupo para evitar experiencias inconsistentes.
"""
# Calcular la puntuación del usuario con hash
hash_input = f"{self.test_name}-{user_id}".encode()
hash_value = hashlib.md5(hash_input).hexdigest()
# Convertir a un número de 0-999
score = int(hash_value[:4], 16) % 1000
# Buscar la variante correspondiente
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

def get_variant_random(self) -> Variant:
"""División aleatoria (adecuada para escenarios que no requieren adherencia)"""
score = random.randint(0, self.total_weight - 1)
for start, end, variant in self.ranges:
if start <= score < end:
return variant
return self.variants[-1]

# Ejemplo de uso
def main():
# Definir tres variantes
variants = [
Variant(
name="control",
model_name="Qwen2.5-7B-Instruct",
weight=50, # 50% del tráfico
config={"temperature": 0.7}
),
Variant(
name="treatment-v1",
model_name="Qwen2.5-14B-Instruct",
weight=30, # 30% del tráfico
config={"temperature": 0.7}
),
Variant(
name="treatment-v2",
model_name="Llama-3.1-8B-Instruct",
weight=20, # 20% del tráfico
config={"temperature": 0.8}
)
]

router = ABTestRouter("model-comparison-v1", variants)

# Probar la división de usuarios
test_users = [f"user_{i}" for i in range(10)]
for user_id in test_users:
variant = router.get_variant_by_user(user_id)
print(f"Usuario {user_id} → Grupo {variant.name} → Modelo {variant.model_name}")

# Estadísticas de distribución
count: Dict[str, int] = {}
for i in range(10000):
variant = router.get_variant_random()
count[variant.name] = count.get(variant.name, 0) + 1
print("\nEstadísticas de división aleatoria (10000 veces):")
for name, cnt in count.items():
print(f" {name}: {cnt} veces ({cnt/100}%)")

if __name__ == "__main__":
main()

Definición y recopilación de métricas

En las pruebas A/B no basta con mirar "qué modelo responde de forma más inteligente"; hay que definir métricas cuantificables.

Tipo de métrica	Métrica específica	Descripción
Métricas del sistema	Latencia, rendimiento, tasa de error	Si el modelo es rápido y estable
Comportamiento del usuario	Satisfacción del usuario, número de turnos de conversación, tasa de renovación	Si al usuario le gusta o no
Métricas de negocio	Tasa de conversión, retención, ingresos	Si ayuda al negocio
Métricas de calidad	Evaluación manual, tasa de contenido dañino, precisión factual	Si la calidad de la salida es buena

Ejemplo

# Ruta del archivo: metrics_collector.py
import time
import json
from dataclasses import dataclass, asdict
from typing import Optional
from datetime import datetime
import uuid

@dataclass
class ChatMetrics:
"""Métricas completas para solicitudes de chat"""
# Información básica
request_id: str
user_id: str
variant_name: str # Nombre del grupo de prueba A/B
model_name: str

# Métricas de tiempo
timestamp: str
ttft: float # Time to First Token
total_time: float # Tiempo total transcurrido

# Métricas de generación
prompt_tokens: int
completion_tokens: int
total_tokens: int

# Métricas de negocio
user_rating: Optional[int] = None # Puntuación del usuario (1-5)
is_error: bool = False
error_message: Optional[str] = None

class MetricsCollector:
"""Recopilador de métricas"""

def __init__(self):
# En un proyecto real, aquí se debería escribir en Kafka o en una base de datos
self.buffer = []

def collect(self, metrics: ChatMetrics):
"""Recopilar una métrica"""
self.buffer.append(asdict(metrics))
# Salida simple; en un proyecto real, escribir en almacenamiento
print(f"[Metrics] {json.dumps(asdict(metrics), ensure_ascii=False)}")

# Cuando el buffer esté lleno, escribir en lote
if len(self.buffer) >= 100:
self.flush()

def flush(self):
"""Escritura por lotes en almacenamiento"""
if self.buffer:
# En un proyecto real: escribir en ClickHouse, BigQuery, etc.
print(f"Flushed {len(self.buffer)} metrics")
self.buffer = []

# Ejemplo de uso
collector = MetricsCollector()

def process_chat_request(user_id: str, prompt: str, router):
"""Procesar una solicitud de chat y registrar métricas"""
request_id = str(uuid.uuid4())

# División de tráfico
variant = router.get_variant_by_user(user_id)

# Registrar la hora de inicio
start_time = time.time()
ttft = 0.0
prompt_tokens = len(prompt)
completion_tokens = 0

try:
# Aquí se reemplaza por la llamada real al modelo
# response = call_model(variant.model_name, prompt)
response = "Esta es la respuesta de la IA"
completion_tokens = len(response)

# Simular TTFT (suponer que el primer token se recibe después de 0,2 segundos)
ttft = 0.2

# Registrar métricas de éxito
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
is_error=False
)
collector.collect(metrics)
return response, request_id

except Exception as e:
# Registrar métricas de error
metrics = ChatMetrics(
request_id=request_id,
user_id=user_id,
variant_name=variant.name,
model_name=variant.model_name,
timestamp=datetime.now().isoformat(),
ttft=ttft,
total_time=time.time() - start_time,
prompt_tokens=prompt_tokens,
completion_tokens=0,
total_tokens=prompt_tokens,
is_error=True,
error_message=str(e)
)
collector.collect(metrics)
raise

# Comentarios del usuario
def submit_user_feedback(request_id: str, rating: int):
"""El usuario envía una retroalimentación con calificación"""
# En un proyecto real: actualizar el registro correspondiente en la base de datos
print(f"Request {request_id} got rating {rating}")

Determinación de la significancia estadística

Una vez obtenidos los datos, no se puede decir que el grupo B es mejor solo porque A tiene 52% y B tiene 55%.

Hay que realizar una prueba de significancia estadística para confirmar que la diferencia no se debe a fluctuaciones aleatorias.

Ejemplo

# Ruta del archivo: ab_test_analysis.py
import math
from typing import Tuple

def calculate_z_test(
control_conv: float,
treatment_conv: float,
control_total: int,
treatment_total: int
) -> Tuple[float, float]:
"""
Calcular el p-value de la prueba Z
control_conv: tasa de conversión del grupo de control
treatment_conv: tasa de conversión del grupo de tratamiento
control_total: tamaño de muestra del grupo de control
treatment_total: tamaño de muestra del grupo de tratamiento
"""
# Tasa de conversión combinada
p_pool = (control_conv * control_total + treatment_conv * treatment_total) / (control_total + treatment_total)
# Error estándar
se_pool = math.sqrt(p_pool * (1 - p_pool) * (1 / control_total + 1 / treatment_total))
# Estadístico Z
z_score = (treatment_conv - control_conv) / se_pool
# Calcular el p-value (prueba de dos colas)
p_value = 2 * (1 - normal_cdf(abs(z_score)))
return z_score, p_value

def normal_cdf(x: float) -> float:
"""Función de distribución acumulada de la normal estándar (aproximación)"""
return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0

def analyze_ab_test(
control_conversions: int,
control_total: int,
treatment_conversions: int,
treatment_total: int,
alpha: float = 0.05
):
"""Analizar los resultados de la prueba A/B"""
control_conv = control_conversions / control_total
treatment_conv = treatment_conversions / treatment_total

z_score, p_value = calculate_z_test(
control_conv, treatment_conv,
control_total, treatment_total
)

relative_lift = (treatment_conv - control_conv) / control_conv

print("Análisis de resultados de la prueba A/B")
print("=" * 50)
print(f"Grupo de control: {control_conversions}/{control_total} ({control_conv:.2%})")
print(f"Grupo de tratamiento: {treatment_conversions}/{treatment_total} ({treatment_conv:.2%})")
print(f"Mejora relativa: {relative_lift:+.2%}")
print("-" * 50)
print(f"Estadístico Z: {z_score:.4f}")
print(f"p-value: {p_value:.4f}")
print("-" * 50)

if p_value < alpha:
print(f"✅ Resultado estadísticamente significativo (p < {alpha})")
if relative_lift > 0:
print(" ¡El grupo de tratamiento se desempeñó mejor!")
else:
print(" ¡El grupo de control se desempeñó mejor!")
else:
print(f"❌ Resultado no significativo (p >= {alpha})")
print(" No se puede determinar qué versión es mejor, se necesitan más datos")

# Ejemplo: supongamos resultados de prueba A/B
if __name__ == "__main__":
# Escenario 1: diferencia significativa
print("Escenario 1: diferencia significativa")
analyze_ab_test(
control_conversions=450, # 450 personas satisfechas en el grupo de control
control_total=1000, # 1000 personas en total en el grupo de control
treatment_conversions=520, # 520 personas satisfechas en el grupo de tratamiento
treatment_total=1000 # 1000 personas en total en el grupo de tratamiento
)

print("\n" + "="*50 + "\n")

# Escenario 2: sin diferencia significativa
print("Escenario 2: sin diferencia significativa")
analyze_ab_test(
control_conversions=48,
control_total=100,
treatment_conversions=52,
treatment_total=100
)

La significancia estadística es importante. Una "mejora" no verificada podría ser solo fluctuación aleatoria.

Seguimiento de experimentos con MLflow

Entrenaste docenas de versiones del modelo, ¿cuál funciona mejor? ¿Qué versión está desplegada?

MLflow te ayuda a gestionar experimentos, rastrear modelos y registrar métricas.

Gestión de experimentos

Ejemplo

# Ruta del archivo: mlflow_experiment.py
import mlflow
import mlflow.pyfunc
import mlflow.sklearn
from mlflow.tracking import MlflowClient
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score
import numpy as np
import time

# Configurar la dirección de seguimiento de MLflow (servidor local o remoto)
mlflow.set_tracking_uri("http://localhost:5000")

# Crear o configurar el experimento
experiment_name = "ejemplo-llm-finetuning"
try:
experiment_id = mlflow.create_experiment(experiment_name)
except mlflow.exceptions.MlflowException:
experiment_id = mlflow.get_experiment_by_name(experiment_name).experiment_id

mlflow.set_experiment(experiment_name)

# Simular un experimento de fine-tuning de LLM
def run_finetuning_experiment(
model_name: str,
learning_rate: float,
batch_size: int,
num_epochs: int,
lora_rank: int,
dataset_name: str = "ejemplo-demo"
):
"""Ejecutar un experimento de fine-tuning y registrarlo en MLflow"""

with mlflow.start_run(run_name=f"{model_name}-lr{learning_rate}-bs{batch_size}") as run:
run_id = run.info.run_id

# Registrar parámetros
mlflow.log_param("model_name", model_name)
mlflow.log_param("learning_rate", learning_rate)
mlflow.log_param("batch_size", batch_size)
mlflow.log_param("num_epochs", num_epochs)
mlflow.log_param("lora_rank", lora_rank)
mlflow.log_param("dataset_name", dataset_name)

# Simular el proceso de entrenamiento, registrar métricas
print(f"Iniciando experimento de entrenamiento {run_id}...")

for epoch in range(num_epochs):
# Simular la pérdida de entrenamiento (disminuye con cada epoch)
train_loss = 2.0 - 0.3 * epoch + np.random.normal(0, 0.1)
val_loss = 2.2 - 0.25 * epoch + np.random.normal(0, 0.15)

# Registrar métricas
mlflow.log_metric("train_loss", train_loss, step=epoch)
mlflow.log_metric("val_loss", val_loss, step=epoch)
mlflow.log_metric("epoch", epoch, step=epoch)

print(f"Epoch {epoch}: train_loss={train_loss:.4f}, val_loss={val_loss:.4f}")
time.sleep(0.5)

# Métricas de evaluación final
final_val_loss = 1.2 + np.random.normal(0, 0.1)
final_accuracy = 0.85 + np.random.normal(0, 0.03)
final_f1 = 0.83 + np.random.normal(0, 0.04)

mlflow.log_metric("final_val_loss", final_val_loss)
mlflow.log_metric("final_accuracy", final_accuracy)
mlflow.log_metric("final_f1", final_f1)

# Registrar archivos (por ejemplo, registros de entrenamiento, salidas de ejemplo)
with open("training_log.txt", "w") as f:
f.write(f"Experiment run_id: {run_id}\n")
f.write(f"Model: {model_name}\n")
f.write(f"Final accuracy: {final_accuracy:.4f}\n")

mlflow.log_artifact("training_log.txt")

# Registrar el modelo (aquí se simula con sklearn, en proyectos reales se registra el LLM real)
# En proyectos reales se puede usar mlflow.transformers.log_model
X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=42)
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

mlflow.sklearn.log_model(
model,
"model",
registered_model_name=f"{model_name.replace('/', '-')}-demo"
)

print(f"¡Experimento completado! run_id: {run_id}")
print(f"Precisión final: {final_accuracy:.4f}")

return run_id, final_accuracy

# Ejecutar varios experimentos
if __name__ == "__main__":
print("Iniciar la interfaz de MLflow (ejecutar en otra terminal):")
print(" mlflow ui --port 5000")
print("\nIniciando experimentos...\n")

# Experimento 1: configuración básica
run_id1, acc1 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()

# Experimento 2: rango de LoRA mayor
run_id2, acc2 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=2e-5,
batch_size=16,
num_epochs=3,
lora_rank=16
)

print()

# Experimento 3: tasa de aprendizaje más alta
run_id3, acc3 = run_finetuning_experiment(
model_name="Qwen2.5-7B-Instruct",
learning_rate=5e-5,
batch_size=16,
num_epochs=3,
lora_rank=8
)

print()
print("¡Todos los experimentos completados!")
print(f"Experimento 1 (run_id={run_id1}): {acc1:.4f}")
print(f"Experimento 2 (run_id={run_id2}): {acc2:.4f}")
print(f"Experimento 3 (run_id={run_id3}): {acc3:.4f}")
print("\nVer resultados: http://localhost:5000")

Registro de modelos

MLflow Model Registry gestiona el ciclo de vida de los modelos: de Staging (pruebas) a Production (producción).

Ejemplo

# Ruta del archivo: mlflow_model_registry.py
import mlflow
from mlflow.tracking import MlflowClient

mlflow.set_tracking_uri("http://localhost:5000")

client = MlflowClient()

def promote_model_to_production(
model_name: str,
version: int,
description: str = ""
):
"""Subir el modelo a la fase de Production"""
# Actualizar la descripción de la versión
client.update_model_version(
name=model_name,
version=version,
description=description
)

# Establecer esta versión como Production
client.transition_model_version_stage(
name=model_name,
version=version,
stage="Production",
archive_existing_versions=True # Archivar la versión anterior
)
print(f"El modelo {model_name} versión {version} se ha subido a Production")

def get_production_model(model_name: str):
"""Obtener el modelo Production actual"""
versions = client.get_latest_versions(model_name, stages=["Production"])
if versions:
return versions[0]
return None

def list_all_models():
"""Listar todos los modelos registrados"""
models = client.search_registered_models()
print("Modelos registrados:")
for model in models:
print(f" - {model.name}")
# Listar todas las versiones de ese modelo
versions = client.search_model_versions(f"name='{model.name}'")
for v in versions:
print(f" Version {v.version}: {v.current_stage}")

# Ejemplo de uso
if __name__ == "__main__":
model_name = "Qwen2.5-7B-Instruct-demo"

print("Listar todos los modelos:")
list_all_models()

print("\n" + "="*50)
# Supongamos que queremos enviar la versión 1 a producción
# promote_model_to_production(
# model_name=model_name,
# version=1,
# description="Modelo de producción: versión optimizada 2024"
# )

print("\nObtener el modelo de producción actual:")
prod_model = get_production_model(model_name)
if prod_model:
print(f" Nombre del modelo: {prod_model.name}")
print(f" Versión: {prod_model.version}")
print(f" Fase: {prod_model.current_stage}")
print(f" Run ID: {prod_model.run_id}")
else:
print(" No hay modelo de producción actualmente")

Monitoreo y alertas

El servicio está en producción, ¿cómo saber si funciona correctamente? ¿Los usuarios encontrarán tiempos de espera? ¿El modelo será cada vez menos preciso?

Necesitas un sistema de monitoreo completo.

Métricas clave: TTFT, TPS, tasa de error

Los servicios de IA tienen varias métricas de monitoreo clave:

Métrica	Nombre completo	Significado	Ejemplo de umbral de alerta
TTFT	Time to First Token	Tiempo desde que el usuario envía una solicitud hasta que recibe la primera palabra	> 3 segundos alerta
TPOT	Time per Output Token	Tiempo promedio por token de salida	> 100 ms alerta
TPS	Tokens Per Second	Cuántos tokens se generan por segundo	< 100 alerta
Tasa de error	Error Rate	Proporción de solicitudes fallidas	> 5% alerta
Utilización de GPU	GPU Utilization	Uso de GPU	< 30% o > 95% alerta
Uso de memoria de video	GPU Memory	Ocupación de memoria de video de GPU	> 90% alerta

Prometheus + Grafana

Prometheus recopila métricas, Grafana muestra paneles; es la combinación perfecta para el monitoreo.

Ejemplo

# Ruta del archivo: metrics_prometheus.py
from prometheus_client import (
Counter, Histogram, Gauge,
start_http_server, generate_latest
)
import time
import random

# ============================================
# Definir métricas
# ============================================

# Counter: contador que solo aumenta
REQUEST_COUNT = Counter(
"llm_requests_total",
"Total number of LLM requests",
["model", "variant", "status"] # Etiquetas
)

# Histogram: distribución estadística (por ejemplo, latencia)
REQUEST_LATENCY = Histogram(
"llm_request_latency_seconds",
"LLM request latency",
["model", "variant"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60] # Buckets
)

TTFT_HISTOGRAM = Histogram(
"llm_ttft_seconds",
"Time to first token",
["model", "variant"],
buckets=[0.05, 0.1, 0.2, 0.5, 1, 2, 5]
)

# Gauge: medidor que puede aumentar o disminuir
ACTIVE_CONNECTIONS = Gauge(
"llm_active_connections",
"Number of active connections",
["model"]
)

GPU_UTILIZATION = Gauge(
"llm_gpu_utilization_percent",
"GPU utilization",
["gpu_id", "model"]
)

GPU_MEMORY_USED = Gauge(
"llm_gpu_memory_used_bytes",
"GPU memory used",
["gpu_id", "model"]
)

TOKENS_PER_SECOND = Gauge(
"llm_tokens_per_second",
"Tokens generated per second",
["model"]
)

# ============================================
# Simular servicio
# ============================================

def process_request(model: str, variant: str):
"""Simular el procesamiento de una solicitud y registrar métricas"""
start_time = time.time()

# Número de conexiones activas +1
ACTIVE_CONNECTIONS.labels(model=model).inc()

try:
# Simular TTFT (tiempo del primer token)
ttft = random.uniform(0.05, 0.3)
TTFT_HISTOGRAM.labels(model=model, variant=variant).observe(ttft)

# Simular tiempo de procesamiento
processing_time = random.uniform(0.5, 3.0)
time.sleep(processing_time)

# Simular éxito
REQUEST_COUNT.labels(
model=model, variant=variant, status="success"
).inc()

# Registrar latencia
latency = time.time() - start_time
REQUEST_LATENCY.labels(model=model, variant=variant).observe(latency)

# Simular TPS
tps = random.uniform(50, 200)
TOKENS_PER_SECOND.labels(model=model).set(tps)

except Exception:
# Simular error
REQUEST_COUNT.labels(
model=model, variant=variant, status="error"
).inc()
raise
finally:
# Número de conexiones activas -1
ACTIVE_CONNECTIONS.labels(model=model).dec()

def update_gpu_metrics():
"""Simular actualización de métricas de GPU"""
# Simular métricas de GPU 0
gpu0_util = random.uniform(40, 90)
gpu0_mem = random.uniform(10, 20) * 1024**3 # 10-20 GB
GPU_UTILIZATION.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_util)
GPU_MEMORY_USED.labels(gpu_id="0", model="Qwen2.5-7B").set(gpu0_mem)

# ============================================
# Iniciar servicio
# ============================================

if __name__ == "__main__":
# Iniciar el endpoint de métricas de Prometheus
start_http_server(8081)
print("Prometheus metrics server started on port 8081")
print("Metrics available at http://localhost:8081/metrics")

# Simular tráfico
models = ["Qwen2.5-7B-Instruct", "Llama-3.1-8B-Instruct"]
variants = ["control", "treatment-v1"]

while True:
model = random.choice(models)
variant = random.choice(variants)
process_request(model, variant)
update_gpu_metrics()
time.sleep(random.uniform(0.1, 0.5))

Archivo de configuración de Prometheus:

Ejemplo

# Ruta del archivo: prometheus.yml
global:
scrape_interval: 15s

scrape_configs:
# Recopilar métricas del servicio LLM
- job_name: "llm-service"
static_configs:
- targets: ["localhost:8081"]
labels:
service: "llm-api"
environment: "production"

# Recopilar métricas de nodos (GPU, CPU, memoria)
- job_name: "node"
static_configs:
- targets: ["localhost:9100"]

alerting:
alertmanagers:
- static_configs:
- targets: ["localhost:9093"]

# Reglas de alerta
rule_files:
- "alerts.yml"

Reglas de alerta:

Ejemplo

# Ruta del archivo: alerts.yml
groups:
- name: llm_alerts
interval: 30s
rules:
# Alerta de alta tasa de error
- alert: HighErrorRate
expr: |
sum(rate(llm_requests_total{status="error"}[5m]))
/
sum(rate(llm_requests_total[5m])) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "Alerta de alta tasa de error"
description: La tasa de error supera el 5%, valor actual: {{ $value | humanizePercentage }}

# Alerta: TTFT demasiado alto
- alert: SlowTTFT
expr: |
histogram_quantile(0.95, sum(rate(llm_ttft_seconds_bucket[5m])) by (le, model)) > 2
for: 5m
labels:
severity: warning
annotations:
summary: "TTFT demasiado alto"
description: "El TTFT del percentil 95 del modelo {{ $labels.model }} supera los 2 segundos"

# Uso de GPU demasiado bajo (desperdicio de recursos)
- alert: LowGPUUtilization
expr: llm_gpu_utilization_percent < 30
for: 30m
labels:
severity: info
annotations:
summary: "Uso de GPU demasiado bajo"
description: "El uso de GPU {{ $labels.gpu_id }} es {{ $value }}%, posible desperdicio de recursos"

# Uso de GPU demasiado alto (riesgo de cuello de botella)
- alert: HighGPUUtilization
expr: llm_gpu_utilization_percent > 95
for: 5m
labels:
severity: warning
annotations:
summary: "Uso de GPU demasiado alto"
description: "El uso de GPU {{ $labels.gpu_id }} es {{ $value }}%, podría convertirse en un cuello de botella"

Monitoreo específico de IA: detección de deriva (drift)

Los modelos de IA no solo se "rompen", también se "desvían": con el tiempo, la distribución de los datos cambia y el rendimiento del modelo disminuye.

Ejemplo

# Ruta del archivo: drift_detection.py
import numpy as np
from scipy import stats
from dataclasses import dataclass
from typing import List, Tuple
import json

@dataclass
class DistributionStats:
"""Información de estadísticas de distribución"""
mean: float
std: float
min: float
max: float
percentiles: dict

class DataDriftDetector:
"""Detector de deriva de datos"""

def __init__(self, baseline_data: np.ndarray):
"""Inicializar con datos de referencia"""
self.baseline_stats = self._compute_stats(baseline_data)
self.baseline_data = baseline_data

def _compute_stats(self, data: np.ndarray) -> DistributionStats:
"""Calcular estadísticas de los datos"""
return DistributionStats(
mean=float(np.mean(data)),
std=float(np.std(data)),
min=float(np.min(data)),
max=float(np.max(data)),
percentiles={
"p10": float(np.percentile(data, 10)),
"p25": float(np.percentile(data, 25)),
"p50": float(np.percentile(data, 50)),
"p75": float(np.percentile(data, 75)),
"p90": float(np.percentile(data, 90))
}
)

def detect(self, new_data: np.ndarray, threshold: float = 0.05) -> dict:
"""
Detectar deriva
Devuelve: si hay deriva, valor p, estadísticas
"""
# Prueba KS: comparar si dos distribuciones son iguales
ks_statistic, ks_pvalue = stats.ks_2samp(self.baseline_data, new_data)

# Calcular estadísticas de los nuevos datos
new_stats = self._compute_stats(new_data)

# Calcular la diferencia de medias (porcentaje)
mean_diff_pct = abs((new_stats.mean - self.baseline_stats.mean) / self.baseline_stats.mean * 100)

# Determinar si hay deriva
is_drift = ks_pvalue < threshold

return {
"is_drift": is_drift,
"ks_statistic": float(ks_statistic),
"ks_pvalue": float(ks_pvalue),
"mean_diff_percent": float(mean_diff_pct),
"baseline_stats": self.baseline_stats,
"new_stats": new_stats,
"threshold": threshold
}

# Simulación: detección de deriva en la longitud de entrada del usuario
def simulate_input_length_drift():
# Datos de referencia: longitud media de entrada del usuario 50
np.random.seed(42)
baseline = np.random.normal(loc=50, scale=15, size=1000)
baseline = np.clip(baseline, 5, 200)

detector = DataDriftDetector(baseline)

print("Ejemplo de detección de deriva de datos")
print("="*50)
print(f"Referencia: media={detector.baseline_stats.mean:.2f}, desviación estándar={detector.baseline_stats.std:.2f}")
print()

# Escenario 1: datos normales (sin deriva)
print("Escenario 1: datos normales")
normal_data = np.random.normal(loc=50, scale=15, size=500)
normal_data = np.clip(normal_data, 5, 200)
result = detector.detect(normal_data)
print(f" Media de nuevos datos: {result['new_stats'].mean:.2f}")
print(f" KS p-value: {result['ks_pvalue']:.4f}")
print(f" ¿Deriva? {'Sí' if result['is_drift'] else 'No'}")
print()

# Escenario 2: con deriva (la entrada se ha hecho más larga)
print("Escenario 2: la entrada se ha hecho más larga")
drift_data = np.random.normal(loc=100, scale=20, size=500)
drift_data = np.clip(drift_data, 5, 300)
result = detector.detect(drift_data)
print(f" Media de nuevos datos: {result['new_stats'].mean:.2f}")
print(f" KS p-value: {result['ks_pvalue']:.4f}")
print(f" ¿Deriva? {'Sí' if result['is_drift'] else 'No'}")
print(f" Diferencia de medias: {result['mean_diff_percent']:.1f}%")

if __name__ == "__main__":
simulate_input_length_drift()

Estrategias de optimización de costos

Las GPU son caras; optimizar costos no es "tacañería", es "capacidad de ingeniería".

Caché de solicitudes: caché semántica

El caché tradicional es de coincidencia exacta: si la pregunta es exactamente la misma, devuelve directamente la respuesta anterior.

El caché semántico va más allá: preguntas similares también pueden reutilizar respuestas anteriores o resultados intermedios.

Ejemplo

# Ruta del archivo: semantic_cache.py
import numpy as np
from typing import Optional, Tuple
import time

class SemanticCache:
"""Implementación simple de caché semántica"""

def __init__(self, threshold: float = 0.9):
self.threshold = threshold
self.queries = [] # Almacenar los vectores de las preguntas
self.answers = [] # Almacenar las respuestas correspondientes
self.timestamps = [] # Almacenar el tiempo

def _simulate_embedding(self, text: str) -> np.ndarray:
"""
Simula embeddings; en proyectos reales se usan OpenAI Embeddings, Sentence-Transformers, etc.
"""
# Simulación simple con hash; en proyectos reales se usa un modelo de embedding real
np.random.seed(hash(text) % (2**32))
return np.random.randn(128) # Vector de 128 dimensiones

def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""Calcular la similitud del coseno"""
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def get(self, query: str) -> Optional[str]:
"""Consultar el caché; si se encuentra una pregunta similar, devolver la respuesta"""
query_vec = self._simulate_embedding(query)

for i, (q_vec, answer, ts) in enumerate(zip(self.queries, self.answers, self.timestamps)):
# Limpieza por expiración (más de 1 hora)
if time.time() - ts > 3600:
continue

similarity = self._cosine_similarity(query_vec, q_vec)
if similarity >= self.threshold:
return answer

return None

def put(self, query: str, answer: str):
"""Guardar en el caché"""
query_vec = self._simulate_embedding(query)
self.queries.append(query_vec)
self.answers.append(answer)
self.timestamps.append(time.time())

# Control de capacidad simple: máximo 1000 entradas
if len(self.queries) > 1000:
self.queries = self.queries[-1000:]
self.answers = self.answers[-1000:]
self.timestamps = self.timestamps[-1000:]

# Ejemplo de uso
def main():
cache = SemanticCache(threshold=0.9)

# Primera solicitud
query1 = "¿Cómo leer un archivo CSV con Python?"
print(f"Consulta 1: {query1}")
answer1 = cache.get(query1)
if answer1:
print(f"Acierto de caché: {answer1}")
else:
print("Fallo de caché, llamando al modelo...")
answer1 = "Se puede usar pandas.read_csv('file.csv') para leer un archivo CSV."
cache.put(query1, answer1)
print(f"Guardado en caché: {answer1}")

print()

# Segunda vez: pregunta similar
query2 = "¿Cómo se lee un CSV con Python?"
print(f"Consulta 2: {query2}")
answer2 = cache.get(query2)
if answer2:
print(f"Acierto de caché: {answer2}")
else:
print("Fallo de caché, llamando al modelo...")

print()

# Tercera vez: pregunta completamente diferente
query3 = "¿Qué es el aprendizaje automático?"
print(f"Consulta 3: {query3}")
answer3 = cache.get(query3)
if answer3:
print(f"Acierto de caché: {answer3}")
else:
print("Fallo de caché, llamando al modelo...")

if __name__ == "__main__":
main()

Enrutamiento de modelos: modelo pequeño como respaldo

No todas las solicitudes necesitan un modelo grande. Para preguntas simples se usa un modelo pequeño; para preguntas complejas, un modelo grande.

Ejemplo

# Ruta del archivo: model_router.py
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class ModelConfig:
name: str
cost_per_1k_tokens: float
max_tokens: int
capabilities: List[str] # Lista de capacidades

class SmartModelRouter:
"""Enrutador inteligente de modelos"""

def __init__(self):
self.models = {
"small": ModelConfig(
name="Qwen2.5-0.5B-Instruct",
cost_per_1k_tokens=0.001,
max_tokens=4096,
capabilities=["chat", "simple_qa", "summarization"]
),
"medium": ModelConfig(
name="Qwen2.5-7B-Instruct",
cost_per_1k_tokens=0.01,
max_tokens=8192,
capabilities=["chat", "qa", "coding", "reasoning"]
),
"large": ModelConfig(
name="Qwen2.5-72B-Instruct",
cost_per_1k_tokens=0.1,
max_tokens=32768,
capabilities=["complex_qa", "advanced_coding", "deep_reasoning"]
)
}

def _classify_query(self, query: str) -> str:
"""
Clasificación simple de preguntas (en un proyecto real se puede usar un clasificador)
Devuelve: small/medium/large
"""
query_lower = query.lower()

# Palabras clave para preguntas simples
simple_keywords = [
"Hola", "hello", "Gracias", "hi", "Adiós",
"El clima de hoy", "Qué hora es", "Fecha",
"Ayuda con traducción", "translate",
"Breve introducción", "Qué es"
]

# Palabras clave para preguntas complejas
complex_keywords = [
"Código", "Escribe un", "Implementa", "code", "python",
"Analiza", "Explica por qué", "Deducir", "Demostrar",
"Comparar", "Contrastar", "Diferencia",
"Detallado", "Profundo", "Complejo"
]

# Comprobar si es una pregunta simple
for keyword in simple_keywords:
if keyword in query_lower:
return "small"

# Comprobar si es una pregunta compleja
for keyword in complex_keywords:
if keyword in query_lower:
return "large"

# Por defecto, usar el modelo mediano
return "medium"

def route(self, query: str, user_tier: str = "free") -> str:
"""
Decisión de enrutamiento
"""
# Primero clasificar la pregunta
category = self._classify_query(query)

# Los usuarios de pago pueden usar un modelo mejor
if user_tier == "pro":
upgrade_map = {
"small": "small",
"medium": "medium",
"large": "large"
}
category = upgrade_map[category]
elif user_tier == "enterprise":
upgrade_map = {
"small": "medium",
"medium": "large",
"large": "large"
}
category = upgrade_map.get(category, category)

# Devolver el nombre del modelo
return self.models[category].name

def get_cost_estimate(self, model_name: str, input_tokens: int, output_tokens: int) -> float:
"""Estimar el costo"""
for model in self.models.values():
if model.name == model_name:
return (input_tokens + output_tokens) * model.cost_per_1k_tokens / 1000
return 0.0

# Ejemplo de uso
def main():
router = SmartModelRouter()

test_queries = [
"Hola, me gustaría conocer algo",
"¿Qué es Python?",
"Escribe un quicksort en Python por favor",
"Analiza en detalle la arquitectura Transformer"
]

print("Ejemplo de enrutamiento inteligente")
print("="*50)

for query in test_queries:
model = router.route(query, user_tier="free")
cost = router.get_cost_estimate(model, 100, 200)
print(f"Pregunta: {query}")
print(f" Enrutado a: {model}")
print(f" Costo estimado: ${cost:.4f}")
print()

# Diferentes niveles de usuario
print("Comparación de niveles de usuario")
print("-"*50)
query = "Escribe un quicksort en Python por favor"
for tier in ["free", "pro", "enterprise"]:
model = router.route(query, user_tier=tier)
print(f"Usuario {tier}: {model}")

if __name__ == "__main__":
main()

Instancias Spot y procesamiento por lotes

Para solicitudes en tiempo real se usan instancias On-Demand (estables pero caras); para procesamiento por lotes offline se usan instancias Spot (baratas pero pueden ser interrumpidas).

Resumen de las formas de optimización de costos:

Estrategia	Escenario	Ahorro de costos
Caché semántica	Muchas preguntas repetidas	30-70%
Enrutamiento de modelos	Alta proporción de preguntas simples	50-80%
Cuantización de modelos	Todos los escenarios	30-50%
Instancias Spot	Procesamiento por lotes offline	60-90%
Autoescalado	Alta fluctuación de tráfico	30-60%

Manejo de fallos y degradación

En producción no existe "no tener fallos", solo existe "cómo responder con elegancia cuando hay fallos".

Estrategias de degradación

Ejemplo

# Ruta del archivo: fallback_handler.py
import time
from typing import Optional
from dataclasses import dataclass
from enum import Enum

class ServiceStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"

@dataclass
class FallbackResponse:
content: str
is_fallback: bool
fallback_reason: Optional[str] = None

class FallbackHandler:
"""Manejador de degradación"""

def __init__(self):
self.error_count = 0
self.last_error_time = 0
self.circuit_open = False
self.circuit_open_time = 0

# Configuración del interruptor de circuito
self.error_threshold = 5 # Abrir el interruptor tras 5 errores consecutivos
self.circuit_timeout = 30 # Intentar recuperar 30 segundos después de abrir el interruptor

def _call_primary_service(self, query: str) -> str:
"""Simular la llamada al servicio principal"""
# Aquí se reemplaza por una llamada real al modelo
return "Esta es la respuesta del servicio principal"

def _call_fallback_service(self, query: str) -> str:
"""Simular llamada al servicio de degradación (más rápido pero de menor calidad)"""
# Puede ser un modelo más pequeño, respuestas en caché o incluso plantillas predefinidas
return "Lo sentimos, el servicio está temporalmente ocupado. Esta es una respuesta simplificada."

def _call_static_fallback(self, query: str) -> str:
"""Respuesta de degradación estática (última línea de defensa)"""
return (
"Lo sentimos mucho, nuestro servicio está experimentando algunos problemas actualmente.\n"
"Por favor, inténtelo de nuevo más tarde. Para problemas urgentes, contacte con [email protected]."
)

def process_request(self, query: str) -> FallbackResponse:
"""Procesar la solicitud, con lógica de degradación completa"""

# 1. Comprobar si el cortacircuitos está abierto
if self.circuit_open:
if time.time() - self.circuit_open_time > self.circuit_timeout:
# Tiempo agotado, probar estado semiabierto, permitir que pasen pocas solicitudes
print("Cortacircuitos semiabierto, intentando recuperar...")
self.circuit_open = False
else:
# Degradación directa
print("Cortacircuitos abierto, degradación directa")
return FallbackResponse(
content=self._call_fallback_service(query),
is_fallback=True,
fallback_reason="circuit_open"
)

# 2. Intentar con el servicio principal
try:
result = self._call_primary_service(query)
# Éxito: restablecer el contador de errores
self.error_count = 0
return FallbackResponse(
content=result,
is_fallback=False
)

except Exception as e:
print(f"Error en la llamada al servicio principal: {e}")
self.error_count += 1
self.last_error_time = time.time()

# 3. Comprobar si es necesario activar el cortacircuitos
if self.error_count >= self.error_threshold:
print("Demasiados errores, abrir el cortacircuitos")
self.circuit_open = True
self.circuit_open_time = time.time()

# 4. Intentar con el servicio de degradación
try:
print("Intentando llamar al servicio de degradación")
result = self._call_fallback_service(query)
return FallbackResponse(
content=result,
is_fallback=True,
fallback_reason="primary_failed"
)
except Exception as e2:
print(f"El servicio de degradación también falló: {e2}")

# 5. Respuesta estática final
return FallbackResponse(
content=self._call_static_fallback(query),
is_fallback=True,
fallback_reason="all_failed"
)

# Ejemplo de uso
def main():
handler = FallbackHandler()

print("Ejemplo de manejo de degradación")
print("="*50)

# Solicitud normal
print("Solicitud normal:")
resp = handler.process_request("Hola")
print(f" Respuesta: {resp.content}")
print(f" ¿Está degradada? {resp.is_fallback}")
print()

# Simular error (en proyectos reales, se lanzaría una excepción de verdad)
# Aquí simplemente demostramos la lógica
print("Resumen de la estrategia de degradación:")
print(" 1. Priorizar el servicio principal")
print(" 2. Si el servicio principal falla → usar modelo pequeño/caché")
print(" 3. Fallos consecutivos → cortar el circuito, evitar avalanchas")
print(" 4. Durante el corte del circuito → devolver rápidamente respuesta degradada")
print(" 5. Última línea de defensa → respuesta estática")

Lanzamiento canario y reversión

No publiques la nueva versión toda de golpe; primero asígnale el 1% del tráfico, si no hay problemas, luego 5%, 10%, 100%.

Si hay problemas, debes poder hacer rollback rápidamente; no esperes a que los usuarios se quejen durante mucho tiempo antes de cambiar la configuración apresuradamente.

Otras extensiones