Llamadas a la API de LLM Asincrónicas en Python: Una Guía Completa

Como desarrolladores y científicos de datos, a menudo nos encontramos necesitando interactuar con estos poderosos modelos a través de API. Sin embargo, a medida que nuestras aplicaciones crecen en complejidad y escala, la necesidad de interacciones de API eficientes y de alto rendimiento se vuelve crucial. Es aquí donde la programación asincrónica brilla, lo que nos permite maximizar el rendimiento y minimizar la latencia al trabajar con API de LLM.

En esta guía completa, exploraremos el mundo de las llamadas a la API de LLM asincrónicas en Python. Cubriremos todo, desde los conceptos básicos de la programación asincrónica hasta técnicas avanzadas para el control de flujos de trabajo complejos. Al final de este artículo, tendrás una comprensión sólida de cómo aprovechar la programación asincrónica para potenciar tus aplicaciones impulsadas por LLM.

Antes de sumergirnos en los detalles de las llamadas a la API de LLM asincrónicas, estableceremos una base sólida en los conceptos de programación asincrónica.

La programación asincrónica permite que se ejecuten varias operaciones de manera concurrente sin bloquear el hilo principal de ejecución. En Python, esto se logra principalmente a través del módulo asyncio, que proporciona un marco para escribir código concurrente utilizando coroutines, bucles de eventos y futuros.

Conceptos clave:

• Coroutines: Funciones definidas con async def que pueden ser pausadas y reanudadas.
• Bucle de eventos: El mecanismo central de ejecución que gestiona y ejecuta tareas asincrónicas.
• Objetos esperables: Objetos que se pueden utilizar con la palabra clave await (coroutines, tareas, futuros).

Aquí hay un ejemplo simple para ilustrar estos conceptos:

import asyncio

<p>async def greet(name):</p>
<p> await asyncio.sleep(1) # Simula una operación de E/S</p>
<p> print(f"Hello, {name}!")</p>

<p>async def main():</p>
<p> await asyncio.gather(</p>
<p> greet("Alice"),</p>
<p> greet("Bob"),</p>
<p> greet("Charlie")</p>
<p> )</p>

asyncio.run(main())

En este ejemplo, definimos una función asincrónica greet que simula una operación de E/S con asyncio.sleep(). La función main utiliza asyncio.gather() para ejecutar varios saludos de manera concurrente. A pesar del retraso de la función sleep, los tres saludos se imprimirán después de aproximadamente 1 segundo, demostrando el poder de la ejecución asincrónica.

La necesidad de asincronía en las llamadas a la API de LLM

Al trabajar con API de LLM, a menudo nos encontramos con escenarios en los que debemos realizar varias llamadas a la API, ya sea en secuencia o en paralelo. El código sincrónico tradicional puede generar cuellos de botella de rendimiento significativos, especialmente al tratar con operaciones de alta latencia como las solicitudes de red a los servicios de LLM.

Considera un escenario en el que debemos generar resúmenes para 100 artículos diferentes utilizando una API de LLM. Con un enfoque sincrónico, cada llamada a la API bloquearía hasta que se recibiera una respuesta, lo que podría tomar varios minutos para completar todas las solicitudes. Un enfoque asincrónico, por otro lado, nos permite iniciar varias llamadas a la API de manera concurrente, reduciendo drásticamente el tiempo de ejecución total.

Configuración del entorno

Para empezar con las llamadas a la API de LLM asincrónicas, debes configurar tu entorno de Python con las bibliotecas necesarias. Aquí hay lo que necesitarás:

• Python 3.7 o superior (para soporte nativo de asyncio)
• aiohttp: Una biblioteca de cliente HTTP asincrónica
• openai: El cliente de Python oficial de OpenAI (si estás utilizando los modelos GPT de OpenAI)
• langchain: Un marco para construir aplicaciones con LLM (opcional, pero recomendado para flujos de trabajo complejos)

Puedes instalar estas dependencias utilizando pip:

