Асинхронные вызовы API LLM в Python: всесторонний гид

Как разработчики и ученые данных, мы часто сталкиваемся с необходимостью взаимодействия с этими мощными моделями через API. Однако, когда наши приложения растут в сложности и масштабе, необходимость эффективных и производительных взаимодействий с API становится крайне важной. Именно здесь асинхронное программирование сияет, позволяя нам максимизировать пропускную способность и минимизировать задержку при работе с API LLM.

В этом всестороннем руководстве мы исследуем мир асинхронных вызовов API LLM в Python. Мы рассмотрим все, от основ асинхронного программирования до продвинутых техник для обработки сложных рабочих процессов. К концу этой статьи у вас будет прочное понимание того, как использовать асинхронное программирование для ускорения ваших приложений, работающих с LLM.

Прежде чем мы углубимся в конкретные детали асинхронных вызовов API LLM, давайте установим прочную основу в концепциях асинхронного программирования.

Асинхронное программирование позволяет выполнять несколько операций одновременно без блокировки основной нити выполнения. В Python это в основном достигается с помощью модуля asyncio, который предоставляет框架 для написания конкрутного кода с использованием корутин, циклов событий и будущих значений.

Ключевые концепции:

• Корутины: Функции, определенные с помощью async def, которые могут быть приостановлены и возобновлены.
• Цикл событий: Центральный механизм выполнения, который управляет и запускает асинхронные задачи.
• Объекты, ожидающие выполнения: Объекты, которые могут быть использованы с ключевым словом await (корутины, задачи, будущие значения).

Вот простой пример, иллюстрирующий эти концепции:

import asyncio

<p>async def greet(name):</p>
<p> await asyncio.sleep(1) # Симулируем операцию ввода/вывода</p>
<p> print(f"Привет, {name}!")</p>

<p>async def main():</p>
<p> await asyncio.gather(</p>
<p> greet("Алиса"),</p>
<p> greet("Боб"),</p>
<p> greet("Чарли")</p>
<p> )</p>

asyncio.run(main())

В этом примере мы определяем асинхронную функцию greet, которая симулирует операцию ввода/вывода с помощью asyncio.sleep(). Функция main использует asyncio.gather(), чтобы запустить несколько приветствий одновременно. Несмотря на задержку, все три приветствия будут напечатаны после примерно 1 секунды, демонстрируя силу асинхронного выполнения.

Необходимость асинхронности в вызовах API LLM

Когда мы работаем с API LLM, мы часто сталкиваемся с сценариями, в которых нам необходимо сделать несколько вызовов API, либо последовательно, либо параллельно. Традиционный синхронный код может привести к значительным проблемам с производительностью, особенно при работе с операциями высокого уровня задержки, такими как запросы к сетевым службам LLM.

Рассмотрим сценарий, в котором нам необходимо сгенерировать резюме для 100 разных статей с помощью API LLM. С синхронным подходом каждый вызов API будет блокироваться до получения ответа, потенциально занимая несколько минут для завершения всех запросов. Асинхронный подход, с другой стороны, позволяет нам инициировать несколько вызовов API одновременно, значительно уменьшая общее время выполнения.

Настройка среды

Чтобы начать работать с асинхронными вызовами API LLM, вам необходимо настроить среду Python с необходимыми библиотеками. Вот что вам понадобится:

• Python 3.7 или выше (для родной поддержки asyncio)
• aiohttp: Асинхронный HTTP-клиент
• openai: Официальный клиент OpenAI для Python (если вы используете модели GPT от OpenAI)
• langchain: Фреймворк для построения приложений с LLM (необязательно, но рекомендуется для сложных рабочих процессов)

Вы можете установить эти зависимости с помощью pip:

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

Базовые асинхронные вызовы API LLM с asyncio и aiohttp

Давайте начнем с простого асинхронного вызова API LLM с помощью aiohttp. Мы будем использовать API GPT-3.5 от OpenAI в качестве примера, но концепции применяются и к другим API LLM.

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> "Объясните квантовые вычисления в простых терминах.",</p>
<p> "Напишите хайку об искусственном интеллекте.",</p>
<p> "Опишите процесс фотосинтеза."</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}\nОтвет: {result}\n")</p>

asyncio.run(main())

В этом примере мы определяем асинхронную функцию generate_text, которая делает вызов API LLM с помощью клиента AsyncOpenAI. Функция main создает несколько задач для разных промптов и использует asyncio.gather(), чтобы запустить их одновременно.

Этот подход позволяет нам отправлять несколько запросов к API LLM одновременно, значительно уменьшая общее время, необходимое для обработки всех промптов.

Продвинутые техники: пакетная обработка и контроль конкуренции

Хотя предыдущий пример демонстрирует основы асинхронных вызовов API LLM, реальные приложения часто требуют более сложных подходов. Давайте исследуем две важные техники: пакетную обработку и контроль конкуренции.

