摘要当 Agent 需要调用外部 API、查询数据库时同步工具会阻塞整个事件循环。本文教你用async def编写生产级异步工具实现并发执行、结构化错误处理和优雅降级让 Agent 性能提升数倍。引言你有没有遇到过这样的场景Agent 需要同时查询三个新闻源的信息结果工具一个接一个地串行执行每个 API 调用等待 2 秒三个就是 6 秒。用户盯着屏幕干等体验极差。问题的根源在于I/O 密集型操作API 调用、数据库查询、文件读写使用了同步方式。在同步模式下CPU 在等待网络响应时完全空闲事件循环被阻塞其他任务无法执行。解决方案就是异步工具——用async def定义工具让 Agent 在等待 I/O 时切换执行其他任务甚至并发调用多个工具。核心概念同步 vs 异步一张表说清维度同步工具 (def)异步工具 (async def)定义方式def tool_func(...)async def tool_func(...)I/O 等待时阻塞事件循环CPU 空闲释放控制权执行其他任务调用方式tool.invoke({...})await tool.ainvoke({...})并发能力串行执行asyncio.gather并发适用场景CPU 密集型计算API 调用、DB 查询、文件读写tool识别自动识别为同步工具自动识别为异步工具关键一点tool装饰器会自动检测函数是否为async def无需额外配置。为什么异步工具如此重要同步执行 3 个 API 调用每个 2 秒 ├── API 1: ████████ 2s ├── API 2: ████████ 2s ├── API 3: ████████ 2s 总耗时: 6s 异步并发执行 3 个 API 调用每个 2 秒 ├── API 1: ████████ 2s ├── API 2: ████████ 2s ├── API 3: ████████ 2s 总耗时: 2s性能提升3 倍而且 API 调用越多差距越大。代码示例1. 基础异步工具用async def定义异步工具tool自动识别importasyncioimportloggingfromlangchain_core.toolsimporttool loggerlogging.getLogger(__name__)toolasyncdeffetch_news_articles(topic:str,max_results:int5)-list:Fetch latest news articles about a topic. Use this to stay updated on: - Industry trends and news - Company announcements - Market developments Args: topic: News topic or keyword max_results: Maximum number of articles (default: 5, max: 20) Returns: List of article dictionaries with title, source, published_date try:logger.info(fFetching news for topic:{topic})awaitasyncio.sleep(0.5)# 模拟网络延迟# 限制最大结果数防止数据过载max_resultsmin(max_results,20)articles[{title:fArticle{i}about{topic},source:News Source,published_date:2026-06-22}foriinrange(max_results)]returnarticlesexceptasyncio.TimeoutError:logger.error(fTimeout fetching news for:{topic})return[{error:Request timeout, please try again}]exceptExceptionase:logger.error(fError fetching news:{str(e)})return[{error:fFailed to fetch news:{str(e)}}]注意三个关键点async def—tool自动识别为异步工具ainvoke调用—await fetch_news_articles.ainvoke({topic: AI})结构化错误返回— 返回[{error: ...}]而非抛出异常2. 并发执行多个工具用asyncio.gather同时调用多个工具性能翻倍importasyncioimporttimefromlangchain_core.toolsimporttooltoolasyncdefsearch_weather(city:str)-str:查询城市天气awaitasyncio.sleep(1)# 模拟 1 秒 API 延迟returnf{city}: 晴, 25°Ctoolasyncdefsearch_stock(symbol:str)-str:查询股票价格awaitasyncio.sleep(1)# 模拟 1 秒 API 延迟returnf{symbol}: ¥150.00 (2.3%)toolasyncdefsearch_news(keyword:str)-str:搜索新闻awaitasyncio.sleep(1)# 模拟 1 秒 API 延迟returnf关于{keyword}的最新新闻AI 行业持续升温...# 串行执行慢asyncdefrun_sequential():starttime.time()r1awaitsearch_weather.ainvoke({city:北京})r2awaitsearch_stock.ainvoke({symbol:AAPL})r3awaitsearch_news.ainvoke({keyword:AI})elapsedtime.time()-startprint(f串行耗时:{elapsed:.1f}s)# ≈ 3.0s# 并发执行快asyncdefrun_concurrent():starttime.time()resultsawaitasyncio.gather(search_weather.ainvoke({city:北京}),search_stock.ainvoke({symbol:AAPL}),search_news.ainvoke({keyword:AI}),)elapsedtime.time()-startprint(f并发耗时:{elapsed:.1f}s)# ≈ 1.0sforrinresults:print(r)asyncio.run(run_sequential())asyncio.run(run_concurrent())输出串行耗时: 3.0s 并发耗时: 1.0s 北京: 晴, 25°C AAPL: ¥150.00 (2.3%) 关于 AI 的最新新闻AI 行业持续升温...3 个工具各需 1 秒串行 3 秒并发只需1 秒。3. 生产级错误分类生产环境中异步工具需要精细的错误处理。不同类型的异常应该有不同的降级策略importasyncioimportaiohttpimportloggingfromlangchain_core.toolsimporttool loggerlogging.getLogger(__name__)toolasyncdeffetch_web_content(url:str)-dict:Fetch and extract main content from a web page. Args: url: Target web page URL Returns: Dict with content, title on success, or error on failure try:asyncwithaiohttp.ClientSession(timeoutaiohttp.ClientTimeout(total10))assession:asyncwithsession.get(url)asresponse:ifresponse.status200:htmlawaitresponse.text()return{content:html[:1000],status:success}else:return{error:fHTTP{response.status},status:failed}exceptasyncio.TimeoutError:# 超时建议重试return{error:请求超时10s建议稍后重试,status:timeout,retryable:True}exceptaiohttp.ClientErrorase:# 网络错误可能是 DNS、连接等问题return{error:f网络错误:{str(e)},status:network_error,retryable:True}exceptExceptionase:# 未知错误记录日志返回通用信息logger.error(fUnexpected error fetching{url}:{e})return{error:内容获取失败请稍后重试,status:unknown_error,retryable:False}错误分类的价值在于Agent 可以根据retryable字段决定是否重试根据status选择替代策略。4. 性能对比实测用一个完整的对比脚本直观感受同步与异步的差距importasyncioimporttimefromlangchain_core.toolsimporttool# 同步工具tooldefsync_api_call(name:str)-str:同步 API 调用time.sleep(1)# 模拟 I/O 阻塞returnfResult from{name}# 异步工具toolasyncdefasync_api_call(name:str)-str:异步 API 调用awaitasyncio.sleep(1)# 非阻塞等待returnfResult from{name}# 测试同步deftest_sync():starttime.time()fornamein[API-1,API-2,API-3]:sync_api_call.invoke({name:name})print(f同步 3 次调用:{time.time()-start:.1f}s)# ≈ 3.0s# 测试异步asyncdeftest_async():starttime.time()awaitasyncio.gather(async_api_call.ainvoke({name:API-1}),async_api_call.ainvoke({name:API-2}),async_api_call.ainvoke({name:API-3}),)print(f异步 3 次调用:{time.time()-start:.1f}s)# ≈ 1.0stest_sync()asyncio.run(test_async())调用次数同步耗时异步耗时性能提升1 次1.0s1.0s无差异3 次3.0s1.0s3x5 次5.0s1.0s5x10 次10.0s1.0s10x调用次数越多异步的优势越明显。最佳实践1. I/O 密集型工具一律用async def判断标准很简单工具内部是否有网络请求、数据库查询、文件读写如果有就用异步。# ✅ 异步调用外部 APItoolasyncdefsearch_web(query:str)-str:awaitaiohttp.get(fhttps://api.example.com/search?q{query})...# ❌ 同步CPU 密集型计算不需要异步tooldefcalculate_fibonacci(n:int)-str:# 纯计算没有 I/O...2. 捕获特定异常返回结构化错误不要只写一个笼统的except Exception。针对不同异常返回不同信息让 Agent 能做出更聪明的决策try:resultawaitapi_call()exceptasyncio.TimeoutError:return{error:超时,retryable:True}# Agent 知道可以重试exceptValueError:return{error:参数错误,retryable:False}# Agent 知道该换参数exceptExceptionase:return{error:str(e),retryable:False}# 兜底3. 添加日志方便调试异步工具的问题往往难以复现日志是排查问题的关键importlogging loggerlogging.getLogger(__name__)toolasyncdeffetch_data(url:str)-dict:logger.info(f开始请求:{url})try:resultawaitdo_fetch(url)logger.info(f请求成功返回{len(result)}条数据)returnresultexceptExceptionase:logger.error(f请求失败:{url}, 错误:{e})return{error:str(e)}4. 限制返回结果数量不限制结果数一次返回上千条数据不仅浪费 Token还会拖慢 Agent 推理。用min()设置上限max_resultsmin(max_results,20)# 硬上限 20 条5. 用ainvoke调用异步工具异步工具必须通过ainvoke调用不能用invoke# ✅ 正确resultawaitfetch_news_articles.ainvoke({topic:AI,max_results:5})# ❌ 错误invoke 无法正确执行异步工具resultfetch_news_articles.invoke({topic:AI})# 可能报错或行为异常总结异步工具是构建高性能 Agent 的基础设施。核心要点回顾要点说明何时用异步I/O 密集型操作API、DB、文件定义方式async deftool自动识别调用方式await tool.ainvoke({...})并发执行asyncio.gather同时调用多个工具错误处理捕获特定异常返回结构化错误信息日志记录logger.info/error便于调试结果限制min(max_results, N)防止数据过载记住一个原则能用async def就用async def。在 Agent 需要同时调用多个工具的场景下异步带来的性能提升是数量级的。下一篇我们将探讨 Agent 的工具编排策略——如何让多个工具协同工作完成复杂的端到端任务。
生产级异步工具：用 async def 打造高性能 Agent
摘要当 Agent 需要调用外部 API、查询数据库时同步工具会阻塞整个事件循环。本文教你用async def编写生产级异步工具实现并发执行、结构化错误处理和优雅降级让 Agent 性能提升数倍。引言你有没有遇到过这样的场景Agent 需要同时查询三个新闻源的信息结果工具一个接一个地串行执行每个 API 调用等待 2 秒三个就是 6 秒。用户盯着屏幕干等体验极差。问题的根源在于I/O 密集型操作API 调用、数据库查询、文件读写使用了同步方式。在同步模式下CPU 在等待网络响应时完全空闲事件循环被阻塞其他任务无法执行。解决方案就是异步工具——用async def定义工具让 Agent 在等待 I/O 时切换执行其他任务甚至并发调用多个工具。核心概念同步 vs 异步一张表说清维度同步工具 (def)异步工具 (async def)定义方式def tool_func(...)async def tool_func(...)I/O 等待时阻塞事件循环CPU 空闲释放控制权执行其他任务调用方式tool.invoke({...})await tool.ainvoke({...})并发能力串行执行asyncio.gather并发适用场景CPU 密集型计算API 调用、DB 查询、文件读写tool识别自动识别为同步工具自动识别为异步工具关键一点tool装饰器会自动检测函数是否为async def无需额外配置。为什么异步工具如此重要同步执行 3 个 API 调用每个 2 秒 ├── API 1: ████████ 2s ├── API 2: ████████ 2s ├── API 3: ████████ 2s 总耗时: 6s 异步并发执行 3 个 API 调用每个 2 秒 ├── API 1: ████████ 2s ├── API 2: ████████ 2s ├── API 3: ████████ 2s 总耗时: 2s性能提升3 倍而且 API 调用越多差距越大。代码示例1. 基础异步工具用async def定义异步工具tool自动识别importasyncioimportloggingfromlangchain_core.toolsimporttool loggerlogging.getLogger(__name__)toolasyncdeffetch_news_articles(topic:str,max_results:int5)-list:Fetch latest news articles about a topic. Use this to stay updated on: - Industry trends and news - Company announcements - Market developments Args: topic: News topic or keyword max_results: Maximum number of articles (default: 5, max: 20) Returns: List of article dictionaries with title, source, published_date try:logger.info(fFetching news for topic:{topic})awaitasyncio.sleep(0.5)# 模拟网络延迟# 限制最大结果数防止数据过载max_resultsmin(max_results,20)articles[{title:fArticle{i}about{topic},source:News Source,published_date:2026-06-22}foriinrange(max_results)]returnarticlesexceptasyncio.TimeoutError:logger.error(fTimeout fetching news for:{topic})return[{error:Request timeout, please try again}]exceptExceptionase:logger.error(fError fetching news:{str(e)})return[{error:fFailed to fetch news:{str(e)}}]注意三个关键点async def—tool自动识别为异步工具ainvoke调用—await fetch_news_articles.ainvoke({topic: AI})结构化错误返回— 返回[{error: ...}]而非抛出异常2. 并发执行多个工具用asyncio.gather同时调用多个工具性能翻倍importasyncioimporttimefromlangchain_core.toolsimporttooltoolasyncdefsearch_weather(city:str)-str:查询城市天气awaitasyncio.sleep(1)# 模拟 1 秒 API 延迟returnf{city}: 晴, 25°Ctoolasyncdefsearch_stock(symbol:str)-str:查询股票价格awaitasyncio.sleep(1)# 模拟 1 秒 API 延迟returnf{symbol}: ¥150.00 (2.3%)toolasyncdefsearch_news(keyword:str)-str:搜索新闻awaitasyncio.sleep(1)# 模拟 1 秒 API 延迟returnf关于{keyword}的最新新闻AI 行业持续升温...# 串行执行慢asyncdefrun_sequential():starttime.time()r1awaitsearch_weather.ainvoke({city:北京})r2awaitsearch_stock.ainvoke({symbol:AAPL})r3awaitsearch_news.ainvoke({keyword:AI})elapsedtime.time()-startprint(f串行耗时:{elapsed:.1f}s)# ≈ 3.0s# 并发执行快asyncdefrun_concurrent():starttime.time()resultsawaitasyncio.gather(search_weather.ainvoke({city:北京}),search_stock.ainvoke({symbol:AAPL}),search_news.ainvoke({keyword:AI}),)elapsedtime.time()-startprint(f并发耗时:{elapsed:.1f}s)# ≈ 1.0sforrinresults:print(r)asyncio.run(run_sequential())asyncio.run(run_concurrent())输出串行耗时: 3.0s 并发耗时: 1.0s 北京: 晴, 25°C AAPL: ¥150.00 (2.3%) 关于 AI 的最新新闻AI 行业持续升温...3 个工具各需 1 秒串行 3 秒并发只需1 秒。3. 生产级错误分类生产环境中异步工具需要精细的错误处理。不同类型的异常应该有不同的降级策略importasyncioimportaiohttpimportloggingfromlangchain_core.toolsimporttool loggerlogging.getLogger(__name__)toolasyncdeffetch_web_content(url:str)-dict:Fetch and extract main content from a web page. Args: url: Target web page URL Returns: Dict with content, title on success, or error on failure try:asyncwithaiohttp.ClientSession(timeoutaiohttp.ClientTimeout(total10))assession:asyncwithsession.get(url)asresponse:ifresponse.status200:htmlawaitresponse.text()return{content:html[:1000],status:success}else:return{error:fHTTP{response.status},status:failed}exceptasyncio.TimeoutError:# 超时建议重试return{error:请求超时10s建议稍后重试,status:timeout,retryable:True}exceptaiohttp.ClientErrorase:# 网络错误可能是 DNS、连接等问题return{error:f网络错误:{str(e)},status:network_error,retryable:True}exceptExceptionase:# 未知错误记录日志返回通用信息logger.error(fUnexpected error fetching{url}:{e})return{error:内容获取失败请稍后重试,status:unknown_error,retryable:False}错误分类的价值在于Agent 可以根据retryable字段决定是否重试根据status选择替代策略。4. 性能对比实测用一个完整的对比脚本直观感受同步与异步的差距importasyncioimporttimefromlangchain_core.toolsimporttool# 同步工具tooldefsync_api_call(name:str)-str:同步 API 调用time.sleep(1)# 模拟 I/O 阻塞returnfResult from{name}# 异步工具toolasyncdefasync_api_call(name:str)-str:异步 API 调用awaitasyncio.sleep(1)# 非阻塞等待returnfResult from{name}# 测试同步deftest_sync():starttime.time()fornamein[API-1,API-2,API-3]:sync_api_call.invoke({name:name})print(f同步 3 次调用:{time.time()-start:.1f}s)# ≈ 3.0s# 测试异步asyncdeftest_async():starttime.time()awaitasyncio.gather(async_api_call.ainvoke({name:API-1}),async_api_call.ainvoke({name:API-2}),async_api_call.ainvoke({name:API-3}),)print(f异步 3 次调用:{time.time()-start:.1f}s)# ≈ 1.0stest_sync()asyncio.run(test_async())调用次数同步耗时异步耗时性能提升1 次1.0s1.0s无差异3 次3.0s1.0s3x5 次5.0s1.0s5x10 次10.0s1.0s10x调用次数越多异步的优势越明显。最佳实践1. I/O 密集型工具一律用async def判断标准很简单工具内部是否有网络请求、数据库查询、文件读写如果有就用异步。# ✅ 异步调用外部 APItoolasyncdefsearch_web(query:str)-str:awaitaiohttp.get(fhttps://api.example.com/search?q{query})...# ❌ 同步CPU 密集型计算不需要异步tooldefcalculate_fibonacci(n:int)-str:# 纯计算没有 I/O...2. 捕获特定异常返回结构化错误不要只写一个笼统的except Exception。针对不同异常返回不同信息让 Agent 能做出更聪明的决策try:resultawaitapi_call()exceptasyncio.TimeoutError:return{error:超时,retryable:True}# Agent 知道可以重试exceptValueError:return{error:参数错误,retryable:False}# Agent 知道该换参数exceptExceptionase:return{error:str(e),retryable:False}# 兜底3. 添加日志方便调试异步工具的问题往往难以复现日志是排查问题的关键importlogging loggerlogging.getLogger(__name__)toolasyncdeffetch_data(url:str)-dict:logger.info(f开始请求:{url})try:resultawaitdo_fetch(url)logger.info(f请求成功返回{len(result)}条数据)returnresultexceptExceptionase:logger.error(f请求失败:{url}, 错误:{e})return{error:str(e)}4. 限制返回结果数量不限制结果数一次返回上千条数据不仅浪费 Token还会拖慢 Agent 推理。用min()设置上限max_resultsmin(max_results,20)# 硬上限 20 条5. 用ainvoke调用异步工具异步工具必须通过ainvoke调用不能用invoke# ✅ 正确resultawaitfetch_news_articles.ainvoke({topic:AI,max_results:5})# ❌ 错误invoke 无法正确执行异步工具resultfetch_news_articles.invoke({topic:AI})# 可能报错或行为异常总结异步工具是构建高性能 Agent 的基础设施。核心要点回顾要点说明何时用异步I/O 密集型操作API、DB、文件定义方式async deftool自动识别调用方式await tool.ainvoke({...})并发执行asyncio.gather同时调用多个工具错误处理捕获特定异常返回结构化错误信息日志记录logger.info/error便于调试结果限制min(max_results, N)防止数据过载记住一个原则能用async def就用async def。在 Agent 需要同时调用多个工具的场景下异步带来的性能提升是数量级的。下一篇我们将探讨 Agent 的工具编排策略——如何让多个工具协同工作完成复杂的端到端任务。
