Llamadas API asincrónicas de LLM en Python: una guía completa (2024)

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

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

Antes de profundizar en los detalles de las llamadas API LLM asincrónicas, establezcamos una base sólida en conceptos de programación asincrónica.

La programación asincrónica permite ejecutar múltiples operaciones simultáneamente sin bloquear el hilo principal de ejecución. En Python, esto se logra principalmente a través de la asíncio módulo, que proporciona un marco para escribir código concurrente utilizando corrutinas, bucles de eventos y futuros.

Conceptos clave:

Corutinas:Funciones definidas con definición asincrónica que se puede pausar y reanudar.
Evento Loop:El mecanismo de ejecución central que administra y ejecuta tareas asincrónicas.
Esperables:Objetos que se pueden utilizar con la palabra clave await (corrutinas, tareas, futuros).

He aquí un ejemplo sencillo para ilustrar estos conceptos:

import asyncioasync def greet(name): await asyncio.sleep(1) # Simulate an I/O operation print(f"Hello, {name}!")async def main(): await asyncio.gather( greet("Alice"), greet("Bob"), greet("Charlie") )asyncio.run(main())

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

La necesidad de Async en las llamadas API de LLM

Cuando trabajamos con API de LLM, a menudo nos encontramos con situaciones en las que necesitamos realizar múltiples llamadas a API, ya sea en secuencia o en paralelo. El código sincrónico tradicional puede generar cuellos de botella de rendimiento importantes, especialmente cuando se trata de operaciones de alta latencia, como solicitudes de red a servicios de LLM.

Configuración de su entorno

Para comenzar a utilizar las llamadas API asincrónicas de LLM, deberá configurar su entorno de Python con las bibliotecas necesarias. Esto es lo que necesitará:

3.7 Python o superior (para compatibilidad nativa con asyncio)
aiohttp:Una biblioteca de cliente HTTP asincrónica
openai: El oficial Cliente Python de OpenAI (si está utilizando los modelos GPT de OpenAI)
cadena larga:Un marco para crear aplicaciones con LLM (opcional, pero recomendado para flujos de trabajo complejos)

Puedes instalar estas dependencias usando pip:

pip install aiohttp openai langchain<div class="relative flex flex-col rounded-lg">

Llamadas API básicas de Async LLM con asyncio y aiohttp

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

import asyncioimport aiohttpfrom openai import AsyncOpenAIasync def generate_text(prompt, client): response = await client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.contentasync def main(): prompts = [ "Explain quantum computing in simple terms.", "Write a haiku about artificial intelligence.", "Describe the process of photosynthesis." ] async with AsyncOpenAI() as client: tasks = [generate_text(prompt, client) for prompt in prompts] results = await asyncio.gather(*tasks) for prompt, result in zip(prompts, results): print(f"Prompt: {prompt}\nResponse: {result}\n")asyncio.run(main())

En este ejemplo, definimos una función asincrónica generate_text que realiza una llamada a la API de OpenAI utilizando el cliente AsyncOpenAI. main La función crea múltiples tareas para diferentes indicaciones y usos. asyncio.gather() para ejecutarlos simultáneamente.

Este enfoque nos permite enviar múltiples solicitudes a la API de LLM simultáneamente, lo que reduce significativamente el tiempo total necesario para procesar todas las solicitudes.

Técnicas avanzadas: procesamiento por lotes y control de concurrencia

Si bien el ejemplo anterior demuestra los conceptos básicos de las llamadas API LLM asincrónicas, las aplicaciones del mundo real suelen requerir enfoques más sofisticados. Exploremos dos técnicas importantes: agrupar solicitudes y controlar la concurrencia.

Agrupamiento de solicitudes: cuando se trabaja con una gran cantidad de solicitudes, suele ser más eficiente agruparlas en lugar de enviar solicitudes individuales para cada solicitud. Esto reduce la sobrecarga de múltiples llamadas a la API y puede generar un mejor rendimiento.

import asynciofrom openai import AsyncOpenAIasync def process_batch(batch, client): responses = await asyncio.gather(*[ client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) for prompt in batch ]) return [response.choices[0].message.content for response in responses]async def main(): prompts = [f"Tell me a fact about number {i}" for i in range(100)] batch_size = 10 async with AsyncOpenAI() as client: results = [] for i in range(0, len(prompts), batch_size): batch = prompts[i:i+batch_size] batch_results = await process_batch(batch, client) results.extend(batch_results) for prompt, result in zip(prompts, results): print(f"Prompt: {prompt}\nResponse: {result}\n")asyncio.run(main())

Control de concurrencia: si bien la programación asincrónica permite la ejecución simultánea, es importante controlar el nivel de concurrencia para evitar sobrecargar el servidor API o exceder los límites de velocidad. Podemos usar asyncio.Semaphore para este propósito.

import asynciofrom openai import AsyncOpenAIasync def generate_text(prompt, client, semaphore): async with semaphore: response = await client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.contentasync def main(): prompts = [f"Tell me a fact about number {i}" for i in range(100)] max_concurrent_requests = 5 semaphore = asyncio.Semaphore(max_concurrent_requests) async with AsyncOpenAI() as client: tasks = [generate_text(prompt, client, semaphore) for prompt in prompts] results = await asyncio.gather(*tasks) for prompt, result in zip(prompts, results): print(f"Prompt: {prompt}\nResponse: {result}\n")asyncio.run(main())

