ainvoke, semáforo y logs429 y un proceso muerto sin haber guardado
nada.
El patrón es contraintuitivo al principio: el grafo procesa un documento. El lote se orquesta fuera.
grafo_unitario = builder.compile()
async def worker(doc):
async with sem:
return await grafo_unitario.ainvoke(doc)
ainvoke es la versión async de invoke, y no
es cosmética: para que la concurrencia sirva de algo, los nodos que hacen E/S
tienen que ser async def y usar await. Un nodo síncrono
que llame a una librería HTTP bloqueante detiene el bucle de eventos entero, y el
resto de tareas espera de brazos cruzados.
sem = asyncio.Semaphore(3)
tareas = [worker(doc) for doc in docs]
resultados = await asyncio.gather(*tareas)
Crear las corrutinas es barato; lo caro es tenerlas todas golpeando la API.
async with sem bloquea la tarea hasta que hay hueco. El límite se
elige mirando el rate limit del proveedor, no a ojo.
async def nodo_procesar_doc(state):
try:
...
return {"resumen": resumen, "status": "completado"}
except Exception as e:
logger.error(f"Doc {state['doc_id']}: {e}")
return {"resumen": "", "status": "error"}El nodo captura y devuelve un estado de error controlado en lugar de propagar. Así todos los resultados del lote tienen la misma forma y el fallo de un documento no tumba los otros 499. Es la misma decisión que el nodo de fallo de la lección 04, aplicada a escala.
| Campo | Por qué |
|---|---|
Identificador de la unidad (doc_id) | Sin él, 500 líneas de log son ruido |
| Nodo y duración | Localiza el cuello de botella sin perfilar |
| Estado de salida | Permite contar éxitos y fallos sin releer los datos |
| Excepción con su tipo | except Exception sin registrar el tipo esconde bugs propios |
print no lleva nivel, ni marca de tiempo, ni se puede apagar.
logging sí. Lo que no se registra: el contenido
íntegro de prompts y respuestas en logs planos. Ahí empieza el módulo
siguiente, que es donde esto se hace bien.
La guía anuncia «procesaremos 20 documentos, pero solo 3 a la vez» y
el código de al lado crea range(1, 11): diez. Además afirma
que en la validación verás los mensajes de éxito «en grupos
de 3», cuando cada tarea duerme un tiempo aleatorio entre 0,5 y 1,5 segundos
— los huecos del semáforo se liberan de forma escalonada, no por tandas.
¿Cuántos documentos ves tú en tus logs, y salen agrupados? Si un texto y su código no coinciden en algo tan comprobable, ¿qué haces con la afirmación de la misma página de que nunca uses hilos para E/S?
Ejecuta el mismo lote con max_concurrency en 1, 3 y 10, y apunta el tiempo
total de cada uno.
asyncio.gather(*tareas, return_exceptions=True), sube la probabilidad de error al 100 % y compara qué devuelve cada versión.