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）

关键变化：

关键概念速查

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) 会阻塞整个程序