第一次把一个 LangChain 链塞进异步服务里时我的第一反应是把 invoke 换成 ainvoke把 stream 换成 astream不就完事了吗实测下来事情确实就这么简单但所有坑也都藏在“只改方法名”的背后。LLM 调用本质上是一段漫长的网络 I/O如果还是按同步方式阻塞等待整个事件循环都会被一次请求拖住并发量一上来服务响应直接雪崩。这篇内容我会从 LangChain 的 Runnable 接口入手拆解异步机制的三个层面并发调用、流式输出、外部服务协作也会分享一些我在本地实验里反复踩过的细节。适合已经能用 LangChain 搭基础链、但想在服务化部署里提升吞吐量的朋友。1. 为什么一个链会卡在 I/O 上异步机制的动机1.1 LLM 调用天然是 I/O 密集任务顺序执行浪费在哪很多人会把“调用大模型”理解成一次普通函数调用传入 prompt返回 answer。但实际拆开看它经历的是网络连接、请求发送、服务端排队、模型推理、流式返回几十个 token、连接关闭。整个过程里真正由我们本地代码执行的 CPU 计算几乎可以忽略绝大多数时间都在等网络。以一次多轮对话为例。假设我要先从向量库里召回相关片段再把片段拼接成 prompt 发给模型最后把模型回复写入日志。用同步写法顺序可能是召回 300ms模型调用 1200ms写日志 50ms总计约 1550ms。这期间线程一直处于阻塞等待状态既没有新计算任务也没有响应新的请求。如果服务同时来了 10 个用户最朴素的实现甚至会一个一个排队处理后面的用户要等前面全部完成。这显然不合理。模型调用的等待时间不是“需要 CPU 处理”的时间而是“外部服务正在干活”的时间。这段时间本地代码什么也不用做只需要等通知。异步机制想解决的问题就是在这段等待里把事件循环让给其他任务让同一个线程可以同时维持大量外部请求。有人会问那用多线程不行吗可以但线程有切换成本和资源占用而且线程池开太多时操作系统调度开销也很可观。异步的做法更像是一个事件循环台账你发出去一堆请求每个请求在事件循环里登记一个“完成回调”。谁先返回谁先触发回调。响应到达的顺序和请求发起的顺序没有必然关系系统也不需要为每个请求分配一个独立线程。1.2 事件循环怎么让多个请求同时“挂着等”我刚开始接触 asyncio 时误以为异步就是“代码跑得快”。后来才理解异步并不改变单个操作的延迟它改变的是“等待期间 CPU 是否被浪费”。事件循环的本质是一个调度器把协程挂起、切换、恢复。你可以这样类比同步模式是去银行柜台办业务一个人办完下一个人才能坐上去异步模式是取号后在休息区等着柜员窗口就那几个但大厅里能同时坐着几百个排队的人。这还没完异步模式里“叫号”也不是完全按取号顺序来的哪个窗口先空出来就喊符合条件的下一位。应用到 LangChain 链上就是多个链同时被挂起执行先完成的先返回结果不必等最慢的那条链全部做完才统一结算。但这里有一个关键点协程里的代码不能自己“卡死”。如果在协程里使用了同步阻塞调用比如 time.sleep(1)事件循环会真的被冻住 1 秒整个大厅的排队系统都停摆。只有在协程里主动让出控制权比如 await asyncio.sleep(1)事件循环才能去处理别的任务。理解了这一层后面很多坑就都能预判了。2. Runnable 的 async 变体ainvoke、astream、abatch 的正确打开方式2.1 别急着把它当成“invoke 的 async 版本”LangChain 的 Runnable 接口把链路抽象成统一操作invoke、batch、stream以及一一对应的 ainvoke、abatch、astream。表面看只是加了个 a 前缀但底层调度逻辑完全不同。先看最小同步写法from langchain_community.llms import SomeLLM llm SomeLLM(...) result llm.invoke(讲个笑话)改成异步import asyncio async def main(): result await llm.ainvoke(讲个笑话) print(result) asyncio.run(main())看起来只是把 invoke 换成 ainvoke再在外面套一个 async 函数。但真正需要注意的是不是所有 Runnable 内部的组件都天生支持异步。比如你用 RunnableLambda 包了一个普通 Python 函数函数内部执行的是同步 I/O那么在 ainvoke 的执行链路里LangChain 会把它丢到线程池里去跑避免阻塞事件循环。所以 ainvoke 的“异步能力”是分层的组件类型异步行为原生支持 async 的组件走真正的协程路径事件循环可调度普通同步函数包成的 RunnableLambda被放入执行器线程池执行同步函数内部还调用了阻塞式网络库线程池能缓解阻塞但仍占用线程资源我踩过的第一个坑就是以为把链上的所有 invoke 都改成 ainvoke就能获得完整的异步能力。实际上如果链里某个 RunnableLambda 调用了一个同步 SDK 的耗时方法这个任务虽然不会阻塞事件循环却会占用一个线程池线程。线程池一旦被打满事件循环照样空转新任务全在排队。2.2 batch 的并发窗口与 abatch 的调度差异同步 batch 是顺序执行还是并发执行LangChain 在同步环境下默认也会对 batch 做一定程度的并发调度但由于没有事件循环通常是把多个输入丢给线程池去跑本质上还是线程池并发。真正高效的是 abatch。它会在一个事件循环内同时启动多个调用不需要为每个调用分配线程。它的并发上限可以用 config 控制results await chain.abatch( [问题一, 问题二, 问题三], config{max_concurrency: 2} )这里的 max_concurrency 限制了同时执行的最大协程数。我实际测试过把它从默认值调低之后模型提供方的限流报错明显变少。不要以为并发越大越好模型服务端通常有 RPM每分钟请求数限制和 TPM每分钟 token 数限制。abatch 同时发出 20 个请求大概率会触发 429。另一件需要留意的事abatch 返回的结果顺序和输入顺序是对应的。这一点很贴心不需要自己去给结果排序。但如果用的是 asyncio.gather 手动并发返回顺序就不一定通常要用 gather 自带的结果顺序来保证对齐。关于这个我后面会专门讲。2.3 流式输出的异步形态astream 与事件回调模型回复本身是 token 一个个生成的。同步 stream 一次只能等一个 token 回来期间无法处理其他事情。astream 的差别在于你可以对多个链同时发起流式请求各自按自己的节奏吐出 token事件循环不会因为某个链的网络延迟而卡住其他链。async for chunk in chain.astream({query: 分析一下这段代码}): print(chunk, end, flushTrue)这段代码的直观效果是用户看到的是逐字输出而不是等全部生成完才一次性显示。真实场景里网页应用通常会把 token 通过 WebSocket 推给前端配合打字机效果。如果服务端用同步方式实现每个 token 回来后需要立刻推送WebSocket 事件和时间循环之间会反复切换异步实现则更自然astream 天然产生异步迭代器token 一到就能触发回调。如果需要在流式输出时做额外处理比如记录每个 token 时间戳、统计首字延迟可以挂回调from langchain_core.callbacks import BaseCallbackHandler class TokenTimeHandler(BaseCallbackHandler): async def on_llm_new_token(self, token: str, **kwargs) - None: # 这里用 await 做别的异步操作也没问题 await log_token(token)线程安全问题也要留个心眼。多个链并行时回调是穿插触发的。如果在回调里写共享状态比如往同一个 list 里 append要确认这个操作是否线程安全。更稳妥的做法是让回调只做分散处理或者用队列把 token 统一收集到消费端。3. 多链并行编排用 asyncio.gather 并联任务并控制并发度3.1 两条链并联的最小写法实际业务里的 LangChain 链很少只有一条。比如一个客服机器人可能需要同时执行三个子任务意图识别、实体抽取、检索知识库。这三个任务之间没有依赖关系串行执行会白白多等两倍时间。用 asyncio.gather 可以把它们同时挂到事件循环上import asyncio async def handle_query(query: str): intent_task asyncio.create_task(intent_chain.ainvoke(query)) entity_task asyncio.create_task(entity_chain.ainvoke(query)) retriever_task asyncio.create_task(retriever_chain.ainvoke(query)) intent_result, entity_result, retriever_result await asyncio.gather( intent_task, entity_task, retriever_task ) return intent_result, entity_result, retriever_result这里使用 asyncio.create_task 把协程包装成任务让它们在后台开始执行。如果不使用 create_task直接写 await intent_chain.ainvoke(query)那就退回了串行第一条链执行完才会开始第二条链。这是新手最容易混淆的地方。关于 gather 的结果顺序gather 返回的结果列表顺序与传入的任务顺序一致并非按完成时间排序。这跟 asyncio.as_completed 的行为不同。如果后续聚合逻辑对顺序有要求用 gather 就足够如果希望“哪个先返回就先用哪个”比如做搜索聚合时想优先展示最快来源那用 asyncio.as_completed 更合适。3.2 并发上限、速率限制与重试策略把 await gather 写出来很简单难的是控制并发度。上面三条链的场景并发任务不多但如果要批量处理 100 条 query每个 query 还要做一轮多路召回那么同时挂起的请求可能瞬间是 300 个。这时候不设上限很可能会被模型提供方限流。控制并发度最直接的工具是 asyncio.Semaphoreimport asyncio sem asyncio.Semaphore(5) async def bounded_chain_invoke(query: str): async with sem: return await chain.ainvoke(query) async def run_queries(queries: list[str]): tasks [asyncio.create_task(bounded_chain_invoke(q)) for q in queries] return await asyncio.gather(*tasks)Semaphore(5) 的意思是同一时间最多有 5 个协程进入 critical section。剩下的协程在 async with sem 那行挂起等待等前面空出名额后自动补充。这比给 abatch 传 max_concurrency 更灵活因为你可以对不同的子调用设置不同的上限比如模型调用上限 5向量库检索上限 10。限流状态下的重试也要格外小心。模型提供方返回 429 时通常会告诉你 Retry-After 时间。但并发场景下所有请求同时失败、同时重试会造成“惊群效应”限制刚松口重试风暴又来了。我建议重试时间加一个随机抖动让请求错峰恢复import random async def invoke_with_retry(query: str, max_retries: int 3): for attempt in range(max_retries): try: return await chain.ainvoke(query) except RateLimitError as e: if attempt max_retries - 1: raise wait_time e.retry_after * (1 random.uniform(0, 0.3)) await asyncio.sleep(wait_time)随机抖动会让整体重试分布更平滑。实测下来添加抖动后整个批量任务的成功率稳定了不少而固定退避时总会在重试阶段再次触发限流。4. 在异步链里接入外部服务HTTP、数据库和任务队列的协作4.1 用异步客户端替换同步客户端避免阻塞事件循环LangChain 的链不可能只在模型之间转悠。真实系统里链经常要读取业务数据库、调用外部搜索接口、拉取订单状态。这些外部服务如果使用同步客户端就会给异步链路带来隐患。一个反面教材是这样的某问答链里有一个检索步骤内部用了同步 HTTP 客户端请求搜索服务。把它包进 RunnableLambda 后直接调用链的 ainvoke。初看没问题因为 RunnableLambda 会被丢到线程池执行。但如果请求量一大线程池线程全部忙于等待网络响应事件循环虽然空闲却没有可用的执行器线程来承接新的同步任务。表现就是 CPU 占用不高但请求延迟无限上涨。正确做法是把涉及的外部调用都切换成异步客户端。比如 HTTP 请求用 httpx 的 AsyncClient数据库操作尽量用支持异步的驱动连缓存读取都可以用 aioredis 之类的异步客户端。看一个对比# 同步客户端在协程里直接调用会阻塞事件循环 # 即使放在 RunnableLambda 里也占线程池资源 def fetch_sync(url: str) - str: import requests resp requests.get(url, timeout10) return resp.text # 异步客户端协程内 await 不会阻塞事件循环 async def fetch_async(url: str) - str: import httpx async with httpx.AsyncClient() as client: resp await client.get(url, timeout10) return resp.text判断一个库是否「异步友好」可以看它是否存在以 async 开头的方法或者文档里是否提到 asyncio 支持。还有一个简单测试在协程里调用它如果整个事件循环被卡住那它就是不友好的。4.2 外部异常传播与超时兜底异步并发场景下异常处理比同步更复杂。gather 默认的机制是只要有一个任务抛出异常整个 gather 立刻抛出异常其他没完成的任务不会被取消而是在后台继续运行。这经常导致诡异的内存泄漏和悬空任务。一个推荐的写法是 return_exceptionsTrueresults await asyncio.gather(*tasks, return_exceptionsTrue) for r in results: if isinstance(r, Exception): # 单独处理这条链的失败 fallback_result await fallback_chain.ainvoke(question)这样单个链失败不会拉着整个批次陪葬。但注意如果任务已经抛出异常return_exceptions 处理后任务内部的资源需要自己清理比如打日志、关闭连接。超时兜底同样重要。模型调用再快也有概率长时间不返回。异步场景里可以使用 asyncio.wait_for 给单个链设置超时try: result await asyncio.wait_for(chain.ainvoke(query), timeout30) except asyncio.TimeoutError: result 当前处理超时请稍后重试超时时间要根据实际链路复杂度调整。如果链里包含多个外部调用总超时应该留足冗余比如每个子调用最多 5 秒链级超时设到 30 秒。不要设得太短否则在高峰期模型响应变慢时大量请求会因超时被提前放弃反而更影响体验。5. 运行时集成asyncio.run 的限制与 Web 服务场景下的正确姿势5.1 asyncio.run 只能在循环外调用本地写异步脚本时最常见的主函数入口就是 asyncio.run(main())。但如果你的代码已经跑在事件循环里比如在处理一个 FastAPI 请求又调用 asyncio.run就会直接抛错asyncio.run() cannot be called from a running event loop。这个错误大家十有八九遇到过。原因很简单一个事件循环已经在这个线程里运行了不能再开一个新的。解决办法是在协程内部直接使用 await而不是再次调用 asyncio.run 去开新循环。以 Web 服务为例接口层本身是异步函数app.post(/ask) async def ask_endpoint(request: Request): body await request.json() result await some_chain.ainvoke(body[query]) return {answer: result}异步端口 ainvoke 是最自然的搭配不需要手动创建事件循环。唯一要留意的是FastAPI 的异步函数如果内部调用了同步阻塞库也会阻塞事件循环因为它在主线程的事件循环里直接跑。所以 Web 服务里的异步链更要坚持「全链路异步」的原则。5.2 在 Web 框架中异步链和后台任务配合有些场景不需要等待整条链执行完才返回响应比如用户提交了一个分析任务系统先返回任务 ID后台慢慢跑完成后通过轮询通知结果。这里同样可以用异步机制FastAPI 的 BackgroundTasks、或者自建一个 asyncio.create_task 队列。background_tasks {} app.post(/submit) async def submit_task(request: Request): body await request.json() task_id generate_task_id() background_tasks[task_id] asyncio.create_task( analysis_chain.ainvoke(body[query]) ) return {task_id: task_id} app.get(/status/{task_id}) async def get_status(task_id: str): task background_tasks.get(task_id) if task is None: return {status: not_found} if task.done(): return {status: done, result: task.result()} return {status: running}这里要注意 create_task 出来的任务如果中途抛异常而没有人 await 它会被事件循环标记为未处理异常甚至打出警告。最好在任务创建时就挂一个回调把异常记录到日志里再做结果缓存。事件循环的生命周期也和 Web 服务一致进程存活期间事件循环一直存在创建的任务由事件循环管理不需要手动关闭。但如果任务太多又没有引用关系管理可能导致后台任务堆积。建议给链调用增加超时并把已完成任务的结果做 TTL 缓存避免任务字典无限膨胀。6. 我从实验里总结的几条硬经验6.1 同步库混入异步链的后果我专门做过一次实验构造一个链内部包含两个步骤步骤 A 用同步 HTTP 库请求一个故意延迟 2 秒的接口步骤 B 调用模型。整个链通过 ainvoke 执行。同时另有一个普通协程任务在循环里每 0.1 秒打印一次时间戳。实验结果是当异步链执行到步骤 A 时打印时间戳的协程几乎停摆最长的停顿接近 2 秒。这说明即使 ainvoke 内部可能把同步函数丢进线程池线程调度依然会竞争 GIL事件循环的响应能力也会被拖累。更糟的是如果你的同步库调用发生在协程里而不是 RunnableLambda 中比如在自定义的 async def 函数里直接用了 requests.get那阻塞的是整个事件循环后果是灾难性的。结论是异步链要成功就必须在整条链的所有 I/O 环节都使用异步库。任何一环是同步阻塞的整条链的性能都会被打回原形。别想着用一两层线程池来兜底线程池只能“不阻塞事件循环”但不能消除线程切换开销和资源占用。6.2 代码层面的注意事项清单最后整理一份我在实践中沉淀的检查清单每一条都来自真实翻车现场ainvoke 不是万能药。先确认链里的每个组件是否真的支持异步调用尤其注意 RunnableLambda 内部包裹的第三方 SDK。并发增大不代表吞吐量线性增长。用 Semaphore 控制并发上限给外部服务和模型调用分别设置不同的限流窗口。在 gather 批量执行时如果要保证结果和输入顺序一致直接用 gather 按原顺序取结果如果无需保证顺序考虑 asyncio.as_completed 做先到先用。asyncio.run 只能出现在协程外的入口处。在 Web 框架里不要尝试“手动开新循环”应该直接 await 目标协程。超时设置不能只靠模型调用本身的参数。要为整条链设置外层兜底超时并考虑用 asyncio.shield 保护关键子任务避免外层超时把正在写日志的收尾协程一并取消。信号量和策略与键里既有连接无关考虑相同并发模型 consistent practices。当我需要动态扩展或评估完整链路吞吐时建议把链拆成三个子域检索域、模型域、后处理域。每个域独立测试异步并发上限后再做整体编排。这样定位性能瓶颈时不会误判是哪个环节拖慢事件循环。异步通信机制在 LangChain 里的价值不在于让单条链跑得更快而在于让系统的整体并发能力向上走一个台阶。把 ainvoke、astream、abatch 和 asyncio 原语配合起来再把外部 I/O 全部换成异步实现之后我手上的服务从单机同时处理几十个请求平滑提升到了能扛住数百个并发请求。这不是夸张而是把等待时间还给事件循环后自然获得的结果。