写 Agent 写到第三版我才真正想明白一件事大部分 Agent 跑得慢背锅的常常不是模型本身而是调度逻辑里那些串行等待。一个 LLM 调用要两三秒一次工具请求又要两三秒一轮子 Agent 协作又是好几个来回。如果每个环节都老老实实排着队跑任务一多时间全花在等上。异步解决的不是能不能跑的问题而是能不能别傻等的问题。这篇文章不打算写一套完整的异步教程只分享我在写 Agent 过程中对异步的几点体会为什么 Agent 天然离不开异步、两条路线的选型取舍、最容易翻车的坑以及几个在真实流程里用得上的编排模式和调试手段。适合已经写过一点 Agent、开始觉得这玩意儿咋这么慢的读者也适合准备把单个 Agent 扩展成多任务协作场景的朋友。1. Agent 跑得慢不是模型慢是你等得太多了1.1 Agent 的一天清一色的等待操作我最早写 Agent 的时候脑子里是把 Agent 当成一个函数来想的输入一个任务输出一个结果。但写多了才发现Agent 运行的绝大部分时间根本不在计算而在等待 IO。具体来说一个典型 Agent 的任务周期大概是这样的接收用户输入把目标拆解成几个子任务。对第一个子任务调用 LLM 生成下一步动作。根据动作去调一个外部工具搜索、查数据库、调内部服务。拿到工具结果后再次调用 LLM 做推理。有时候还要把一部分子任务分给下游 Agent等它们把结果回传。最后汇总所有中间结果再生成最终答案。你留意一下这里面的每一步核心动作都是发一个请求等一个响应。LLM 推理是远程调用工具是远程调用子 Agent 协作也是远程调用。它跟写一个传统 Web 接口不一样传统接口虽然也在等数据库但等的时间通常在几十毫秒级别而 Agent 里的每次等待动辄是几秒甚至几十秒。所以 Agent 本质上是一个 IO 密集型的编排器它的大半辈子都花在等外部依赖返回。这不丢人这是它的工作方式决定的。问题在于你的调度代码是用什么方式去等。1.2 串行等待的浪费一算账就心疼我举个自己的例子。早期我写过一个调研类 Agent流程上要让 Agent 同时查三个外部数据源然后再让模型去汇总这些材料。那时候我的代码长这样def run_research(query: str): results [] for source in [source_a, source_b, source_c]: data call_external_source(source, query) results.append(data) summary call_llm(summarize(results)) return summary看起来逻辑很清楚就是一个循环调用三个数据源最后调一次 LLM。但问题就在这里三个数据源是串行等待的。假设每个外部数据源平均响应时间 3 秒LLM 汇总耗时 4 秒那总耗时是3 3 3 4 13秒。但如果你把三个数据源请求同时丢出去等它们都回来了再汇总那总耗时是max(3, 3, 3) 4 7秒。省出来的 6 秒几乎是白捡的。任务越复杂、调用链越长串行和并行的差距就越夸张。如果有五个子 Agent 各跑 10 秒串行就是 50 秒起步并行最多十几秒。不要觉得反正模型慢差这几秒无所谓。当你的 Agent 从单任务变成多用户、多任务并发的时候这个差距会被放大到让用户明显感觉到这服务卡不卡的程度。2. 异步化的两条路线全局协程 与 线程池包装2.1 路线一全异步链路清爽但要求高第一种做法是整个 Agent 的核心链路都用原生异步来实现。在 Python 里就是async defawait这一套。一个异步的工具调用函数大概长这样import asyncio async def call_external_source(source: str, query: str): # 这里的 HTTP 请求必须是异步客户端 async with httpx.AsyncClient() as client: resp await client.get(fhttps://api.example.com/{source}, params{q: query}) return resp.json() async def run_research(query: str): tasks [ call_external_source(a, query), call_external_source(b, query), call_external_source(c, query), ] results await asyncio.gather(*tasks) summary await call_llm(results) return summary全异步的好处是调度开销极小。事件循环切换任务的成本是微秒级的比线程切换要便宜得多。而且不会有锁竞争、线程安全这些问题代码读起来是一条线性流程只是中间有await跳转。但这条路对依赖库的要求很苛刻。调用链上所有网络请求都必须是异步客户端比如 HTTP 要用httpx或aiohttp数据库驱动要是asyncpg、aiomysql。如果你依赖某个内部库它底层只提供同步接口那你在这个环节就会非常难受。我体会最深的是异步会传染。一旦一个函数是async所有调用它的地方都得变async。如果项目里有一个老客户端库不支援异步要么你把它包在线程池里这就是路线二要么你用asyncio.to_thread把它丢给线程不然整个事件循环都会被它堵住。2.2 路线二同步核心加并发壳接入成本低第二种做法是同步核心不动只在调用外部依赖的时候把它丢进线程池里做并发。比如import asyncio import time from concurrent.futures import ThreadPoolExecutor def call_external_source_sync(source: str, query: str): # 这里用同步的 requests 也行 resp requests.get(fhttps://api.example.com/{source}, params{q: query}) return resp.json() async def run_research_with_pool(query: str, pool: ThreadPoolExecutor): loop asyncio.get_running_loop() tasks [ loop.run_in_executor(pool, call_external_source_sync, s, query) for s in [a, b, c] ] results await asyncio.gather(*tasks) return await call_llm(results)这样做的优点非常现实基本不用改原有同步代码外部依赖的生态没有任何限制。你的 Agent 逻辑可以继续用普通的同步函数写只在最外层把这些一次性丢进线程池做并行。缺点是线程本身是重量级资源。默认线程池一般也就十几个线程如果你同时并发几百个外部调用线程会排队反而把延迟拉高。而且线程池里的共享状态需要自己加锁用不好会遇到各种难以复现的并发 bug。2.3 怎么选别跟风看你的依赖生态我在两个方案之间反复横跳过最后得出的判断标准挺简单如果整个 Agent 的依赖链已经全部是异步的那我优先选全异步。代码干净、可控、调度效率高长期维护省心。如果 Agent 里混着大量老同步库我不会强上全异步。我会把核心编排逻辑做成异步然后用asyncio.to_thread或线程池把这些同步调用包一层让它们在分支任务里并行跑核心流程不受阻塞。如果项目刚起步团队又普遍不熟悉协程我建议先用同步代码把 Agent 的行为逻辑跑通再在外围做一层异步并发优化而不是一上来就写一整套async/await矩阵。有一件事需要提醒异步不是银弹。它解决的是并行等待不解决单个请求本身慢的问题。如果一个外部接口就要 30 秒不管你异步还是同步那一跳都是 30 秒。异步真正的价值是让你在等待 30 秒的同时其他任务还能继续推进。3. 写 Agent 异步代码时最容易翻车的四个地方3.1 事件循环被堵死整个 Agent 假死全异步链路里最经典的翻车现场就是在async函数里混入了同步阻塞操作。有一次我写一个 Agent 的定时巡检逻辑中间想让它休息五秒再继续下一轮我顺手写了time.sleep(5)。结果就是这 5 秒里整个事件循环卡住不动所有正在跑的任务全部暂停。我本地测试时只有一个 Agent 感觉不明显一上测试环境多个 Agent 并发立刻出现了一批请求长时间无响应的现象。原因其实很直白事件循环只有一个线程它轮询所有已注册的协程。如果这个线程自己睡死过去了那谁也别想跑。正确做法是在异步代码里需要等待一律用await asyncio.sleep()。如果你必须调用一个同步阻塞的库至少用asyncio.to_thread把它丢出去别让它侵占主循环。这是一个非常低级但极其常见的错误值得每个写 Agent 的人自查一遍。3.2 超时设置没生效进程被远程调用拖垮第二个坑比第一个隐蔽但也更致命。有段时间我写的 Agent 在线上偶尔会整体卡住几分钟查半天才发现某个外部工具的 SDK 默认超时时间非常长甚至某些情况下等于没有超时。上游服务一挂我的每个请求都要等满 120 秒才报错而并发任务一多全部堵在等待上整个 Agent 就像被冻住了一样。从那以后我给自己定了一条规矩所有外部调用默认都必须设置超时而且超时时间不能比业务容忍度更长。在asyncio里我会习惯性地用asyncio.wait_for包一层import asyncio async def call_with_timeout(coro, timeout: float 10.0): try: return await asyncio.wait_for(coro, timeouttimeout) except asyncio.TimeoutError: # 记录日志并走降级逻辑 return fallback_result()用的时候就是result await call_with_timeout(call_external_source(a, query), timeout5.0)还有一个细节超时之后被 wait_for 包裹的协程虽然会收到取消信号但如果那个协程正在执行某些不可中断的同步操作取消也不会立刻生效。所以底层网络调用尽量选支持超时的 SDK双保险。3.3 任务取消不干净协程在后台偷偷跑第三个坑是关于取消的。asyncio里Task被取消时会往目标协程抛一个CancelledError。如果你在协程里没处理这个异常任务看起来是结束了但协程内部的资源可能没有释放干净。我遇到过一个典型问题一个 Agent 任务因为超时被取消但它在执行过程中已经打开了一个数据库连接这个连接的释放代码写在正常的逻辑流里没放在finally里。结果每次取消都会泄漏一条连接跑一段时间后连接池被打满后续任务全部失败。处理办法很简单所有涉及资源获取的协程都用try/finally来保证清理async def call_and_record(): conn await acquire_conn() try: return await do_remote_work(conn) finally: await conn.close()另外一个让我比较头疼的问题是一个父任务被取消时它通过asyncio.gather拉起来的子任务不一定都会跟着取消。gather的行为是如果某个子任务抛异常其他子任务不会被自动取消但整个 gather 会立刻抛出异常。如果你不显式处理这些子任务会变成孤儿协程继续在后台跑白白占用事件循环。所以在我自己的代码里凡是涉及多任务聚合的地方都会在except分支里显式取消还没完成的任务。这个细节在调试的时候特别难查因为 Agent 的行为看起来也是正常结束的只是偶尔状态更新有点怪。3.4 共享上下文并发修改一句话引发整轮出错第四个坑来自并发共享状态。Agent 通常有一个记忆或者上下文对象承载目标、历史对话、中间结果。在多任务并行的情况下如果多个协程同时往这个上下文里写内容就会出现覆盖和顺序混乱。我犯过一次很隐蔽的错误一个 Agent 同时发了三个工具调用每个工具返回后都会往context[refine_result]里写入自己的加工结果。最后一个返回的协程覆盖掉了前面两个的成果模型最终拿到的是一个残缺的结果整个推理链路都被带偏了。后来我在这种场景下采取了一个很简单的策略协程内部不直接修改共享上下文而是各自返回自己的局部结果由主流程统一收口再写入共享状态。这就是一种单写者模式虽然牺牲了一点点灵活性但换来了非常宝贵的确定性。配合contextvars.ContextVar做链路级上下文传递也可以避免很多手抖造成的全局污染。4. 异步编排 Agent 流程的四种实用模式4.1 扇出汇聚并行调用多个工具再合并结果最常用也最好用的就是扇出汇聚模式。Agent 在推进一个步骤时发现有多条独立信息可以同时获取就把它们全部并行发出等所有结果回来后再汇聚给模型做下一步推理。用代码来写就是asyncio.gatherasync def step_fetch_all(query: str): search_task search_web(query) db_task query_internal_db(query) doc_task load_related_docs(query) search_result, db_result, doc_result await asyncio.gather( search_task, db_task, doc_task ) return merge_as_context(search_result, db_result, doc_result)这个模式之所以可靠是因为它有一个明确的汇聚点gather 返回之后所有子任务都已经完成。这让后续的模型调用拿到的是完整信息不会出现模板变量还没填好就渲染的情况。我通常会在 gather 里配合return_exceptionsTrue避免某个工具挂了导致整个批次全废。然后拿到返回结果后先判断哪些是异常哪些是有效数据再构造提示词。这样单个工具抖动不会把整轮对话搞崩。4.2 竞争回退多个候选谁先成功用谁第二个我经常用的是竞争回退模式。场景是这样的有些外部工具的可用性不稳定Agent 需要调用某一个服务但这个服务偶尔超时。为了降低失败率我同时发起主服务调用和备用服务调用哪个先成功就采用哪个。代码上可以用asyncio.wait来实现import asyncio async def call_with_fallback(primary_coro, fallback_coro, timeout5.0): task_primary asyncio.create_task(asyncio.wait_for(primary_coro, timeout)) task_fallback asyncio.create_task(asyncio.wait_for(fallback_coro, timeout)) done, pending await asyncio.wait( {task_primary, task_fallback}, return_whenasyncio.FIRST_COMPLETED, ) for task in pending: task.cancel() for task in done: if task.exception() is None: return task.result() raise RuntimeError(both primary and fallback failed)这个模式在决策链路里的价值在于它把等待一个不确定的服务变成了让多个候选互相兜底。我实际跑下来成功率能从 80% 左右提升到 95% 以上代价只是偶尔多花一小份资源。不过要注意控制这种冗余调用的数量别搞成十个服务一起跑否则就是拿成本换确定性反而不划算。两路到三路竞争是我用下来比较舒服的平衡点。4.3 依赖链用事件和回调代替嵌套地狱有些 Agent 流程是强依赖的B 步骤依赖 A 步骤的输出C 步骤又依赖 A 和 B 的部分结果。这种链式关系如果单纯用嵌套await写逻辑是清晰的但一旦分支变多代码会变成一团乱麻。我自己的做法是把依赖关系抽象成两个层面。第一层是阶段一个阶段内的任务可以并行阶段之间保持顺序第二层是任务级回调每个任务完成后把自己的结果存入一个结果池后续依赖方从结果池里取数。举个例子Agent 要先收集基础资料→再针对每份资料做分析→最后汇总。基础资料收集是第一阶段三份资料并行取针对每份资料的分析是第二阶段三份分析可以并行但要等第一阶段全部完成汇总是第三阶段。这种情况下我会用两轮 gather 而不是把所有任务揉在一起base_docs await asyncio.gather(load_doc(1), load_doc(2), load_doc(3)) analysis_tasks [analyze(doc) for doc in base_docs] analyses await asyncio.gather(*analysis_tasks) final_report await summarize(analyses)这个写法看起来就是普通的顺序代码但由于每个阶段内部是并行的整体的耗时接近最长的那个阶段而不是所有阶段的总和。而且它还保留了可读性任何一个接手的人都看得出流程长什么样。比事件回调矩阵或者任务队列状态机要直观得多。我在实践中发现绝大多数业务 Agent 的依赖结构都只是层状 DAG用这种分层 gather 已经完全够用不需要引入复杂的编排引擎。4.4 上下文传播与隔离别让本轮状态污染下一轮异步并发环境下上下文传播不只是一个传参问题更是一个隔离问题。我见过不少 Agent 框架里上下文是用一个全局字典存的在并发跑多个任务时A 任务写的中间状态被 B 任务读到两边一起错乱。我在自己的项目里用的是两条腿走路一是直接把上下文对象作为参数显式传入每个子任务这样数据流清晰二是每个 Agent 实例拥有独立的上下文副本任务结束后只把需要的部分合并回主流程。如果你用 Pythoncontextvars.ContextVar是一项好用的工具它可以让同一个任务链里的异步代码共享同一个上下文对象但又不会跨任务泄漏。但需要注意ContextVar的赋值只在当前任务里生效如果你显式创建了新任务而不手动复制上下文新任务读到的可能还是默认值。我自己常用的做法是在创建 Task 时手动传上下文引用不依赖隐式传播避免踩这个隐坑。5. 异步 Agent 的监控、限流与调试实践5.1 可视化等待耗时先找瓶颈再优化异步改造做完之后最重要的事情是看清楚等待时间到底花在了哪里。我在项目里会做一个非常轻量的计时装饰器把所有外部调用的耗时、状态和超时信息记录到结构化日志里import time import asyncio from functools import wraps def async_observe(tag: str): def decorator(func): wraps(func) async def wrapper(*args, **kwargs): start time.perf_counter() try: result await func(*args, **kwargs) elapsed time.perf_counter() - start log_event(async_call_ok, tagtag, cost_ms(elapsed * 1000)) return result except Exception as e: elapsed time.perf_counter() - start log_event(async_call_error, tagtag, cost_ms(elapsed * 1000), errorstr(e)) raise return wrapper return decorator只要给关键调用都加上这么一层把日志拿到类似表格工具或日志平台里一汇总哪个工具耗时最高、哪个工具超时率最大一目了然。我做过一次排查发现某个 Agent 流程整体耗时 20 秒其中 18 秒都耗在数据库查询上模型调用反而只占了 2 秒。但那一次我原以为瓶颈出在模型端差点给模型加价换高档版本。查完日志才发现换模型毫无意义应该去优化的是数据库索引和查询逻辑。这一步省下的钱比省下的时间更值。5.2 用信号量给并发刹刹车别把上游打爆异步释放了并发能力之后很容易手一滑就让并发量冲到几十上百。我在早期就是这么干的以为并发越高越快结果把外部模型供应商的接口限流给触发了反而导致大量请求排队重试整体吞吐急剧下降。后来我养成了一个习惯所有访问外部 LLM 或共享资源的并发调用都用一个Semaphore限一下。import asyncio class AgentRateLimiter: def __init__(self, limit: int 3): self._sem asyncio.Semaphore(limit) async def call_llm_limited(self, prompt: str): async with self._sem: return await call_llm(prompt)为什么限制并发反而能提升整体体验因为对于大多数外部服务并发一旦超过某个阈值服务端的排队和限流会吃掉你多出来的并发收益。我自己的经验值访问中大型模型接口时并发控制在 3 到 5 个比较舒服访问轻度查询类接口可以放宽到两位数但也不是无上限。这里有一个希望对你有用的判断方法先以一个较小并发跑一组压测请求记录平均延迟和成功率然后逐步加大并发。如果请求的延迟不再明显下降甚至成功率开始出现抖动那说明你已经摸到阈值了把它作为你的信号量上限。5.3 让异步代码可以测试的几条实战经验最后说说异步代码的可测试性。Agent 的异步逻辑最大的痛点是外部依赖太多测试起来不稳定。我踩过不少次这里分享三条心得。第一所有 IO 调用最好都做成可注入的。不要让 Agent 内部的函数直接硬编码调用某个外部 HTTP 服务而是通过依赖参数传入调用器。测试的时候传一个模拟的那一套进去立刻就能跑通全链路。第二不要在测试里用真正的 sleep 来模拟等待。我会用可控的事件对象来触发特定时序。比如我想测试超时分支就传一个永远不会完成的伪调用函数然后用wait_for设置一个非常短的超时时间这样测试能快速落下确定性结果也不再白白等几秒。第三失败注入很值得做。写一个测试用例让某个外部工具随机抛异常观察 Agent 是否走降级逻辑、是否正确重试、是否清理了资源。这个测试能精准地保护你前面踩过的大多数坑。我现在的项目在每个 Pull Request 里都会跑一轮这样的失败注入用例把那种线上突然大面积报错的体验提前按死在测试阶段。调试异步代码还有一个朴实的建议日志里打上任务 ID 和父任务 ID。异步代码的执行顺序和书写顺序经常不一致如果日志不带任务标识你会看到几路 log 混在一起完全理不清因果关系。我之前就被这种絮乱日志折磨过加上 task_id 字段之后排查问题的速度翻倍不止。最后的一点体会写 Agent 这几年我的整体心得是先把行为逻辑串行跑通再谈异步优化。不要一开始就陷入协程嵌套、事件驱动、任务编排的汪洋大海里先写一个最简单的原始同步版本让 Agent 的表现符合预期然后再逐步引入并发。每一次异步改造都应该伴随明确的衡量标准改完之后端到端平均耗时降低了多少超时率是否上升并发打上去后上游是否稳定没有这些数字支撑的异步优化极有可能是自我安慰。如果只让我送出一个建议那就是给所有外部调用统一加上两样东西一个超时控制一个耗时统计。这两样东西在平时看起来不起眼但当 Agent 在一线环境中表现糟糕时它们几乎能帮你迅速定位到 80% 的问题。我的线上事故排查经验里有太多场景最后都归结为某一次调用忘记了超时而不是模型本身出了问题。异步对 Agent 来说不是高级优化它是基础能力。把等待方式想清楚Agent 才能真正跑得起来而不是把时间花在排队发呆上。