Python 异步编程：await、asyncio.gather 和 asyncio.create_task 的区别与最佳实践

在现代 Python 开发中，异步编程几乎已经成为标配，尤其是在需要处理 I/O 密集型任务（如网络请求、数据库查询、LLM 调用）的场景。
但是很多同学在写异步代码时，经常搞不清楚 await、asyncio.gather 和 asyncio.create_task 的区别。

1. await 是什么？

await 的本质是：运行一个协程，等待它返回结果。

如果你写：

result = await some_async_function()

等价于“我要执行 some_async_function，并且在它完成之前，当前协程不往下走”。
如果不写 await，拿到的只是一个 coroutine 对象，它不会真正执行。

示例

import asyncioasync def foo():print("foo start")await asyncio.sleep(1)print("foo end")return 42async def main():coro = foo()print("没有 await:", coro)result = await foo()print("有 await:", result)asyncio.run(main())

输出：

没有 await: <coroutine object foo at 0x...>
foo start
foo end
有 await: 42

 结论：

• await 是必须的，否则协程不会执行。

• 如果有多个 await，它们是 顺序执行 的。

2. asyncio.gather：让多个任务并发执行

如果你有多个异步任务，但希望它们并发执行（而不是顺序），就用 asyncio.gather。

示例

import asyncioasync def foo(name, delay):print(f"{name} start")await asyncio.sleep(delay)print(f"{name} end")return nameasync def main():results = await asyncio.gather(foo("A", 2),foo("B", 2),)print("结果:", results)asyncio.run(main())

输出：

A start
B start
A end
B end
结果: ['A', 'B']

两个任务 并发执行，总耗时 2 秒，而不是 4 秒。

3. asyncio.create_task：更灵活的任务调度

gather 适合“一次性并发并收集结果”。
如果你希望任务 后台运行，并在合适的时机再获取结果，就要用 create_task。

示例

import asyncioasync def foo(name, delay):print(f"{name} start")await asyncio.sleep(delay)print(f"{name} end")return nameasync def main():task1 = asyncio.create_task(foo("A", 2))task2 = asyncio.create_task(foo("B", 2))print("任务已创建，可以先做点别的事...")# 等待单个任务result1 = await task1result2 = await task2print("结果:", result1, result2)asyncio.run(main())

输出：

A start
B start
任务已创建，可以先做点别的事...
A end
B end
结果: A B

create_task 的好处：

• 任务一旦创建就开始执行（不像 await 要等到被调用时才跑）。

• 可以分开等待结果，适合 流式处理 或 长时间运行的后台任务。

4. 三者的对比

5. 实战：并发调用 LLM 接口

假设我们要并发调用大语言模型，问三个问题：

import asyncio
from autogen_core.models import UserMessage
from autogen_learn.llm_client import model_clientasync def ask(question: str):result = await model_client.create([UserMessage(content=question, source="user")])return resultasync def main():# 方法一：顺序（效率低）r1 = await ask("Capital of France?")r2 = await ask("Capital of Germany?")r3 = await ask("Capital of Japan?")print(r1, r2, r3)# 方法二：gather 并发（推荐）results = await asyncio.gather(ask("Capital of France?"),ask("Capital of Germany?"),ask("Capital of Japan?"))print(results)# 方法三：create_task（更灵活）task1 = asyncio.create_task(ask("Capital of France?"))task2 = asyncio.create_task(ask("Capital of Germany?"))task3 = asyncio.create_task(ask("Capital of Japan?"))print("任务已发出，可以先处理别的逻辑…")answers = [await task1, await task2, await task3]print(answers)await model_client.close()asyncio.run(main())

• 如果要 一次性并发多个请求 → gather 更合适。

• 如果要 分批处理结果，或边发边收 → create_task 更灵活。

6. 总结

• await：运行并等待一个任务，默认是顺序执行。

• asyncio.gather：并发执行多个任务，等待所有完成。

• asyncio.create_task：任务一创建就执行，可以灵活等待结果，适合流式或后台任务。

如果你只是想并发几个请求，用 gather 就够了。
如果你要调度复杂任务，create_task 会更好。