前言做 RAG 的同学90% 都踩过这个坑文档切分写好了向量库搭好了测试的时候十几条文档跑得好好的一到生产环境传几百上千条文档直接给你抛个 400 错误batch size is invalid, it should not be larger than 20你一脸懵啥意思我传的内容不对参数写错了翻半天文档才反应过来哦原来向量模型接口有批次限制单次最多传 20 条文本。新手第一反应那我循环一条一条传不就行了结果呢几百条文档传了十分钟慢得要死还容易超时。还有的直接写个计数器满 20 条传一次代码写得乱七八糟每个地方都要重复写一遍维护起来想死。今天我就把这个问题彻底讲透从问题本质到几种实现方案再到生产级的最佳实践给你一套拿过去就能用的分批工具以后再遇到任何批次限制的问题直接套用就行写一次用一辈子。一、先搞懂为什么会有批次限制很多人不理解为什么要限制单次传的条数我一次性传完不行吗其实很简单接口服务端也要考虑性能和稳定性防止过载一次性传几千条服务器处理不过来容易 OOM 或者超时公平性防止单个用户占满所有资源影响其他用户成本控制批量太大服务端资源消耗太快成本扛不住不只是阿里云百炼几乎所有大模型厂商的接口都有批次限制只是限制条数不一样阿里云百炼通义千问单次最多 20 条OpenAI单次最多 2048 条同时还有 token 数限制其他厂商从 10 条到 100 条不等划重点不管你用哪家的向量模型都要先看文档确认批次限制不要想当然地一次性全传不然肯定报错。这是新手最容易踩的坑测试的时候数据少测不出来一上生产就炸。二、错误写法直接全量传必报错先给大家看最典型的错误写法90% 的新手第一次写都是这么写的# 错误写法一次性把所有文本都传进去embeddingget_embeddings()content_list[chunk.contentforchunkinchunks]vectorsembedding.embed_documents(content_list)# 超过20条直接报错测试的时候 chunks 只有三五条啥问题没有一到生产环境几百条直接给你甩个 400 错误{detail:status_code: 400 \n code: InvalidParameter \n message: 400 InternalError.Algo.InvalidParameter: Value error, batch size is invalid, it should not be larger than 20.: input.contents}很多人看到这个错误还懵半天不知道啥意思其实就是你传的批次太大了超过了接口限制。三、入门级解决计数器分批简单粗暴知道了问题原因怎么解决最容易想到的就是我凑够一批就传一次传完清空继续凑下一批。就是你题目里写的这种写法# 入门级写法计数器分批doc_list[]vector_storeget_vector_store()forchunk_id,content,metadatainzip(chunk_ids,texts,metadatas):docDocument(page_contentcontent,metadatametadata,idchunk_id)doc_list.append(doc)# 凑够10条就存一次iflen(doc_list)10:vector_store.add_documents(doc_list)doc_list[]# 最后剩下的不足10条也要存一次ifdoc_list:vector_store.add_documents(doc_list)这种写法的优点简单直观一看就懂新手也能写出来不需要额外工具纯循环就能实现缺点也很明显复用性差每个需要分批的地方都要写一遍计数器代码重复容易漏最后一批很多人忘了循环结束后还要处理剩下的不足一批的数据导致最后几条丢了批次大小写死换个厂商限制不一样要到处改代码只能处理简单场景遇到需要重试、进度显示、异常处理的场景代码会越写越乱大佬经验这种写法适合临时写个脚本用一次用完就扔。要是项目里多处都要用到分批千万别这么写后面维护起来烦死你。四、进阶级解决通用分批工具函数写一次用一辈子真正优雅的做法是写一个通用的分批工具函数任何列表都能分不只是 Embedding 能用入库、批量请求、批量处理都能用。用 Python 的生成器yield实现非常简洁通用分批工具函数fromtypingimportList,Generatordefbatchify(data_list:List,batch_size:int)-Generator[List,None,None]: 通用分批工具把大列表拆成指定大小的小批次 :param data_list: 要拆分的原始列表 :param batch_size: 每批的大小 :return: 生成器每次返回一个批次的列表 foriinrange(0,len(data_list),batch_size):yielddata_list[i:ibatch_size]就这么几行代码搞定所有分批需求。怎么用用在 Embedding 向量化embeddingget_embeddings()content_list[chunk.contentforchunkinchunks]all_vectors[]# 每批10条留余量不卡着20的上限forbatchinbatchify(content_list,batch_size10):vectorsembedding.embed_documents(batch)all_vectors.extend(vectors)用在向量库批量入库vector_storeget_vector_store()doc_list[Document(page_contentcontent,metadatameta,iddoc_id)forcontent,meta,doc_idinzip(texts,metadatas,chunk_ids)]# 每批10条入库forbatchinbatchify(doc_list,batch_size10):vector_store.add_documents(batch)是不是干净多了逻辑清晰代码复用批次大小想改就改也不会漏最后一批因为生成器会自动把剩下的不足一批的也返回。大佬经验这种通用工具函数一定要放在项目的 utils 目录里到处都能用。不只是 Embedding以后遇到任何批量处理的场景比如批量查数据库、批量发请求、批量写文件直接调用就行省超多重复代码。五、生产级方案加重试、加进度、加异常处理上面的方案已经够用了但生产环境还要考虑更多问题网络波动某一批请求失败了怎么办几千条数据处理到哪了不知道干等着着急某一批失败了整个任务都挂了前面处理的都白跑了所以生产级的分批处理还要加上重试、进度显示、异常捕获这些能力。完整生产级工具代码importtimefromtypingimportList,Generator,Callablefromtqdmimporttqdmimportlogging loggerlogging.getLogger(__name__)defbatchify(data_list:List,batch_size:int)-Generator[List,None,None]:通用分批工具foriinrange(0,len(data_list),batch_size):yielddata_list[i:ibatch_size]defbatch_process_with_retry(data_list:List,process_func:Callable,batch_size:int10,max_retries:int3,retry_delay:float1.0,show_progress:boolTrue,desc:str处理中)-List: 带重试、带进度的批量处理工具生产级 :param data_list: 要处理的原始数据列表 :param process_func: 处理单批数据的函数接收一个批次的列表返回处理结果 :param batch_size: 每批大小 :param max_retries: 单批最大重试次数 :param retry_delay: 重试间隔秒 :param show_progress: 是否显示进度条 :param desc: 进度条描述 :return: 所有批次的处理结果合并后的列表 all_results[]batcheslist(batchify(data_list,batch_size))total_batcheslen(batches)# 进度条iteratortqdm(batches,descdesc,totaltotal_batches)ifshow_progresselsebatchesforbatch_idx,batchinenumerate(iterator):# 单批处理带重试last_errorNoneforretryinrange(max_retries):try:batch_resultprocess_func(batch)# 如果返回的是列表就合并不是列表就直接加ifisinstance(batch_result,list):all_results.extend(batch_result)else:all_results.append(batch_result)last_errorNonebreakexceptExceptionase:last_errore logger.warning(f第{batch_idx1}/{total_batches}批处理失败f第{retry1}次重试{str(e)})time.sleep(retry_delay*(retry1))# 重试间隔递增iflast_errorisnotNone:# 重试次数用完还是失败记录错误继续处理下一批不要整个任务挂掉logger.error(f第{batch_idx1}/{total_batches}批处理失败已重试{max_retries}次跳过该批{str(last_error)})# 也可以选择把失败的批次存下来后面单独处理continuereturnall_results怎么用举个例子批量向量化# 定义单批处理函数defembed_batch(batch_texts:List[str])-List[List[float]]:embeddingget_embeddings()returnembedding.embed_documents(batch_texts)# 批量处理content_list[chunk.contentforchunkinchunks]all_vectorsbatch_process_with_retry(data_listcontent_list,process_funcembed_batch,batch_size10,max_retries3,desc向量化中)print(f成功向量化{len(all_vectors)}条文本)再举个例子批量入库defadd_docs_batch(batch_docs:List[Document])-List[str]:vector_storeget_vector_store()returnvector_store.add_documents(batch_docs)# 批量入库doc_list[...]# 要入库的Document列表doc_idsbatch_process_with_retry(data_listdoc_list,process_funcadd_docs_batch,batch_size10,max_retries3,desc向量入库中)print(f成功入库{len(doc_ids)}条文档)就这么简单几行代码就有了自动分批失败自动重试重试间隔递增避免打挂服务进度条显示知道处理到哪了异常捕获某一批失败不影响其他批通用任何批量处理场景都能用大佬经验生产环境一定要加重试和异常处理网络波动、接口限流都是常事不要因为一批失败整个任务都挂了前面白跑几十分钟。进度条也很重要尤其是处理大量数据的时候用户看着进度条心里有底不会以为程序卡死了。六、避坑指南这些坑90%的人都踩过坑1卡着上限传容易触发其他限制很多人说限制 20 条那我就设 batch_size20刚好满不浪费。千万别这么干因为接口不只有条数限制还有 token 数限制、总字符数限制。你卡着 20 条传如果某一批的文本都特别长总 token 数超了还是会报错。最佳实践批次大小留 30%-50% 的余量限制 20 就设 10 或者 15不要卡着上限。牺牲一点速度换稳定性非常值。坑2只分 Embedding 的批入库不分批很多人只注意到向量模型有批次限制忘了向量数据库也有批次限制。比如 Milvus 单次插入也有数量限制还有大小限制你一次性传几千条照样报错。所以不光向量化要分批入库也要分批而且两个的批次大小最好一致方便对应。坑3没有重试机制一遇网络波动就全挂生产环境网络波动、接口限流、服务端临时故障都是常事偶尔失败一次很正常。你不加重试一批失败整个任务就挂了前面跑了半小时的都白跑哭都来不及。一定要加重试而且重试间隔要递增不要失败了立刻重试容易把服务打挂也容易触发限流。坑4并发开太高触发限流有人为了快开多线程并发请求结果并发太高触发接口限流直接被封一段时间更慢了。不要盲目追求速度要根据接口的 QPS 限制来控制并发数稳比快重要。一般来说串行分批处理就够快了真要并发也要控制好并发数不要开太多。坑5忘了处理最后一批用计数器写法的新手最容易犯这个错循环里凑够一批处理一次循环结束后剩下的不足一批的忘了处理导致最后几条数据丢了。用生成器的写法就不会有这个问题因为data_list[i:i batch_size]就算最后不足 batch_size也会返回剩下的所有元素。所以优先用生成器的写法少踩坑。七、给新手的 5 条实战心法1. 遇到批量处理先想有没有批次限制不管是调用接口、写数据库、发消息只要是批量操作先问一句有没有批次限制不要等报错了才去查提前看文档提前做好分批少走很多弯路。2. 通用逻辑要抽成工具函数不要到处复制粘贴分批这种通用逻辑一定要抽成工具函数放在 utils 里到处都能用。不要每个地方都写一遍计数器代码重复不说改的时候还要改好多地方容易漏。写一次用一辈子香得很。3. 稳定性永远比速度重要不要为了快卡着上限传、开超高并发看起来快了一点其实很容易出问题一出问题就要返工反而更慢。留余量、加重试、稳扎稳打才是生产环境的正确做法。稳比快重要这是无数次踩坑换来的经验。4. 进度条是个好东西用户体验提升巨大处理大量数据的时候一定要加进度条。不然用户看着黑框框半天没反应以为程序卡死了直接给你关了白跑半天。一个进度条几行代码用户体验提升不止一个档次。5. 失败不可怕可怕的是失败了前面白跑批量处理的时候一定要考虑失败的情况不要因为一批失败整个任务都挂了。能重试的重试重试不行的跳过记录下来后面单独处理不要让一批失败影响整个任务。前面跑了多久的成果不能因为最后一批失败就全白费。最后Embedding 批次限制这个问题说大不大说小不小新手很容易踩坑。但只要你掌握了分批处理的思路有一套通用的工具函数这都不是事。而且这个思路不只是用在 Embedding 上任何批量处理的场景都能用一通百通。把今天的工具函数存到你的 utils 目录里以后再遇到任何批次限制的问题直接拿出来用省超多时间。如果文章对你有帮助欢迎点赞收藏有问题评论区交流。