文本一多就报超出最大行数——阿里云向量模型焊死分批调用，上千条照样一次跑完

本文基于 DashScope SDK（pip install -U dashscope）与 OpenAI SDK 编写，所有代码均在 Python 3.12+ 环境实跑验证。Embedding 模型接口偶有微调，若官方更新请以阿里云百炼文档为准。

做 RAG、语义检索、文档聚类时，第一件事就是「把文本变成向量」。阿里云的通用文本向量模型（text-embedding 系列）效果不错、性价比也高，但它有个坑：单次 API 调用不是无限量喂的，每个请求能处理的「行数」有硬上限。

你兴冲冲把几百条、上千条切好的文本一次性塞进input列表，结果接口直接甩回来一个错误：code非 0、message提示超出单次最大行数。要么你被迫手写循环一条条调（慢到怀疑人生），要么干脆放弃批量。

这篇就把这件事焊死：写一个batch_embed()，把大列表按模型上限切成小块，循环（或并发）逐个批次调用，最后把向量按原顺序拼回去。列表再长，也能一次跑完。

一、先搞清楚：你家模型到底一次能吃几条

很多人栽在这里——默认「API 嘛，列表多长都行」，其实阿里云向量模型每个请求有「最大行数」。而且不同模型上限不一样，不是统一的 20、也不是统一的 10。我先给你一张核实过的对照表，你按自己用的模型填数字。

重点提醒：如果你和我一样用的是text-embedding-v3，上限是 10 条/次，不是 20。网上常说的「20 条」对应的是qwen3.7-text-embedding；「25 条」对应的是老模型 v1/v2。把批次上限写死成一个 magic number 是大忌——一定做成可配置常量，按模型改。

另外两件事必须记住，否则向量白算：

1. 维度一致性：建索引（离线）和线上查询（query）必须用同一个模型 + 同一个dimension，否则两个向量不在同一个空间，算出来的相似度毫无意义。
2. 检索场景用text_type：做问答/检索时，给 query 设text_type="query"、给文档设text_type="document"，官方说能明显提升检索准确率；对称任务（聚类、分类）用默认document即可。

二、最朴素的解法：for 循环切块

别被「分批」吓到，本质就是range(0, len, BATCH)切片 + 循环调用。三步搞定。

第 1 步：装 SDK、配密钥

pipinstall-Udashscope

importos# 把密钥放到环境变量，千万别 hardcoded 进代码# 在 ~/.bashrc 或系统环境变量里：export DASHSCOPE_API_KEY="sk-xxxx"os.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_SIZE=10# 本文以 text-embedding-v3 为例MODEL=TextEmbedding.Models.text_embedding_v3 DIMENSION=1024# 必须和你的向量库字段维度一致dashscope.api_key=os.getenv("DASHSCOPE_API_KEY")

第 3 步：写一个batch_embed

defbatch_embed(texts,model=MODEL,dimension=DIMENSION,batch_size=MAX_BATCH_SIZE):""" 把大列表按 batch_size 切块，逐批调用 Embedding。 返回：(向量列表, 总 token 数) 向量列表长度和输入 texts 完全一致、顺序一致。 """all_embeddings=[]total_tokens=0forstartinrange(0,len(texts),batch_size):batch=texts[start:start+batch_size]# 切片，最后一块自动变短resp=TextEmbedding.call(model=model,input=batch,# 直接传 Python list 即可dimension=dimension,)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_tokens+=resp.usage["total_tokens"]returnall_embeddings,total_tokens

跑一下看看：

texts=[f"这是第{i}条测试文本，用于演示分批向量化。"foriinrange(57)]vectors,tokens=batch_embed(texts)print(len(vectors))# 57，和输入等长print(len(vectors[0]))# 1024，维度正确print(tokens)# 总消耗的 token 数

57 条文本，按batch_size=10会被切成 6 批（10+10+10+10+10+7），循环 6 次就跑完了。就这么简单。

三、进阶：并发 + 重试，又快又稳

朴素循环有个问题：串行。57 条跑 6 批还行，5 万条就是 5000 个网络往返，纯串行能跑到天亮。解决办法是并发调用+失败重试（限流 429、网络抖动太常见了）。

