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_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ，创建后复制保存，它只显示一次。如果你还没注册，官网入口在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，注册后进控制台即可。
接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面写了 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:pass@localhost: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_KEY="sk-你的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_url=settings["api"]["base_url"], api_key=os.environ[settings["api"]["api_key_env"]], timeout=settings["api"]["timeout_seconds"], ) qdrant = QdrantClient(url=cfg["target"]["url"]) def load_batch(offset, limit): # 这里换成你的真实查询，示例用伪代码 import psycopg2 conn = psycopg2.connect(cfg["source"]["dsn"]) cur = conn.cursor() cur.execute( f"SELECT {cfg['source']['id_column']}, {cfg['source']['text_column']} " f"FROM {cfg['source']['table']} " f"WHERE {cfg['source']['filter']} " f"ORDER 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( model=settings["embedding"]["model"], input=texts, dimensions=settings["embedding"]["dimensions"], ) return [d.embedding for d in resp.data] def upsert_vectors(ids, vectors): points = [ PointStruct(id=i, vector=v, payload={"version": settings["embedding"]["version_tag"]}) for i, v in zip(ids, vectors) ] qdrant.upsert(collection_name=cfg["target"]["collection"], points=points) 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(f"batch offset={offset} attempt={attempt} error={e}") time.sleep(2 ** attempt) else: raise RuntimeError(f"batch offset={offset} 重试耗尽") offset += batch_size time.sleep(cfg["batch"]["sleep_seconds"]) print(f"已回填 {offset} 条") if __name__ == "__main__": main()几个关键点。第一，embedding 调用走的是 client.embeddings.create，base_url 指向 TaoToken，模型名写在 settings.json，换模型只改这一处。第二，批大小 64 是保守值，你可以根据 API 限流调整，但别一上来就 1000，容易触发限流还不好定位。第三，payload 里打了 version 标记，回填完成后可以按版本过滤，方便灰度切换。第四，重试用了指数退避，网络抖动或临时限流都能扛过去。
如果你还想在回填前用对话模型清洗文本，比如去掉简历里的乱码，可以复用同一个 client：
def clean_text(text): resp = client.chat.completions.create( model=settings["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_KEY="sk-你的key" python backfill.py正常输出类似：
已回填 10 条 已回填 20 条 已回填 30 条第三步，核对向量库。用 Qdrant 的查询接口确认写入：
from qdrant_client import QdrantClient qdrant = QdrantClient(url="http://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_name="resumes_v2", query_vector=query_vec, limit=3) for h in hits: print(h.id, h.score)如果 top 结果的 score 明显高于随机，说明新向量空间工作正常。如果 score 都在 0.1 以下，大概率是维度或模型配错了，回去检查 settings.json 的 dimensions 和 collection 的 vector_size 是否一致。
第五步，核对版本标记。查一下 payload：
points = qdrant.retrieve(collection_name="resumes_v2", ids=[1, 2, 3]) for p in points: print(p.id, p.payload)应该看到{"version": "v2"}。这样回填完成后，你可以用 filter 只搜 v2 向量，旧向量保留做回滚兜底。
6. 本篇常见错排查
回填任务跑挂，八成是下面几个原因。
报错 401 Unauthorized：Key 没注入或写错。检查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，要么在请求里传 dimensions=1536 做降维。两者必须一致。
报错 429 Too Many Requests：批太大或没休眠。把 batch.size 降到 32，sleep_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_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 。通道没问题再回头查脚本。
7. 把回填做成可重复的工程动作
回填任务最怕的是一次性脚本：跑完就扔，下次换模型又从头写。更好的做法是把它固化成可重复的流程——配置外置、Key 统一、断点续跑、版本标记。这样下次从 text-embedding-3-large 换到更新的模型时，你只需要改 settings.json 里的模型名和 dimensions，重建一个 collection，重跑同一个脚本。
如果你回填之后还要长期跑 coding agent 来维护这套批处理代码，可以看看 Coding Plan，地址在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite ，它和 embedding 调用共用同一个 Key 通道，不用再单独配一套。接入细节和参数说明都在文档里：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。Key 管理和新建入口在控制台：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite 。
最后留一个实操建议：回填前先备份原 collection 的快照，Qdrant 支持 snapshot，一条命令的事。回填过程中保持旧 collection 可读，新 collection 写满并验证通过后再切流量。这样即使回填中途出问题，搜索服务也不受影响，回滚就是切回旧 collection 的事。
