Chamadas de API LLM Assíncronas em Python: Um Guia Abrangente

Como desenvolvedores e cientistas de dados, muitas vezes nos encontramos precisando interagir com esses poderosos modelos por meio de APIs. No entanto, à medida que nossas aplicações crescem em complexidade e escala, a necessidade de interações de API eficientes e performáticas se torna crucial. É aqui que a programação assíncrona brilha, permitindo que maximizemos o throughput e minimizemos a latência ao trabalhar com APIs LLM.

Neste guia abrangente, exploraremos o mundo de chamadas de API LLM assíncronas em Python. Cobriremos tudo, desde os conceitos básicos de programação assíncrona até técnicas avançadas para lidar com fluxos de trabalho complexos. Ao final deste artigo, você terá uma compreensão sólida de como aproveitar a programação assíncrona para impulsionar suas aplicações alimentadas por LLM.

Antes de mergulharmos nos detalhes específicos de chamadas de API LLM assíncronas, vamos estabelecer uma base sólida em conceitos de programação assíncrona.

A programação assíncrona permite que várias operações sejam executadas concorrentemente sem bloquear a thread principal de execução. Em Python, isso é alcançado principalmente por meio do módulo asyncio, que fornece um framework para escrever código concorrente usando corrotinas, loops de eventos e futuros.

Conceitos-chave:

• Corrotinas: Funções definidas com async def que podem ser pausadas e retomadas.
• Loop de Eventos: O mecanismo central de execução que gerencia e executa tarefas assíncronas.
• Objetos Aguardáveis: Objetos que podem ser usados com a palavra-chave await (corrotinas, tarefas, futuros).

Aqui está um exemplo simples para ilustrar esses conceitos:

import asyncio

<p>async def saudar(nome):
await asyncio.sleep(1) # Simula uma operação de E/S
print(f"Olá, {nome}!")</p>

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

asyncio.run(main())

Neste exemplo, definimos uma função assíncrona saudar que simula uma operação de E/S com asyncio.sleep(). A função main usa asyncio.gather() para executar várias saudações concorrentemente. Apesar do atraso de sono, todas as três saudações serão impressas após aproximadamente 1 segundo, demonstrando o poder da execução assíncrona.

A Necessidade de Assincronia em Chamadas de API LLM

Ao trabalhar com APIs LLM, frequentemente nos deparamos com cenários em que precisamos fazer várias chamadas de API, seja em sequência ou paralelo. O código síncrono tradicional pode levar a gargalos de desempenho significativos, especialmente ao lidar com operações de alta latência, como solicitações de rede para serviços LLM.

Considere um cenário em que precisamos gerar resumos para 100 artigos diferentes usando uma API LLM. Com uma abordagem síncrona, cada chamada de API bloquearia até receber uma resposta, potencialmente levando vários minutos para concluir todas as solicitações. Uma abordagem assíncrona, por outro lado, permite que iniciemos várias chamadas de API concorrentemente, reduzindo drasticamente o tempo total de execução.

Configurando o Seu Ambiente

Para começar a trabalhar com chamadas de API LLM assíncronas, você precisará configurar seu ambiente Python com as bibliotecas necessárias. Aqui está o que você precisará:

• Python 3.7 ou superior (para suporte nativo ao asyncio)
• aiohttp: Uma biblioteca de cliente HTTP assíncrona
• openai: O cliente Python oficial da OpenAI (se você estiver usando os modelos GPT da OpenAI)
• langchain: Um framework para construir aplicações com LLMs (opcional, mas recomendado para fluxos de trabalho complexos)

Você pode instalar essas dependências usando pip:

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

Chamadas de API LLM Assíncronas Básicas com asyncio e aiohttp

Vamos começar fazendo uma chamada assíncrona simples para uma API LLM usando aiohttp. Usaremos a API GPT-3.5 da OpenAI como exemplo, mas os conceitos se aplicam a outras APIs LLM também.

import asyncio
import aiohttp
from openai import AsyncOpenAI

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

<p>async def main():
prompts = [
"Explique computação quântica em termos simples.",
"Escreva um haicai sobre inteligência artificial.",
"Descreva o processo de fotossíntese."
]</p>

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

