第2篇异步编程精要 —— Agent开发的并发基石在Agent系统中LLM API调用、工具搜索、数据库读写几乎全是I/O密集型操作。如果你的代码还在串行等待每个HTTP响应那大部分时间都浪费在了“空转”上。本文将从事件循环原理讲起深入剖析async/await的运行时机制并通过真实的压测数据揭示异步并发如何将Agent系统的吞吐量提升数倍。一、Agent开发为什么要“异步优先”在同步编程模型中代码的执行顺序就是代码的书写顺序。当程序执行到response requests.get(api_url)这一行时整个进程会被“挂起”——它只能傻等网络数据包从远端服务器传回来期间CPU处于空闲状态。对于Agent而言一次典型的请求链路涉及调用LLM生成计划约800ms2s调用搜索工具获取上下文约300ms1s再次调用LLM进行推理约500ms1.5s如果这些调用串行执行一个完整Agent任务可能需要35秒。试想如果将单用户的三个请求拆分为三个独立的并发调用总耗时将从“三者之和”变成“三者之中最慢的那个”——这就是异步编程最直接的价值。异步的本质在等待I/O结果时主动让出CPU控制权让事件循环去执行其他任务。这不是多线程没有线程切换开销而是单线程内的任务调度。二、核心机制从async def到事件循环2.1 协程对象 vs 普通函数当你在函数前加上async def它就变成了一个协程函数。调用它并不会立即执行函数体而是返回一个协程对象coroutine object。async def call_llm_async(question: str): # 这里的代码不会立即执行 return await client.chat.completions.create(...) # 这一行运行极快没有网络请求发生 coro call_llm_async(什么是递归) print(type(coro)) # class coroutine这个协程对象就像一个“冷冻状态的执行蓝图”——它知道要做什么但还没开始做。只有当它被提交给事件循环Event Loop并得到调度时内部的代码才会真正运行。2.2await交出控制权的关键指令await是异步编程中的“让位”操作符。当执行到await client.chat.completions.create(...)时协程向事件循环注册一个I/O等待任务。协程主动挂起suspension把CPU控制权交还给事件循环。事件循环可以去执行其他就绪的协程。当网络响应返回时事件循环恢复该协程的执行。这个机制使得单线程内可以并发处理成千上万个I/O任务且没有线程上下文切换的开销。三、运行时剖析tasks与*tasks的真相理解了基础概念后我们来看这两行经典的并发代码。它们常常让初学者感到困惑tasks [call_llm_async(q) for q in QUESTIONS] results await asyncio.gather(*tasks)3.1 第一行构建协程列表无事发生tasks [call_llm_async(q) for q in QUESTIONS]运行时实际发生的事情遍历QUESTIONS列表假设有三个问题依次调用call_llm_async(q)生成三个协程对象将这三个协程对象存入tasks列表此时没有任何网络请求发出也没有任何函数体被执行。tasks列表里只是存储了三个“待执行的任务蓝图”。3.2 解包操作符*拆开列表asyncio.gather的定义是def gather(*aws, return_exceptionsFalse): # 接受零个或多个可等待对象awaitable objects它不接受一个列表只接受多个独立参数。*tasks的作用是把[coro1, coro2, coro3]展开为coro1, coro2, coro3满足函数签名要求。如果不加*错误写法是results await asyncio.gather(tasks) # ❌ 传入了一个列表而不是三个协程3.3await asyncio.gather(...)并发执行这一行是真正的执行入口。其运行时行为包装成Taskgather会将每个协程对象封装成Task对象并立即提交到当前事件循环中。并发调度main协程被挂起等待gather完成。事件循环开始轮流驱动三个Task。每个任务执行到await client.chat.completions.create(...)时由于网络IO尚未就绪任务主动挂起事件循环立即切换到下一个任务。三个任务“交替前进”网络请求在底层实际上是并行发出的由操作系统和网络栈处理。结果汇聚每个任务完成后结果被gather收集。gather等待所有任务完成后返回一个有序列表——结果顺序与传入参数的顺序严格一致与完成顺序无关。3.4 运行时可视化假设每个LLM请求需要2秒网络推理同步串行与异步并发的时序差异如下[同步串行]总耗时约6秒 main : ---[req1]---[req2]---[req3]--- 时间 2s 2s 2s [异步并发]总耗时约2秒 main : ---[gather]--------------------- task1: ---[等待IO]--------------------- task2: ---[等待IO]--------------------- task3: ---[等待IO]--------------------- 网络IO重叠并发总耗时≈最慢任务四、实战对比三请求压测数据我们使用真实的DeepSeek API进行同步与异步的性能对比测试。4.1 测试代码import asyncio, time, os from openai import OpenAI, AsyncOpenAI from dotenv import load_dotenv load_dotenv() api_key os.getenv(DEEPSEEK_API_KEY) base_url https://api.deepseek.com/v1 model deepseek-chat QUESTIONS [ 用一句话解释什么是递归。, 用一句话解释什么是闭包。, 用一句话解释什么是异步编程。, ] # ------ 同步串行 ------ sync_client OpenAI(api_keyapi_key, base_urlbase_url) def sync_serial(): start time.time() for q in QUESTIONS: resp sync_client.chat.completions.create(modelmodel, messages[{role: user, content: q}]) print(f同步回复: {resp.choices[0].message.content[:30]}...) return time.time() - start # ------ 异步并发 ------ async_client AsyncOpenAI(api_keyapi_key, base_urlbase_url) async def call_async(q): resp await async_client.chat.completions.create(modelmodel, messages[{role: user, content: q}]) return resp.choices[0].message.content async def async_concurrent(): tasks [call_async(q) for q in QUESTIONS] results await asyncio.gather(*tasks) for r in results: print(f异步回复: {r[:30]}...) return time.time() - start # 实际需在函数内记录时间 async def main(): sync_time sync_serial() async_time await async_concurrent() print(f同步耗时: {sync_time:.2f}s, 异步耗时: {async_time:.2f}s, 加速比: {sync_time/async_time:.2f}x)4.2 预期结果模式耗时说明同步串行~5.4s三个请求依次等待总耗时约为三者之和异步并发~1.9s三个请求几乎同时发出总耗时≈最慢的一个请求加速比~2.8x网络延迟越大并发优势越明显实际运行时加速比通常在2.5x4x之间具体取决于网络延迟和模型服务的并发处理能力。若单个请求耗时1.5s三请求并发的异步方案理论上可达到接近3x的加速。五、Agent场景下的异步应用模式5.1 批量用户请求当多个用户同时向你的Agent服务发送请求时如果使用同步模型每个请求会独占一个线程线程数随并发用户量线性增长最终导致系统崩溃。使用异步模型单线程即可处理数千个并发请求。5.2 一个Agent内部的工具并行调用当一个Agent决策需要同时“搜索”和“查数据库”时可以用asyncio.gather让两个工具调用并发执行search_task search_tool.ainvoke(query) db_task db_tool.ainvoke(query) search_result, db_result await asyncio.gather(search_task, db_task)5.3 流式输出Streaming使用异步流式接口可以在生成完整响应之前就开始向客户端输出Token大幅改善用户体验。配合SSEServer-Sent Events可以做到“一边生成、一边传输、一边显示”。六、陷阱与避坑指南常见误用原因正确做法在async函数中混用requests库requests是同步阻塞的会卡住事件循环统一使用httpx.AsyncClient或aiohttp忘记await直接调用协程函数只返回协程对象不执行必须加上await或提交给gather在循环中串行awaitfor q in questions: await call(q)仍是串行用列表推导式先生成协程再统一gather无限创建Task没有控制并发数可能导致API限流使用asyncio.Semaphore控制并发度七、思考与动手建议在实际项目中将同步的LLM调用改造为异步调用并用time.perf_counter()测量前后的耗时差异。尝试用asyncio.Semaphore(2)限制并发数为2观察总耗时的变化理解限流的意义。
第2篇异步编程精要 —— Agent开发的并发基石在Agent系统中LLM API调用、工具搜索、数据库读写几乎全是I/O密集型操作。如果你的代码还在串行等待每个HTTP响应那大部分时间都浪费在了“空转”上。本文将从事件循环原理讲起深入剖析async/await的运行时机制并通过真实的压测数据揭示异步并发如何将Agent系统的吞吐量提升数倍。一、Agent开发为什么要“异步优先”在同步编程模型中代码的执行顺序就是代码的书写顺序。当程序执行到response requests.get(api_url)这一行时整个进程会被“挂起”——它只能傻等网络数据包从远端服务器传回来期间CPU处于空闲状态。对于Agent而言一次典型的请求链路涉及调用LLM生成计划约800ms2s调用搜索工具获取上下文约300ms1s再次调用LLM进行推理约500ms1.5s如果这些调用串行执行一个完整Agent任务可能需要35秒。试想如果将单用户的三个请求拆分为三个独立的并发调用总耗时将从“三者之和”变成“三者之中最慢的那个”——这就是异步编程最直接的价值。异步的本质在等待I/O结果时主动让出CPU控制权让事件循环去执行其他任务。这不是多线程没有线程切换开销而是单线程内的任务调度。二、核心机制从async def到事件循环2.1 协程对象 vs 普通函数当你在函数前加上async def它就变成了一个协程函数。调用它并不会立即执行函数体而是返回一个协程对象coroutine object。async def call_llm_async(question: str): # 这里的代码不会立即执行 return await client.chat.completions.create(...) # 这一行运行极快没有网络请求发生 coro call_llm_async(什么是递归) print(type(coro)) # class coroutine这个协程对象就像一个“冷冻状态的执行蓝图”——它知道要做什么但还没开始做。只有当它被提交给事件循环Event Loop并得到调度时内部的代码才会真正运行。2.2await交出控制权的关键指令await是异步编程中的“让位”操作符。当执行到await client.chat.completions.create(...)时协程向事件循环注册一个I/O等待任务。协程主动挂起suspension把CPU控制权交还给事件循环。事件循环可以去执行其他就绪的协程。当网络响应返回时事件循环恢复该协程的执行。这个机制使得单线程内可以并发处理成千上万个I/O任务且没有线程上下文切换的开销。三、运行时剖析tasks与*tasks的真相理解了基础概念后我们来看这两行经典的并发代码。它们常常让初学者感到困惑tasks [call_llm_async(q) for q in QUESTIONS] results await asyncio.gather(*tasks)3.1 第一行构建协程列表无事发生tasks [call_llm_async(q) for q in QUESTIONS]运行时实际发生的事情遍历QUESTIONS列表假设有三个问题依次调用call_llm_async(q)生成三个协程对象将这三个协程对象存入tasks列表此时没有任何网络请求发出也没有任何函数体被执行。tasks列表里只是存储了三个“待执行的任务蓝图”。3.2 解包操作符*拆开列表asyncio.gather的定义是def gather(*aws, return_exceptionsFalse): # 接受零个或多个可等待对象awaitable objects它不接受一个列表只接受多个独立参数。*tasks的作用是把[coro1, coro2, coro3]展开为coro1, coro2, coro3满足函数签名要求。如果不加*错误写法是results await asyncio.gather(tasks) # ❌ 传入了一个列表而不是三个协程3.3await asyncio.gather(...)并发执行这一行是真正的执行入口。其运行时行为包装成Taskgather会将每个协程对象封装成Task对象并立即提交到当前事件循环中。并发调度main协程被挂起等待gather完成。事件循环开始轮流驱动三个Task。每个任务执行到await client.chat.completions.create(...)时由于网络IO尚未就绪任务主动挂起事件循环立即切换到下一个任务。三个任务“交替前进”网络请求在底层实际上是并行发出的由操作系统和网络栈处理。结果汇聚每个任务完成后结果被gather收集。gather等待所有任务完成后返回一个有序列表——结果顺序与传入参数的顺序严格一致与完成顺序无关。3.4 运行时可视化假设每个LLM请求需要2秒网络推理同步串行与异步并发的时序差异如下[同步串行]总耗时约6秒 main : ---[req1]---[req2]---[req3]--- 时间 2s 2s 2s [异步并发]总耗时约2秒 main : ---[gather]--------------------- task1: ---[等待IO]--------------------- task2: ---[等待IO]--------------------- task3: ---[等待IO]--------------------- 网络IO重叠并发总耗时≈最慢任务四、实战对比三请求压测数据我们使用真实的DeepSeek API进行同步与异步的性能对比测试。4.1 测试代码import asyncio, time, os from openai import OpenAI, AsyncOpenAI from dotenv import load_dotenv load_dotenv() api_key os.getenv(DEEPSEEK_API_KEY) base_url https://api.deepseek.com/v1 model deepseek-chat QUESTIONS [ 用一句话解释什么是递归。, 用一句话解释什么是闭包。, 用一句话解释什么是异步编程。, ] # ------ 同步串行 ------ sync_client OpenAI(api_keyapi_key, base_urlbase_url) def sync_serial(): start time.time() for q in QUESTIONS: resp sync_client.chat.completions.create(modelmodel, messages[{role: user, content: q}]) print(f同步回复: {resp.choices[0].message.content[:30]}...) return time.time() - start # ------ 异步并发 ------ async_client AsyncOpenAI(api_keyapi_key, base_urlbase_url) async def call_async(q): resp await async_client.chat.completions.create(modelmodel, messages[{role: user, content: q}]) return resp.choices[0].message.content async def async_concurrent(): tasks [call_async(q) for q in QUESTIONS] results await asyncio.gather(*tasks) for r in results: print(f异步回复: {r[:30]}...) return time.time() - start # 实际需在函数内记录时间 async def main(): sync_time sync_serial() async_time await async_concurrent() print(f同步耗时: {sync_time:.2f}s, 异步耗时: {async_time:.2f}s, 加速比: {sync_time/async_time:.2f}x)4.2 预期结果模式耗时说明同步串行~5.4s三个请求依次等待总耗时约为三者之和异步并发~1.9s三个请求几乎同时发出总耗时≈最慢的一个请求加速比~2.8x网络延迟越大并发优势越明显实际运行时加速比通常在2.5x4x之间具体取决于网络延迟和模型服务的并发处理能力。若单个请求耗时1.5s三请求并发的异步方案理论上可达到接近3x的加速。五、Agent场景下的异步应用模式5.1 批量用户请求当多个用户同时向你的Agent服务发送请求时如果使用同步模型每个请求会独占一个线程线程数随并发用户量线性增长最终导致系统崩溃。使用异步模型单线程即可处理数千个并发请求。5.2 一个Agent内部的工具并行调用当一个Agent决策需要同时“搜索”和“查数据库”时可以用asyncio.gather让两个工具调用并发执行search_task search_tool.ainvoke(query) db_task db_tool.ainvoke(query) search_result, db_result await asyncio.gather(search_task, db_task)5.3 流式输出Streaming使用异步流式接口可以在生成完整响应之前就开始向客户端输出Token大幅改善用户体验。配合SSEServer-Sent Events可以做到“一边生成、一边传输、一边显示”。六、陷阱与避坑指南常见误用原因正确做法在async函数中混用requests库requests是同步阻塞的会卡住事件循环统一使用httpx.AsyncClient或aiohttp忘记await直接调用协程函数只返回协程对象不执行必须加上await或提交给gather在循环中串行awaitfor q in questions: await call(q)仍是串行用列表推导式先生成协程再统一gather无限创建Task没有控制并发数可能导致API限流使用asyncio.Semaphore控制并发度七、思考与动手建议在实际项目中将同步的LLM调用改造为异步调用并用time.perf_counter()测量前后的耗时差异。尝试用asyncio.Semaphore(2)限制并发数为2观察总耗时的变化理解限流的意义。
)
)
