Agent开发——Day 04 async/await 异步编程
Day 04 案例分析：async/await 异步编程
案例一：LLM 并发调用（最贴近 Agent 开发的场景）
场景描述
你在开发一个 Agent，用户问了一个复杂问题，你需要同时调用 3 个 LLM：
- LLM-A：生成初稿
- LLM-B：检查事实
- LLM-C：评估语气
同步写法（慢）：
result_a=call_llm_a(prompt)# 等 2 秒result_b=call_llm_b(prompt)# 等 2 秒result_c=call_llm_c(prompt)# 等 2 秒# 总共等 6 秒异步写法（快）：
result_a,result_b,result_c=awaitasyncio.gather(call_llm_a(prompt),call_llm_b(prompt),call_llm_c(prompt),)# 总共等 2 秒（最慢的那个）结论：asyncio.gather()是 Agent 开发中最常用的并发工具。
案例二：FastAPI 的本质（理解"为什么框架要求 async"）
FastAPI、LangGraph 的 Server 模式，都要求你写async def。原因是：
# FastAPI 路由（必须是 async def）@app.post("/chat")asyncdefchat(request:ChatRequest):# 当这里在等 LLM 响应时，FastAPI 可以去处理其他用户的请求response=awaitcall_llm(request.message)return{"reply":response}如果用同步def，服务器处理一个用户时，其他用户的请求全部排队等待。
用async def，等待 IO 的时候 CPU 可以去处理其他请求，吞吐量大幅提升。
案例三：Day 3 代码的异步改造
原始同步版本
importrequestsdefcall_api_with_auth(api_key:str,prompt:str)->str:url="https://httpbin.org/post"headers={"Authorization":f"Bearer{api_key}","Content-Type":"application/json"}body={"model":"some-model","messages":[{"role":"user","content":prompt}]}response=requests.post(url,headers=headers,json=body,timeout=10)response.raise_for_status()echo=response.json()print("服务器收到的 headers:",echo["headers"].get("Authorization"))return"调用成功"异步改造版本（见 day004.py）
关键变化：
| 同步 | 异步 |
|---|---|
import requests | import httpx或import aiohttp |
def | async def |
requests.post() | await client.post() |
| 直接调用函数 | asyncio.run(main()) |
关键概念速查
async def— 声明一个协程函数
asyncdefmy_function():return"结果"调用它不会立刻执行，而是返回一个"协程对象"（承诺将来会执行）。
await— 等待协程完成
result=awaitmy_function()# 挂起当前函数，等 my_function 完成await只能在async def函数内部使用。
asyncio.run()— 启动事件循环（程序入口）
if__name__=="__main__":asyncio.run(main())# 在同步代码中运行异步函数的唯一正确方式asyncio.gather()— 并发执行多个协程
results=awaitasyncio.gather(task1(),task2(),task3())# results 是一个列表，顺序与传入顺序一致asyncio.sleep()— 异步等待（不阻塞其他任务）
awaitasyncio.sleep(1)# 等1秒，但期间其他协程可以运行# 对比：time.sleep(1) 会阻塞整个程序