Ускорьте работу с LLM с помощью asyncio: асинхронный Python для быстрого AI

сентября 27, 2025 · 4 min

Зачем нужен asyncio в AI

Во многих AI-проектах большая часть времени уходит на ожидание — ответа API, завершения операций ввода-вывода или нескольких вызовов. asyncio в Python помогает не терять это время: с async/await вы можете запускать I/O-ориентированные задачи конкурентно в одном потоке.

Как работает asyncio

asyncio планирует awaitable-объекты (обычно корутины) в событийном цикле. Вместо того чтобы блокировать выполнение при ожидании I/O, корутина отдает управление через await, и могут выполняться другие корутины. Проще говоря: синхронный код — одна касса в магазине, асинхронный — несколько самообслуживаний, что повышает пропускную способность при ожиданиях.

Примеры: синхронно против асинхронно

Синхронный пример: три последовательных вызова функции, которая ждет 2 секунды каждый раз. Общее время равно сумме ожиданий.

import time

def say_hello():
    print("Hello...")
    time.sleep(2)  # simulate waiting (like an API call)
    print("...World!")

def main():
    say_hello()
    say_hello()
    say_hello()

if __name__ == "__main__":
    start = time.time()
    main()
    print(f"Finished in {time.time() - start:.2f} seconds")

Асинхронная версия: три корутины стартуют почти одновременно и ждут параллельно, поэтому общее время примерно равно самому долгому ожиданию, а не сумме.

import nest_asyncio, asyncio
nest_asyncio.apply()
import time

async def say_hello():
    print("Hello...")
    await asyncio.sleep(2)  # simulate waiting (like an API call)
    print("...World!")

async def main():
    # Run tasks concurrently
    await asyncio.gather(
        say_hello(),
        say_hello(),
        say_hello()
    )

if __name__ == "__main__":
    start = time.time()
    asyncio.run(main())
    print(f"Finished in {time.time() - start:.2f} seconds")

Симуляция загрузки: несколько загрузок выполняются одновременно. Каждая загрузка имитирует разное время с помощью asyncio.sleep, и общая продолжительность приблизительно равна самой длительной загрузке.

import asyncio
import random
import time

async def download_file(file_id: int):
    print(f"Start downloading file {file_id}")
    download_time = random.uniform(1, 3)  # simulate variable download time
    await asyncio.sleep(download_time)    # non-blocking wait
    print(f"Finished downloading file {file_id} in {download_time:.2f} seconds")
    return f"File {file_id} content"

async def main():
    files = [1, 2, 3, 4, 5]

    start_time = time.time()
    
    # Run downloads concurrently
    results = await asyncio.gather(*(download_file(f) for f in files))
    
    end_time = time.time()
    print("\nAll downloads completed.")
    print(f"Total time taken: {end_time - start_time:.2f} seconds")
    print("Results:", results)

if __name__ == "__main__":
    asyncio.run(main())

Применение asyncio в LLM-воркфлоу

Вызовы LLM (OpenAI, Anthropic, Hugging Face и т. п.) — это классические I/O-операции: каждый запрос большую часть времени ждет ответа сервера. Если выполнять много промптов по очереди, общее время складывается. Ниже приведено сравнение: синхронный подход и асинхронный.

Требования и настройка (пример):

!pip install openai

import asyncio
from openai import AsyncOpenAI


import os
from getpass import getpass
os.environ['OPENAI_API_KEY'] = getpass('Enter OpenAI API Key: ')

Синхронный клиент (запросы по очереди):

import time
from openai import OpenAI

# Create sync client
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 = [
    "Briefly explain quantum computing.",
    "Write a 3-line haiku about AI.",
    "List 3 startup ideas in agri-tech.",
    "Summarize Inception in 2 sentences.",
    "Explain blockchain in 2 sentences.",
    "Write a 3-line story about a robot.",
    "List 5 ways AI helps healthcare.",
    "Explain Higgs boson in simple terms.",
    "Describe neural networks in 2 sentences.",
    "List 5 blog post ideas on renewable energy.",
    "Give a short metaphor for time.",
    "List 3 emerging trends in ML.",
    "Write a short limerick about programming.",
    "Explain supervised vs unsupervised learning in one sentence.",
    "List 3 ways to reduce urban traffic."
]

    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--- Response {i} ---")
        print(res)

    print(f"\n[Synchronous] Finished in {end - start:.2f} seconds")

if __name__ == "__main__":
    main()

Асинхронный клиент (запускает все запросы практически одновременно):

from openai import AsyncOpenAI

# Create async client
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 = [
    "Briefly explain quantum computing.",
    "Write a 3-line haiku about AI.",
    "List 3 startup ideas in agri-tech.",
    "Summarize Inception in 2 sentences.",
    "Explain blockchain in 2 sentences.",
    "Write a 3-line story about a robot.",
    "List 5 ways AI helps healthcare.",
    "Explain Higgs boson in simple terms.",
    "Describe neural networks in 2 sentences.",
    "List 5 blog post ideas on renewable energy.",
    "Give a short metaphor for time.",
    "List 3 emerging trends in ML.",
    "Write a short limerick about programming.",
    "Explain supervised vs unsupervised learning in one sentence.",
    "List 3 ways to reduce urban traffic."
]

    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--- Response {i} ---")
        print(res)

    print(f"\n[Asynchronous] Finished in {end - start:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

Практические преимущества в AI-воркфлоу

Повышенная пропускная способность: параллельные вызовы API сокращают общее время выполнения.
Экономия ресурсов и затрат: быстрее — значит дешевле при оплате по времени или квотам.
Лучший пользовательский опыт: асинхронность делает приложения отзывчивее.
Масштабируемость: асинхронные шаблоны позволяют обрабатывать больше одновременных задач без линейного роста ресурсов.

Когда выбирать asyncio

Применяйте asyncio для I/O-ориентированных задач (вызовы API, сетевые запросы, обращения к БД). Для CPU-ограниченных задач лучше подходят multiprocessing или другие способы параллелизации.

Если ваше AI-приложение выполняет много запросов к LLM, комбинирует данные из нескольких API или обслуживает множество пользователей одновременно, asyncio поможет значительно сократить время ожидания и ускорить работу системы.