Пакетная обработка: когда мы имеем дело с большим количеством промптов, часто более эффективно объединять их в группы, а не отправлять отдельные запросы для каждого промпта. Это уменьшает накладные расходы на несколько вызовов API и может привести к лучшей производительности.

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"Расскажите факт о числе {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}\nОтвет: {result}\n")</p>

asyncio.run(main())

Контроль конкуренции: хотя асинхронное программирование позволяет выполнять задачи одновременно, важно контролировать уровень конкуренции, чтобы не перегружать сервер API или превышать лимиты скорости. Мы можем использовать asyncio.Semaphore для этой цели.

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"Расскажите факт о числе {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}\nОтвет: {result}\n")</p>

asyncio.run(main())

В этом примере мы используем семафор, чтобы ограничить количество одновременных запросов до 5, гарантируя, что мы не перегружаем сервер API.

Обработка ошибок и повторные вызовы в асинхронных вызовах LLM

Когда мы работаем с внешними API, крайне важно реализовать надежную обработку ошибок и механизмы повторных вызовов. Давайте улучшим наш код, чтобы обработать общие ошибки и реализовать экспоненциальную задержку для повторных вызовов.

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"Ошибка: {e}")</p>
<p> raise APIError("Не удалось сгенерировать текст")</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, "Не удалось сгенерировать ответ после нескольких попыток."</p>

<p>async def main():</p>
<p> prompts = [f"Расскажите факт о числе {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}\nОтвет: {result}\n")</p>

asyncio.run(main())

Эта улучшенная версия включает:

• Пользовательское исключение APIError для ошибок, связанных с API.
• Функцию generate_text_with_retry, украшенную @retry из библиотеки tenacity, реализующей экспоненциальную задержку.
• Обработку ошибок в функции process_prompt, чтобы поймать и сообщить о неудачах.

Оптимизация производительности: потоковая передача ответов

Для генерации длинного контента потоковая передача ответов может значительно улучшить воспринимаемую производительность вашего приложения. Вместо ожидания всего ответа вы можете обрабатывать и отображать фрагменты текста по мере их поступления.

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 = "Напишите короткий рассказ о путешественнике во времени."</p>

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

<p>print(f"Полный ответ:\n{result}")</p>

asyncio.run(main())

Этот пример демонстрирует, как потоково передавать ответ от API, печатая каждый фрагмент по мере его поступления. Этот подход особенно полезен для чат-приложений или любых сценариев, в которых вы хотите предоставить обратную связь пользователю в режиме реального времени.

Создание асинхронных рабочих процессов с LangChain

Для более сложных приложений, работающих с LLM, фреймворк LangChain предоставляет высокоуровневую абстракцию, которая упрощает процесс цепочки нескольких вызовов LLM и интеграции других инструментов. Давайте рассмотрим пример использования LangChain с асинхронными возможностями:

Этот пример показывает, как LangChain можно использовать для создания более сложных рабочих процессов с потоковой передачей и асинхронным выполнением. AsyncCallbackManager и StreamingStdOutCallbackHandler позволяют потоково передавать сгенерированный контент.

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="Напишите короткий рассказ о {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 = ["магический лес", "будущий город", "подводная цивилизация"]</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"\nТема: {topic}\nИстория: {story}\n{'='*50}\n")</p>

asyncio.run(main())

Обслуживание асинхронных приложений LLM с FastAPI

Чтобы сделать ваше асинхронное приложение LLM доступным в качестве веб-сервиса, FastAPI является отличным выбором благодаря своей родной поддержке асинхронных операций. Вот пример создания простого API-эндпоинта для генерации текста:

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> # Симулируем некоторые задачи в фоне</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> # Симулируем логирование или дополнительную обработку</p>
<p> await asyncio.sleep(2)</p>
<p> print(f"Лог: Промпт '{prompt}' сгенерировал текст длиной {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>

Этот пример FastAPI создает эндпоинт /generate, который принимает промпт и возвращает сгенерированный текст. Он также демонстрирует, как использовать задачи в фоне для дополнительной обработки без блокировки ответа.

Лучшие практики и распространенные ошибки

Когда вы работаете с асинхронными вызовами API LLM, помните о следующих лучших практиках:

1. Используйте пул соединений: при выполнении нескольких запросов повторно используйте соединения, чтобы уменьшить накладные расходы.
2. Реализуйте правильную обработку ошибок: всегда учитывайте проблемы с сетью, ошибки API и непредвиденные ответы.
3. Уважайте лимиты скорости: используйте семафоры или другие механизмы контроля конкуренции, чтобы не перегружать сервер API.
4. Мониторьте и логгируйте: реализуйте всестороннее логирование, чтобы отслеживать производительность и выявлять проблемы.
5. Используйте потоковую передачу для длинного контента: она улучшает пользовательский опыт и позволяет обрабатывать фрагменты ответа по мере их поступления.