阿里云向量模型 Embedding 单次只能处理 20 条文本的解决方案

一、问题背景

在 RAG（Retrieval-Augmented Generation，检索增强生成）项目中，我们需要把文档切分后的 Chunk 转换成向量，并保存到向量数据库（如 Milvus、Chroma）。

完整流程：

flowchart LR
A[文档] –> B[解析]
B –> C[分块 Chunk]
C –> D[Embedding 向量化]
D –> E[向量数据库]
E –> F[检索问答]

在调用阿里云向量模型时发现：单次请求最多只能处理 20 条文本。

二、问题原因

Embedding 模型一次请求处理的数据量存在 Batch 限制。如果一次发送大量文本，例如 1000 个 Chunk 一次调用 Embedding，可能出现：

• 请求超过模型限制
• API 返回 batch size 超限错误
• 请求失败后需要全部重试

云服务限制主要为了控制计算资源、请求稳定性和响应时间。

三、核心解决方案：Batch 分批处理

不要一次提交全部文本，而是按照固定数量切分。例如 100 条文本：

flowchart TD
A[100 条文本] –> B[Batch 1：20条]
A –> C[Batch 2：20条]
A –> D[Batch 3：20条]
A –> E[Batch 4：20条]
A –> F[Batch 5：20条]
B –> G[合并生成的向量结果]
C –> G
D –> G
E –> G
F –> G

四、Python 分批代码示例

def batch_split(data, batch_size=20):
for i in range(0, len(data), batch_size):
yield data[i:i+batch_size]

for batch in batch_split(texts, 20):
vectors = embedding.embed(batch)

通过生成器逐批处理，避免一次加载和发送大量数据。

五、RAG 项目中的应用

文档上传后：

flowchart TD
A[PDF/DOCX] –> B[文本解析]
B –> C[Chunk切分]
C –> D[Batch Embedding]
D –> E[生成向量]
E –> F[写入 Milvus]

例如：1000 个 Chunk 不会一次调用 Embedding，而是每 20 个 Chunk 调用一次。

六、LangChain 中的处理方式

如果直接使用 vectorstore.from_documents()，内部可能一次调用大量文本。建议手动控制：

for i in range(0, len(documents), 20):
batch_docs = documents[i:i+20]
vectors = embedding.embed_documents(
[doc.page_content for doc in batch_docs]
)

然后再保存到向量数据库。

七、增加重试机制

网络请求可能失败，需要增加重试：失败 → 等待 → 再次请求。建议：

• 设置最大重试次数
• 添加等待时间
• 记录失败 Batch

八、动态 Batch 优化

固定 20 条并不一定是最佳方案。因为文本长度不同：短文本 20 条可能很轻松，长文本 20 条可能超过 Token 限制。可以根据 Token 数量动态调整 Batch 大小。

九、企业级 RAG 架构建议

推荐流程：

flowchart TD
A[文档上传] –> B[任务队列]
B –> C[文本切分]
C –> D[Embedding Batch任务]
D –> E[阿里云向量模型]
E –> F[Milvus向量库]

不要上传大量文件后立即全部 Embedding，应通过任务调度控制。

十、总结

阿里云向量模型一次只能接受 20 条文本，是模型接口的 Batch 限制。解决方式：

核心思想：把大任务拆成小任务，让 Embedding 服务稳定、高效运行。