DashScope Embedding 分批处理实战——单批 20 条的完整解决方案

在使用阿里云 DashScope 向量模型（如text-embedding-v2）进行向量化时，你是否遇到过batch size is invalid, it should not be larger than 20的报错？本文从源码层面剖析问题根因，并提供一套完整的分批处理方案，让你的 embedding 流程稳如磐石。

问题现象

当你在项目中调用 DashScope Embedding API 时，突然收到如下报错：

status_code: 400 code: InvalidParameter message: <400> InternalError.Algo.InvalidParameter: Value error, batch size is invalid, it should not be larger than 20.: input.contents

关键信息：batch size is invalid, it should not be larger than 20

这意味着 DashScope Embedding API 对单次请求的文本数量有硬性限制：最多 20 条。

根因分析

为什么会触发这个报错？

在典型的 RAG 项目或知识库构建场景中，我们通常会这样写代码：

fromlangchain.embeddingsimportDashScopeEmbeddingsfromlangchain.vectorstoresimportMilvus# 1. 初始化 embedding 模型embeddings=DashScopeEmbeddings(model="text-embedding-v2",dashscope_api_key="sk-xxxx")# 2. 文档切分后得到大量 chunkdoc_list=[...]# 假设有 100+ 个子块content_list=[doc.page_contentfordocindoc_list]# 3. 一次性调用 embedding → 💥 报错vectors=embeddings.embed_documents(content_list)# 超过 20 条就炸# 4. 一次性写入 Milvus → 💥 同样可能炸vector_store.add_documents(doc_list)# 内部也会调 embedding

两处高危调用：

只要content_list或doc_list的长度超过 20，就会触发400 InvalidParameter。

解决方案：分批处理

Step 1：写一个通用的分批工具函数

fromtypingimportList,Generatordefbatch_split(lst:list,batch_size:int=20)->Generator[List,None,None]:""" 将列表切分为多个小批次 Args: lst: 待切分的原始列表 batch_size: 每批最大数量，默认 20（适配 DashScope 限制） Yields: 每批子列表 Example: >>> items = list(range(45)) >>> for batch in batch_split(items, 20): ... print(f"本批 {len(batch)} 条") ... 本批 20 条 本批 20 条 本批 5 条 """foriinrange(0,len(lst),batch_size):yieldlst[i:i+batch_size]

设计要点：

• 使用Generator惰性产出，不占用额外内存
• 默认batch_size=20，与 DashScope 限制对齐
• 最后一批不足 20 条时自动兜底

Step 2：分批调用 Embedding

fromlangchain.embeddingsimportDashScopeEmbeddingsdefbatch_embed_documents(embeddings:DashScopeEmbeddings,texts:List[str],batch_size:int=20)->List[List[float]]:""" 分批向量化文本，规避 DashScope 单批 20 条限制 Args: embeddings: DashScope Embedding 实例 texts: 待向量化的文本列表 batch_size: 每批数量，默认 20 Returns: 所有文本的向量结果，顺序与输入一致 """all_vectors=[]foridx,batchinenumerate(batch_split(texts,batch_size)):print(f"正在处理第{idx+1}批，共{len(batch)}条...")# 分批调用，每次最多 20 条batch_vectors=embeddings.embed_documents(batch)all_vectors.extend(batch_vectors)# 可选：添加短暂延时，避免 QPS 超限# time.sleep(0.1)returnall_vectors# 使用示例texts=[doc.page_contentfordocindoc_list]# 假设有 100 条vectors=batch_embed_documents(embeddings,texts)print(f"共生成{len(vectors)}个向量，维度{len(vectors[0])}")

关键点：

• 使用batch_split将文本切成 ≤20 的小批
• 每批调用embed_documents()，结果用extend()合并
• 顺序与输入完全一致，不影响后续检索匹配

Step 3：分批写入向量库（Milvus / Chroma / FAISS）

方案 A：手动分批写入 Milvus

fromlangchain.schemaimportDocumentfromlangchain.vectorstoresimportMilvusdefbatch_add_documents(vector_store:Milvus,documents:List[Document],batch_size:int=20)->None:""" 分批将文档写入 Milvus，规避 DashScope embedding 限制 Args: vector_store: Milvus 向量库实例 documents: 待写入的 Document 列表 batch_size: 每批数量，默认 20 """foridx,batchinenumerate(batch_split(documents,batch_size)):print(f"正在写入第{idx+1}批，共{len(batch)}条...")# 分批写入，框架内部会自动调 embeddingvector_store.add_documents(batch)# 使用示例batch_add_documents(vector_store,doc_list)

方案 B：如果你手动计算了向量，直接分批插入

defbatch_insert_with_vectors(vector_store:Milvus,documents:List[Document],vectors:List[List[float]],batch_size:int=20)->None:""" 已知向量，直接分批插入 Milvus """# 确保文档和向量一一对应assertlen(documents)==len(vectors)foridx,(doc_batch,vec_batch)inenumerate(zip(batch_split(documents,batch_size),batch_split(vectors,batch_size))):print(f"正在插入第{idx+1}批，共{len(doc_batch)}条...")# Milvus 的 from_texts 或 add_texts 可以直接传入向量vector_store.add_texts(texts=[d.page_contentfordindoc_batch],metadatas=[d.metadatafordindoc_batch],embeddings=vec_batch)

完整实战代码

将上述方案整合为一个完整的工具类：

