老实说我刚开始写 Agent 时的代码全是同步的。意图很单纯LLM 调用一个接一个工具调用排队执行整个主循环一眼看得穿调试也省心。那时候我觉得异步是个“锦上添花”的东西等同步跑不通了再聊。直到有次给一个调研型 Agent 加了五个信息源查询每个请求要两三秒用户在页面上干等二十多秒期间程序就像死了一样连个流式过程都输出不了。那次之后我才真正意识到写 Agent 和写普通 CRUD 接口有个巨大的差异Agent 的大部分时间都花在“等待”上而同步模型恰恰把等待变成了浪费。今天这篇就是我在把 Agent 主循环全面切到异步之后攒下来的一些体会和踩坑实录希望能帮同样在做 Agent 开发的你少走几步弯路。先说一个我验证过很多次的结论Agent 和异步是天生一对。这不是说同步不能写 Agent而是说只要你的 Agent 要面对真实场景——流式输出、并发工具调用、多用户并发、任务中断恢复——同步方案就会处处碰壁。这篇文章不讲某个现成框架怎么用而是从零拆解异步 Agent 的关键环节、常见误区以及我现在每天都在用的工程化套路。1. Agent 循环的本质其实是编排“等待”1.1 Agent 到底在等着什么任何 Agent无论包装得多复杂底层都在跑同一个循环观察、思考、行动。先观察环境收集工具返回的信息再把观察结果交给 LLM 思考让它决策下一步然后执行决策调用对应的工具更新环境回到观察。你可能见过各种花哨的名字什么 ReAct、Plan-and-Execute剥开来都是这个循环的变体。这三步有一个几乎所有 Agent 都绕不开的共同点每一步都在等待外部资源。观察要等网络请求返回思考要等 LLM 生成完 token行动又要等工具执行完毕。整个循环里 CPU 真正干活的时间可能不到总耗时的 10%剩下 90% 都在等待 IO。如果你用同步风格实现就意味着进程在这 90% 时间里彻底闲着用户被迫看着空白界面。举个再实在不过的例子。假设 Agent 需要查三个信息源才能回答一个问题查站 A等 3 秒查站 B等 2 秒查站 C等 2 秒。同步串行总耗时至少 7 秒。异步并发则是一条命令同时发出三个请求总耗时只由最慢的那个决定也就是 3 秒左右。这还只是三个工具真实 Agent 动辄就是五六个待调工具串行和并行的差距会成倍放大。1.2 同步方案的三个绕不开的痛点我在同步风格里写过整套 Agent 流程踩过的坑归纳下来主要就三个。第一个是无法流式输出。用户说了一句话Agent 要内部观察、思考好几轮才肯吐出第一个字这中间用户只能一直盯着“等待中”。同步接口意味着 LLM 必须完整生成完一整段才能返回用户等到花都谢了。异步配合流式接口可以把边生成边返回的 token 实时推给用户至少让人知道程序在干活这在产品体验上是天壤之别。第二个是多工具并发能力基本为零。真实 Agent 经常需要同时查天气、查日历、查地图这些工具之间往往没有依赖本来可以并行。同步实现却只能挨个执行所有延迟累加而且只要中间某个工具抽风变慢后面所有工具都会被拖住。第三个是资源效率难看。为了模拟并发有人会引入多线程每个任务开一个线程。线程可不是免费的线程创建切换有开销锁竞争和共享内存同步容易出问题。更麻烦的是Python 的 GIL 让纯 CPU 密集的多线程也拿不到真正的并行IO 密集型场景用线程池凑合能跑但线程数一上来内存和调度开销直接把人看傻眼。Agent 这种成千上万并发 IO 的场景正确姿势是单线程加事件循环加协程也就是 asyncio 那套模型。1.3 事件循环好懂吗一个服务员照看全场的模型讲事件循环之前先给一个生活化的类比。同步风格的服务员是这样的走到一桌客人面前站着不动等这桌客人慢慢点完、吃完、结账再去服务下一桌。这样当然也能服务好但效率很低大部分时间服务员都杵在那干等。异步的事件循环则像一个熟练的值班经理他在不同桌之间快速流转A 桌点完菜在等后厨的时候他先去给 B 桌上菜B 桌在等加菜他又去回应 C 桌的呼唤。关键在于在一次等待期间调度器会去做别的事。代码层面这个“等待”就是一个await表达式——协程把 CPU 控制权还给事件循环事件循环去唤醒其他就绪的协程。再往深一层看asyncio 通常跑在单线程上它靠非阻塞 IO 和系统事件通知机制比如大家常说的 epoll、kqueue来判断“哪些协程已经可以继续了”。所以你在 async 函数里写await some_io()真实含义是“我现在不占用 CPU等 IO 结果出来再叫我”。从阅读代码的角度整个流程看起来几乎是同步的顺序执行、连续、可读。但在等待的间隙里别的协程正在飞跑。这就是协程被称为“协作式多任务”的原因你的协程主动让出调度器才有机会运行别人。如果有哪个函数不守规矩在事件循环里搞真正的阻塞那后面所有协程都会陪葬这点我在第 3 节会展开讲。2. 异步写 Agent 的正确姿势主循环、流式与并发调用2.1 最小可用的异步 Agent 主循环初学者容易一上来就套现成框架反而把核心机制包在黑盒里。我更推荐先拿一个极简的异步循环把概念打通。下面是一段最基础的异步 Agent 主循环骨架它不是完整业务代码但结构是真实 Agent 的缩小版import asyncio from dataclasses import dataclass, field from typing import Optional dataclass class AgentState: query: str history: list field(default_factorylist) observations: list field(default_factorylist) done: bool False reply: Optional[str] None async def observe(state: AgentState) - str: # 模拟工具调用搜索、查询数据库、读文件……本质都是 IO await asyncio.sleep(0.5) return f找到与 {state.query} 相关的资料 async def think(state: AgentState, observation: str) - str: # 模拟 LLM 推理耗时 await asyncio.sleep(1) return 先调用搜索工具再汇总结论 async def act(state: AgentState, decision: str) - str: # 模拟执行某个工具动作 await asyncio.sleep(0.5) return f已执行{decision} async def run_agent(query: str) - str: state AgentState(queryquery) for _ in range(5): observation await observe(state) state.observations.append(observation) decision await think(state, observation) result await act(state, decision) state.history.append((decision, result)) if 结束 in result: state.done True break return state.reply or 完成 async def main(): reply await run_agent(本周天气适合露营吗) print(reply) if __name__ __main__: asyncio.run(main())这个骨架的核心在于每一处 IO 都是await整个循环读起来像同步但等待期间事件循环可以去调度别的任务。你现在跑它可能看不出并发优势因为只有一个 Agent 在跑。但它已经是一个可改造的地基——把observe换成真正的并发工具调用把think换成流式 LLM SDK它就是一个完整的异步 Agent。2.2 流式输出把“等结果”变成“持续反馈”接入真实 LLM 后最能影响用户体验的就是流式输出。同步写法是等 LLM 把一整段回答全生成完再一次性返回异步写法则是通过流式接口持续推送增量 token。我用一个模拟生成器的例子描述这层行为async def llm_stream(prompt: str): for word in [今天, 天气, 适合, 露营, 但注意, 防风]: await asyncio.sleep(0.2) yield word async def process(): async for token in llm_stream(给个建议): print(token, end) print()在 Agent 场景里流式输出的价值不光是“输出快一点”。更关键的是它允许我们在生成过程中做判断。比如解析输出内容时发现“调用工具”的关键词已经出现就可以提前中断生成而不是傻等整段生成完再行动。这个能力在 Agent 里极其实用能省下大量 token 和等待时间。很多 Agent 框架把流式封装到了内部你从外面看不到但只要你从头写过一遍就会明白异步在这里不是优化项而是必需品。2.3 并行工具调用与限流让 Agent 同时干几件事Agent 场景天然适合并行调用工具因为工具之间常常没有依赖——查天气、查航班、查酒店完全可以同时进行。最直接的方法是asyncio.gatherasync def fetch_weather(city: str) - dict: await asyncio.sleep(1) return {city: city, condition: 晴} async def fetch_flight(city: str) - dict: await asyncio.sleep(1.5) return {city: city, price: 1200} async def fetch_hotel(city: str) - dict: await asyncio.sleep(2) return {city: city, star: 4} async def parallel_lookup(city: str) - dict: weather, flight, hotel await asyncio.gather( fetch_weather(city), fetch_flight(city), fetch_hotel(city), ) return {weather: weather, flight: flight, hotel: hotel}gather会等所有协程完成再返回返回顺序和传入顺序一致。但真实环境里有个很尴尬的情况三个工具里有一个特别慢另外两个早就完成了你却必须一直等那一个拖后腿的。如果你希望“完成一个就处理一个”用asyncio.as_completed更合适async def process_all(city: str): coros [fetch_weather(city), fetch_flight(city), fetch_hotel(city)] for coro in asyncio.as_completed(coros): result await coro # 完成一个就先落库或做轻量处理不用等全部完成 await handle_partial_result(result)工具并发数量上去以后还要考虑限流。Agent 一次可能发起五六个候选工具调用但下游 API 往往有速率限制无脑并发会把上游打爆。这时可以用Semaphore限制并发数我几乎在每个生产级 Agent 里都会加这一层sem asyncio.Semaphore(3) async def bounded_call(tool_name: str, *args): async with sem: return await call_tool(tool_name, *args) results await asyncio.gather(*[bounded_call(t, *a) for t, a in tasks])提示信号量只限制并发数量不保证顺序。如果需要按顺序输出结果用 gather 配套 index 记录或者走 as_completed 并按业务字段归组都比假设顺序来得靠谱。2.4 超时、取消与 TaskGroup让 Agent 变得可控异步世界里最容易被忽略的管理对象是时间。LLM 可能网络抖动、外部工具可能永远不返回、用户可能中途取消对话。没有超时机制的 Agent一旦某个任务挂死整个系统会积累一堆僵尸协程资源被白白占着。Python 3.11 之后asyncio.timeout和asyncio.TaskGroup是处理这类问题的标准姿势try: async with asyncio.timeout(10): result await call_llm(prompt) except TimeoutError: result LLM 调用超时改用缓存结果async def run_task_group(): async with asyncio.TaskGroup() as tg: tg.create_task(fetch_weather(北京)) tg.create_task(fetch_flight(北京)) tg.create_task(fetch_hotel(北京)) # 任何一个任务抛异常TaskGroup 会取消其他任务并向上传播TaskGroup 有一个行为特别适合 Agent 编排只要组里任何一个任务抛异常其他未完成任务会被自动取消。举个例子三个工具调用里有一个失败其他两个查询再跑下去已经没有意义与其等它们慢慢结束不如整体取消快速失败。这在资源节省和错误传播上都非常有价值。如果你还在用 Python 3.10 或更早版本那只能用asyncio.wait(..., timeout...)加显式task.cancel()来组合实现代码会啰嗦不少。所以我一直建议新项目尽量直接上 Python 3.11。“取消”还有一个重要来源是用户主动停止。Agent 跑到一半用户点了取消按钮同步调用栈往往很难干净中止正在进行的网络请求而异步里通常只需要持有主任务引用在取消入口调用task.cancel()事件循环会向目标协程注入CancelledError让它有机会做清理工作关闭子任务、记日志、释放资源然后正常终止。这个能力在对话类产品里几乎是刚需。2.5 状态管理别在异步里用共享可变对象Agent 天然是有状态的当前目标、已观察信息、历史记录、中间决策、待处理的工具结果。多个协程之间共享这个状态时最常见的错误是直接修改同一个可变对象——一个dict或一个类实例多个协程同时往里写可能互相覆盖也可能读到半成品。我现在的做法是把 Agent 状态设计成不可变的数据结构每次更新生成新实例。虽然多花一点内存但换来的是极高的确定性尤其在并行工具调用的场景下每个协程只负责返回自己的结果不碰共享状态最后收集齐了再一次性推进状态。这样即使某一步失败状态也不会被污染回滚和重试容易得多。dataclass(frozenTrue) class AgentState: version: int query: str observations: tuple[str, ...] history: tuple[tuple[str, str], ...] done: bool False def with_observation(self, obs: str) - AgentState: return AgentState( versionself.version 1, queryself.query, observationsself.observations (obs,), historyself.history, )这里有个容易被忽略的小细节frozen dataclass 的哈希要求所有字段可哈希而 tuple 天然可哈希list 不行。所以状态里的观测列表和历史列表我都用 tuple 而不是 list。这样设计出来的状态同时满足不可变、可序列化、可哈希三个需求既能安全地在并发协程间传递也能随时持久化保存、跨进程恢复。3. 我在异步 Agent 开发中踩过的坑3.1 time.sleep 是如何把整个事件循环“哄睡”的我早期犯过一个特别典型但后果很严重的错误在 async 函数里用time.sleep(1)模拟耗时操作。结果整个程序直接卡住——注意不是只卡当前协程而是所有协程全部卡住。连那些早就该继续运行的任务也像被按了暂停键一样。原因前面解释过事件循环通常只有一个线程time.sleep是真正的线程阻塞它把唯一的工作线程按在原地而asyncio.sleep只是向事件循环注册一个定时器协程立刻让出控制权事件循环可以继续跑别的任务。两者在同步代码里几乎等价在异步世界里完全不是一回事。排查方法其实很简单如果某个协程特别慢而且其他协程也被拖累十有八九就是有真正的线程阻塞混进来了。去代码里搜可疑的time.sleep、requests.get、read()阻塞点把它们替换成异步版本。这个坑我在论坛里见过无数人踩包括我自己谁也别笑谁。3.2 fire-and-forget 的任务去哪儿了异步代码里有个诱惑很大的写法用asyncio.create_task创建一个后台任务但不去await它。比如 Agent 在回复用户的同时想偷偷把中间结果写进缓存于是随手写一句asyncio.create_task(save_cache(result)) return result看起来没毛病但在 Python 的某些运行环境下这个任务如果没有任何外部引用可能被垃圾回收器直接回收连执行的机会都没有。更稳妥的做法是持有任务引用并在主流程末尾显式await或cancel。如果你确实希望任务在后台慢慢跑也应该把它挂到一个全局任务注册表或TaskGroup里至少保证它有引用、能被追踪、能被取消。这个坑的隐蔽点在于本地调试时程序很快就跑完了后台任务好像也执行了线上任务一多引用丢失导致的静默失败会让你百思不得其解。异步世界里看似“丢出去不用管”的任务其实是有生命周期的设计时必须明确它属于谁、何时结束、异常怎么处理。3.3 并发更新同一个状态的竞态Agent 天然会把任务拆给多个协程。例如同时查 5 个工具每个工具返回后都想往state.observations追加结果。如果是同一个 list并发append表面上没问题——GIL 会保护字节码级的操作——但如果你在追加之后还做了判断、用新结果去派生下一步动作就可能读到半更新状态的中间态。我试过三种解法一是给状态更新加asyncio.Lock简单但容易把并发降回串行二是用不可变状态每次生成新实例最后批次汇总三是在并行工具调用时每个协程只返回纯结果由主循环统一收齐后再一次性更新状态。对 Agent 这种高轮次场景第三种做法效果最好因为它天然规避了锁竞争而且失败回滚非常干净。3.4 traceback 又长又乱给所有流程打上 trace id异步代码调试最大的痛点是 traceback 又长又乱极难定位是哪一步出的错。原因在于协程的调用链被事件循环打散了你很难看出“这个协程是谁发起的中间经过哪几个状态”。我被折磨几轮之后养成了一个习惯Agent 入口生成一个全局trace_id通过日志上下文或显式参数传给所有协程让所有日志行都带上它。import uuid trace_id uuid.uuid4().hex logger.info(agent start, extra{trace_id: trace_id})这样多个用户同时跑多个 Agent 时从日志里就能按 trace_id 串联出单个 Agent 的完整生命周期。异步代码里手动维护上下文确实有点烦但值得。某些库能用 contextvars 来自动传递上下文不过显式传参会更可控尤其是当你需要把 Agent 状态持久化保存时trace_id也一并存下来才能保证跨进程、跨时间点都能追踪。3.5 同步库混进异步里的边界我在异步代码里直接调用过同步 IO 库结果就是把整个事件循环卡死。这个和time.sleep是同一类问题事件循环线程里不允许出现真正的阻塞。但现实情况是不是所有库都有异步版本。我的经验是分情况处理如果有异步替代库优先换比如同步 HTTP 库换成异步客户端。如果没有异步替身但任务是 CPU 密集或纯计算型用asyncio.to_thread扔到默认线程池里不阻塞事件循环。如果底层库为了性能使用了阻塞原生的调用比如某些 C 扩展在 async 函数里裸调要格外小心尽量隔离到专用线程或干脆拆分到独立进程。这个边界没法靠死记规则核心是一根弦事件循环线程里不允许出现真正的阻塞。一旦出现所有并发协程都会串行等待异步就退化成更慢的同步。4. 异步 Agent 的工程化建议与观测设计4.1 同步壳加异步核的接口设计做工程时经常会遇到一种矛盾团队其他模块还没适应异步 API或者调用方就想要一句result agent.run(query)的简单接口。这时候我推荐“同步壳加异步核”的模式def run_sync(query: str) - str: return asyncio.run(run_agent(query))但这个模式有个大坑如果run_sync在一个已经运行的事件循环里被调用比如在某个 Web 框架的异步视图里asyncio.run会因为无法嵌套事件循环而直接报错。所以正确的做法是对外尽量暴露异步主接口同步便捷接口只提供给那些确定不在事件循环线程中调用的调用方。否则接口设计得不干净迟早会踩到嵌套事件循环的雷。4.2 让 Agent 像状态机一样可观测异步重构之后我最大的变化是把 Agent 理解成一个显式状态机。每一次主循环就是一次状态迁移观察 → 思考 → 行动 → 更新状态。每一步都记录结构化日志配上耗时、token 消耗、工具调用参数。这样线上出问题时我能快速定位是卡在 LLM 调用还是某个外部工具没返回。我通常给每个 Agent 执行实例定义一个结构化事件对象dataclass(frozenTrue) class Event: step: str # observe / think / act / finish duration: float trace_id: str detail: dict把这些事件写入日志或时序数据库后续就能统计平均每轮耗时、工具调用成功率、哪类查询最耗时。异步 Agent 执行路径多、分支广这种观测能力不是可选项是必需品。没有观测的异步 Agent 就像一个没有仪表盘的飞机飞得再快也让人不踏实。4.3 测试异步 Agent 的实用套路测试异步代码是很多人头疼的环节因为测试框架需要管理事件循环。我用的是 pytest 加异步插件那套配合固定超时避免测试卡死pytest.mark.asyncio async def test_agent_parallel_tools(): result await run_agent(查三个城市天气) assert 北京 in result如果不想每次测试都真实调用 LLM就做一个 fake LLM 协程返回预设内容。测试异步代码的重点是每个测试都该有超时否则一个坏协程能让整个测试套件挂到天荒地老。主逻辑里的超时控制也要一并测进去比如模拟 LLM 超时断言 Agent 能走兜底分支。异步的好处这时候也体现出来了控制协程的调度时机远比控制真实线程简单你可以很稳定地复现竞态、超时、取消这些边界场景。5. 多 Agent 协作异步从“可选项”变成“必备项”5.1 多 Agent 编排本质上是一场事件流当你从单个 Agent 走向多 Agent 协作异步的权重会再次上升。两个 Agent 并行调研、一个 Agent 把结果抛给另一个 Agent 做裁决、多个 Agent 竞争回答同一个问题再由汇总器挑选最佳结果这些模式在架构上本质都是事件流。事件流天然适合异步每个 Agent 是一个协程或一个任务它们通过队列、消息通道来协作。我之前做过一个多角色讨论型 Agent 项目三个角色各持立场轮流发言然后由一个总结者收尾。如果每个角色都占一个线程线程调度复杂还容易抢锁用异步协程加队列每个角色就是一个while True循环加await queue.get()代码甚至比同步版更简洁。5.2 扇出、扇入与队列的取舍多 Agent 协作里最常用到的模式是扇出和扇入。扇出就是把一个任务拆给多个 Agent 并行处理扇入就是把多个 Agent 的结果收集起来做汇总。扇出可以用gather或TaskGroup实现扇入则可以依赖asyncio.Queue来做结果归集。async def fan_out(query: str, agents: list): results await asyncio.gather(*[agent.run(query) for agent in agents]) return await summarize(results) async def fan_in_with_queue(agents, task_queue, result_queue): # 每个 agent 从 task_queue 拿任务结果丢进 result_queue async def worker(agent): while True: task await task_queue.get() result await agent.run(task) await result_queue.put(result) await asyncio.gather(*[worker(agent) for agent in agents])队列模式的好处是天然支持动态扩缩容、背压控制。任务太多时队列会积压Agent 不会过载某个 Agent 崩溃了它只是不再从队列取任务其他 Agent 的工作不受影响。这些能力在单 Agent 时代只是“加分项”到了多 Agent 协作阶段就是“保命项”。我体会最深的是只要涉及多个 Agent 的编排异步其实是你唯一务实的选择因为线程模型下的锁、调度、资源占用会让你在还没跑通业务逻辑之前就先陷进并发泥潭里。最后分享一点个人心得。虽然这篇文章一直在讲异步的好处但我并不劝你把所有代码都改成异步。如果你的 Agent 只跑一轮、只调一个工具、用户也不在乎多等几秒同步完全够用。我自己现在的判断标准很简单只要出现“一个 Agent 在同时等两件事”或者“用户需要看流式输出”或者“可能有多个 Agent 并发跑”我直接上异步。写得多了你会发现异步从来不是一种炫技它是一种让时间变得可控的基层能力。它解决不了所有问题但恰好解决了 Agent 开发里绕不开的大部分问题。另外如果你正打算从同步切到异步我的建议是别一次性把整个系统翻个底朝天。先把最痛的一条链路改成异步拿到立竿见影的体验提升再逐步铺开。一次改一个环每改一环都能验证比大爆炸式重构稳妥得多。这个路子反正我试过挺香。