企业官网建设流程全解析

050 异步编程在Agent中的应用——从一次“假死”说起

昨天凌晨两点，我盯着终端里那个Agent的日志，它卡在“正在调用天气API”这一行已经整整8分钟了。没有报错，没有输出，CPU占用率低得像在养老。按常理，LLM等待响应顶多几十秒，这次却像被谁施了定身术。我下意识按下Ctrl+C，堆栈里清清楚楚写着——requests.get()阻塞在urllib3的socket读操作里。那一刻我就明白了，Agent不是“假死”，它只是被我自己写的同步代码活活堵住了。

说起来丢人，这个Agent号称能自主规划任务、调用多个工具，结果我图省事，在async函数里直接用了requests.get()。你猜怎么着？Python的事件循环是单线程的，requests.get()会死等网络响应，这期间循环连呼吸的功夫都没有。其他本该并发的子任务、心跳检测、甚至超时取消的定时器，全都排着队干瞪眼。这病根不在Agent的“脑子”，而在于它的“血管”被同步代码堵死了。

咱们先看一段反面教材。写Agent时，你可能会觉得工具函数就是普通函数，加不加async无所谓，于是写出这种：

# 别这样写！这是阻塞事件循环的万恶之源deffetch_weather(city):importrequests resp=requests.get(f"https://api.weather.com/{city}")returnresp.json()asyncdefrun_agent(city):weather=fetch_weather(city)# 这一行卡住，整个Agent瘫痪...

这段代码挂在async体系里，却用同步IO。事件循环看到fetch_weather()返回的是普通值，根本没机会切换上下文，它就是死等。解决方式很简单——换httpx.AsyncClient，或者用asyncio.to_thread把同步函数丢进线程池。我后来把工具函数全改成异步版，Agent立刻活了过来。

importhttpxasyncdeffetch_weather(city):# 这才对嘛，异步非阻塞，事件循环可以随时切走asyncwithhttpx.AsyncClient()asclient:resp=awaitclient.get(f"https://api.weather.com/{city}")returnresp.json()

但这只是第一步。Agent通常要同时调用多个工具，比如查天气、查日历、查机票。三个请求如果串行，耗时就是三倍；写成并行，可能一秒多搞定。于是你自然会想到asyncio.gather()。这里也有坑——gather()默认是“全都要”，任何一个子任务抛出异常，整个gather立刻取消，其他任务跟着遭殃。而且异常会直接冒泡，如果你没抓，Agent整个就崩了。

# 踩过坑的写法：一个失败，全盘崩溃results=awaitasyncio.gather(fetch_weather(city),fetch_calendar(user_id),fetch_flights(dep,arr))

更好的方式是把异常留给各任务自己处理，或者用return_exceptions=True。我习惯封装一个safe_call，让工具函数永不上抛异常，只返回一个带状态的结构。

asyncdefsafe_call(coro,*args,**kwargs):try:result=awaitcoro(*args,**kwargs)return{"ok":True,"data":result}exceptExceptionase:# 这里记日志，别静默吞掉，不然排查问题要哭logger.exception("tool call failed")return{"ok":False,"error":str(e)}

另一个容易忽略的点是并发数。Agent要是规划出20个工具调用，你一股脑全用gather并发，几个免费API分分钟给你限流或封IP。这时候就得用信号量asyncio.Semaphore，把同时执行的协程数量压住。

sem=asyncio.Semaphore(5)# 最多同时跑5个asyncdeflimited_call(coro):asyncwithsem:returnawaitcoro# 用法tasks=[limited_call(fetch_weather(city))forcityincity_list]results=awaitasyncio.gather(*tasks,return_exceptions=True)

说到超时，这是Agent必须有的保命手段。LLM有时候会抽风，工具API也可能一直不响应。假如没有超时控制，你的Agent就会像那次一样卡到天荒地老。asyncio.wait_for是救星，给它一个协程和最大等待秒数，超时直接抛TimeoutError，你可以捕获后给出降级方案。

try:result=awaitasyncio.wait_for(fetch_weather(city),timeout=5.0)exceptasyncio.TimeoutError:# 超时了，给个默认值，别让Agent死等result={"default":"sunny"}

如果你需要更精细的控制，可以用asyncio.wait配合FIRST_COMPLETED，比如同时发起三个提供相似服务的API，谁先回来用谁，剩下两个直接取消。这在Agent决策链路里特别有用，能用更低延迟换来更强的鲁棒性。

tasks=[fetch_from_provider_a(),fetch_from_provider_b(),fetch_from_provider_c()]done,pending=awaitasyncio.wait(tasks,return_when=asyncio.FIRST_COMPLETED)# 拿到第一个完成的任务，剩下的统统取消，别浪费资源forpinpending:p.cancel()result=list(done)[0].result()

还有一个细节，很多人在Agent里写time.sleep(1)来“等待重试”或“限流”。这又是同步阻塞的函数，在异步环境里照样冻住事件循环。换成await asyncio.sleep(1)，才合规矩。

再聊聊和Agent框架的结合。像LangChain、AutoGen这些主流框架，它们的核心AgentExecutor、BaseTool都提供了异步接口。你自定义工具时，别只实现_run，顺手把_arun也实现掉。否则Agent在异步执行流程里调用你的工具，绕回同步方法，不仅可能阻塞，还会多出线程切换的开销。我见过不少项目，明明框架是异步的，自定义工具是同步的，整个流程的性能被拖到惨不忍睹。记住，工具函数能异步就不要同步。

调试异步Agent，经验比什么都宝贵。我自己的习惯是给每个重要协程都打上“入口/出口”日志，格式带上协程名和当前时间，千万别嫌日志多，异步调度的顺序有时候诡异到你根本猜不到谁先谁后。

另一个坑在asyncio.run()上。它只能调用一次，而且会创建新事件循环。如果你在Jupyter或某些脚本环境里多次执行，会报“event loop is already running”的错。我的做法是写一个main()，在if __name__ == "__main__":里调用，不要把asyncio.run()包在其他地方。

最后说说取消。Agent如果收到用户中断信号，或者主任务超时，你需要优雅地让所有子协程停止。不要只靠Task.cancel()，还得处理CancelledError，在清理阶段释放资源，比如关闭数据库连接、关闭HTTP客户端。别小看这个，否则你跑一晚上Agent，第二天发现文件句柄泄漏到报警。

回到我最初那次假死，修复其实只花了十分钟。把requests换成了httpx，给所有工具调用加了超时，再引入信号量控制并发，Agent从此再也没卡死过。可那晚的教训一直刻在脑子里：异步编程不是“for循环加并发”的炫技，它是一套关于资源调度、异常隔离、超时兜底的心法。Agent再聪明，它的骨架也是由一个个异步任务拼起来的。骨架松了，再强的模型也跑不动。

写给你几条实在话。第一，Agent里所有的IO都得异步化，包括HTTP、数据库、文件读写，一个漏网之鱼就能堵死全家。第二，不要相信上游API的响应时间，所有外部调用必须配超时。第三，给每个异步任务想好它被取消时该怎么办。第四，遇到莫名其妙的问题，先打印事件循环的任务堆栈，很多时候答案就在你没异步干净的那个函数里。

异步编程就像给Agent建排水系统。平时看不出来，一旦暴雨来了，谁家的井盖没打开、管道里堵了石头，一眼就能现形。愿你写的每个Agent，都是雨夜里的通途。