如何通过异步调用更新列表、字典或Pandas DataFrame？

你的场景属于IO密集型任务（API请求），用线程池或异步IO能大幅提升执行效率，下面给出两种无需修改DataFrame更新逻辑的可行方案，核心都是先批量并行计算结果，再统一赋值给DataFrame列：

方案1：用线程池并行（无需修改原有calculate函数） #

适合不想改动现有calculate同步逻辑的场景，利用concurrent.futures.ThreadPoolExecutor实现并行请求：

from concurrent.futures import ThreadPoolExecutor
from pandas import DataFrame

# 你的原有calculate函数（无需修改）
def calculate(text: str):
    # 示例API请求逻辑
    # import requests
    # resp = requests.get(f"https://your-api-url.com?text={text}")
    # return resp.json()
    pass

def perform_calculations(df: DataFrame):
    # 提取所有待计算的文本列表
    texts = df["text"].tolist()
    
    # 初始化线程池，max_workers根据API并发限制调整（别超过接口限流）
    with ThreadPoolExecutor(max_workers=10) as executor:
        # executor.map会严格保持结果顺序与输入texts一致，避免行错位
        results = list(executor.map(calculate, texts))
    
    # 统一将结果赋值给DataFrame新列
    df["calculations"] = results
    return df

方案2：用异步IO（性能更优，需修改calculate为异步） #

如果追求更高并发效率，可以将同步API请求改为异步（比如用aiohttp替代requests），搭配asyncio批量执行：

import asyncio
import aiohttp
from pandas import DataFrame

# 将原有calculate改为异步版本
async def async_calculate(session: aiohttp.ClientSession, text: str):
    try:
        async with session.get(f"https://your-api-url.com?text={text}") as resp:
            return await resp.json()
    except Exception as e:
        # 捕获请求异常，返回默认值避免中断整个任务
        return None

async def batch_async_calculations(texts: list):
    async with aiohttp.ClientSession() as session:
        # 创建所有异步任务
        tasks = [async_calculate(session, text) for text in texts]
        # 等待所有任务完成，结果顺序与任务列表一致
        results = await asyncio.gather(*tasks)
    return results

def perform_calculations(df: DataFrame):
    texts = df["text"].tolist()
    # 运行异步任务并获取结果
    results = asyncio.run(batch_async_calculations(texts))
    df["calculations"] = results
    return df

注意事项 #

• 两种方案都会严格保证结果顺序与原DataFrame行顺序一致，直接赋值不会出现行错位问题
• 务必根据目标API的并发/频率限制调整max_workers（线程池）或任务数量（异步），避免被封禁
• 建议在计算函数中添加异常捕获逻辑，防止单个请求失败导致整个任务中断

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