如何在Python中高效处理6000个API请求以缩短总耗时？

我有一个Python脚本，可向API发送请求并根据输入提示生成响应，每个请求通常耗时约10秒。现有一份6000行数据的CSV文件，需要将其中text列的值作为输入提示发送至API，并将生成的响应追加至CSV的新列中。

由于请求数量庞大，串行处理需要6000*10秒，耗时过长。以下是我当前的基础代码片段：

import pandas as pd
from requests_futures.sessions import FuturesSession
import time

# Define your existing function for sending a request
def send_request(session, input_prompt):
    url = 'https://ai-chat-test.***.workers.dev/'
    consolidated_request = f"""From the input "{input_prompt}", extract key descriptions, deduplicate them, and use these to create a compelling, concise description of the fashion item."""
    return session.post(url, json={"messages": [{"role": "user", "content": consolidated_request}]}, headers={"Content-Type": "application/json"})

# Function to apply send_request concurrently and save intermittently
def apply_concurrent_requests(df, function):
    session = FuturesSession(max_workers=5)
    futures = {function(session, text): index for index, text in enumerate(df['text'])}
    results = [None] * len(df)
    batch_counter = 0
    start_time = time.time()

    for future in as_completed(futures):
        index = futures[future]
        try:
            response = future.result()
            if response.status_code == 200:
                data = response.json()
                for message in data:
                    if message['role'] == 'assistant':
                        results[index] = message['content']
                        break
                else:
                    results[index] = "No assistant message found."
            else:
                results[index] = f"Failed to get a response, status code: {response.status_code}"
        except Exception as e:
            results[index] = f"Exception occurred: {e}"

        # Save after processing each batch of 100 or the last response
        batch_counter += 1
        if batch_counter % 100 == 0 or index + 1 == len(df):
            # Save the responses so far to the DataFrame
            for j, res in enumerate(results[:index+1]):
                df.at[j, 'response'] = res

            # Save to CSV
            df.to_csv('intermediate_metadata.csv', index=False)
            print(f"Processed {index+1} requests in {time.time() - start_time:.2f} seconds and saved to CSV.")
            start_time = time.time()  # Reset the timer for the next batch

    return df

df = pd.read_csv('metadata.csv')
updated_df = apply_concurrent_requests(df, send_request)
updated_df.to_csv('updated_metadata.csv', index=False)

优化方案 #

1. 合理调整并发线程数 #

原代码max_workers=5的并发数偏低，可根据API的限流规则（比如每秒请求数限制）适当调高至20-50区间。但需注意：

• 过高的并发可能触发API限流（返回429状态码），建议先小批量测试，找到最优并发数。
• 搭配重试机制，处理限流或临时网络故障。

2. 优化中间结果保存逻辑 #

原代码每次保存时循环遍历results[:index+1]更新DataFrame，效率极低。直接通过索引更新对应行即可：

# 替换原保存逻辑中的循环
df.at[index, 'response'] = results[index]

每次处理完一条请求就更新DataFrame对应行，批量保存时直接写入CSV即可。

3. 添加请求重试机制 #

使用requests.adapters.HTTPAdapter添加重试策略，处理临时失败的请求：

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session(max_workers, retries=3):
    session = FuturesSession(max_workers=max_workers)
    retry_strategy = Retry(
        total=retries,
        backoff_factor=1,  # 重试间隔依次为1s、2s、4s...
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

4. 改用异步IO框架（aiohttp） #

线程池的并发效率不如异步IO，对于大规模IO密集型任务，推荐使用aiohttp+asyncio实现更高并发：

异步版本示例代码

import pandas as pd
import aiohttp
import asyncio
import time
from tqdm import tqdm

async def send_request(session, input_prompt):
    url = 'https://ai-chat-test.***.workers.dev/'
    consolidated_request = f"""From the input "{input_prompt}", extract key descriptions, deduplicate them, and use these to create a compelling, concise description of the fashion item."""
    async with session.post(
        url,
        json={"messages": [{"role": "user", "content": consolidated_request}]},
        headers={"Content-Type": "application/json"}
    ) as response:
        if response.status == 200:
            data = await response.json()
            for message in data:
                if message['role'] == 'assistant':
                    return message['content']
            return "No assistant message found."
        else:
            return f"Failed to get a response, status code: {response.status}"

async def process_batch(session, batch):
    tasks = [send_request(session, text) for idx, text in batch]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [(idx, res) for (idx, _), res in zip(batch, results)]

async def apply_async_requests(df, max_concurrent=30, batch_size=100):
    df['response'] = None
    total_rows = len(df)
    start_time = time.time()

    async with aiohttp.ClientSession() as session:
        for i in tqdm(range(0, total_rows, batch_size)):
            batch = list(df.iloc[i:i+batch_size].iterrows())
            batch_results = await process_batch(session, batch)
            for idx, res in batch_results:
                if isinstance(res, Exception):
                    df.at[idx, 'response'] = f"Exception occurred: {res}"
                else:
                    df.at[idx, 'response'] = res
            
            # 保存中间结果
            df.to_csv('intermediate_metadata.csv', index=False)
            elapsed = time.time() - start_time
            print(f"Processed {min(i+batch_size, total_rows)}/{total_rows} requests in {elapsed:.2f} seconds")
    
    return df

if __name__ == "__main__":
    df = pd.read_csv('metadata.csv')
    updated_df = asyncio.run(apply_async_requests(df))
    updated_df.to_csv('updated_metadata.csv', index=False)

5. 进度可视化 #

使用tqdm库添加进度条，直观显示处理进度（如上面异步代码所示）。

关键注意事项 #

• API限流：务必遵守API的使用规则，避免因过高并发导致账号被封禁或请求被拒绝。
• 中间结果保存：定期保存中间结果，防止程序崩溃导致数据丢失。
• 异常处理：完善的异常处理能避免单个请求失败导致整个流程中断。

内容的提问来源于stack exchange，提问作者sdave