摘要在 RAG 项目对接阿里云向量模型时会遇到接口单批次最大 20 条限制一次性传入上百个文档块直接报错。本文结合 FastAPIMilvus 真实业务代码讲解分批切片方案改造原有add_chunks入库逻辑解决大文档批量向量化超限问题同时给出坑点与注意事项。背景阿里云 Embedding 服务限制单次调用embed_documents最多传入 20 条文本如果一次性传入几十上百个 chunk直接抛出参数超限异常。一、原代码问题原始逻辑把全部 chunk 一次性组装成doc_list完整列表一次性调用vector_store.add_documents()。# 问题当文档分块数量 20阿里云embedding接口报错 async def add_chunks( self, document: PolicyDocument, chunks: list[DocumentChunk] ) - None: embeddingget_embeddings() content_list[ chunk.content for chunk in chunks ] vectorsawait asyncio.to_thread(embedding.embed_documents,content_list) texts [chunk.content for chunk in chunks] chunk_ids [chunk.id for chunk in chunks] metadatas[] for chunk in chunks: if document.effective_date: effective_datedocument.effective_date.isoformat() else: effective_date metadatas.append( { document_id: document.id, filename: document.filename, topic: document.topic or , effective_date: effective_date, section: chunk.section or , page: chunk.page or 0, parent_id: chunk.parent_id or , } ) doc_list:list[Document][] for chunk_id,content,metadata in zip(chunk_ids,texts,metadatas): docDocument(page_contentcontent,metadatametadata,idchunk_id) doc_list.append(doc) # 一次性全部写入数量超过20触发阿里云接口报错 vector_storeget_vector_store() vector_store.add_documents(doc_list)故障现象小文档分块少一切正常PDF 长文档分块 50‑200 个调用embed_documents直接返回错误batch size exceed limit, max 20。底层原因LangChain 的add_documents内部会把传入的所有文本一次性丢给 Embedding 接口阿里云侧限制单批最大 20 条。二、第一版改造示例中每 10 条一批提交示例代码设置阈值为 10生产环境建议改成20贴合阿里云上限。async def add_chunks( self, document: PolicyDocument, chunks: list[DocumentChunk] ) - None: texts [chunk.content for chunk in chunks] chunk_ids [chunk.id for chunk in chunks] metadatas[] for chunk in chunks: if document.effective_date: effective_datedocument.effective_date.isoformat() else: effective_date metadatas.append( { document_id: document.id, filename: document.filename, topic: document.topic or , effective_date: effective_date, section: chunk.section or , page: chunk.page or 0, parent_id: chunk.parent_id or , } ) doc_list:list[Document][] vector_store get_vector_store() # 遍历组装Document达到批次阈值就执行入库 for chunk_id,content,metadata in zip(chunk_ids,texts,metadatas): docDocument(page_contentcontent,metadatametadata,idchunk_id) doc_list.append(doc) # 每满10条就写入向量库 写入后列表清空 if len(doc_list)10: vector_store.add_documents(doc_list) doc_list[]⚠这里存在一个严重 BUG循环结束后如果剩余不足 10 条的数据不会执行add_documents剩余 chunk 直接丢失例如总共有 23 条 chunk10 条写入、10 条写入剩下 3 条留在doc_list没有入库。修复补写剩余批次在 for 循环结束之后必须判断列表不为空执行最后一批写入for chunk_id,content,metadata in zip(chunk_ids,texts,metadatas): docDocument(page_contentcontent,metadatametadata,idchunk_id) doc_list.append(doc) if len(doc_list)10: vector_store.add_documents(doc_list) doc_list[] # 关键处理尾部不足批次的数据 if doc_list: vector_store.add_documents(doc_list)三、更优雅通用写法按批次切片推荐生产使用不用在循环内判断长度直接对 chunk 列表做切片batch_size20对齐阿里云上限。async def add_chunks( self, document: PolicyDocument, chunks: list[DocumentChunk] ) - None: BATCH_SIZE 20 #阿里云embedding最大单批20 texts [chunk.content for chunk in chunks] chunk_ids [chunk.id for chunk in chunks] metadatas[] for chunk in chunks: effective_date document.effective_date.isoformat() if document.effective_date else metadatas.append({ document_id: document.id, filename: document.filename, topic: document.topic or , effective_date: effective_date, section: chunk.section or , page: chunk.page or 0, parent_id: chunk.parent_id or , }) vector_store get_vector_store() # 切片分批 for i in range(0, len(chunks), BATCH_SIZE): slice_ids chunk_ids[i:iBATCH_SIZE] slice_texts texts[i:iBATCH_SIZE] slice_metas metadatas[i:iBATCH_SIZE] batch_docs [] for cid, txt, meta in zip(slice_ids, slice_texts, slice_metas): batch_docs.append(Document(page_contenttxt, metadatameta, idcid)) vector_store.add_documents(batch_docs)四、重点踩坑总结尾部残留数据问题循环内满 N 条才入库循环结束一定要处理剩下不足一批的数据否则丢数据。batch_size 取值阿里云限制 20不要写超过 20可以设置 15‑20 留余量。异步注意add_documents底层 Embedding 是同步 IOFastAPI 中建议套asyncio.to_thread避免阻塞事件循环。异常处理补充生产环境建议加上 try‑except单批失败做好日志区分哪些 chunk 入库成功哪些失败。不要自己手动调用 embed_documentsLangChain 的add_documents内部已经调用 embedding手动调用会重复向量化浪费 token。五、扩展增加异常日志简单示例for i in range(0, len(chunks), BATCH_SIZE): try: #组装batch_docs vector_store.add_documents(batch_docs) print(f批次 {i}‑{iBATCH_SIZE} 入库成功) except Exception as e: print(f批次 {i} 入库失败{str(e)}) raise e六、总结对接第三方 Embedding 服务几乎都会遇到单批数量限制。原始代码一次性全量提交短文档没问题长文档直接报错简单循环判断长度要注意尾部剩余数据推荐使用range切片方式分批代码可读性高不容易漏数据。