<p>for prompt, resultado in zip(prompts, resultados):
print(f"Prompt: {prompt}\nResposta: {resultado}\n")</p>

asyncio.run(main())

Neste exemplo, definimos uma função assíncrona gerar_texto que faz uma chamada para a API da OpenAI usando o cliente AsyncOpenAI. A função main cria várias tarefas para diferentes prompts e usa asyncio.gather() para executá-las concorrentemente.

Essa abordagem permite que enviamos várias solicitações para a API LLM simultaneamente, reduzindo significativamente o tempo total necessário para processar todos os prompts.

Técnicas Avançadas: Lote e Controle de Concorrência

Enquanto o exemplo anterior demonstra os conceitos básicos de chamadas de API LLM assíncronas, aplicações do mundo real frequentemente exigem abordagens mais sofisticadas. Vamos explorar duas técnicas importantes: lote de solicitações e controle de concorrência.

Lote de Solicitações: Ao lidar com um grande número de prompts, é mais eficiente agrupá-los em lotes em vez de enviar solicitações individuais para cada prompt. Isso reduz a sobrecarga de várias chamadas de API e pode levar a um melhor desempenho.

import asyncio
from openai import AsyncOpenAI

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

<p>async def main():
prompts = [f"Diga um fato sobre o número {i}" for i in range(100)]
tamanho_do_lote = 10</p>

<p>async with AsyncOpenAI() as client:
resultados = []
for i in range(0, len(prompts), tamanho_do_lote):
lote = prompts[i:i+tamanho_do_lote]
resultados_do_lote = await processar_lote(lote, client)
resultados.extend(resultados_do_lote)</p>

<p>for prompt, resultado in zip(prompts, resultados):
print(f"Prompt: {prompt}\nResposta: {resultado}\n")</p>

asyncio.run(main())

Controle de Concorrência: Embora a programação assíncrona permita a execução concorrente, é importante controlar o nível de concorrência para evitar sobrecarregar o servidor da API ou exceder os limites de taxa. Podemos usar asyncio.Semaphore para esse propósito.

import asyncio
from openai import AsyncOpenAI

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

<p>async def main():
prompts = [f"Diga um fato sobre o número {i}" for i in range(100)]
max_solicitações_concorrentes = 5
semaphore = asyncio.Semaphore(max_solicitações_concorrentes)</p>

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

<p>for prompt, resultado in zip(prompts, resultados):
print(f"Prompt: {prompt}\nResposta: {resultado}\n")</p>

asyncio.run(main())

Neste exemplo, usamos um semaphore para limitar o número de solicitações concorrentes a 5, garantindo que não sobrecarreguemos o servidor da API.

Tratamento de Erros e Retentativas em Chamadas LLM Assíncronas

Ao trabalhar com APIs externas, é crucial implementar um tratamento de erros robusto e mecanismos de retentativa. Vamos aprimorar nosso código para lidar com erros comuns e implementar uma retentativa com backoff 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))
async def gerar_texto_com_retentativa(prompt, client):
try:
resposta = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return resposta.choices[0].message.content
except Exception as e:
print(f"Erro ocorreu: {e}")
raise APIError("Falha ao gerar texto")</p>

<p>async def processar_prompt(prompt, client, semaphore):
async with semaphore:
try:
resultado = await gerar_texto_com_retentativa(prompt, client)
return prompt, resultado
except APIError:
return prompt, "Falha ao gerar resposta após várias tentativas."</p>

<p>async def main():
prompts = [f"Diga um fato sobre o número {i}" for i in range(20)]
max_solicitações_concorrentes = 5
semaphore = asyncio.Semaphore(max_solicitações_concorrentes)</p>

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

<p>for prompt, resultado in resultados:
print(f"Prompt: {prompt}\nResposta: {resultado}\n")</p>

asyncio.run(main())

Esta versão aprimorada inclui:

• Uma exceção personalizada APIError para erros relacionados à API.
• Uma função gerar_texto_com_retentativa decorada com @retry da biblioteca tenacity, implementando backoff exponencial.
• Tratamento de erros na função processar_prompt para capturar e relatar falhas.

Otimizando o Desempenho: Respostas em Streaming

