1. 回填任务到底在补什么从一次 embedding 模型切换说起回填任务Backfill Task说白了就是对历史已有的数据补跑某个新增的处理流程。放到 embedding 场景里它有个更具体的名字——re-embed backfill也就是把存量数据重新拉出来用新的 embedding 模型再生成一遍向量覆盖或补充到向量数据库里。为什么这件事绕不开因为向量维度是绑定 collection 的。Qdrant、Milvus、pgvector 这类向量库在建 collection 时就把维度写死了比如 1536 维。如果你原来用 text-embedding-ada-002 生成 1536 维向量后来想换成 text-embedding-3-large 的 3072 维两者根本没法塞进同一个 collection 做相似度搜索——维度和语义空间都不一样硬混进去匹配直接失效。所以 embedding 模型锁死在部署级是合理设计真正的坑在于系统只在简历上传时生成向量没有批量回填能力。等你要换模型时几十万份历史数据全得重新跑一遍。这个场景的工程难点不在算法而在调用链。回填脚本要遍历存量数据、分批调用 embedding API、写回向量库中间可能还夹着清洗、去重、版本标记。如果每个环节各用各的 Key配置散落在环境变量、脚本参数、CI 配置里排查一次 401 就得翻半天。我试过把这类批处理统一走一个 API 通道Key 集中管理脚本只认一个 base_url 和一个 token配置混乱的问题基本消失。下面就把这套流程拆成可复制的步骤。2. 前置准备用 TaoToken 统一 Key 与 API 通道回填任务的第一原则是批处理脚本不要自己管一堆厂商 Key。你可能有 embedding 模型、偶尔还要调对话模型做数据清洗、再顺手跑个 coding agent 改脚本如果每个都单独配 Key回填跑到一半某个 Key 过期整批任务就卡住了。TaoToken 在这里的角色是统一入口一个 API Key 走一个 base_url就能覆盖模型对话、embedding、coding plan 等调用。对回填任务来说最直接的好处是脚本里只需要维护一份凭证换模型时改的是模型名不是 Key。你需要先拿到 Key。打开控制台创建 API Key地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 创建后复制保存它只显示一次。如果你还没注册官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后进控制台即可。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面写了 base_url 和兼容格式。API 基础地址是 https://taotoken.net/api 注意这个地址不带 UTM 参数直接填进配置即可。它兼容 OpenAI 风格的接口所以你的批处理脚本如果原来用的是 openai 的 Python SDK基本只需要改 base_url 和 api_key 两行。这里有个前置判断回填任务适合谁如果你只是几十条数据手动跑一次脚本就行但只要是上千条以上、或者未来还会反复换模型就值得把 Key 和通道统一起来否则每次回填都是一次配置考古。3. 可复制配置config.toml 与 settings.json 骨架回填脚本的配置我习惯拆成两层一层是项目级的 config.toml管数据源、向量库、批大小一层是工具级的 settings.json管 API 通道和模型名。这样换模型只动 settings.json换数据表只动 config.toml。先看 config.toml# config.toml —— 回填任务项目配置 [source] # 存量数据来源示例用 PostgreSQL dsn postgresql://user:passlocalhost:5432/app table resumes id_column id text_column content # 只回填还没打过新版本标记的行 filter embedding_version IS NULL OR embedding_version v2 [target] # 向量库示例用 Qdrant url http://localhost:6333 collection resumes_v2 vector_size 3072 distance Cosine [batch] # 每批处理条数别一次拉太多 size 64 # 批间休眠秒数给 API 留余量 sleep_seconds 0.5 # 失败重试次数 max_retries 3 [checkpoint] # 断点续跑用记录已处理的 id file ./backfill_checkpoint.json再看 settings.json这里放 TaoToken 通道和模型{ api: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, timeout_seconds: 60 }, embedding: { model: text-embedding-3-large, version_tag: v2, dimensions: 3072 }, chat: { model: gpt-4o-mini, purpose: clean_text } }注意 api_key 我没有写死在文件里而是用环境变量名引用。回填脚本启动前 export 一下export TAOTOKEN_API_KEYsk-你的key这样配置文件可以进版本库Key 不会泄露。如果你在 CI 里跑回填把 Key 配成 secret 注入同名环境变量即可。4. 批处理脚本分批拉取、调用 embedding、写回向量库配置就绪后核心脚本逻辑分四步读一批源数据、调 embedding、写向量库、记断点。下面是一个可运行的 Python 骨架用 openai SDK 和 qdrant-client。import os import json import time import tomllib from openai import OpenAI from qdrant_client import QdrantClient from qdrant_client.models import PointStruct # 读配置 with open(config.toml, rb) as f: cfg tomllib.load(f) with open(settings.json, r) as f: settings json.load(f) # 初始化客户端统一走 TaoToken 通道 client OpenAI( base_urlsettings[api][base_url], api_keyos.environ[settings[api][api_key_env]], timeoutsettings[api][timeout_seconds], ) qdrant QdrantClient(urlcfg[target][url]) def load_batch(offset, limit): # 这里换成你的真实查询示例用伪代码 import psycopg2 conn psycopg2.connect(cfg[source][dsn]) cur conn.cursor() cur.execute( fSELECT {cfg[source][id_column]}, {cfg[source][text_column]} fFROM {cfg[source][table]} fWHERE {cfg[source][filter]} fORDER BY {cfg[source][id_column]} LIMIT %s OFFSET %s, (limit, offset), ) rows cur.fetchall() cur.close() conn.close() return rows def embed_texts(texts): resp client.embeddings.create( modelsettings[embedding][model], inputtexts, dimensionssettings[embedding][dimensions], ) return [d.embedding for d in resp.data] def upsert_vectors(ids, vectors): points [ PointStruct(idi, vectorv, payload{version: settings[embedding][version_tag]}) for i, v in zip(ids, vectors) ] qdrant.upsert(collection_namecfg[target][collection], pointspoints) def main(): offset 0 batch_size cfg[batch][size] while True: rows load_batch(offset, batch_size) if not rows: break ids [r[0] for r in rows] texts [r[1] for r in rows] for attempt in range(cfg[batch][max_retries]): try: vectors embed_texts(texts) upsert_vectors(ids, vectors) break except Exception as e: print(fbatch offset{offset} attempt{attempt} error{e}) time.sleep(2 ** attempt) else: raise RuntimeError(fbatch offset{offset} 重试耗尽) offset batch_size time.sleep(cfg[batch][sleep_seconds]) print(f已回填 {offset} 条) if __name__ __main__: main()几个关键点。第一embedding 调用走的是 client.embeddings.createbase_url 指向 TaoToken模型名写在 settings.json换模型只改这一处。第二批大小 64 是保守值你可以根据 API 限流调整但别一上来就 1000容易触发限流还不好定位。第三payload 里打了 version 标记回填完成后可以按版本过滤方便灰度切换。第四重试用了指数退避网络抖动或临时限流都能扛过去。如果你还想在回填前用对话模型清洗文本比如去掉简历里的乱码可以复用同一个 clientdef clean_text(text): resp client.chat.completions.create( modelsettings[chat][model], messages[{role: user, content: f清理以下文本的乱码只返回正文\n{text}}], ) return resp.choices[0].message.content同一个 Key、同一个通道不用再配第二套凭证。5. 验证请求小批量回填与结果核对别一上来就跑全量。先取 100 条做小批量验证确认链路通了再放开。第一步把 config.toml 的 filter 临时改成只选少量数据比如加AND id 1000或者直接把 batch.size 设成 10跑几轮看输出。第二步跑脚本观察日志export TAOTOKEN_API_KEYsk-你的key python backfill.py正常输出类似已回填 10 条 已回填 20 条 已回填 30 条第三步核对向量库。用 Qdrant 的查询接口确认写入from qdrant_client import QdrantClient qdrant QdrantClient(urlhttp://localhost:6333) info qdrant.get_collection(resumes_v2) print(info.points_count) # 应等于已回填条数第四步做一次相似度搜索验证语义空间正确。取一条已知文本生成向量后搜 top3看返回的 id 是否语义相近query_vec embed_texts([五年后端开发经验熟悉 Go 和 Kubernetes])[0] hits qdrant.search(collection_nameresumes_v2, query_vectorquery_vec, limit3) for h in hits: print(h.id, h.score)如果 top 结果的 score 明显高于随机说明新向量空间工作正常。如果 score 都在 0.1 以下大概率是维度或模型配错了回去检查 settings.json 的 dimensions 和 collection 的 vector_size 是否一致。第五步核对版本标记。查一下 payloadpoints qdrant.retrieve(collection_nameresumes_v2, ids[1, 2, 3]) for p in points: print(p.id, p.payload)应该看到{version: v2}。这样回填完成后你可以用 filter 只搜 v2 向量旧向量保留做回滚兜底。6. 本篇常见错排查回填任务跑挂八成是下面几个原因。报错 401 UnauthorizedKey 没注入或写错。检查echo $TAOTOKEN_API_KEY是否有值settings.json 里的 api_key_env 名字是否和环境变量一致。注意别把 Key 直接写进 config.toml 提交到仓库。报错 400 dimensions mismatch模型返回的维度和 collection 的 vector_size 对不上。text-embedding-3-large 默认 3072 维如果你 collection 建的是 1536要么重建 collection要么在请求里传 dimensions1536 做降维。两者必须一致。报错 429 Too Many Requests批太大或没休眠。把 batch.size 降到 32sleep_seconds 提到 1重试次数保留 3 次。回填是后台任务慢一点没关系别把 API 打爆。写入成功但搜索质量差检查是不是新旧向量混在同一个 collection。回填期间建议用独立 collection比如 resumes_v2回填完再切流量。如果必须同 collection靠 payload 的 version 字段过滤别让两种向量参与同一次搜索。断点续跑重复处理checkpoint 文件没更新。脚本里每批成功后写一次 offset重启时先读 checkpoint 跳过已处理部分。上面骨架为了简洁没展开你可以在 main 循环里加json.dump({offset: offset}, open(cfg[checkpoint][file], w))。文本为空导致 embedding 报错源数据里有空字符串。在 load_batch 后加一层过滤texts [t if t.strip() else 空 for t in texts]或者直接跳过空行并记录 id。排障时如果拿不准是通道问题还是脚本问题可以先用模型对话页发一条测试请求确认 Key 和通道本身是通的地址在 https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。通道没问题再回头查脚本。7. 把回填做成可重复的工程动作回填任务最怕的是一次性脚本跑完就扔下次换模型又从头写。更好的做法是把它固化成可重复的流程——配置外置、Key 统一、断点续跑、版本标记。这样下次从 text-embedding-3-large 换到更新的模型时你只需要改 settings.json 里的模型名和 dimensions重建一个 collection重跑同一个脚本。如果你回填之后还要长期跑 coding agent 来维护这套批处理代码可以看看 Coding Plan地址在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它和 embedding 调用共用同一个 Key 通道不用再单独配一套。接入细节和参数说明都在文档里https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。Key 管理和新建入口在控制台https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。最后留一个实操建议回填前先备份原 collection 的快照Qdrant 支持 snapshot一条命令的事。回填过程中保持旧 collection 可读新 collection 写满并验证通过后再切流量。这样即使回填中途出问题搜索服务也不受影响回滚就是切回旧 collection 的事。