AI API异步调用与批处理优化：提升10倍吞吐量

import requests

def process_text(text): response = requests.post( "https://api.openai.com/v1/chat/completions", headers={"Authorization": "Bearer sk-xxx"}, json={ "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": text}] } ) return response.json()

# 处理1000条数据
for text in texts: result = process_text(text)  # 阻塞等待，一条一条处理

import asyncio

async def say_hello(): await asyncio.sleep(1)  # 模拟IO操作 print("Hello!")

# 这样不会执行
# say_hello()

# 这样才会执行
asyncio.run(say_hello())

async def main(): # 同时启动3个任务 await asyncio.gather( say_hello(), say_hello(), say_hello() )

asyncio.run(main())  # 总共只等1秒，不是3秒

pip install aiohttp

import aiohttp
import asyncio

async def call_api(session, text): url = "https://api.openai.com/v1/chat/completions" headers = {"Authorization": "Bearer sk-xxx"} payload = { "model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": text}] } async with session.post(url, headers=headers, json=payload) as response: return await response.json()

async def main(): texts = ["文本1", "文本2", "文本3"] # 复用session，启用连接池 async with aiohttp.ClientSession() as session: tasks = [call_api(session, text) for text in texts] results = await asyncio.gather(*tasks) return results

results = asyncio.run(main())

async def call_api_with_limit(session, semaphore, text): async with semaphore:  # 获取信号量，超过限制就等待 result = await call_api(session, text) return result

async def main(): # 限制最大并发数为50 semaphore = asyncio.Semaphore(50) async with aiohttp.ClientSession() as session: tasks = [ call_api_with_limit(session, semaphore, text) for text in texts ] results = await asyncio.gather(*tasks)

def chunk_list(lst, chunk_size): """把列表分块""" for i in range(0, len(lst), chunk_size): yield lst[i:i + chunk_size]

async def process_chunk(session, semaphore, chunk): """处理一个数据块（内部串行）""" results = [] for text in chunk: async with semaphore: result = await call_api(session, text) results.append(result) return results

async def main(): texts = [...]  # 10000条数据 chunks = list(chunk_list(texts, 100))  # 分成100块 semaphore = asyncio.Semaphore(10)  # 每块内部限制10并发 async with aiohttp.ClientSession() as session: tasks = [process_chunk(session, semaphore, chunk) for chunk in chunks] results = await asyncio.gather(*tasks)

from aiohttp import ClientSession, TCPConnector

# 自定义连接池配置
connector = TCPConnector( limit=100, # 总连接数限制 limit_per_host=50, # 每个主机的连接限制 enable_cleanup_closed=True, force_close=False, # 保持连接
)

async with ClientSession(connector=connector) as session: # 你的代码... pass

import random
from aiohttp import ClientError

async def call_api_with_retry(session, text, max_retries=3): for attempt in range(max_retries): try: return await call_api(session, text) except ClientError as e: if attempt == max_retries - 1: raise  # 最后一次失败，抛出异常 # 指数退避 + 随机抖动，避免惊群效应 wait_time = (2 ** attempt) + random.uniform(0, 1) print(f"请求失败，{wait_time:.1f}秒后重试...") await asyncio.sleep(wait_time)

import asyncio
import random
from typing import List, Dict, Any, Callable
from aiohttp import ClientSession, TCPConnector, ClientError

class AsyncAPIProcessor: """异步API批量处理器""" def __init__( self, api_key: str, base_url: str = "https://api.openai.com/v1", max_concurrent: int = 50, max_retries: int = 3, chunk_size: int = 100 ): self.api_key = api_key self.base_url = base_url self.max_concurrent = max_concurrent self.max_retries = max_retries self.chunk_size = chunk_size self.semaphore = asyncio.Semaphore(max_concurrent) async def _call_api( self, session: ClientSession, text: str, model: str = "gpt-3.5-turbo" ) -> Dict[Any, Any]: """单次API调用""" url = f"{self.base_url}/chat/completions" headers = {"Authorization": f"Bearer {self.api_key}"} payload = { "model": model, "messages": [{"role": "user", "content": text}] } async with session.post(url, headers=headers, json=payload) as resp: return await resp.json() async def _call_with_retry( self, session: ClientSession, text: str ) -> Dict[Any, Any]: """带重试的API调用""" for attempt in range(self.max_retries): try: async with self.semaphore: return await self._call_api(session, text) except ClientError as e: if attempt == self.max_retries - 1: return {"error": str(e), "text": text} wait_time = (2 ** attempt) + random.uniform(0, 1) await asyncio.sleep(wait_time) async def process( self, texts: List[str], progress_callback: Callable = None ) -> List[Dict[Any, Any]]: """批量处理文本""" # 连接池配置 connector = TCPConnector( limit=self.max_concurrent * 2, limit_per_host=self.max_concurrent, enable_cleanup_closed=True ) async with ClientSession(connector=connector) as session: tasks = [ self._call_with_retry(session, text) for text in texts ] # 分批处理，避免内存爆炸 results = [] for i in range(0, len(tasks), self.chunk_size): chunk = tasks[i:i + self.chunk_size] chunk_results = await asyncio.gather(*chunk) results.extend(chunk_results) if progress_callback: progress_callback(len(results), len(texts)) return results

# 使用示例
async def main(): processor = AsyncAPIProcessor( api_key="your-api-key", max_concurrent=50, max_retries=3 ) texts = ["文本1", "文本2", ...]  # 你的数据 def show_progress(current, total): print(f"进度: {current}/{total} ({current/total*100:.1f}%)") results = await processor.process(texts, show_progress) return results

# 运行
results = asyncio.run(main())