Concurrencia y contrapresión en llamadas a un LLM

El gather que tumbó el proveedor

Hay un patrón que aparece en casi todo proyecto con LLM. Tienes mil documentos que clasificar, escribes un bucle, lo conviertes en asyncio.gather porque “es más rápido” y lo lanzas. Durante diez segundos todo va de maravilla. Después llegan los errores 429, los reintentos se apilan encima de las peticiones originales y el proceso termina con la mitad de los resultados y la cuota del minuto agotada.

El fallo no está en el proveedor. Está en que has confundido concurrencia con paralelismo sin freno. Una llamada a un LLM no cuesta CPU: cuesta un hueco en dos presupuestos ajenos, el de peticiones por minuto y el de tokens por minuto. Si no los modelas en tu código, los modela el proveedor, y lo hace rechazándote.

Dos límites, no uno

Casi todos los proveedores miden dos cosas a la vez:

Un semáforo que limita a veinte peticiones simultáneas controla el primero y deja el segundo al azar. Veinte peticiones de 40 000 tokens cada una te sacan del TPM aunque el RPM esté en el 5 %. Por eso hacen falta dos mecanismos distintos: uno para la concurrencia (cuántas llamadas vuelan a la vez) y otro para el caudal (cuántos tokens por minuto salen).

El semáforo: concurrencia acotada

Lo primero es un tope duro de llamadas en vuelo. En Python, un asyncio.Semaphore basta:

import asyncio

# Tope de llamadas simultáneas. Se ajusta con medidas, no a ojo:
# empieza bajo y sube mientras la latencia p95 no se deteriore.
MAX_EN_VUELO = 16
semaforo = asyncio.Semaphore(MAX_EN_VUELO)

async def llamar(cliente, prompt: str) -> str:
    # Si hay 16 llamadas en vuelo, la 17 espera aquí sin consumir nada.
    async with semaforo:
        resp = await cliente.generar(prompt, timeout=30)
        return resp.texto

async def procesar(cliente, prompts: list[str]) -> list[str]:
    # gather sigue siendo válido: el semáforo es quien frena, no el bucle.
    return await asyncio.gather(*(llamar(cliente, p) for p in prompts))

Esto ya evita la avalancha. Pero tiene un defecto que se nota en lotes grandes: gather crea mil corrutinas de golpe. Cada una guarda su prompt en memoria mientras espera, y si un fallo cancela el conjunto pierdes todo el trabajo. Para lotes de miles, mejor un conjunto de workers que consumen de una cola.

La cola acotada: contrapresión de verdad

La contrapresión consiste en que, cuando el sistema no da abasto, el productor se entera y frena. Una cola con maxsize hace exactamente eso: put se bloquea cuando está llena.

import asyncio

async def productor(cola: asyncio.Queue, documentos):
    for doc in documentos:
        # Bloquea si la cola está llena: el productor nunca va por delante
        # de lo que los workers pueden digerir.
        await cola.put(doc)
    # Una señal de parada por worker para cerrar limpiamente.
    for _ in range(N_WORKERS):
        await cola.put(None)

async def worker(cola: asyncio.Queue, cliente, resultados: dict):
    while (doc := await cola.get()) is not None:
        try:
            resultados[doc.id] = await cliente.generar(doc.prompt, timeout=30)
        except Exception as e:
            # Un fallo en un documento no tumba el lote: se anota y se sigue.
            resultados[doc.id] = e
        finally:
            cola.task_done()

N_WORKERS = 16
async def ejecutar(cliente, documentos):
    cola = asyncio.Queue(maxsize=N_WORKERS * 2)   # cola corta a propósito
    resultados: dict = {}
    workers = [asyncio.create_task(worker(cola, cliente, resultados))
               for _ in range(N_WORKERS)]
    await productor(cola, documentos)
    await asyncio.gather(*workers)
    return resultados

Dos detalles de criterio. La cola es corta a propósito: una cola de diez mil elementos no es contrapresión, es un almacén donde los problemas se esconden. Y los errores se guardan por documento: un lote de mil con tres fallos tiene que devolver 997 resultados y una lista de tres pendientes, no una excepción.

El caudal: un cubo de tokens

Falta el segundo límite. El mecanismo clásico es un token bucket: un cubo que se rellena a ritmo constante y del que cada llamada retira su coste estimado antes de salir.

import asyncio, time

class CuboDeTokens:
    def __init__(self, tokens_por_minuto: int):
        self.capacidad = tokens_por_minuto
        self.disponibles = float(tokens_por_minuto)
        self.ritmo = tokens_por_minuto / 60      # tokens que entran por segundo
        self.ultimo = time.monotonic()
        self.cerrojo = asyncio.Lock()

    async def retirar(self, coste: int):
        while True:
            async with self.cerrojo:
                ahora = time.monotonic()
                self.disponibles = min(
                    self.capacidad,
                    self.disponibles + (ahora - self.ultimo) * self.ritmo,
                )
                self.ultimo = ahora
                if self.disponibles >= coste:
                    self.disponibles -= coste
                    return
                falta = (coste - self.disponibles) / self.ritmo
            await asyncio.sleep(falta)   # espera fuera del cerrojo

El coste se estima antes de enviar: tokens del prompt más el max_tokens que pides de salida. Estimar de más es más seguro que estimar de menos. Cuando llega la respuesta puedes reponer la diferencia con el uso real que devuelve la API.

Qué hacer cuando la cola se llena

Si el productor es un lote, bloquearse está bien. Si el productor es un usuario esperando, no. Ahí la contrapresión tiene que subir hasta la interfaz con una decisión explícita:

• Rechazar rápido con un 503 y un Retry-After, mejor que dejar al usuario treinta segundos mirando un spinner.
• Degradar: un modelo más pequeño, una respuesta en caché o una versión sin el paso opcional del flujo.
• Priorizar: dos colas, una interactiva y otra de lotes, y los workers sirven primero la interactiva.

Lo importante es que la decisión sea tuya y esté escrita en código, no un efecto secundario de que el proveedor te devuelva 429. Junto con los reintentos y timeouts, este control de entrada es lo que separa un script de un servicio.

Cómo calibrarlo

1. Mide la latencia p50 y p95 de una llamada típica con concurrencia 1.
2. Sube MAX_EN_VUELO de cuatro en cuatro mientras el p95 no se dispare y no aparezcan 429.
3. Fija el cubo de tokens al 80 % del límite contratado, para dejar margen a picos y a otros procesos que comparten la cuota.
4. Registra la profundidad de la cola y el tiempo de espera en ella. Si la cola vive llena, el cuello de botella está aguas abajo y no se arregla añadiendo workers.

El lunes, abre el sitio donde haces gather sobre llamadas a un modelo y añade un semáforo con el valor más bajo que te dé un rendimiento aceptable. Son tres líneas, y es lo que más se agradece el día que el volumen se multiplica por diez.