个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangChain 批量调用与并发控制batch、abatch 与速率限制一、四种调用方式二、batch 的参数2.1 max_concurrency 必设2.2 return_exceptionsTrue 很重要三、异步为什么快3.1 同步 batch 的真相3.2 abatch 才是真并发3.3 Web 服务里必须用异步四、速率限制按 RPM 控制五、进度与断点六、长任务别用同步 batch七、常见问题八、一份生产级批处理模板九、小结LangChain 批量调用与并发控制batch、abatch 与速率限制有 1000 条数据要处理一条条invoke要跑到天亮。但直接batch全发出去又会 429 限流。这篇讲清楚怎么又快又不被限流以及异步到底快在哪。一、四种调用方式# 单个chain.invoke({q:...})# 批量同步chain.batch([{q:a},{q:b},{q:c}])# 单个异步awaitchain.ainvoke({q:...})# 批量异步 ← 生产首选awaitchain.abatch([{q:a},{q:b},...])方式并发适合invoke无单条、交互式batch线程池小批量、脚本ainvoke需配合 gather并发单条abatch真并发大批量、Web 服务二、batch 的参数resultschain.batch(inputs_list,config{max_concurrency:5},# 最多同时 5 个return_exceptionsTrue,# 单条失败不中断整体)2.1 max_concurrency 必设不设的话1000 条输入会一次性发出 1000 个请求。结果必然是429 Too Many Requests 429 Too Many Requests ...然后触发重试反而比串行还慢。# ❌ 危险chain.batch(one_thousand_inputs)# ✅ 安全chain.batch(one_thousand_inputs,config{max_concurrency:5})2.2 return_exceptionsTrue 很重要默认情况任何一条失败都会让整个 batch 抛异常前面已经算完的结果全丢了。resultschain.batch(inputs,config{max_concurrency:5},return_exceptionsTrue)# 逐个检查结果ok,bad[],[]forinp,resinzip(inputs,results):ifisinstance(res,Exception):bad.append((inp,res))else:ok.append(res)print(f成功{len(ok)}失败{len(bad)})# 失败的可以单独重试处理 1000 条数据时因为 1 条失败而丢掉 999 条的结果是很疼的。三、异步为什么快3.1 同步 batch 的真相chain.batch()用的是线程池。线程有开销而且 Python 有 GIL实际加速有限。3.2 abatch 才是真并发importasyncioimporttimeasyncdefmain():t0time.time()resultsawaitchain.abatch(inputs_list,config{max_concurrency:20},return_exceptionsTrue,)print(f耗时{time.time()-t0:.1f}s)asyncio.run(main())LLM 调用是IO 等待发出请求 → 等响应异步能在等待时去发下一个请求并发 20 就意味着 20 倍加速。实测对比100 条单条约 2 秒方式耗时循环invoke~200 秒batch(max_concurrency5)~40 秒abatch(max_concurrency20)~10 秒3.3 Web 服务里必须用异步app.post(/ask)asyncdefask(req:AskRequest):# ❌ 同步调用会阻塞整个事件循环# 一个慢请求会让所有其他请求排队returnchain.invoke({q:req.question})# ✅returnawaitchain.ainvoke({q:req.question})这是 FastAPI 里最容易犯的性能错误。同步调用在 async 函数里会阻塞事件循环QPS 直接掉到个位数。四、速率限制按 RPM 控制max_concurrency控制同时几个但 API 通常限制的是每分钟多少个请求RPM。importasyncioimporttimefromcollectionsimportdequeclassRateLimiter:限制每分钟最多 n 个请求def__init__(self,max_per_minute:int):self.max_per_minutemax_per_minute self.callsdeque()asyncdefacquire(self):nowtime.time()# 清掉 60 秒之前的记录whileself.callsandnow-self.calls[0]60:self.calls.popleft()iflen(self.calls)self.max_per_minute:wait60-(now-self.calls[0])awaitasyncio.sleep(wait)returnawaitself.acquire()self.calls.append(now)limiterRateLimiter(max_per_minute500)asyncdeflimited_call(item):awaitlimiter.acquire()returnawaitchain.ainvoke(item)resultsawaitasyncio.gather(*[limited_call(i)foriininputs])两个参数配合max_concurrency控制瞬时并发防打爆速率限制器控制长期速率防超限流。五、进度与断点处理 1000 条时你一定想知道跑到哪了以及崩了能不能续跑。importjsonfrompathlibimportPath OUTPath(results.jsonl)asyncdefprocess_all(inputs,limit20):doneset()ifOUT.exists():withOUT.open(encodingutf-8)asf:forlineinf:done.add(json.loads(line)[id])todo[iforiininputsifi[id]notindone]print(f总共{len(inputs)}已完成{len(done)}待处理{len(todo)})semasyncio.Semaphore(limit)asyncdefone(item):asyncwithsem:try:resawaitchain.ainvoke(item)return{id:item[id],ok:True,result:res}exceptExceptionase:return{id:item[id],ok:False,error:str(e)}withOUT.open(a,encodingutf-8)asf:fori,coroinenumerate(asyncio.as_completed([one(t)fortintodo]),1):recawaitcoro f.write(json.dumps(rec,ensure_asciiFalse)\n)f.flush()# 立刻落盘ifi%500:print(f进度{i}/{len(todo)})print(完成)三个要点每处理完一条就写盘flush崩了能从断点继续用asyncio.Semaphore控制并发启动时读已有结果跳过已完成的。六、长任务别用同步 batch一个常见错误在 Web 请求里处理 1000 条数据。app.post(/batch)asyncdefbatch_endpoint(req):resultsawaitchain.abatch(req.items,config{max_concurrency:20})returnresults# 请求挂在这里几分钟网关先超时了正确做法改成异步任务 轮询/回调。app.post(/batch)asyncdefbatch_endpoint(req,background_tasks:BackgroundTasks):job_idstr(uuid.uuid4())background_tasks.add_task(process_all,req.items,job_id)return{job_id:job_id}app.get(/batch/{job_id})asyncdefget_status(job_id:str):return{progress:get_progress(job_id)}七、常见问题现象原因解法大量 429并发太高降max_concurrency加速率限制一条失败全批失败没设return_exceptions设成 True跑一半崩了要重来没落盘逐条写 jsonl断点续跑异步还是慢中间有同步阻塞调用检查链里是否有requests等同步库内存爆了一次性构造 1000 个任务分块处理每块 100 条网关超时长任务同步等改后台任务 轮询特别提醒第 4 条如果你的链里某个工具用的是同步 HTTP 库如requestsabatch的并发会被它卡住。工具要提供 async 实现前面自定义 Tool 那篇讲过coroutine参数。八、一份生产级批处理模板asyncdefbatch_process(chain,items,*,concurrency10,rpmNone,out_pathout.jsonl):semasyncio.Semaphore(concurrency)limiterRateLimiter(rpm)ifrpmelseNoneasyncdefone(item):asyncwithsem:iflimiter:awaitlimiter.acquire()try:return{id:item.get(id),ok:True,result:awaitchain.ainvoke(item)}exceptExceptionase:return{id:item.get(id),ok:False,error:str(e)}withopen(out_path,a,encodingutf-8)asf:forcoroinasyncio.as_completed([one(i)foriinitems]):recawaitcoro f.write(json.dumps(rec,ensure_asciiFalse)\n)f.flush()returnout_path九、小结四种方式invoke单条、batch同步线程池、ainvoke异步单条、abatch真并发生产首选max_concurrency必设否则一次打爆 APIreturn_exceptionsTrue必设否则一条失败丢掉整批Web 服务里必须用ainvoke/abatch同步调用会阻塞事件循环max_concurrency管瞬时并发速率限制器管长期 RPM两者配合大批量任务逐条落盘 断点续跑别等全部完成再写长任务改后台任务 轮询别让 HTTP 请求挂几分钟。下一篇讲缓存——同样的请求别反复花钱。