<p>pip install aiohttp openai langchain
&lt;div class=&quot;relative flex flex-col rounded-lg&quot;&gt;

Llamadas a la API de LLM asincrónicas básicas con asyncio y aiohttp

Comencemos haciendo una llamada asincrónica simple a una API de LLM utilizando aiohttp. Utilizaremos la API GPT-3.5 de OpenAI como ejemplo, pero los conceptos se aplican a otras API de LLM también.

import asyncio
import aiohttp
from openai import AsyncOpenAI

<p>async def generate_text(prompt, client):</p>
<p> response = await client.chat.completions.create(</p>
<p> model="gpt-3.5-turbo",</p>
<p> messages=[{"role": "user", "content": prompt}]</p>
<p> )</p>
<p> return response.choices[0].message.content</p>

<p>async def main():</p>
<p> prompts = [</p>
<p> "Explica la computación cuántica en términos simples.",</p>
<p> "Escribe un haikú sobre la inteligencia artificial.",</p>
<p> "Describe el proceso de la fotosíntesis."</p>
<p> ]</p>

<p>async with AsyncOpenAI() as client:</p>
<p> tasks = [generate_text(prompt, client) for prompt in prompts]</p>
<p> results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p> print(f"Prompt: {prompt}\nRespuesta: {result}\n")</p>

asyncio.run(main())

En este ejemplo, definimos una función asincrónica generate_text que hace una llamada a la API de OpenAI utilizando el cliente AsyncOpenAI. La función main crea varias tareas para diferentes prompts y utiliza asyncio.gather() para ejecutarlas de manera concurrente.

Este enfoque nos permite enviar varias solicitudes a la API de LLM simultáneamente, reduciendo significativamente el tiempo total necesario para procesar todos los prompts.

Técnicas avanzadas: Lotes y control de concurrencia

Mientras que el ejemplo anterior demuestra los conceptos básicos de las llamadas a la API de LLM asincrónicas, las aplicaciones del mundo real a menudo requieren enfoques más sofisticados. Exploraremos dos técnicas importantes: lotes de solicitudes y control de concurrencia.

Lotes de solicitudes: Cuando se trata de un gran número de prompts, a menudo es más eficiente agruparlos en lotes en lugar de enviar solicitudes individuales para cada prompt. Esto reduce la sobrecarga de múltiples llamadas a la API y puede conducir a un mejor rendimiento.

import asyncio
from openai import AsyncOpenAI

<p>async def process_batch(batch, client):</p>
<p> responses = await asyncio.gather(*[</p>
<p> client.chat.completions.create(</p>
<p> model="gpt-3.5-turbo",</p>
<p> messages=[{"role": "user", "content": prompt}]</p>
<p> ) for prompt in batch</p>
<p> ])</p>
<p> return [response.choices[0].message.content for response in responses]</p>

<p>async def main():</p>
<p> prompts = [f"Cuéntame un hecho sobre el número {i}" for i in range(100)]</p>
<p> batch_size = 10</p>

<p>async with AsyncOpenAI() as client:</p>
<p> results = []</p>
<p> for i in range(0, len(prompts), batch_size):</p>
<p> batch = prompts[i:i+batch_size]</p>
<p> batch_results = await process_batch(batch, client)</p>
<p> results.extend(batch_results)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p> print(f"Prompt: {prompt}\nRespuesta: {result}\n")</p>

asyncio.run(main())

Control de concurrencia: Si bien la programación asincrónica permite la ejecución concurrente, es importante controlar el nivel de concurrencia para evitar abrumar al servidor de la API o exceder los límites de velocidad. Podemos utilizar asyncio.Semaphore para este propósito.

import asyncio
from openai import AsyncOpenAI

<p>async def generate_text(prompt, client, semaphore):</p>
<p> async with semaphore:</p>
<p> response = await client.chat.completions.create(</p>
<p> model="gpt-3.5-turbo",</p>
<p> messages=[{"role": "user", "content": prompt}]</p>
<p> )</p>
<p> return response.choices[0].message.content</p>

