Python后端AI专题29缓存、配额、幂等与 Token 成本如何一起设计模型调用既慢又按量计费。用户双击发送、网关重试、20 个并发请求同时检查剩余额度如果只写“先 GET 用量再 1”很容易超卖如果缓存键没有租户与知识版本又会把旧答案甚至别人的答案返回。四个概念要按请求生命周期一起设计。隔离矩阵与 RBAC 答案请求者本人会话他人会话同租户普通用户200/201404不同租户用户404404管理员共享读取需要数据库 membership/role 表作为实时事实Token 只携身份或短寿命角色快照Service 查询当前 membershipRepository 条件变成(user_id本人 OR role permits tenant scope)仍强制 tenant_id。只信长期 JWT 的admin撤权后 Token 到期前仍有权限。请求进入时的正确顺序配图建议顺序鉴权并形成 tenant/user用tenant endpoint idempotency-key payload hash查幂等记录已完成则返回原响应处理中则返回一致状态原子预留最坏 Token/任务额度检查带知识版本与权限范围的缓存执行业务并记录真实 usage结算/释放差额保存幂等结果。缓存命中是否消耗额度取决于产品计费规则但必须明确不能一会儿按请求计、一会儿按模型 Token 计。原子配额为什么要用一段 LuaRedis 的 GET 与 INCR 分开执行会竞态。Lua 在单个 Redis 执行线程内完成“读当前值—判断上限—递增—设置 TTL”localcurrenttonumber(redis.call(GET,KEYS[1])or0)localamounttonumber(ARGV[1])locallimittonumber(ARGV[2])ifcurrentamountlimitthenreturn-1endlocalupdatedredis.call(INCRBY,KEYS[1],amount)ifupdatedamountthenredis.call(EXPIRE,KEYS[1],tonumber(ARGV[3]))endreturnupdated20 个并发对 limit5 的内存参考实现请求恰好 5 个成功真实 Redis 两个独立实例共享 limit3也没有超卖。完整配额模块from__future__importannotationsimportasyncioimporttimefromdataclassesimportdataclassfromuuidimportuuid4fromredis.asyncioimportRedisclassQuotaExceeded(RuntimeError):passdataclass(frozenTrue,slotsTrue)classReservation:id:strtenant_id:stramount:intbucket_key:str|NoneNoneclassInMemoryQuota:def__init__(self,*,limit:int)-None:self.limitlimit self._usage:dict[str,int]{}self._active:dict[str,Reservation]{}self._lockasyncio.Lock()asyncdefreserve(self,tenant_id:str,*,amount:int)-Reservation:ifamount0:raiseValueError(reservation amount must be positive)asyncwithself._lock:currentself._usage.get(tenant_id,0)ifcurrentamountself.limit:raiseQuotaExceeded(fquota{self.limit}exceeded)reservationReservation(str(uuid4()),tenant_id,amount)self._usage[tenant_id]currentamount self._active[reservation.id]reservationreturnreservationasyncdefrelease(self,reservation:Reservation)-None:asyncwithself._lock:activeself._active.pop(reservation.id,None)ifactive:self._usage[active.tenant_id]-active.amountasyncdefused(self,tenant_id:str)-int:asyncwithself._lock:returnself._usage.get(tenant_id,0)classRedisQuota:_reserve_script local current tonumber(redis.call(GET, KEYS[1]) or 0) local amount tonumber(ARGV[1]) local limit tonumber(ARGV[2]) if current amount limit then return -1 end local updated redis.call(INCRBY, KEYS[1], amount) if updated amount then redis.call(EXPIRE, KEYS[1], tonumber(ARGV[3])) end return updated def__init__(self,redis:Redis,*,limit:int,window_seconds:int)-None:self.redisredis self.limitlimit self.window_secondswindow_secondsdef_key(self,tenant_id:str)-str:windowint(time.time())//self.window_secondsreturnfknowflow:quota:{tenant_id}:{window}asyncdefreserve(self,tenant_id:str,*,amount:int)-Reservation:keyself._key(tenant_id)resultawaitself.redis.eval(self._reserve_script,1,key,amount,self.limit,self.window_seconds5,)ifint(result)0:raiseQuotaExceeded(fquota{self.limit}exceeded)returnReservation(str(uuid4()),tenant_id,amount,key)asyncdefused(self,tenant_id:str)-int:returnint(awaitself.redis.get(self._key(tenant_id))or0)asyncdefrelease(self,reservation:Reservation)-None:script(local vtonumber(redis.call(GET,KEYS[1]) or 0); local nmath.max(0,v-tonumber(ARGV[1])); redis.call(SET,KEYS[1],n,KEEPTTL); return n)keyreservation.bucket_keyorself._key(reservation.tenant_id)awaitself.redis.eval(script,1,key,reservation.amount)本次审查还修复了一个跨窗口 bug若预留发生在第 N 窗口、失败释放时已进入 N1重新计算 key 会扣错桶。Reservation 现在保存实际bucket_key时钟前移测试证明 release 仍操作原 key。幂等不是缓存答案InMemoryIdempotencyStore对同一 key 加锁只有一个 factory 执行asyncdefget_or_create(self,key,factory):lockself._locks.setdefault(key,asyncio.Lock())asyncwithlock:ifkeyinself._values:returnself._values[key]valueawaitfactory()self._values[key]valuereturnvalue生产需要 Redis/数据库持久化状态、TTL、payload hash 和失败语义。若同一个 key 配不同请求体应返回 409不能把第一次响应错误套给不同操作。RAG 缓存键必须包含什么至少tenant_id、knowledge_base_id、权限版本、规范化问题、检索参数、索引 pipeline fingerprint、Prompt hash、模型/温度。文档重新索引后旧缓存应自然失效。缓存正文也要连同结构化 citations 缓存不能只缓存一段可能指向旧页码的字符串。当前配额/MinIO/幂等测试结果...... [100%] 6 passed in 0.84s本篇最终完整模块quotas.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsimportasyncioimporttimefromdataclassesimportdataclassfromuuidimportuuid4fromredis.asyncioimportRedisclassQuotaExceeded(RuntimeError):passdataclass(frozenTrue,slotsTrue)classReservation:id:strtenant_id:stramount:intbucket_key:str|NoneNoneclassInMemoryQuota:Atomic reference implementation; Redis uses the same reserve semantics.def__init__(self,*,limit:int)-None:self.limitlimit self._usage:dict[str,int]{}self._active:dict[str,Reservation]{}self._lockasyncio.Lock()asyncdefreserve(self,tenant_id:str,*,amount:int)-Reservation:ifamount0:raiseValueError(reservation amount must be positive)asyncwithself._lock:currentself._usage.get(tenant_id,0)ifcurrentamountself.limit:raiseQuotaExceeded(fquota{self.limit}exceeded)reservationReservation(str(uuid4()),tenant_id,amount)self._usage[tenant_id]currentamount self._active[reservation.id]reservationreturnreservationasyncdefrelease(self,reservation:Reservation)-None:asyncwithself._lock:activeself._active.pop(reservation.id,None)ifactive:self._usage[active.tenant_id]-active.amountasyncdefused(self,tenant_id:str)-int:asyncwithself._lock:returnself._usage.get(tenant_id,0)classRedisQuota:Cross-process quota implemented as one atomic Lua operation._reserve_script local current tonumber(redis.call(GET, KEYS[1]) or 0) local amount tonumber(ARGV[1]) local limit tonumber(ARGV[2]) if current amount limit then return -1 end local updated redis.call(INCRBY, KEYS[1], amount) if updated amount then redis.call(EXPIRE, KEYS[1], tonumber(ARGV[3])) end return updated def__init__(self,redis:Redis,*,limit:int,window_seconds:int)-None:self.redisredis self.limitlimit self.window_secondswindow_secondsdef_key(self,tenant_id:str)-str:windowint(time.time())//self.window_secondsreturnfknowflow:quota:{tenant_id}:{window}asyncdefreserve(self,tenant_id:str,*,amount:int)-Reservation:keyself._key(tenant_id)resultawaitself.redis.eval(self._reserve_script,1,key,amount,self.limit,self.window_seconds5,)ifint(result)0:raiseQuotaExceeded(fquota{self.limit}exceeded)returnReservation(str(uuid4()),tenant_id,amount,key)asyncdefused(self,tenant_id:str)-int:returnint(awaitself.redis.get(self._key(tenant_id))or0)asyncdefrelease(self,reservation:Reservation)-None:# Clamp to zero because the reservation may expire with its window.scriptlocal vtonumber(redis.call(GET,KEYS[1]) or 0); local nmath.max(0,v-tonumber(ARGV[1])); redis.call(SET,KEYS[1],n,KEEPTTL); return nkeyreservation.bucket_keyorself._key(reservation.tenant_id)awaitself.redis.eval(script,1,key,reservation.amount)本篇练习设计幂等键冲突扩展InMemoryIdempotencyStore同一个 key 第一次绑定 payload hashaaa第二次若传bbb必须抛IdempotencyConflict相同 hash 返回原值。先写两个并发测试20 个相同 payload 只调用 factory 一次不同 payload 冲突且不会覆盖原响应。说明生产记录应保存哪些字段和 TTL 怎样选。下一篇给出完整实现并进入 Prompt Injection 攻防为什么知识库里的文字属于不可信输入即使它来自“内部文档”。