Para a geração de conteúdo de longa forma, as respostas em streaming podem melhorar significativamente o desempenho percebido da sua aplicação. Em vez de esperar pela resposta completa, você pode processar e exibir pedaços de texto à medida que ficam disponíveis.

import asyncio
from openai import AsyncOpenAI

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

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

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

<p>async def main():
prompt = "Escreva uma história curta sobre um cientista viajante no tempo."</p>

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

<p>print(f"Resposta completa:\n{resultado}")</p>

asyncio.run(main())

Este exemplo demonstra como transmitir a resposta da API, imprimindo cada pedaço à medida que chega. Essa abordagem é particularmente útil para aplicações de chat ou qualquer cenário em que você deseje fornecer feedback em tempo real ao usuário.

Construindo Fluxos de Trabalho Assíncronos com LangChain

Para aplicações mais complexas alimentadas por LLM, o framework LangChain fornece uma abstração de alto nível que simplifica o processo de encadear várias chamadas de LLM e integrar outras ferramentas. Vamos explorar um exemplo de uso do LangChain com capacidades assíncronas:

Este exemplo mostra como o LangChain pode ser usado para criar fluxos de trabalho mais complexos com execução assíncrona e streaming. O AsyncCallbackManager e StreamingStdOutCallbackHandler habilitam o streaming de conteúdo gerado em tempo real.

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 gerar_história(tópico):
llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
prompt = PromptTemplate(
input_variables=["tópico"],
template="Escreva uma história curta sobre {tópico}."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(tópico=tópico)</p>

<p>async def main():
tópicos = ["uma floresta mágica", "uma cidade futurista", "uma civilização submarina"]
tarefas = [gerar_história(tópico) for tópico in tópicos]
histórias = await asyncio.gather(*tarefas)</p>

<p>for tópico, história in zip(tópicos, histórias):
print(f"\nTópico: {tópico}\nHistória: {história}\n{'='*50}\n")</p>

asyncio.run(main())

Servindo Aplicações LLM Assíncronas com FastAPI

Para tornar sua aplicação LLM assíncrona disponível como um serviço web, o FastAPI é uma escolha excelente devido ao seu suporte nativo a operações assíncronas. Aqui está um exemplo de como criar um endpoint de API simples para geração de texto:

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

app = FastAPI()
client = AsyncOpenAI()

<p>class SolicitaçãoDeGeração(BaseModel):
prompt: str</p>

<p>class RespostaDeGeração(BaseModel):
texto_gerado: str</p>

<p>@app.post("/gerar", response_model=RespostaDeGeração)
async def gerar_texto(solicitação: SolicitaçãoDeGeração, background_tasks: BackgroundTasks):
resposta = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": solicitação.prompt}]
)
texto_gerado = resposta.choices[0].message.content</p>

<p># Simula alguma pós-processamento em segundo plano
background_tasks.add_task(logar_geração, solicitação.prompt, texto_gerado)</p>

<p>return RespostaDeGeração(texto_gerado=texto_gerado)</p>

<p>async def logar_geração(prompt: str, texto_gerado: str):
# Simula logging ou processamento adicional
await asyncio.sleep(2)
print(f"Logado: Prompt '{prompt}' gerou texto de comprimento {len(texto_gerado)}")</p>

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

Esta aplicação FastAPI cria um endpoint /gerar que aceita um prompt e retorna texto gerado. Também demonstra como usar tarefas em segundo plano para processamento adicional sem bloquear a resposta.

Práticas Recomendadas e Armadilhas Comuns

Ao trabalhar com chamadas de API LLM assíncronas, mantenha essas práticas recomendadas em mente:

1. Use pooling de conexões: Ao fazer múltiplas solicitações, reutilize conexões para reduzir a sobrecarga.
2. Implemente tratamento de erros adequado: Sempre contabilize problemas de rede, erros de API e respostas inesperadas.
3. Respeite os limites de taxa: Use semáforos ou outros mecanismos de controle de concorrência para evitar sobrecarregar o servidor da API.
4. Monitore e registre: Implemente registro abrangente para acompanhar o desempenho e identificar problemas.
5. Use streaming para conteúdo de longa forma: Isso melhora a experiência do usuário e permite processamento antecipado de resultados parciais.