文本一多就报超出最大行数——阿里云向量模型焊死分批调用上千条照样一次跑完本文基于 DashScope SDKpip install -U dashscope与 OpenAI SDK 编写所有代码均在 Python 3.12 环境实跑验证。Embedding 模型接口偶有微调若官方更新请以阿里云百炼文档为准。做 RAG、语义检索、文档聚类时第一件事就是「把文本变成向量」。阿里云的通用文本向量模型text-embedding 系列效果不错、性价比也高但它有个坑单次 API 调用不是无限量喂的每个请求能处理的「行数」有硬上限。你兴冲冲把几百条、上千条切好的文本一次性塞进input列表结果接口直接甩回来一个错误code非 0、message提示超出单次最大行数。要么你被迫手写循环一条条调慢到怀疑人生要么干脆放弃批量。这篇就把这件事焊死写一个batch_embed()把大列表按模型上限切成小块循环或并发逐个批次调用最后把向量按原顺序拼回去。列表再长也能一次跑完。一、先搞清楚你家模型到底一次能吃几条很多人栽在这里——默认「API 嘛列表多长都行」其实阿里云向量模型每个请求有「最大行数」。而且不同模型上限不一样不是统一的 20、也不是统一的 10。我先给你一张核实过的对照表你按自己用的模型填数字。模型单次最大行数单行最大 Token可选维度默认粗体text-embedding-v3108,1921024/ 768 / 512 / 256 / 128 / 64text-embedding-v4108,1922048 / 1536 /1024/ 768 / 512 / 256 / 128 / 64text-embedding-v2252,048512 / 768 / 1024text-embedding-v1252,0481536qwen3.7-text-embedding20128,0002560 / 2048 / 1536 /1024/ 768 / 512 / 256重点提醒如果你和我一样用的是text-embedding-v3上限是 10 条/次不是 20。网上常说的「20 条」对应的是qwen3.7-text-embedding「25 条」对应的是老模型 v1/v2。把批次上限写死成一个 magic number 是大忌——一定做成可配置常量按模型改。另外两件事必须记住否则向量白算维度一致性建索引离线和线上查询query必须用同一个模型 同一个dimension否则两个向量不在同一个空间算出来的相似度毫无意义。检索场景用text_type做问答/检索时给 query 设text_typequery、给文档设text_typedocument官方说能明显提升检索准确率对称任务聚类、分类用默认document即可。二、最朴素的解法for 循环切块别被「分批」吓到本质就是range(0, len, BATCH)切片 循环调用。三步搞定。第 1 步装 SDK、配密钥pipinstall-Udashscopeimportos# 把密钥放到环境变量千万别 hardcoded 进代码# 在 ~/.bashrc 或系统环境变量里export DASHSCOPE_API_KEYsk-xxxxos.environ[DASHSCOPE_API_KEY]os.getenv(DASHSCOPE_API_KEY,)密钥优先走环境变量DASHSCOPE_API_KEY不要写死在源码里也别提交到 Git。第 2 步确定批次上限常量fromhttpimportHTTPStatusimportdashscopefromdashscopeimportTextEmbedding# 按你自己用的模型改这个数见第一节对照表# text-embedding-v3 / v4 - 10# text-embedding-v1 / v2 - 25# qwen3.7-text-embedding - 20MAX_BATCH_SIZE10# 本文以 text-embedding-v3 为例MODELTextEmbedding.Models.text_embedding_v3 DIMENSION1024# 必须和你的向量库字段维度一致dashscope.api_keyos.getenv(DASHSCOPE_API_KEY)第 3 步写一个batch_embeddefbatch_embed(texts,modelMODEL,dimensionDIMENSION,batch_sizeMAX_BATCH_SIZE): 把大列表按 batch_size 切块逐批调用 Embedding。 返回(向量列表, 总 token 数) 向量列表长度和输入 texts 完全一致、顺序一致。 all_embeddings[]total_tokens0forstartinrange(0,len(texts),batch_size):batchtexts[start:startbatch_size]# 切片最后一块自动变短respTextEmbedding.call(modelmodel,inputbatch,# 直接传 Python list 即可dimensiondimension,)ifresp.status_code!HTTPStatus.OK:raiseRuntimeError(f第{start//batch_size}批调用失败: f{resp.code}{resp.message})# 接口返回顺序与输入的 batch 顺序一致直接 extendall_embeddings.extend(e[embedding]foreinresp.output[embeddings])total_tokensresp.usage[total_tokens]returnall_embeddings,total_tokens跑一下看看texts[f这是第{i}条测试文本用于演示分批向量化。foriinrange(57)]vectors,tokensbatch_embed(texts)print(len(vectors))# 57和输入等长print(len(vectors[0]))# 1024维度正确print(tokens)# 总消耗的 token 数57 条文本按batch_size10会被切成 6 批10101010107循环 6 次就跑完了。就这么简单。三、进阶并发 重试又快又稳朴素循环有个问题串行。57 条跑 6 批还行5 万条就是 5000 个网络往返纯串行能跑到天亮。解决办法是并发调用失败重试限流 429、网络抖动太常见了。第 1 步区分「参数错误」和「临时错误」批次超过上限是参数错误重试多少次都没用必须立刻停下来让你改batch_size而限流、超时是临时错误才值得重试。先把这两种异常分开classBatchSizeError(Exception):批次超上限属于参数错误不该重试。passdef_embed_one_batch(batch,model,dimension):respTextEmbedding.call(modelmodel,inputbatch,dimensiondimension)ifresp.status_code!HTTPStatus.OK:msgf{resp.code}{resp.message}# 命中超出最大行数之类直接抛特定异常终止if超出inresp.messageorexceedinresp.message.lower():raiseBatchSizeError(msg)raiseRuntimeError(msg)# 其余当作临时错误交给重试装饰器return[e[embedding]foreinresp.output[embeddings]]第 2 步用 tenacity 加重试pipinstalltenacityfromtenacityimport(retry,stop_after_attempt,wait_exponential,retry_if_exception_type,)retry(stopstop_after_attempt(3),# 最多重试 3 次waitwait_exponential(multiplier1,min1,max8),# 1s、2s、4s 退避retryretry_if_exception_type(RuntimeError),# 只重试临时错误reraiseTrue,)def_embed_with_retry(batch,model,dimension):return_embed_one_batch(batch,model,dimension)第 3 步用线程池并发fromconcurrent.futuresimportThreadPoolExecutor,as_completeddefbatch_embed_concurrent(texts,modelMODEL,dimensionDIMENSION,batch_sizeMAX_BATCH_SIZE,max_workers4): 并发分批向量化。max_workers 别开太大 阿里云有 QPS 限流并发过高反而疯狂 429。一般 2~4 足够。 # 先切片记下每块的原始下标保证结果顺序和输入一致batches[(i,texts[i:ibatch_size])foriinrange(0,len(texts),batch_size)]buckets[None]*len(batches)withThreadPoolExecutor(max_workersmax_workers)aspool:future_to_idx{pool.submit(_embed_with_retry,b,model,dimension):ifori,(_,b)inenumerate(batches)}forfutinas_completed(future_to_idx):idxfuture_to_idx[fut]buckets[idx]fut.result()# 用下标回填顺序不乱# 按原始顺序摊平return[embforvec_listinbucketsforembinvec_list]并发版把 5000 次请求压到几十秒而且任一批超上限会立刻抛BatchSizeError终止不会傻乎乎重试。四、如果你用 LangChain / OpenAI 生态兼容写法很多人的 RAG 栈是用langchain OpenAI SDK 搭的。阿里云百炼提供了OpenAI 兼容接口只要把base_url和model改一下原来的client.embeddings.create照用同样要切块。fromopenaiimportOpenAI clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),# 百炼的 OpenAI 兼容域名华北2北京其他地区见官方文档base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,)defbatch_embed_openai(texts,modeltext-embedding-v3,batch_sizeMAX_BATCH_SIZE,dimension1024):out[]forstartinrange(0,len(texts),batch_size):batchtexts[start:startbatch_size]respclient.embeddings.create(modelmodel,inputbatch,# 同样是 listdimensionsdimension,# 仅 v3/v4 支持该参数)out.extend(d.embeddingfordinresp.data)returnout坑dimensions参数只有text-embedding-v3/v4支持v1/v2 不支持传了会报错。五、三种写法横向对比写法速度复杂度容错适用场景朴素循环batch_embed慢串行低无重试单批失败全停数据量小几百条、脚本一次性跑并发 重试batch_embed_concurrent快并发中自动重试限流/超时超上限立刻终止生产首选大批量索引OpenAI 兼容batch_embed_openai慢串行低无重试已有 OpenAI/LangChain 栈不想引 DashScope SDK生产环境直接上「并发 重试」既快又不会因为偶发 429 把整批任务搞挂。六、排坑表现象原因解决接口返回错误message提示超出单次最大行数单批文本数超过模型上限调小batch_size按第一节对照表填v3/v410v1/v225qwen3.720检索时相似度奇低、答非所问建索引和查询用了不同模型/维度索引与 query 必须用同一model 同一dimension高频调用后大量失败触发 QPS 限流429降max_workers、加tenacity退避重试传了dimensions却报错v1/v2 不支持该参数只有 v3/v4 能传dimensions老模型删掉传入空字符串/超长行报错单行有最小/最大约束调用前清洗去空、截断超单行最大 Token的文本检索效果一般没区分 query / document检索场景给 query 设text_typequery、文档设text_typedocument七、核心知识点回顾阿里云向量模型单次调用有「最大行数」上限且因模型而异v3/v410、v1/v225、qwen3.720必须做成可配置常量。分批的本质就是range(0, len, BATCH)切片 循环调用再把结果按原顺序拼回去。维度一致性是铁律索引和查询必须用同一模型 同一维度否则向量不在同一空间。区分两类错误批次超上限是参数错误立刻停限流/超时是临时错误重试。并发别贪多max_workers开 2~4 即可开太大反而疯狂触发限流。八、速查表# 1. 装包# pip install -U dashscope tenacity# 2. 批次上限按模型改# text-embedding-v3 / v4 - 10# text-embedding-v1 / v2 - 25# qwen3.7-text-embedding - 20MAX_BATCH_SIZE10# 3. 朴素分批小数据vectors,tokensbatch_embed(texts)# 4. 并发分批生产首选vectorsbatch_embed_concurrent(texts,max_workers4)# 5. OpenAI 兼容已有 LangChain 栈vectorsbatch_embed_openai(texts,modeltext-embedding-v3)一句话总结单次有上限 不等于 不能批量。把大列表切成 小于等于 上限的小块循环/并发调用顺序拼回——上千条文本一次跑完。 文本一多就报超出最大行数——阿里云向量模型焊死分批调用，上千条照样一次跑完
文本一多就报超出最大行数——阿里云向量模型焊死分批调用上千条照样一次跑完本文基于 DashScope SDKpip install -U dashscope与 OpenAI SDK 编写所有代码均在 Python 3.12 环境实跑验证。Embedding 模型接口偶有微调若官方更新请以阿里云百炼文档为准。做 RAG、语义检索、文档聚类时第一件事就是「把文本变成向量」。阿里云的通用文本向量模型text-embedding 系列效果不错、性价比也高但它有个坑单次 API 调用不是无限量喂的每个请求能处理的「行数」有硬上限。你兴冲冲把几百条、上千条切好的文本一次性塞进input列表结果接口直接甩回来一个错误code非 0、message提示超出单次最大行数。要么你被迫手写循环一条条调慢到怀疑人生要么干脆放弃批量。这篇就把这件事焊死写一个batch_embed()把大列表按模型上限切成小块循环或并发逐个批次调用最后把向量按原顺序拼回去。列表再长也能一次跑完。一、先搞清楚你家模型到底一次能吃几条很多人栽在这里——默认「API 嘛列表多长都行」其实阿里云向量模型每个请求有「最大行数」。而且不同模型上限不一样不是统一的 20、也不是统一的 10。我先给你一张核实过的对照表你按自己用的模型填数字。模型单次最大行数单行最大 Token可选维度默认粗体text-embedding-v3108,1921024/ 768 / 512 / 256 / 128 / 64text-embedding-v4108,1922048 / 1536 /1024/ 768 / 512 / 256 / 128 / 64text-embedding-v2252,048512 / 768 / 1024text-embedding-v1252,0481536qwen3.7-text-embedding20128,0002560 / 2048 / 1536 /1024/ 768 / 512 / 256重点提醒如果你和我一样用的是text-embedding-v3上限是 10 条/次不是 20。网上常说的「20 条」对应的是qwen3.7-text-embedding「25 条」对应的是老模型 v1/v2。把批次上限写死成一个 magic number 是大忌——一定做成可配置常量按模型改。另外两件事必须记住否则向量白算维度一致性建索引离线和线上查询query必须用同一个模型 同一个dimension否则两个向量不在同一个空间算出来的相似度毫无意义。检索场景用text_type做问答/检索时给 query 设text_typequery、给文档设text_typedocument官方说能明显提升检索准确率对称任务聚类、分类用默认document即可。二、最朴素的解法for 循环切块别被「分批」吓到本质就是range(0, len, BATCH)切片 循环调用。三步搞定。第 1 步装 SDK、配密钥pipinstall-Udashscopeimportos# 把密钥放到环境变量千万别 hardcoded 进代码# 在 ~/.bashrc 或系统环境变量里export DASHSCOPE_API_KEYsk-xxxxos.environ[DASHSCOPE_API_KEY]os.getenv(DASHSCOPE_API_KEY,)密钥优先走环境变量DASHSCOPE_API_KEY不要写死在源码里也别提交到 Git。第 2 步确定批次上限常量fromhttpimportHTTPStatusimportdashscopefromdashscopeimportTextEmbedding# 按你自己用的模型改这个数见第一节对照表# text-embedding-v3 / v4 - 10# text-embedding-v1 / v2 - 25# qwen3.7-text-embedding - 20MAX_BATCH_SIZE10# 本文以 text-embedding-v3 为例MODELTextEmbedding.Models.text_embedding_v3 DIMENSION1024# 必须和你的向量库字段维度一致dashscope.api_keyos.getenv(DASHSCOPE_API_KEY)第 3 步写一个batch_embeddefbatch_embed(texts,modelMODEL,dimensionDIMENSION,batch_sizeMAX_BATCH_SIZE): 把大列表按 batch_size 切块逐批调用 Embedding。 返回(向量列表, 总 token 数) 向量列表长度和输入 texts 完全一致、顺序一致。 all_embeddings[]total_tokens0forstartinrange(0,len(texts),batch_size):batchtexts[start:startbatch_size]# 切片最后一块自动变短respTextEmbedding.call(modelmodel,inputbatch,# 直接传 Python list 即可dimensiondimension,)ifresp.status_code!HTTPStatus.OK:raiseRuntimeError(f第{start//batch_size}批调用失败: f{resp.code}{resp.message})# 接口返回顺序与输入的 batch 顺序一致直接 extendall_embeddings.extend(e[embedding]foreinresp.output[embeddings])total_tokensresp.usage[total_tokens]returnall_embeddings,total_tokens跑一下看看texts[f这是第{i}条测试文本用于演示分批向量化。foriinrange(57)]vectors,tokensbatch_embed(texts)print(len(vectors))# 57和输入等长print(len(vectors[0]))# 1024维度正确print(tokens)# 总消耗的 token 数57 条文本按batch_size10会被切成 6 批10101010107循环 6 次就跑完了。就这么简单。三、进阶并发 重试又快又稳朴素循环有个问题串行。57 条跑 6 批还行5 万条就是 5000 个网络往返纯串行能跑到天亮。解决办法是并发调用失败重试限流 429、网络抖动太常见了。第 1 步区分「参数错误」和「临时错误」批次超过上限是参数错误重试多少次都没用必须立刻停下来让你改batch_size而限流、超时是临时错误才值得重试。先把这两种异常分开classBatchSizeError(Exception):批次超上限属于参数错误不该重试。passdef_embed_one_batch(batch,model,dimension):respTextEmbedding.call(modelmodel,inputbatch,dimensiondimension)ifresp.status_code!HTTPStatus.OK:msgf{resp.code}{resp.message}# 命中超出最大行数之类直接抛特定异常终止if超出inresp.messageorexceedinresp.message.lower():raiseBatchSizeError(msg)raiseRuntimeError(msg)# 其余当作临时错误交给重试装饰器return[e[embedding]foreinresp.output[embeddings]]第 2 步用 tenacity 加重试pipinstalltenacityfromtenacityimport(retry,stop_after_attempt,wait_exponential,retry_if_exception_type,)retry(stopstop_after_attempt(3),# 最多重试 3 次waitwait_exponential(multiplier1,min1,max8),# 1s、2s、4s 退避retryretry_if_exception_type(RuntimeError),# 只重试临时错误reraiseTrue,)def_embed_with_retry(batch,model,dimension):return_embed_one_batch(batch,model,dimension)第 3 步用线程池并发fromconcurrent.futuresimportThreadPoolExecutor,as_completeddefbatch_embed_concurrent(texts,modelMODEL,dimensionDIMENSION,batch_sizeMAX_BATCH_SIZE,max_workers4): 并发分批向量化。max_workers 别开太大 阿里云有 QPS 限流并发过高反而疯狂 429。一般 2~4 足够。 # 先切片记下每块的原始下标保证结果顺序和输入一致batches[(i,texts[i:ibatch_size])foriinrange(0,len(texts),batch_size)]buckets[None]*len(batches)withThreadPoolExecutor(max_workersmax_workers)aspool:future_to_idx{pool.submit(_embed_with_retry,b,model,dimension):ifori,(_,b)inenumerate(batches)}forfutinas_completed(future_to_idx):idxfuture_to_idx[fut]buckets[idx]fut.result()# 用下标回填顺序不乱# 按原始顺序摊平return[embforvec_listinbucketsforembinvec_list]并发版把 5000 次请求压到几十秒而且任一批超上限会立刻抛BatchSizeError终止不会傻乎乎重试。四、如果你用 LangChain / OpenAI 生态兼容写法很多人的 RAG 栈是用langchain OpenAI SDK 搭的。阿里云百炼提供了OpenAI 兼容接口只要把base_url和model改一下原来的client.embeddings.create照用同样要切块。fromopenaiimportOpenAI clientOpenAI(api_keyos.getenv(DASHSCOPE_API_KEY),# 百炼的 OpenAI 兼容域名华北2北京其他地区见官方文档base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1,)defbatch_embed_openai(texts,modeltext-embedding-v3,batch_sizeMAX_BATCH_SIZE,dimension1024):out[]forstartinrange(0,len(texts),batch_size):batchtexts[start:startbatch_size]respclient.embeddings.create(modelmodel,inputbatch,# 同样是 listdimensionsdimension,# 仅 v3/v4 支持该参数)out.extend(d.embeddingfordinresp.data)returnout坑dimensions参数只有text-embedding-v3/v4支持v1/v2 不支持传了会报错。五、三种写法横向对比写法速度复杂度容错适用场景朴素循环batch_embed慢串行低无重试单批失败全停数据量小几百条、脚本一次性跑并发 重试batch_embed_concurrent快并发中自动重试限流/超时超上限立刻终止生产首选大批量索引OpenAI 兼容batch_embed_openai慢串行低无重试已有 OpenAI/LangChain 栈不想引 DashScope SDK生产环境直接上「并发 重试」既快又不会因为偶发 429 把整批任务搞挂。六、排坑表现象原因解决接口返回错误message提示超出单次最大行数单批文本数超过模型上限调小batch_size按第一节对照表填v3/v410v1/v225qwen3.720检索时相似度奇低、答非所问建索引和查询用了不同模型/维度索引与 query 必须用同一model 同一dimension高频调用后大量失败触发 QPS 限流429降max_workers、加tenacity退避重试传了dimensions却报错v1/v2 不支持该参数只有 v3/v4 能传dimensions老模型删掉传入空字符串/超长行报错单行有最小/最大约束调用前清洗去空、截断超单行最大 Token的文本检索效果一般没区分 query / document检索场景给 query 设text_typequery、文档设text_typedocument七、核心知识点回顾阿里云向量模型单次调用有「最大行数」上限且因模型而异v3/v410、v1/v225、qwen3.720必须做成可配置常量。分批的本质就是range(0, len, BATCH)切片 循环调用再把结果按原顺序拼回去。维度一致性是铁律索引和查询必须用同一模型 同一维度否则向量不在同一空间。区分两类错误批次超上限是参数错误立刻停限流/超时是临时错误重试。并发别贪多max_workers开 2~4 即可开太大反而疯狂触发限流。八、速查表# 1. 装包# pip install -U dashscope tenacity# 2. 批次上限按模型改# text-embedding-v3 / v4 - 10# text-embedding-v1 / v2 - 25# qwen3.7-text-embedding - 20MAX_BATCH_SIZE10# 3. 朴素分批小数据vectors,tokensbatch_embed(texts)# 4. 并发分批生产首选vectorsbatch_embed_concurrent(texts,max_workers4)# 5. OpenAI 兼容已有 LangChain 栈vectorsbatch_embed_openai(texts,modeltext-embedding-v3)一句话总结单次有上限 不等于 不能批量。把大列表切成 小于等于 上限的小块循环/并发调用顺序拼回——上千条文本一次跑完。