Python后端AI专题19：批量 Embedding 与断点续建：第三批失败后从哪里重来

Python后端AI专题19批量 Embedding 与断点续建第三批失败后从哪里重来一份 500 页手册产生 1800 个 chunk。前两批向量已经写入第三批遇到供应商 503。若任务从头重跑不仅浪费 Token还可能写出重复记录若直接把状态改成 ready又会留下残缺索引。本篇用持久化next_batch让失败可恢复。overlap 边界实验完整答案我们用自然的售后长句构造边界目标短语起始位置是 251长度 14因此在 256 硬边界被切开。检查脚本核心ANSWER定制商品不适用七天无理由退款definspect(overlap:int)-dict[str,object]:chunkschunk_blocks([ParsedBlock(TEXT,0)],chunk_size256,overlapoverlap)totalsum(len(chunk.text)forchunkinchunks)return{overlap:overlap,chunk_count:len(chunks),answer_covered:any(ANSWERinchunk.textforchunkinchunks),matching_chunks:[{ordinal:c.ordinal,start:c.start_char,end:c.end_char}forcinchunksif定制商品inc.textor无理由退款inc.text],duplicated_characters:total-len(TEXT),}执行python scripts/run_overlap_boundary_demo.py的真实输出{overlap: 0, chunk_count: 2, answer_covered: False, matching_chunks: [{ordinal: 0, start: 0, end: 256}, {ordinal: 1, start: 256, end: 289}], duplicated_characters: 0} {overlap: 64, chunk_count: 2, answer_covered: True, matching_chunks: [{ordinal: 0, start: 0, end: 256}, {ordinal: 1, start: 192, end: 289}], duplicated_characters: 64}64 overlap 以 64 个重复字符换回完整答案但更好的结构化分块会识别条款句而不是依赖固定重叠。overlap 是保险带不是方向盘。为什么要批量 Embedding单条请求有固定网络、鉴权和调度开销批量可以提高吞吐。但批次越大单次失败重做越多也可能触发供应商输入上限。batch_size32只是初始配置应根据 Token 总量、API 限制、延迟和内存评测。本项目把批次构造成稳定顺序batches[chunks[start:startself.batch_size]forstartinrange(0,len(chunks),self.batch_size)]恢复的前提是同一文档、同一清洗/分块参数和同一模型版本会产生相同批次。若这些条件改变就应创建新索引版本不能沿用旧游标。游标表示“下一批”不是“最后一批”成功写入 batch 0 后保存next_batch1成功写入 batch 1 后保存 2。恢复循环forbatch_numberinrange(job.next_batch,len(batches)):这避免了1/-1的歧义。更重要的是向量写入成功后才能推进游标若先推进再写进程在两者之间崩溃会永久跳过一批。awaitself.vector_store.upsert(records)jobreplace(job,next_batchbatch_number1)awaitself.jobs.save(job)数据库与外部向量库无法天然共享事务所以upsert的记录 ID 必须稳定且幂等document_id:chunk.ordinal。若写入成功、保存游标前崩溃恢复会再写同一 ID而不是产生副本。状态机让失败可见uploaded → parsing → chunking → embedding → ready 任一步骤 → failed状态不是 UI 装饰。它帮助运维区分解析错误、向量服务错误和完成状态。异常时保存当前 job、错误文本和游标后继续抛出让 Celery 的重试机制看到失败。第三批失败测试测试 Provider 在第三次调用时失败classFailThirdBatchEmbedding(FakeEmbeddingProvider):asyncdefembed(self,texts):ifself.calls2:self.calls1raiseRuntimeError(temporary embedding failure)returnawaitsuper().embed(texts)第一次运行后withpytest.raises(RuntimeError,matchtemporary):awaitfailing.run(job.id)failedawaitjobs.get(job.id)assertfailed.statusfailedassertfailed.next_batch2换成正常 Provider 恢复再验证总数与 ID 唯一awaitresumed.run(job.id)readyawaitjobs.get(job.id)recordsawaitvectors.records_for(tenant_idtenant,knowledge_base_idkb)assertready.statusreadyassertlen(records)ready.total_chunksassertlen({record.idforrecordinrecords})len(records)真实结果1 passed in 0.37s。完整索引编排模块from__future__importannotationsfromcontextlibimportcontextmanagerfromdataclassesimportdataclass,replacefromtimeimportperf_counterfromtypingimportLiteral,Protocolfromapp.core.metricsimportobserve_index_stagefromapp.providers.contractsimportEmbeddingProviderfromapp.providers.object_storageimportObjectStoragefromapp.providers.vector_storeimportVectorRecord,VectorStorefromapp.services.ingestion.chunkingimportchunk_blocksfromapp.services.ingestion.cleaningimportclean_blocksfromapp.services.ingestion.documentsimportparse_document JobStatusLiteral[uploaded,parsing,chunking,embedding,ready,failed]contextmanagerdef_measure_stage(stage:str):# type: ignore[no-untyped-def]Record the real elapsed time even when a stage raises.startedperf_counter()try:yieldfinally:observe_index_stage(stage,perf_counter()-started)dataclass(frozenTrue,slotsTrue)classDocumentIndexJob:id:strtenant_id:strknowledge_base_id:strdocument_id:strfilename:strobject_key:strstatus:JobStatusuploadednext_batch:int0total_chunks:int0error:str|NoneNoneclassJobStore(Protocol):asyncdefsave(self,job:DocumentIndexJob)-None:...asyncdefget(self,job_id:str)-DocumentIndexJob:...classInMemoryJobStore:def__init__(self)-None:self._jobs:dict[str,DocumentIndexJob]{}asyncdefsave(self,job:DocumentIndexJob)-None:self._jobs[job.id]jobasyncdefget(self,job_id:str)-DocumentIndexJob:returnself._jobs[job_id]classIndexer:Idempotent index orchestration with a persisted batch cursor.def__init__(self,*,storage:ObjectStorage,embedder:EmbeddingProvider,vector_store:VectorStore,jobs:JobStore,chunk_size:int800,overlap:int120,batch_size:int32,)-None:self.storagestorage self.embedderembedder self.vector_storevector_store self.jobsjobs self.chunk_sizechunk_size self.overlapoverlap self.batch_sizebatch_sizeasyncdefrun(self,job_id:str)-DocumentIndexJob:jobawaitself.jobs.get(job_id)# Queue delivery is at-least-once. A completed job is immutable; an# intentional rebuild must create/reset a versioned job explicitly.ifjob.statusready:returnjobtry:jobreplace(job,statusparsing,errorNone)awaitself.jobs.save(job)with_measure_stage(parsing):dataawaitself.storage.get(job.object_key)parsedparse_document(job.filename,data)jobreplace(job,statuschunking)awaitself.jobs.save(job)with_measure_stage(chunking):chunkschunk_blocks(clean_blocks(parsed),chunk_sizeself.chunk_size,overlapself.overlap,)jobreplace(job,statusembedding,total_chunkslen(chunks))awaitself.jobs.save(job)batches[chunks[start:startself.batch_size]forstartinrange(0,len(chunks),self.batch_size)]with_measure_stage(embedding):forbatch_numberinrange(job.next_batch,len(batches)):batchbatches[batch_number]vectorsawaitself.embedder.embed([chunk.textforchunkinbatch])awaitself.vector_store.upsert([VectorRecord(idf{job.document_id}:{chunk.ordinal},tenant_idjob.tenant_id,knowledge_base_idjob.knowledge_base_id,textchunk.text,vectorvector,documentjob.filename,pagechunk.page,sectionchunk.section,metadata{document_id:job.document_id,ordinal:chunk.ordinal,embedding_model:self.embedder.model,},)forchunk,vectorinzip(batch,vectors,strictTrue)])jobreplace(job,next_batchbatch_number1)awaitself.jobs.save(job)jobreplace(job,statusready,errorNone)awaitself.jobs.save(job)returnjobexceptExceptionasexc:failedreplace(job,statusfailed,errorstr(exc))awaitself.jobs.save(failed)raise这份模块在运行时怎样走Indexer.run(job_id)是 Worker 真正调用的入口不是为了检查点额外造出的函数。它先通过JobStore.get()读取持久游标状态已经是ready时直接返回避免队列重复投递造成二次写入。没有完成时按下面顺序推进storage.get → parse_document → clean_blocks → chunk_blocks → 分批 embed → vector_store.upsert → 每成功一批就 save(next_batch 1)JobStore使用Protocol是因为同一编排逻辑需要两种调用者单测和离线示例使用InMemoryJobStore生产 Worker 使用持久化实现。协议只规定save/get没有把 SQLAlchemy 会话塞进索引算法。_measure_stage()也已经进入这条真实调用链。它在finally中记录 parsing、chunking、embedding 的实际耗时所以阶段即使抛错也能留下观测值这比在测试里手工调用一次observe_index_stage(embedding, 0.25)更有意义。对应回归测试替换指标观察器完整跑一次索引并断言三个阶段按顺序出现。异常处理只做两件确定的事把最新job保存成failed再原样抛出异常交给 Worker 决定是否重试。它不会把错误吞掉后返回一个假ready。重放时从next_batch开始向量 id 又由document_id ordinal稳定生成因此同一批意外重复提交也会覆盖而不是新增副本。本篇练习判断哪些变更允许续建对以下变更分别回答“继续旧 job”还是“创建新索引版本”并说明原因重试同一 503、batch_size 32 改 16、chunk_size 800 改 512、Embedding 模型名不变但维数 384 改 768、只修改 Celery 并发数。然后给DocumentIndexJob设计一个pipeline_fingerprint列出哈希至少应包含哪些配置。下一篇给出决策表和指纹实现思路并把记录真正落到 PostgreSQL/pgvector表结构、余弦距离与 HNSW 索引应该怎样设计。