Python后端AI专题27把索引搬出请求线程Celery 任务、进度与重放解析 200 页 PDF、生成几百批向量不应占住上传 HTTP 请求。API 的职责是安全保存原文件、创建 durable job 并入队Worker 根据 job id 重建依赖并执行同一个 Indexer客户端轮询文档状态。异步不是“起一个后台协程”而是任务可重启、可观察、可幂等重放。本篇最终完整模块index_tasks.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsimportasynciofromapp.core.configimportSettingsfromapp.services.production_backendimportProductionBackendfromapp.workers.celery_appimportcelery_appcelery_app.task(bindTrue,nameknowflow.index_document,autoretry_for(ConnectionError,TimeoutError),retry_backoffTrue,retry_jitterTrue,retry_kwargs{max_retries:3},max_retries3,)defindex_document(self,job_id:str)-dict[str,str]:# type: ignore[no-untyped-def]Queue contract for durable indexing. The production composition resolves the SQL-backed job and calls the same Indexer service used by offline tests. Keeping only a job id in the Celery payload avoids serializing credentials, bytes or SQLAlchemy objects. settingsSettings(persistence_modepostgres,index_modeinline)asyncdefrun()-None:backendProductionBackend(settings)try:awaitbackend.process_job(job_id)finally:awaitbackend.engine.dispose()asyncio.run(run())return{job_id:job_id,status:ready}celery_app.task(nameknowflow.dispatch_index_outbox,autoretry_for(ConnectionError,TimeoutError),retry_backoffTrue,retry_kwargs{max_retries:3},)defdispatch_index_outbox()-dict[str,int]:Periodic recovery for committed jobs that were never published.settingsSettings(persistence_modepostgres,index_modecelery)asyncdefrun()-dict[str,int]:backendProductionBackend(settings)try:returnawaitbackend.dispatch_pending_index_jobs()finally:awaitbackend.engine.dispose()returnasyncio.run(run())会话越权练习完整答案Alice/Bob 分别注册得到合法 TokenAlice 创建会话。Bob 的读写攻击read_attackawaitclient.get(f/api/v1/conversations/{conversation_id}/messages,headersbob)write_attackawaitclient.post(f/api/v1/conversations/{conversation_id}/messages,headersbob,json{question:窃取答案},)assertread_attack.status_code404assertwrite_attack.status_code404assertAlicenotinread_attack.textwrite_attack.text真实结果1 passed in 4.12s。知识库可能是团队共享资源而会话包含个人问题、上下文和引用因此项目仍按 user_id 进一步隔离未来若支持共享会话应引入显式成员/角色而不是删除过滤条件。API 为什么只传 job id不要把文件字节、JWT、SQLAlchemy 对象或 Provider 客户端序列化进 Celery 消息。队列载荷只有稳定的job_idWorker 再从数据库取得租户、知识库、对象 key、游标和状态。好处消息小、JSON 可序列化、凭据不在 Broker 复制、重试读取最新状态。此前代码把这个参数误命名为document_id虽然值正好传到process_job()但日志语义错误本篇已用签名测试修成job_id。late acknowledgement 意味着什么celery_app.conf.update(task_serializerjson,accept_content[json],task_acks_lateTrue,task_reject_on_worker_lostTrue,task_track_startedTrue,worker_prefetch_multiplier1,)late ack 让 Worker 完成后再确认进程中途死亡Broker 可重新投递。代价是任务可能至少执行一次所以业务必须幂等。worker_prefetch_multiplier1避免一个 Worker 抢走大量长任务其他 Worker 空闲。“至少一次”不等于“恰好一次”。Exactly-once 不能靠 Celery 配置获得而是稳定记录 ID、数据库约束、状态机和幂等 upsert 共同实现业务效果。自动重试只捕获临时故障celery_app.task(bindTrue,nameknowflow.index_document,autoretry_for(ConnectionError,TimeoutError),retry_backoffTrue,retry_jitterTrue,retry_kwargs{max_retries:3},max_retries3,)格式损坏、OCR_REQUIRED、维数不匹配不应无限重试连接与超时可能恢复。更完整系统会把 Provider 的 retryable 领域错误纳入而不是用autoretry_for(Exception,)。完整 Celery 应用配置from__future__importannotationsimportosfromceleryimportCelery brokeros.getenv(KNOWFLOW_CELERY_BROKER_URL,redis://localhost:6389/1)celery_appCelery(knowflow,brokerbroker,backendbroker,include[app.workers.index_tasks],)celery_app.conf.update(task_serializerjson,result_serializerjson,accept_content[json],task_acks_lateTrue,task_reject_on_worker_lostTrue,task_track_startedTrue,worker_prefetch_multiplier1,timezoneAsia/Shanghai,)完整 Worker 任务from__future__importannotationsimportasynciofromapp.core.configimportSettingsfromapp.services.production_backendimportProductionBackendfromapp.workers.celery_appimportcelery_appcelery_app.task(bindTrue,nameknowflow.index_document,autoretry_for(ConnectionError,TimeoutError),retry_backoffTrue,retry_jitterTrue,retry_kwargs{max_retries:3},max_retries3,)defindex_document(self,job_id:str)-dict[str,str]:Resolve a durable job; never serialize credentials or file bytes.settingsSettings(persistence_modepostgres,index_modeinline)asyncdefrun()-None:backendProductionBackend(settings)try:awaitbackend.process_job(job_id)finally:awaitbackend.engine.dispose()asyncio.run(run())return{job_id:job_id,status:ready}为什么 Worker 内设置index_modeinline它只是让 Worker 处理 job 时不要再次把同一任务入队持久化仍是 PostgreSQL文件仍从 MinIO 读取。进度怎样到客户端上传接口在 Celery 模式必须返回{document_id:...,status:queued}项目此前硬编码ready真实链路发现后已通过失败测试修复。客户端 GET/documents/{id}读取 SQL 中 document/job 状态可观察queued → parsing → chunking → embedding → ready/failed不依赖 Celery result backend 作为业务事实源。Worker 合同测试当前结果.. [100%] 2 passed in 2.28s本篇练习模拟重复投递使用同一个 job id 连续调用Indexer.run()两次第一次已经 ready断言第二次结束后向量记录总数不变、所有 id 唯一、状态仍 ready。然后回答当前实现会重新解析和 Embedding虽然结果不重复是否算高效幂等应在什么状态下直接短路什么情况下允许显式 reindex下一篇给出测试与改进方案并从 API、Repository、向量 SQL 三层拆解多租户隔离。