企业官网建设流程全解析

写 Agent 写到第三版，我才真正想明白一件事：大部分 Agent 跑得慢，背锅的常常不是模型本身，而是调度逻辑里那些串行等待。一个 LLM 调用要两三秒，一次工具请求又要两三秒，一轮子 Agent 协作又是好几个来回。如果每个环节都老老实实排着队跑，任务一多，时间全花在"等"上。异步解决的不是"能不能跑"的问题，而是"能不能别傻等"的问题。

这篇文章不打算写一套完整的异步教程，只分享我在写 Agent 过程中对异步的几点体会：为什么 Agent 天然离不开异步、两条路线的选型取舍、最容易翻车的坑，以及几个在真实流程里用得上的编排模式和调试手段。适合已经写过一点 Agent、开始觉得"这玩意儿咋这么慢"的读者，也适合准备把单个 Agent 扩展成多任务协作场景的朋友。

1. Agent 跑得慢，不是模型慢，是你等得太多了

1.1 Agent 的一天：清一色的等待操作

我最早写 Agent 的时候，脑子里是把 Agent 当成"一个函数"来想的：输入一个任务，输出一个结果。但写多了才发现，Agent 运行的绝大部分时间根本不在计算，而在等待 IO。

具体来说，一个典型 Agent 的任务周期大概是这样的：

1. 接收用户输入，把目标拆解成几个子任务。
2. 对第一个子任务，调用 LLM 生成下一步动作。
3. 根据动作去调一个外部工具（搜索、查数据库、调内部服务）。
4. 拿到工具结果后，再次调用 LLM 做推理。
5. 有时候还要把一部分子任务分给下游 Agent，等它们把结果回传。
6. 最后汇总所有中间结果，再生成最终答案。

你留意一下，这里面的每一步，核心动作都是"发一个请求，等一个响应"。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 def+await这一套。一个异步的工具调用函数大概长这样：

import asyncio async def call_external_source(source: str, query: str): # 这里的 HTTP 请求必须是异步客户端 async with httpx.AsyncClient() as client: resp = await client.get(f"https://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(f"https://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, timeout=timeout) except asyncio.TimeoutError: # 记录日志并走降级逻辑 return fallback_result()

用的时候就是：

result = await call_with_timeout(call_external_source("a", query), timeout=5.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.gather：

async 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_exceptions=True，避免某个工具挂了导致整个批次全废。然后拿到返回结果后，先判断哪些是异常，哪些是有效数据，再构造提示词。这样单个工具抖动不会把整轮对话搞崩。

4.2 竞争回退：多个候选谁先成功用谁

第二个我经常用的是"竞争回退"模式。场景是这样的：有些外部工具的可用性不稳定，Agent 需要调用某一个服务，但这个服务偶尔超时。为了降低失败率，我同时发起主服务调用和备用服务调用，哪个先成功就采用哪个。

代码上可以用asyncio.wait来实现：

import asyncio async def call_with_fallback(primary_coro, fallback_coro, timeout=5.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_when=asyncio.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 实例拥有独立的上下文副本，任务结束后只把需要的部分合并回主流程。

如果你用 Python，contextvars.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", tag=tag, cost_ms=(elapsed * 1000)) return result except Exception as e: elapsed = time.perf_counter() - start log_event("async_call_error", tag=tag, cost_ms=(elapsed * 1000), error=str(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 才能真正跑得起来，而不是把时间花在排队发呆上。