一、先通俗讲两个核心概念

1. `asyncio`：Python 协程，用来做IO 密集型并发

你调用图片生成 API 属于网络 IO：发请求→等待 LLM 服务返回图片，CPU 全程空闲在等待。

串行：一张一张调用 API，等上一张返回才发下一张，耗时叠加；
协程并发：一次性批量发起 N 个图片请求，所有网络等待并行进行，总耗时≈最慢那一张的耗时。

2. `run_in_executor` 解决的痛点：同步 API 不能直接放进协程

大部分第三方 LLM、图片生成 SDK 提供的都是同步阻塞接口（调用时代码卡住，直到接口返回才往下走），这种普通同步函数不能直接丢给 asyncio.gather 并发执行，会阻塞整个事件循环，并发直接失效。

loop.run_in_executor(线程池, 同步函数, 参数)

作用：把同步阻塞 API 放到独立线程里运行，包装成可被协程调度的异步任务，既保留原有同步 SDK 写法，又能实现并发。

3. `asyncio.gather`：批量收集多个异步任务，统一并发执行

把一堆包装好的异步任务丢进去，同时调度全部任务并发跑，等所有任务全部执行完毕后，统一收集所有返回结果。

二、结合你的业务场景拆解

你的需求：批量生成多张自媒体配图

改造前（串行）

plaintext

第1张图 → 调用同步图片API，等待3s返回
第2张图 → 调用同步图片API，等待3s返回
第3张图 → 调用同步图片API，等待3s返回
总耗时 ≈ 3+3+3 = 9s

改造方案：`run_in_executor` + `asyncio.gather`

每张图片的同步图片生成 API，用 run_in_executor 丢进线程池，封装成异步可调度任务；
所有图片的异步任务交给 asyncio.gather 一次性并发执行；
三张图片请求同时发出去，全部在网络等待，总耗时≈最慢单张 3s，耗时直接大幅下降。

核心逻辑代码示例

import asyncio
import requests# 原生同步：调用图片生成API（第三方SDK大多是同步阻塞）
def sync_generate_image(prompt):resp = requests.post("图片生成API地址", json={"prompt": prompt})return resp.content# 封装：把同步函数转为可协程调度的任务
async def async_generate_image(prompt):loop = asyncio.get_running_loop()# run_in_executor：同步函数丢线程池，变成异步任务result = await loop.run_in_executor(None, sync_generate_image, prompt)return resultasync def batch_generate_images(prompt_list):# 批量构建所有异步任务tasks = [async_generate_image(prompt) for prompt in prompt_list]# gather：所有任务并发执行，等待全部完成results = await asyncio.gather(*tasks)return results# 批量生成3张配图
asyncio.run(batch_generate_images(["封面1", "封面2", "封面3"]))