""" DashScope Embedding 分批处理工具 解决 batch size > 20 导致的 400 报错 """fromtypingimportList,Generator,Optionalimporttimefromlangchain.embeddings.baseimportEmbeddingsfromlangchain.schemaimportDocumentfromlangchain.vectorstores.baseimportVectorStoreclassBatchEmbeddingProcessor:"""DashScope Embedding 分批处理器"""def__init__(self,embeddings:Embeddings,batch_size:int=20,sleep_interval:Optional[float]=None):""" Args: embeddings: Embedding 模型实例（如 DashScopeEmbeddings） batch_size: 每批最大数量，默认 20 sleep_interval: 每批处理后的休眠秒数（防 QPS 超限） """self.embeddings=embeddings self.batch_size=batch_size self.sleep_interval=sleep_interval@staticmethoddefbatch_split(lst:list,batch_size:int=20)->Generator[List,None,None]:"""切分列表为多个小批次"""foriinrange(0,len(lst),batch_size):yieldlst[i:i+batch_size]defembed_texts(self,texts:List[str])->List[List[float]]:""" 分批向量化文本 Args: texts: 待向量化的文本列表 Returns: 向量列表，顺序与输入一致 """all_vectors=[]total=len(texts)foridx,batchinenumerate(self.batch_split(texts,self.batch_size)):print(f"[Embedding] 批次{idx+1}/{(total-1)//self.batch_size+1}，"f"本批{len(batch)}条")batch_vectors=self.embeddings.embed_documents(batch)all_vectors.extend(batch_vectors)ifself.sleep_interval:time.sleep(self.sleep_interval)returnall_vectorsdefadd_to_vector_store(self,vector_store:VectorStore,documents:List[Document])->None:""" 分批将文档写入向量库 Args: vector_store: 向量库实例（Milvus / Chroma / FAISS） documents: 待写入的 Document 列表 """total=len(documents)foridx,batchinenumerate(self.batch_split(documents,self.batch_size)):print(f"[VectorStore] 批次{idx+1}/{(total-1)//self.batch_size+1}，"f"本批{len(batch)}条")vector_store.add_documents(batch)ifself.sleep_interval:time.sleep(self.sleep_interval)defprocess_documents(self,vector_store:VectorStore,documents:List[Document])->List[List[float]]:""" 一站式处理：向量化 + 写入向量库 Args: vector_store: 目标向量库 documents: 待处理的文档列表 Returns: 所有向量 """texts=[doc.page_contentfordocindocuments]# Step 1: 分批向量化vectors=self.embed_texts(texts)# Step 2: 分批写入（如果向量库支持直接传向量）# 否则让 vector_store.add_documents 内部自行计算self.add_to_vector_store(vector_store,documents)returnvectors# ============ 使用示例 ============if__name__=="__main__":fromlangchain.embeddingsimportDashScopeEmbeddingsfromlangchain.vectorstoresimportMilvus# 初始化embeddings=DashScopeEmbeddings(model="text-embedding-v2",dashscope_api_key="sk-xxxx")vector_store=Milvus(embedding_function=embeddings,connection_args={"host":"localhost","port":"19530"},collection_name="my_docs")# 假设 doc_list 是从文件/网页切分出来的 100+ 个子块doc_list=[...]# List[Document]# 使用分批处理器processor=BatchEmbeddingProcessor(embeddings=embeddings,batch_size=20,# DashScope 限制sleep_interval=0.1# 防 QPS 超限，可选)# 一站式处理vectors=processor.process_documents(vector_store,doc_list)print(f"完成！共处理{len(vectors)}条文档")

方案对比

进阶：兼容其他模型

如果你的项目需要同时支持多个 Embedding 模型，可以进一步封装：

classUnifiedEmbeddingProcessor:"""统一的分批处理器，兼容多种 Embedding 模型"""# 各模型的批量限制BATCH_LIMITS={"dashscope":20,"openai":2048,# OpenAI text-embedding-3 支持大批量"qwen":25,# 通义千问"bge":32,# BGE 模型}def__init__(self,embeddings:Embeddings,model_type:str="dashscope"):self.embeddings=embeddings self.batch_size=self.BATCH_LIMITS.get(model_type,20)

这样无论后端用哪个模型，都能自动适配对应的 batch limit。

常见坑点总结

❌ 坑点 1：忘了给add_documents分批

# 错误：内部会调 embedding，同样触发 20 条限制vector_store.add_documents(doc_list)# doc_list 有 100 条 → 💥

✅ 正确：手动分批或自定义 VectorStore 子类

forbatchinbatch_split(doc_list,20):vector_store.add_documents(batch)

❌ 坑点 2：只分批了 embedding，没分批写入

# 错误：向量化分批了，但写入还是一次性vectors=batch_embed_documents(embeddings,texts)# ✅ 分批了vector_store.add_documents(doc_list)# ❌ 这里又炸了

✅ 正确：两个步骤都分批

processor=BatchEmbeddingProcessor(embeddings,batch_size=20)processor.process_documents(vector_store,doc_list)# ✅ 全程分批

❌ 坑点 3：并发请求导致 QPS 超限

# 错误：如果用了多线程/异步，可能 QPS 超限# DashScope 免费版通常限制 20 QPS

✅ 正确：添加延时或使用限速器

processor=BatchEmbeddingProcessor(embeddings=embeddings,batch_size=20,sleep_interval=0.1# 每批间隔 100ms，控制 QPS < 10)

一句话总结

DashScope Embedding 单批最多 20 条，超过必报 400。写一个batch_split工具函数，在向量化（embed_documents）和写入（add_documents）两个环节都分批处理，即可彻底解决。

如果这篇文章对你有帮助，欢迎点赞、收藏、关注！有问题可以在评论区留言讨论。