阿里云向量模型单次最多 20 条分批处理的完整解决方案问题背景在用阿里云百炼DashScope的文本向量模型做 RAG 知识库时很多人第一步就踩坑# 假设切分出了 100 个文档块直接一次性丢进去embeddings.embed_documents(chunk_texts)# ❌ 报错超过单次批量上限报错原因很简单——向量模型的单次 API 调用有批量条数上限。根据阿里云官方文档不同模型版本的上限不一样模型单次最多条数单条最大 Tokenqwen3.7-text-embedding20128,000text-embedding-v3 / v4108,192text-embedding-v1 / v2252,048这个限制对字符串数组和文件输入同样生效。也就是说只要你的文档块数量超过上限就必须自己想办法分批。本文给出一套生产可用的方案分批切分 限流重试 并发加速并说明 LangChain 场景下怎么接入。核心思路把大列表切成小批次思路非常朴素——一次吃不下就分几次吃defbatched(items,batch_size):把列表切成固定大小的小批次foriinrange(0,len(items),batch_size):yielditems[i:ibatch_size]texts[f第{i}条文档foriinrange(47)]print([len(b)forbinbatched(texts,20)])# [20, 20, 7] ← 47 条被切成 3 批用生成器yield而不是一次性构造列表处理几十万条文档时也不会占用额外内存。基础版分批调用 结果拼接以 DashScope SDK 为例importdashscopefromdashscopeimportTextEmbedding dashscope.api_keysk-xxxBATCH_SIZE20# 按所用模型的实际上限调整v3/v4 为 10v1/v2 为 25defembed_documents(texts,batch_sizeBATCH_SIZE):分批调用向量接口拼回完整结果all_embeddings[]foridx,batchinenumerate(batched(texts,batch_size),1):respTextEmbedding.call(modeltext-embedding-v4,inputbatch,)ifresp.status_code!200:raiseRuntimeError(f批次{idx}调用失败:{resp.code}{resp.message})# 按 text_index 排序保证与输入顺序一一对应vecssorted(resp.output[embeddings],keylambdax:x[text_index])all_embeddings.extend(v[embedding]forvinvecs)print(f批次{idx}: 处理{len(batch)}条累计{len(all_embeddings)}/{len(texts)})returnall_embeddings两个关键细节batch_size写成常量/配置项而不是写死 20。换模型时改一个数字就行。按text_index排序后再拼接。返回结果虽然一般按顺序但显式排序可以避免某些异常情况下的错位——向量库最怕的就是第 5 条文本配了第 6 条的向量这种 bug 不报错但检索结果全错。进阶版限流重试必加批量处理几百上千条时几乎必然会撞上 QPS 限流Throttling.RateQuota。不加重试跑到第 300 条挂了前功尽弃。推荐指数退避策略importtimeimportrandomdefembed_with_retry(batch,max_retries5):带指数退避的重试forattemptinrange(max_retries):try:respTextEmbedding.call(modeltext-embedding-v4,inputbatch)ifresp.status_code200:returnresp# 限流或服务端错误才重试参数错误直接抛ifresp.codenotin(Throttling.RateQuota,InternalError):raiseRuntimeError(f不可重试的错误:{resp.code}{resp.message})exceptException:ifattemptmax_retries-1:raisewait2**attemptrandom.random()# 1s, 2s, 4s, 8s... 加随机抖动print(f 第{attempt1}次失败{wait:.1f}s 后重试)time.sleep(wait)要点只对限流类错误重试。如果是 API Key 错误、参数错误重试一万次也没用应该立刻失败。加随机抖动random.random()。多个并发任务同时失败时如果都按同样的节奏重试会形成惊群抖动可以把重试时间点打散。提速版多线程并发分批是串行的1000 条要发 50 次请求每次 1 秒就要等 50 秒。既然是 I/O 密集型任务用线程池并发可以大幅提速fromconcurrent.futuresimportThreadPoolExecutordefembed_documents_concurrent(texts,batch_sizeBATCH_SIZE,max_workers5):并发分批处理注意 max_workers 不要超过账号 QPS 上限batcheslist(batched(texts,batch_size))results[None]*len(batches)defwork(i,batch):respembed_with_retry(batch)vecssorted(resp.output[embeddings],keylambdax:x[text_index])results[i][v[embedding]forvinvecs]# 按批次下标写入保证顺序withThreadPoolExecutor(max_workersmax_workers)aspool:list(pool.map(lambdaargs:work(*args),enumerate(batches)))return[vecforbatch_vecsinresultsforvecinbatch_vecs]注意两个坑max_workers不是越大越好。阿里云账号有 QPS 限制并发太猛会大面积触发限流反而更慢。一般 5~10 起步配合上面的重试机制观察调整。结果按下标写入预分配的列表而不是append。并发完成顺序是乱的直接 append 会导致向量和文本错位。LangChain 场景自己包一层 EmbeddingsLangChain 的DashScopeEmbeddings或langchain_community.embeddings.DashScopeEmbeddings在老版本里不会自动分批直接embed_documents(100条)就会报错。最干净的解法是继承后重写embed_documentsfromlangchain_core.embeddingsimportEmbeddingsfromlangchain_community.embeddingsimportDashScopeEmbeddingsclassBatchedDashScopeEmbeddings(Embeddings):def__init__(self,modeltext-embedding-v4,batch_size10):self.innerDashScopeEmbeddings(modelmodel)self.batch_sizebatch_sizedefembed_documents(self,texts):all_embeddings[]forbatchinbatched(texts,self.batch_size):all_embeddings.extend(self.inner.embed_documents(batch))returnall_embeddingsdefembed_query(self,text):returnself.inner.embed_query(text)# 之后和原来用法完全一样FAISS/Chroma 内部循环时就不会超限了embeddingsBatchedDashScopeEmbeddings(modeltext-embedding-v4,batch_size10)vectorstoreFAISS.from_documents(chunks,embeddings)这样做的好处是对上层完全透明——FAISS.from_documents、Chroma.from_documents这些接口内部会反复调用embed_documents分批逻辑封装在底层上层代码一行不用改。顺便提醒如果你的报错其实发生在Milvus / Zilliz 插入阶段upsert/insert也有批量限制同样的分批函数batched()可以原样复用。大规模离线场景用 Batch 批量调用如果你要处理的是几十万条、不要求实时的数据比如全量重建知识库逐条同步调用又慢又贵。阿里云为 text-embedding-v4 提供了OpenAI 兼容的 Batch 调用把任务提交成文件异步跑完后取结果价格约为实时调用的一半。适用判断很简单实时检索、在线问答 → 同步接口 本文的分批方案离线全量灌库、定期重建索引 → Batch 批量调用省钱总结问题方案单次最多 N 条batched()分批切分batch_size 做成可配置常量结果错位风险按text_index排序并发时按下标写入预分配列表限流报错指数退避重试 随机抖动只重试限流类错误速度太慢ThreadPoolExecutor并发 5~10注意 QPS 上限LangChain 集成继承Embeddings重写embed_documents对上层透明海量离线数据用 Batch 批量调用成本减半核心代码不到 30 行但分批、排序、重试、并发这四个细节一个都不能少——缺了任何一个都可能在数据量上来之后给你一个惊喜。本文分批与重试逻辑已在本地通过模拟 API 实测47 条文档自动分为 20/20/7 三批限流重试正常生效全部结果无超限、无错位。