050 异步编程在Agent中的应用——从一次“假死”说起昨天凌晨两点我盯着终端里那个Agent的日志它卡在“正在调用天气API”这一行已经整整8分钟了。没有报错没有输出CPU占用率低得像在养老。按常理LLM等待响应顶多几十秒这次却像被谁施了定身术。我下意识按下CtrlC堆栈里清清楚楚写着——requests.get()阻塞在urllib3的socket读操作里。那一刻我就明白了Agent不是“假死”它只是被我自己写的同步代码活活堵住了。说起来丢人这个Agent号称能自主规划任务、调用多个工具结果我图省事在async函数里直接用了requests.get()。你猜怎么着Python的事件循环是单线程的requests.get()会死等网络响应这期间循环连呼吸的功夫都没有。其他本该并发的子任务、心跳检测、甚至超时取消的定时器全都排着队干瞪眼。这病根不在Agent的“脑子”而在于它的“血管”被同步代码堵死了。咱们先看一段反面教材。写Agent时你可能会觉得工具函数就是普通函数加不加async无所谓于是写出这种# 别这样写这是阻塞事件循环的万恶之源deffetch_weather(city):importrequests resprequests.get(fhttps://api.weather.com/{city})returnresp.json()asyncdefrun_agent(city):weatherfetch_weather(city)# 这一行卡住整个Agent瘫痪...这段代码挂在async体系里却用同步IO。事件循环看到fetch_weather()返回的是普通值根本没机会切换上下文它就是死等。解决方式很简单——换httpx.AsyncClient或者用asyncio.to_thread把同步函数丢进线程池。我后来把工具函数全改成异步版Agent立刻活了过来。importhttpxasyncdeffetch_weather(city):# 这才对嘛异步非阻塞事件循环可以随时切走asyncwithhttpx.AsyncClient()asclient:respawaitclient.get(fhttps://api.weather.com/{city})returnresp.json()但这只是第一步。Agent通常要同时调用多个工具比如查天气、查日历、查机票。三个请求如果串行耗时就是三倍写成并行可能一秒多搞定。于是你自然会想到asyncio.gather()。这里也有坑——gather()默认是“全都要”任何一个子任务抛出异常整个gather立刻取消其他任务跟着遭殃。而且异常会直接冒泡如果你没抓Agent整个就崩了。# 踩过坑的写法一个失败全盘崩溃resultsawaitasyncio.gather(fetch_weather(city),fetch_calendar(user_id),fetch_flights(dep,arr))更好的方式是把异常留给各任务自己处理或者用return_exceptionsTrue。我习惯封装一个safe_call让工具函数永不上抛异常只返回一个带状态的结构。asyncdefsafe_call(coro,*args,**kwargs):try:resultawaitcoro(*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把同时执行的协程数量压住。semasyncio.Semaphore(5)# 最多同时跑5个asyncdeflimited_call(coro):asyncwithsem:returnawaitcoro# 用法tasks[limited_call(fetch_weather(city))forcityincity_list]resultsawaitasyncio.gather(*tasks,return_exceptionsTrue)说到超时这是Agent必须有的保命手段。LLM有时候会抽风工具API也可能一直不响应。假如没有超时控制你的Agent就会像那次一样卡到天荒地老。asyncio.wait_for是救星给它一个协程和最大等待秒数超时直接抛TimeoutError你可以捕获后给出降级方案。try:resultawaitasyncio.wait_for(fetch_weather(city),timeout5.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,pendingawaitasyncio.wait(tasks,return_whenasyncio.FIRST_COMPLETED)# 拿到第一个完成的任务剩下的统统取消别浪费资源forpinpending:p.cancel()resultlist(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都是雨夜里的通途。 050 异步编程在Agent中的应用——从一次“假死”说起
050 异步编程在Agent中的应用——从一次“假死”说起昨天凌晨两点我盯着终端里那个Agent的日志它卡在“正在调用天气API”这一行已经整整8分钟了。没有报错没有输出CPU占用率低得像在养老。按常理LLM等待响应顶多几十秒这次却像被谁施了定身术。我下意识按下CtrlC堆栈里清清楚楚写着——requests.get()阻塞在urllib3的socket读操作里。那一刻我就明白了Agent不是“假死”它只是被我自己写的同步代码活活堵住了。说起来丢人这个Agent号称能自主规划任务、调用多个工具结果我图省事在async函数里直接用了requests.get()。你猜怎么着Python的事件循环是单线程的requests.get()会死等网络响应这期间循环连呼吸的功夫都没有。其他本该并发的子任务、心跳检测、甚至超时取消的定时器全都排着队干瞪眼。这病根不在Agent的“脑子”而在于它的“血管”被同步代码堵死了。咱们先看一段反面教材。写Agent时你可能会觉得工具函数就是普通函数加不加async无所谓于是写出这种# 别这样写这是阻塞事件循环的万恶之源deffetch_weather(city):importrequests resprequests.get(fhttps://api.weather.com/{city})returnresp.json()asyncdefrun_agent(city):weatherfetch_weather(city)# 这一行卡住整个Agent瘫痪...这段代码挂在async体系里却用同步IO。事件循环看到fetch_weather()返回的是普通值根本没机会切换上下文它就是死等。解决方式很简单——换httpx.AsyncClient或者用asyncio.to_thread把同步函数丢进线程池。我后来把工具函数全改成异步版Agent立刻活了过来。importhttpxasyncdeffetch_weather(city):# 这才对嘛异步非阻塞事件循环可以随时切走asyncwithhttpx.AsyncClient()asclient:respawaitclient.get(fhttps://api.weather.com/{city})returnresp.json()但这只是第一步。Agent通常要同时调用多个工具比如查天气、查日历、查机票。三个请求如果串行耗时就是三倍写成并行可能一秒多搞定。于是你自然会想到asyncio.gather()。这里也有坑——gather()默认是“全都要”任何一个子任务抛出异常整个gather立刻取消其他任务跟着遭殃。而且异常会直接冒泡如果你没抓Agent整个就崩了。# 踩过坑的写法一个失败全盘崩溃resultsawaitasyncio.gather(fetch_weather(city),fetch_calendar(user_id),fetch_flights(dep,arr))更好的方式是把异常留给各任务自己处理或者用return_exceptionsTrue。我习惯封装一个safe_call让工具函数永不上抛异常只返回一个带状态的结构。asyncdefsafe_call(coro,*args,**kwargs):try:resultawaitcoro(*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把同时执行的协程数量压住。semasyncio.Semaphore(5)# 最多同时跑5个asyncdeflimited_call(coro):asyncwithsem:returnawaitcoro# 用法tasks[limited_call(fetch_weather(city))forcityincity_list]resultsawaitasyncio.gather(*tasks,return_exceptionsTrue)说到超时这是Agent必须有的保命手段。LLM有时候会抽风工具API也可能一直不响应。假如没有超时控制你的Agent就会像那次一样卡到天荒地老。asyncio.wait_for是救星给它一个协程和最大等待秒数超时直接抛TimeoutError你可以捕获后给出降级方案。try:resultawaitasyncio.wait_for(fetch_weather(city),timeout5.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,pendingawaitasyncio.wait(tasks,return_whenasyncio.FIRST_COMPLETED)# 拿到第一个完成的任务剩下的统统取消别浪费资源forpinpending:p.cancel()resultlist(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都是雨夜里的通途。