如何基于Python asyncio构建API调用流水线？

你的需求是让第一个API的调用完成一个就立刻启动第二个API调用，不用等所有第一个API都结束。核心思路是把每个输入对应的两个API调用封装成一个独立的协程，这样多个输入的处理可以并发执行，单个输入的两个步骤则是串行的（第一个完成后立刻走第二个）。

修正后的完整代码示例 #

首先，确保你的API调用协程复用aiohttp.ClientSession（原代码里没传session，这是实际使用时的错误，需要修正）：

import asyncio
import aiohttp

# 第一个API调用协程
async def getsomething(input_data, session):
    async with session.get(f"https://api.example.com/first/{input_data}") as resp:
        return await resp.json()

# 第二个API调用协程，用第一个的结果作为参数
async def get_next_thing(first_result, session):
    async with session.post("https://api.example.com/second", json=first_result) as resp:
        return await resp.json()

# 封装单个输入的完整流水线
async def process_single_input(input_data, session):
    # 第一步：调用第一个API
    first_output = await getsomething(input_data, session)
    # 第一步完成后立刻调用第二个API
    second_output = await get_next_thing(first_output, session)
    # 返回两个结果，或者按需只返回第二个
    return (first_output, second_output)

async def main(inputs):
    async with aiohttp.ClientSession() as session:
        # 为每个输入创建流水线任务
        tasks = [process_single_input(input, session) for input in inputs]
        # 等待所有流水线任务完成
        outputs = await asyncio.gather(*tasks)
    return outputs

if __name__ == "__main__":
    inputs = ["input1", "input2", "input3"]
    results = asyncio.run(main(inputs))
    print(results)

逻辑说明 #

• 每个process_single_input协程对应一个输入的完整流程：先跑第一个API，拿到结果后立刻跑第二个API。
• 多个process_single_input任务是并发执行的：当其中一个任务完成第一个API调用时，会自动切换到执行第二个API，而其他任务可能还在处理第一个API，完全不需要等所有第一个API都结束。
• 最终asyncio.gather会收集所有输入的最终结果（这里是第一个和第二个API的结果对）。

这种方式既保留了并发的效率，又实现了你要的“流水线”效果——第一个API完成一个就启动第二个。

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