6.1 异步编程基础

async/await 语法

🎯 小白理解指南：async/await 怎么读？

• async def：定义一个异步函数（也叫"协程"）
• await：暂停并等待某个操作完成

把异步函数想象成一个可以暂停的任务：

• 遇到 await 就暂停，让出时间给其他任务
• 等待的操作完成后，从暂停的地方继续执行

就像你在等外卖时先去刷手机，外卖到了再继续吃饭。

import asyncio

async def fetch_data(id: int) -> str:
    """异步获取数据"""
    # await asyncio.sleep(1) 模拟等待网络请求
    # 🎯 这1秒内，程序可以去做别的事，不会傻等
    await asyncio.sleep(1)
    return f"数据 {id}"

async def main():
    # 顺序执行（慢）—— 一个等完再执行下一个
    result1 = await fetch_data(1)  # 等 1 秒
    result2 = await fetch_data(2)  # 再等 1 秒
    print(f"顺序: {result1}, {result2}")  # 总共 2 秒

    # 并发执行（快）—— 同时开始，一起等
    # 🎯 gather() 会同时启动所有任务，等全部完成后返回结果列表
    results = await asyncio.gather(
        fetch_data(1),  # 同时开始
        fetch_data(2),  # 同时开始
        fetch_data(3)   # 同时开始
    )
    print(f"并发: {results}")  # 总共只要 1 秒！

# 🎯 asyncio.run() 是启动异步程序的入口
# 它会创建一个"事件循环"来调度所有异步任务
asyncio.run(main())

异步 Agent 实现

🎯 小白理解指南：异步 Agent 有什么好处？

假设你有 10 个搜索任务，每个需要 0.5 秒：

• 同步 Agent：一个一个做，总共 5 秒
• 异步 Agent：同时发起 10 个请求，总共只要 0.5 秒！

这就是为什么 LangChain 和 LangGraph 都支持异步——让 AI Agent 更快！

import asyncio
from typing import List

class AsyncAgent:
    """异步 Agent"""

    def __init__(self, name: str):
        self.name = name

    async def process_task(self, task: str) -> str:
        """处理单个任务"""
        await asyncio.sleep(0.5)  # 模拟网络请求或 API 调用
        return f"[{self.name}] 完成: {task}"

    async def process_batch(self, tasks: List[str]) -> List[str]:
        """批量处理任务"""
        # 🎯 *[...] 是解包语法，把列表变成多个参数
        # gather() 会同时启动所有任务
        results = await asyncio.gather(
            *[self.process_task(task) for task in tasks]
        )
        return results

async def main():
    agent = AsyncAgent("WorkerBot")
    tasks = ["任务1", "任务2", "任务3"]

    import time
    start = time.time()
    results = await agent.process_batch(tasks)
    end = time.time()

    print(f"结果: {results}")
    print(f"耗时: {end - start:.2f}秒")  # ~0.5秒（并发）
    # 🎯 如果用同步方式，3个任务×0.5秒=1.5秒
    # 用异步，3个任务同时做，只要0.5秒！

asyncio.run(main())

🎯 小白常见问题

Q: 什么时候用异步？ A: 当你的程序需要等待外部操作时：网络请求、文件读写、数据库查询。这些操作 CPU 不忙，只是在等。

Q: 什么时候不用异步？ A: 当你的程序需要大量计算时（CPU 密集型），异步帮不上忙。这时候应该用多进程（multiprocessing）。

Q: 异步代码好像很复杂？ A: 刚开始确实需要适应。记住核心规则：

1. 异步函数用 async def 定义
2. 调用异步函数用 await
3. 并发执行用 asyncio.gather()
4. 启动程序用 asyncio.run()

---

下一节：6.2 异步工具和上下文