Объедините асинхронность с параллелизмом ⇐ Python

[Anonymous]

У меня есть следующий код:

Код: Выделить всё

async def run_task(...):
...
semaphore = asyncio.Semaphore(cfg.concurrency_limit)

async def run_single_sample(task_sample: TaskSample):
async with semaphore:
await run_agent(cfg, task_sample, cfg.output_dir / task.value)

samples = [run_single_sample(task_sample) for task_sample in sliced_samples]
await tqdm.gather(*samples, desc=f"Task: {task.value}")

Это часть приложения машинного обучения. Итак, представьте, что run_agent — это некая асинхронная чат-система, в которой LLM решают проблему. В этом процессе выполняется несколько вызовов API к OpenAI, некоторым локальным модальным окнам, а также немного процессам ЦП для выполнения локального кода.
Хотя асинхронность/блокировка имеет смысл для вызовов API , Я не уверен, что это имеет смысл для процессов ЦП. Потому что я так понимаю, что все еще последовательно. Например. пока образец 1 может ожидать вызова ЦП, начинается образец 2, но затем снова продолжается образец 1.
Глядя на мой вывод htop, кажется, что он использует 1 процессор на 100%, поэтому я решил возможно, из-за локального выполнения кода при запуске агента процессор может быть здесь узким местом.
Поэтому я попробовал объединить его с чем-то вроде ThreadPoolExecutor, asyncio.to_thread, asyncio.run_in_executor, ...
но кажется, что все они предназначены для запуска методов синхронизации.
Как я могу сделать это правильно? Или мой вариант использования какой-то ошибочный.

Подробнее здесь: https://stackoverflow.com/questions/793 ... -parallism