这类"大部分时间在等外部（网络、磁盘、数据库）"的任务，叫 I/O 密集型。Agent 干的活几乎全是这种——调大模型、查向量库、访问工具 API，无一不是"发出去、等回来"。所以异步对 Agent不是锦上添花，是刚需。

同一个服务员招呼 3 桌，两种干法的时间线对比：

await 就是服务员说"这桌的菜我先挂着等，趁这空儿我去招呼别桌"的那个动作：

import asyncio                        # Python 自带的异步库

async def fetch_weather(city):        # async def = 这是一个协程(可让出的函数)
    print(f"开始查 {city} 天气…")
    await asyncio.sleep(1)            # await = 等 1 秒(模拟网络)，等的时候让出控制权
    print(f"{city} 天气拿到了")
    return f"{city}: 晴 28℃"          # 和普通函数一样能 return

import asyncio

async def main():                     # 通常把顶层逻辑放进一个叫 main 的协程
    result = await fetch_weather("北京")   # 这里能用 await，因为在 async def 里
    print("结果:", result)

asyncio.run(main())                   # ← 唯一的“启动”入口，从普通代码进入异步世界
# 输出：
# 开始查 北京 天气…
# (等约1秒)
# 北京 天气拿到了
# 结果: 北京: 晴 28℃

import asyncio, time

async def fetch_weather(city):
    await asyncio.sleep(1)            # 每个都“等”1 秒
    return f"{city}: 晴"

async def main():
    t = time.time()
    # gather 把 3 个协程同时发起，一起等 → 总共约 1 秒而不是 3 秒
    results = await asyncio.gather(
        fetch_weather("北京"),
        fetch_weather("上海"),
        fetch_weather("广州"),
    )
    print(results)                    # ['北京: 晴', '上海: 晴', '广州: 晴']（顺序和传入一致）
    print(f"耗时 {time.time()-t:.1f} 秒")   # 耗时 1.0 秒 ← 三个等待重叠了！

asyncio.run(main())

把今天的知识对到 Agent 的真实场景，你会发现异步无处不在：

• 并行取证：一个"研究 Agent"要同时查 5 个资料源，gather 一起查，5 秒变 1 秒；
• 多智能体协作：主管同时派 3 个专家 Agent 干活，它们并发跑，最后汇总（后面阶段 8 会讲）；
• 流式输出：大模型一个字一个字往外吐，你要一边收一边处理，天生就是异步的活（明天/Day 12 会用到）；
• 高并发服务：你的 Agent 做成 API 后要同时接待很多用户——异步能用一个进程扛住大量"都在等模型回话"的请求。

第②坑最隐蔽也最重要：异步世界里混进一个"阻塞"调用，会把整条流水线全堵死——就像服务员本该去招呼别桌，却站在后厨前发呆。所以昨天的同步 requests 在异步代码里要换成它的异步兄弟 httpx：

import httpx                          # pip install httpx（requests 的异步版）

async def get_json(url):
    async with httpx.AsyncClient() as client:   # async with = 异步版的 with
        resp = await client.get(url, timeout=10)  # await 发请求，等的时候让出控制权
        return resp.json()

# 之后就能 await asyncio.gather(get_json(u1), get_json(u2), ...) 真正并发拉取

✋ 动手：亲手把 3 秒压成 1 秒

# 无需任何第三方库，asyncio 是 Python 自带的
python async_demo.py

# 存成 async_demo.py
import asyncio, time

async def task(name, seconds):
    print(f"{name} 开始…")
    await asyncio.sleep(seconds)      # 模拟等网络
    print(f"{name} 完成")
    return name

async def main():
    # —— 对照 1：串行(连续 await)，看它花约 3 秒 ——
    t = time.time()
    await task("串行A", 1); await task("串行B", 1); await task("串行C", 1)
    print(f"串行耗时 {time.time()-t:.1f} 秒\n")

    # —— 对照 2：并发(gather)，看它只花约 1 秒 ——
    t = time.time()
    await asyncio.gather(task("并发A", 1), task("并发B", 1), task("并发C", 1))
    print(f"并发耗时 {time.time()-t:.1f} 秒")

asyncio.run(main())

进阶挑战：把 gather 里的任务改成 5 个、每个睡 2 秒，先猜总耗时，再运行验证。（答案：约 2 秒，因为 5 个 2 秒的等待全重叠了。）