_{🧠 图解记忆：为什么 AI 应用需要 asyncio；点击图片可查看原图。}

💡 答案要点

为什么 AI 应用需要 asyncio？

LLM API 调用 = I/O 密集型（网络等待远大于计算）。asyncio 让单线程并发成为可能：

传统同步（串行）：
请求1 → 等待API 3s → 请求2 → 等待API 3s → 总计 6s
  ↓ 不高效，CPU 全在等

异步并发：
请求1 ──等待API 3s──→ 处理结果
请求2 ──等待API 3s──→ 处理结果   总计 3s（并行）

核心概念：

概念	说明
async def	定义协程函数，不能用 return，要用 await
await	等待另一个协程完成，释放控制权
asyncio.gather()	并发执行多个协程
asyncio.create_task()	创建任务（后台执行）
asyncio.Semaphore	控制并发数（限流）

典型 AI 应用场景：

展开 Python 代码示例（37 行）

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

# 场景1：并发调用多个 LLM（批量生成）
async def batch_generate(prompts: list[str]):
    tasks = [client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": p}]
    ) for p in prompts]
    
    responses = await asyncio.gather(*tasks)
    return [r.choices[0].message.content for r in responses]

# 场景2：带并发限制的调用（Semaphore）
semaphore = asyncio.Semaphore(5)  # 最多5个并发

async def limited_call(prompt: str):
    async with semaphore:
        return await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )

# 场景3：超时控制
async def call_with_timeout(prompt: str, timeout=30):
    try:
        return await asyncio.wait_for(
            client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": prompt}]
            ),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        return {"error": "timeout"}

生产级异步封装：

class AsyncLLMClient:
    """带重试 + 并发控制 + 超时的异步 LLM 客户端"""
    
    def __init__(self, max_concurrent=10, timeout=60):
        self.client = AsyncOpenAI()
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.timeout = timeout
    
    async def chat(self, prompt: str, retry=3) -> str:
        for attempt in range(retry):
            try:
                async with self.semaphore:
                    response = await asyncio.wait_for(
                        self.client.chat.completions.create(
                            model="gpt-4o",
                            messages=[{"role": "user", "content": prompt}]
                        ),
                        timeout=self.timeout
                    )
                    return response.choices[0].message.content
            except (asyncio.TimeoutError, Exception) as e:
                if attempt == retry - 1:
                    raise
                await asyncio.sleep(2 ** attempt)  # 指数退避
        
    async def batch_chat(self, prompts: list[str]) -> list[str]:
        tasks = [self.chat(p) for p in prompts]
        return await asyncio.gather(*tasks)

面试话术：

"LLM API 调用通常是 I/O 密集型，asyncio 可以让等待网络响应的时间服务其他请求。但异步不等于无限并发：还需要 Semaphore 或连接池限制、超时、取消传播、针对可重试错误的退避，以及下游过载时的背压。收益应在供应商限流和目标流量下压测。"

📚 参考：Python asyncio 官方文档

---