Что такое Asyncio?
В современном мире разработки AI-приложений производительность играет ключевую роль. При работе с большими языковыми моделями (LLMs) разработчики часто тратят много времени в ожидании — ожидание ответов API, завершения множества вызовов или операций ввода-вывода. Здесь на помощь приходит библиотека asyncio. Удивительно, но многие разработчики используют LLM, не осознавая, что могут улучшить свои приложения с помощью асинхронного программирования.
Что такое Asyncio?
Библиотека asyncio в Python позволяет писать параллельный код с использованием синтаксиса async/await, что позволяет эффективно выполнять несколько задач, связанных с вводом-выводом, в одном потоке. В своей основе asyncio работает с объектами, которые можно ожидать — обычно это корутины, которые планируются и выполняются циклом событий без блокировки.
Проще говоря, синхронный код выполняет задачи одну за другой, как если бы вы стояли в одной очереди в магазине, в то время как асинхронный код выполняет задачи одновременно, как если бы вы использовали несколько автоматов для самообслуживания. Это особенно полезно для API-вызовов (например, OpenAI, Anthropic, Hugging Face), где большая часть времени уходит на ожидание ответов, что позволяет значительно ускорить выполнение.
Начало работы с асинхронным Python
Пример: Запуск задач с использованием и без использования Asyncio
В этом примере мы запустим простую функцию три раза синхронно. Вывод показывает, что каждый вызов функции say_hello() печатает «Hello…», ждет 2 секунды, а затем печатает «…World!». Поскольку вызовы происходят один за другим, общее время ожидания составляет 6 секунд.
import time
def say_hello():
print("Hello...")
time.sleep(2) # симуляция ожидания (как вызов API)
print("...World!")
def main():
say_hello()
say_hello()
say_hello()
if __name__ == "__main__":
start = time.time()
main()
print(f"Завершено за {time.time() - start:.2f} секунд")
Ниже приведен код, который показывает, что все три вызова функции say_hello() начинаются почти одновременно. Каждый из них сразу печатает «Hello…», затем ждет 2 секунды одновременно, прежде чем напечатать «…World!».
import nest_asyncio, asyncio
nest_asyncio.apply()
import time
async def say_hello():
print("Hello...")
await asyncio.sleep(2) # симуляция ожидания (как вызов API)
print("...World!")
async def main():
# Выполнение задач одновременно
await asyncio.gather(
say_hello(),
say_hello(),
say_hello()
)
if __name__ == "__main__":
start = time.time()
asyncio.run(main())
print(f"Завершено за {time.time() - start:.2f} секунд")
Пример: Симуляция загрузки
Представьте, что вам нужно загрузить несколько файлов. Каждая загрузка занимает время, но в это время ваша программа может работать над другими загрузками, а не простаивать.
import asyncio
import random
import time
async def download_file(file_id: int):
print(f"Начинаю загрузку файла {file_id}")
download_time = random.uniform(1, 3) # симуляция времени загрузки
await asyncio.sleep(download_time) # неблокирующее ожидание
print(f"Завершена загрузка файла {file_id} за {download_time:.2f} секунд")
return f"Содержимое файла {file_id}"
async def main():
files = [1, 2, 3, 4, 5]
start_time = time.time()
# Запуск загрузок одновременно
results = await asyncio.gather(*(download_file(f) for f in files))
end_time = time.time()
print("\nВсе загрузки завершены.")
print(f"Общее время загрузки: {end_time - start_time:.2f} секунд")
print("Результаты:", results)
if __name__ == "__main__":
asyncio.run(main())
Использование Asyncio в AI-приложении с LLM
Теперь, когда мы понимаем, как работает asyncio, давайте применим это на практике в реальном AI-примере. Большие языковые модели (LLMs), такие как модели GPT от OpenAI, часто требуют множества вызовов API, каждый из которых требует времени для завершения. Если мы будем выполнять эти вызовы один за другим, мы потеряем драгоценное время, ожидая ответов.
В этом разделе мы сравним выполнение нескольких запросов с использованием и без использования asyncio с клиентом OpenAI. Мы используем 15 коротких запросов, чтобы наглядно продемонстрировать разницу в производительности.
!pip install openai
import asyncio
from openai import AsyncOpenAI
import os
from getpass import getpass
os.environ['OPENAI_API_KEY'] = getpass('Введите API-ключ OpenAI: ')
import time
from openai import OpenAI
# Создание синхронного клиента
client = OpenAI()
def ask_llm(prompt: str):
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
def main():
prompts = [
"Кратко объясните квантовые вычисления.",
"Напишите хайку о ИИ.",
"Перечислите 3 идеи стартапов в агритехе.",
"Сделайте краткое резюме фильма 'Начало'.",
"Объясните блокчейн за 2 предложения.",
"Напишите 3-строчное рассказ о роботе.",
"Перечислите 5 способов, как ИИ помогает здравоохранению.",
"Объясните бозон Хиггса простыми словами.",
"Опишите нейронные сети за 2 предложения.",
"Перечислите 5 идей для блога по возобновляемым источникам энергии.",
"Приведите короткую метафору для времени.",
"Перечислите 3 новые тенденции в машинном обучении.",
"Напишите короткий лимрик о программировании.",
"Объясните контролируемое и неконтролируемое обучение в одном предложении.",
"Перечислите 3 способа уменьшить городской трафик."
]
start = time.time()
results = []
for prompt in prompts:
results.append(ask_llm(prompt))
end = time.time()
for i, res in enumerate(results, 1):
print(f"\n--- Ответ {i} ---")
print(res)
print(f"\n[Синхронно] Завершено за {end - start:.2f} секунд")
if __name__ == "__main__":
main()
Синхронная версия обрабатывает все 15 запросов один за другим, поэтому общее время составляет сумму времени каждого запроса. Поскольку каждый запрос потребовал времени на выполнение, общее время выполнения оказалось значительно больше — в этом случае 49.76 секунд.
from openai import AsyncOpenAI
# Создание асинхронного клиента
client = AsyncOpenAI()
async def ask_llm(prompt: str):
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
async def main():
prompts = [
"Кратко объясните квантовые вычисления.",
"Напишите хайку о ИИ.",
"Перечислите 3 идеи стартапов в агритехе.",
"Сделайте краткое резюме фильма 'Начало'.",
"Объясните блокчейн за 2 предложения.",
"Напишите 3-строчное рассказ о роботе.",
"Перечислите 5 способов, как ИИ помогает здравоохранению.",
"Объясните бозон Хиггса простыми словами.",
"Опишите нейронные сети за 2 предложения.",
"Перечислите 5 идей для блога по возобновляемым источникам энергии.",
"Приведите короткую метафору для времени.",
"Перечислите 3 новые тенденции в машинном обучении.",
"Напишите короткий лимрик о программировании.",
"Объясните контролируемое и неконтролируемое обучение в одном предложении.",
"Перечислите 3 способа уменьшить городской трафик."
]
start = time.time()
results = await asyncio.gather(*(ask_llm(p) for p in prompts))
end = time.time()
for i, res in enumerate(results, 1):
print(f"\n--- Ответ {i} ---")
print(res)
print(f"\n[Асинхронно] Завершено за {end - start:.2f} секунд")
if __name__ == "__main__":
asyncio.run(main())
Асинхронная версия обрабатывает все 15 запросов одновременно, начиная их почти в одно и то же время, а не один за другим. В результате общее время выполнения оказалось близким к времени самого медленного отдельного запроса — 8.25 секунд, а не сумме всех запросов.
Почему это важно в AI-приложениях
В реальных AI-приложениях ожидание завершения каждого запроса перед началом следующего может быстро стать узким местом, особенно при работе с несколькими запросами или источниками данных. Это особенно распространено в таких рабочих процессах, как:
- Генерация контента для нескольких пользователей одновременно — например, чат-боты, рекомендательные системы или панели управления для нескольких пользователей.
- Множественные вызовы LLM в одном рабочем процессе — таких как резюмирование, уточнение, классификация или многошаговое рассуждение.
- Получение данных из нескольких API — например, комбинирование вывода LLM с информацией из векторной базы данных или внешних API.
Использование asyncio в этих случаях приносит значительные преимущества:
- Улучшенная производительность — благодаря параллельным вызовам API вместо ожидания каждого из них по отдельности ваша система может обрабатывать больше задач за меньшее время.
- Экономия затрат — более быстрое выполнение может снизить операционные расходы, а пакетирование запросов, где это возможно, может дополнительно оптимизировать использование платных API.
- Лучший пользовательский опыт — параллельность делает приложения более отзывчивыми, что критически важно для систем реального времени, таких как AI-ассистенты и чат-боты.
- Масштабируемость — асинхронные паттерны позволяют вашему приложению обрабатывать гораздо больше одновременных запросов без пропорционального увеличения потребления ресурсов.
Часто задаваемые вопросы (FAQ)
1. Что такое асинхронное программирование?
Асинхронное программирование позволяет выполнять несколько операций одновременно, не дожидаясь завершения каждой из них, что улучшает производительность и отзывчивость приложений.
2. Как asyncio работает в Python?
Библиотека asyncio позволяет писать асинхронный код, используя синтаксис async/await, что упрощает управление параллельными задачами.
3. В чем преимущества использования asyncio в AI-приложениях?
Использование asyncio позволяет значительно сократить время ожидания, улучшить производительность и снизить затраты на использование API.
4. Какие ошибки обычно совершают разработчики при использовании asyncio?
Частые ошибки включают неправильное понимание корутин, блокировку основного потока и неправильное управление состояниями задач.
5. Как начать использовать asyncio в своем проекте?
Начните с установки библиотеки, изучите базовые концепции синтаксиса async/await и попробуйте реализовать простые примеры, как показано в этой статье.
6. Есть ли лайфхаки для оптимизации кода с использованием asyncio?
Используйте asyncio.gather() для параллельного выполнения задач, избегайте блокирующих вызовов и следите за управлением ресурсами, чтобы избежать утечек памяти.