En este ejemplo, utilizamos un semáforo para limitar la cantidad de solicitudes simultáneas a 5, lo que garantiza que no sobrecarguemos el servidor API.

Manejo de errores y reintentos en llamadas LLM asíncronas

Al trabajar con API externas, es fundamental implementar mecanismos sólidos de manejo de errores y reintentos. Mejoremos nuestro código para manejar errores comunes e implementar una reducción exponencial de los reintentos.

import asyncioimport randomfrom openai import AsyncOpenAIfrom tenacity import retry, stop_after_attempt, wait_exponentialclass APIError(Exception): pass@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))async def generate_text_with_retry(prompt, client): try: response = await client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content except Exception as e: print(f"Error occurred: {e}") raise APIError("Failed to generate text")async def process_prompt(prompt, client, semaphore): async with semaphore: try: result = await generate_text_with_retry(prompt, client) return prompt, result except APIError: return prompt, "Failed to generate response after multiple attempts."async def main(): prompts = [f"Tell me a fact about number {i}" for i in range(20)] max_concurrent_requests = 5 semaphore = asyncio.Semaphore(max_concurrent_requests) async with AsyncOpenAI() as client: tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts] results = await asyncio.gather(*tasks) for prompt, result in results: print(f"Prompt: {prompt}\nResponse: {result}\n")asyncio.run(main())

Esta versión mejorada incluye:

Una costumbre APIError excepción para errores relacionados con la API.
A generate_text_with_retry función decorada con @retry de la biblioteca tenacidad, implementando retroceso exponencial.
Manejo de errores en el process_prompt Función para detectar y reportar fallas.

Optimización del rendimiento: respuestas en streaming

Para la generación de contenido extenso, las respuestas en tiempo real pueden mejorar significativamente el rendimiento percibido de su aplicación. En lugar de esperar la respuesta completa, puede procesar y mostrar fragmentos de texto a medida que estén disponibles.

import asynciofrom openai import AsyncOpenAIasync def stream_text(prompt, client): stream = await client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], stream=True ) full_response = "" async for chunk in stream: if chunk.choices[0].delta.content is not None: content = chunk.choices[0].delta.content full_response += content print(content, end='', flush=True) print("\n") return full_responseasync def main(): prompt = "Write a short story about a time-traveling scientist." async with AsyncOpenAI() as client: result = await stream_text(prompt, client) print(f"Full response:\n{result}")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 desee proporcionar comentarios en tiempo real al usuario.

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

Para aplicaciones más complejas impulsadas por LLM, el Marco LangChain Proporciona una abstracción de alto nivel que simplifica el proceso de encadenamiento de múltiples llamadas LLM e integración de otras herramientas. Veamos un ejemplo de uso de LangChain con capacidades asincrónicas:

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

import asynciofrom langchain.llms import OpenAIfrom langchain.prompts import PromptTemplatefrom langchain.chains import LLMChainfrom langchain.callbacks.manager import AsyncCallbackManagerfrom langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandlerasync def generate_story(topic): llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()])) prompt = PromptTemplate( input_variables=["topic"], template="Write a short story about {topic}." ) chain = LLMChain(llm=llm, prompt=prompt) return await chain.arun(topic=topic)async def main(): topics = ["a magical forest", "a futuristic city", "an underwater civilization"] tasks = [generate_story(topic) for topic in topics] stories = await asyncio.gather(*tasks) for topic, story in zip(topics, stories): print(f"\nTopic: {topic}\nStory: {story}\n{'='*50}\n")asyncio.run(main())

Servicio de aplicaciones LLM asincrónicas con FastAPI

Para que su aplicación LLM asincrónica esté disponible como un servicio web, FastAPI es una excelente opción debido a su compatibilidad nativa con operaciones asincrónicas. A continuación, se muestra un ejemplo de cómo crear un punto final de API simple para la generación de texto:

from fastapi import FastAPI, BackgroundTasksfrom pydantic import BaseModelfrom openai import AsyncOpenAIapp = FastAPI()client = AsyncOpenAI()class GenerationRequest(BaseModel): prompt: strclass GenerationResponse(BaseModel): generated_text: str@app.post("/generate", response_model=GenerationResponse)async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks): response = await client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": request.prompt}] ) generated_text = response.choices[0].message.content # Simulate some post-processing in the background background_tasks.add_task(log_generation, request.prompt, generated_text) return GenerationResponse(generated_text=generated_text)async def log_generation(prompt: str, generated_text: str): # Simulate logging or additional processing await asyncio.sleep(2) print(f"Logged: Prompt '{prompt}' generated text of length {len(generated_text)}")if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

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

Mejores prácticas y errores comunes

A medida que trabaje con API LLM asincrónicas, tenga en cuenta estas prácticas recomendadas:

Utilice la agrupación de conexiones:Al realizar múltiples solicitudes, reutilice las conexiones para reducir la sobrecarga.
Implementar un manejo adecuado de errores:Tenga siempre en cuenta los problemas de red, los errores de API y las respuestas inesperadas.
Respetar los límites de velocidad:Utilice semáforos u otros mecanismos de control de concurrencia para evitar saturar la API.
Monitorizar y registrar:Implementar un registro integral para realizar un seguimiento del rendimiento e identificar problemas.
Utilice la transmisión para contenido de larga duración:Mejora la experiencia del usuario y permite el procesamiento temprano de resultados parciales.