<p>async def main():</p>
<p> prompts = [f"Cuéntame un hecho sobre el número {i}" for i in range(100)]</p>
<p> max_concurrent_requests = 5</p>
<p> semaphore = asyncio.Semaphore(max_concurrent_requests)</p>

<p>async with AsyncOpenAI() as client:</p>
<p> tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]</p>
<p> results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in zip(prompts, results):</p>
<p> print(f"Prompt: {prompt}\nRespuesta: {result}\n")</p>

asyncio.run(main())

En este ejemplo, utilizamos un semáforo para limitar el número de solicitudes concurrentes a 5, asegurando que no abrumemos al servidor de la API.

Control de errores y reintentos en llamadas a la API de LLM asincrónicas

Al trabajar con API externas, es crucial implementar un control de errores y mecanismos de reintento robustos. Mejoraremos nuestro código para manejar errores comunes e implementar un reintento con retroceso exponencial.

import asyncio
import random
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential

class APIError(Exception):
pass

<p>@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))</p>
<p>async def generate_text_with_retry(prompt, client):</p>
<p> try:</p>
<p> response = await client.chat.completions.create(</p>
<p> model="gpt-3.5-turbo",</p>
<p> messages=[{"role": "user", "content": prompt}]</p>
<p> )</p>
<p> return response.choices[0].message.content</p>
<p> except Exception as e:</p>
<p> print(f"Ocurrió un error: {e}")</p>
<p> raise APIError("Falló la generación de texto")</p>

<p>async def process_prompt(prompt, client, semaphore):</p>
<p> async with semaphore:</p>
<p> try:</p>
<p> result = await generate_text_with_retry(prompt, client)</p>
<p> return prompt, result</p>
<p> except APIError:</p>
<p> return prompt, "Falló la generación de respuesta después de varios intentos."</p>

<p>async def main():</p>
<p> prompts = [f"Cuéntame un hecho sobre el número {i}" for i in range(20)]</p>
<p> max_concurrent_requests = 5</p>
<p> semaphore = asyncio.Semaphore(max_concurrent_requests)</p>

<p>async with AsyncOpenAI() as client:</p>
<p> tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]</p>
<p> results = await asyncio.gather(*tasks)</p>

<p>for prompt, result in results:</p>
<p> print(f"Prompt: {prompt}\nRespuesta: {result}\n")</p>

asyncio.run(main())

Esta versión mejorada incluye:

• Una excepción personalizada APIError para errores relacionados con la API.
• Una función generate_text_with_retry decorada con @retry de la biblioteca tenacity, que implementa un reintento con retroceso exponencial.
• Manejo de errores en la función process_prompt para capturar y reportar fallas.

Optimización del rendimiento: Respuestas en streaming

Para la generación de contenido de larga forma, las respuestas en streaming pueden mejorar significativamente el rendimiento percibido de tu aplicación. En lugar de esperar a que se complete toda la respuesta, puedes procesar y mostrar fragmentos de texto a medida que están disponibles.

import asyncio
from openai import AsyncOpenAI

<p>async def stream_text(prompt, client):</p>
<p> stream = await client.chat.completions.create(</p>
<p> model="gpt-3.5-turbo",</p>
<p> messages=[{"role": "user", "content": prompt}],</p>
<p> stream=True</p>
<p> )</p>

<p>full_response = ""</p>
<p>async for chunk in stream:</p>
<p> if chunk.choices[0].delta.content is not None:</p>
<p> content = chunk.choices[0].delta.content</p>
<p> full_response += content</p>
<p> print(content, end="", flush=True)</p>

<p>print("\n")</p>
<p>return full_response</p>

<p>async def main():</p>
<p> prompt = "Escribe una historia corta sobre un científico que viaja en el tiempo."</p>

<p>async with AsyncOpenAI() as client:</p>
<p> result = await stream_text(prompt, client)</p>

<p>print(f"Respuesta completa:\n{result}")</p>

asyncio.run(main())

