Python异步操作后列表顺序丢失问题及解决咨询

问题出在你直接用共享列表reply_list让异步任务append结果，而异步任务的完成时间不一致——不符合条件的任务（直接返回空字符串）执行更快，会先被添加到列表，导致顺序混乱。asyncio.gather本身会按任务提交的顺序返回结果，所以正确的做法是让每个异步任务返回自己的结果，再通过gather收集，而非修改共享列表。

修改方案 #

1. 调整异步任务函数，返回结果而非修改共享列表 #

@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
async def async_generate(chain, query, line):
    if query and not query.isupper() and len(line) <= 6:
        resp = await chain.arun(query=query)
        # debugging
        print("query", query)
        print("response", resp)
        return resp
    else:
        return ""

2. 利用asyncio.gather的顺序特性收集结果 #

@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
async def generate_concurrently(line_list, value_list):
    # --- langchain boilerplate ---
    # 保留你的chain初始化代码

    # 创建任务列表，顺序与原数据一致
    tasks = [async_generate(chain, query, line) for query, line in zip(value_list, line_list)]
    # gather会按任务创建顺序返回结果，自动对应原数据行
    reply_list = await asyncio.gather(*tasks)

    return reply_list

为什么这样有效？ #

asyncio.gather(*tasks)会等待所有任务完成，然后返回一个列表，列表中元素的顺序完全等于你传入tasks的顺序，和任务实际完成的先后无关。这样不管某个任务执行快慢，结果都会和原line_list/value_list的行一一对应。

额外优化建议 #

• 如果DataFrame行数极多（上千行），可以考虑将任务分批处理，避免一次性创建过多异步任务导致资源耗尽
• 可以给每个任务添加超时控制，避免单个任务卡住影响整体流程：

  async with asyncio.timeout(30):
      resp = await chain.arun(query=query)
  

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