第 1 步：区分「参数错误」和「临时错误」

批次超过上限是参数错误，重试多少次都没用，必须立刻停下来让你改batch_size；而限流、超时是临时错误，才值得重试。先把这两种异常分开：

classBatchSizeError(Exception):"""批次超上限，属于参数错误，不该重试。"""passdef_embed_one_batch(batch,model,dimension):resp=TextEmbedding.call(model=model,input=batch,dimension=dimension)ifresp.status_code!=HTTPStatus.OK:msg=f"{resp.code}{resp.message}"# 命中"超出最大行数"之类，直接抛特定异常终止if"超出"inresp.messageor"exceed"inresp.message.lower():raiseBatchSizeError(msg)raiseRuntimeError(msg)# 其余当作临时错误，交给重试装饰器return[e["embedding"]foreinresp.output["embeddings"]]

第 2 步：用 tenacity 加重试

pipinstalltenacity

fromtenacityimport(retry,stop_after_attempt,wait_exponential,retry_if_exception_type,)@retry(stop=stop_after_attempt(3),# 最多重试 3 次wait=wait_exponential(multiplier=1,min=1,max=8),# 1s、2s、4s 退避retry=retry_if_exception_type(RuntimeError),# 只重试临时错误reraise=True,)def_embed_with_retry(batch,model,dimension):return_embed_one_batch(batch,model,dimension)

第 3 步：用线程池并发

fromconcurrent.futuresimportThreadPoolExecutor,as_completeddefbatch_embed_concurrent(texts,model=MODEL,dimension=DIMENSION,batch_size=MAX_BATCH_SIZE,max_workers=4):""" 并发分批向量化。max_workers 别开太大， 阿里云有 QPS 限流，并发过高反而疯狂 429。一般 2~4 足够。 """# 先切片，记下每块的原始下标，保证结果顺序和输入一致batches=[(i,texts[i:i+batch_size])foriinrange(0,len(texts),batch_size)]buckets=[None]*len(batches)withThreadPoolExecutor(max_workers=max_workers)aspool:future_to_idx={pool.submit(_embed_with_retry,b,model,dimension):ifori,(_,b)inenumerate(batches)}forfutinas_completed(future_to_idx):idx=future_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 client=OpenAI(api_key=os.getenv("DASHSCOPE_API_KEY"),# 百炼的 OpenAI 兼容域名（华北2北京；其他地区见官方文档）base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",)defbatch_embed_openai(texts,model="text-embedding-v3",batch_size=MAX_BATCH_SIZE,dimension=1024):out=[]forstartinrange(0,len(texts),batch_size):batch=texts[start:start+batch_size]resp=client.embeddings.create(model=model,input=batch,# 同样是 listdimensions=dimension,# 仅 v3/v4 支持该参数)out.extend(d.embeddingfordinresp.data)returnout

坑：dimensions参数只有text-embedding-v3/v4支持，v1/v2 不支持，传了会报错。

五、三种写法横向对比

生产环境直接上「并发 + 重试」：既快又不会因为偶发 429 把整批任务搞挂。

六、排坑表

七、核心知识点回顾

1. 阿里云向量模型单次调用有「最大行数」上限，且因模型而异（v3/v4=10、v1/v2=25、qwen3.7=20），必须做成可配置常量。
2. 分批的本质就是range(0, len, BATCH)切片 + 循环调用，再把结果按原顺序拼回去。
3. 维度一致性是铁律：索引和查询必须用同一模型 + 同一维度，否则向量不在同一空间。
4. 区分两类错误：批次超上限是参数错误（立刻停），限流/超时是临时错误（重试）。
5. 并发别贪多：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_SIZE=10# 3. 朴素分批（小数据）vectors,tokens=batch_embed(texts)# 4. 并发分批（生产首选）vectors=batch_embed_concurrent(texts,max_workers=4)# 5. OpenAI 兼容（已有 LangChain 栈）vectors=batch_embed_openai(texts,model="text-embedding-v3")

一句话总结：单次有上限 不等于 不能批量。把大列表切成 小于等于 上限的小块，循环/并发调用，顺序拼回——上千条文本，一次跑完。