Este ejemplo demuestra cómo transmitir la respuesta desde la API, imprimiendo cada fragmento a medida que llega. Este enfoque es particularmente útil para aplicaciones de chat o cualquier escenario en el que desees proporcionar retroalimentación en tiempo real al usuario.

Construcción de flujos de trabajo asincrónicos con LangChain

Para aplicaciones de LLM más complejas, el marco de LangChain proporciona una abstracción de alto nivel que simplifica el proceso de encadenar varias llamadas a LLM y la integración de otras herramientas. Veamos un ejemplo de cómo utilizar LangChain con capacidades asincrónicas:

Este ejemplo muestra cómo LangChain se puede utilizar para crear flujos de trabajo más complejos con transmisión y ejecución asincrónica. El AsyncCallbackManager y StreamingStdOutCallbackHandler permiten la transmisión en tiempo real del contenido generado.

import asyncio
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.callbacks.manager import AsyncCallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

<p>async def generate_story(topic):</p>
<p> llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))</p>
<p> prompt = PromptTemplate(</p>
<p> input_variables=["topic"],</p>
<p> template="Escribe una historia corta sobre {topic}."</p>
<p> )</p>
<p> chain = LLMChain(llm=llm, prompt=prompt)</p>
<p> return await chain.arun(topic=topic)</p>

<p>async def main():</p>
<p> topics = ["un bosque mágico", "una ciudad futurista", "una civilización submarina"]</p>
<p> tasks = [generate_story(topic) for topic in topics]</p>
<p> stories = await asyncio.gather(*tasks)</p>

<p>for topic, story in zip(topics, stories):</p>
<p> print(f"\nTema: {topic}\nHistoria: {story}\n{'='*50}\n")</p>

asyncio.run(main())

Servir aplicaciones de LLM asincrónicas con FastAPI

Para hacer que tu aplicación de LLM asincrónica esté disponible como un servicio web, FastAPI es una excelente opción debido a su soporte nativo para operaciones asincrónicas. Aquí hay un ejemplo de cómo crear un punto de acceso de API simple para generación de texto:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

<p>class GenerationRequest(BaseModel):</p>
<p> prompt: str</p>

<p>class GenerationResponse(BaseModel):</p>
<p> generated_text: str</p>

<p>@app.post("/generate", response_model=GenerationResponse)</p>
<p>async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks):</p>
<p> response = await client.chat.completions.create(</p>
<p> model="gpt-3.5-turbo",</p>
<p> messages=[{"role": "user", "content": request.prompt}]</p>
<p> )</p>
<p> generated_text = response.choices[0].message.content</p>

<p> # Simula algún procesamiento posterior en segundo plano</p>
<p> background_tasks.add_task(log_generation, request.prompt, generated_text)</p>

<p> return GenerationResponse(generated_text=generated_text)</p>

<p>async def log_generation(prompt: str, generated_text: str):</p>
<p> # Simula registro o procesamiento adicional</p>
<p> await asyncio.sleep(2)</p>
<p> print(f"Registrado: Prompt '{prompt}' generó texto de longitud {len(generated_text)}")</p>

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

Esta aplicación FastAPI crea un punto de acceso /generate que acepta un prompt y devuelve texto generado. También demuestra cómo utilizar tareas en segundo plano para procesamiento adicional sin bloquear la respuesta.

Mejores prácticas y errores comunes

Al trabajar con llamadas a la API de LLM asincrónicas, mantén en mente estas mejores prácticas:

1. Utiliza agrupación de conexiones: Cuando realices múltiples solicitudes, reutiliza conexiones para reducir la sobrecarga.
2. Implementa un control de errores adecuado: Siempre ten en cuenta problemas de red, errores de API y respuestas inesperadas.
3. Respeta los límites de velocidad: Utiliza semáforos u otros mecanismos de control de concurrencia para evitar abrumar al servidor de la API.
4. Monitorea y registra: Implementa un registro exhaustivo para rastrear el rendimiento y identificar problemas.
5. Utiliza transmisión para contenido de larga forma: Mejora la experiencia del usuario y permite el procesamiento temprano de resultados parciales.