前六期下来问答机器人已经像个正经产品了50行检索引擎第2期接大模型第3期健康监控第4期重试降级熔断第5期SSE流式第6期多轮记忆。功能全了可以上线了。上线第一周我就吃了个大亏。有个用户写了个死循环脚本一晚上调了我几万次接口。账单出来那一刻我手都在抖大模型API是按token收费的没有限流就等于把钱包敞开放在公网上。这就是今天第7期要干的事给机器人加个限流器。方案用令牌桶60行代码标准库搞定不依赖任何第三方包。先说思路。想象一个桶每秒往里滴5滴水桶最多装10滴。每次请求进来先从桶里舀一勺有水就放行没水就拒绝。正常用户问一句答一句水滴得上毫无感觉。脚本一秒发50个请求桶里10滴水瞬间舀空剩下的全被拦下。突发流量能扛10个持续流量卡死在每秒5个。这就是令牌桶的精髓允许小突发掐死大洪水。环境很简单Python 3.10以上threading是标准库不用装任何东西。代码来了完整可运行importtimeimportthreadingclassTokenBucket:def__init__(self,rate,capacity):self.raterate# 每秒补充几个令牌self.capacitycapacity# 桶最多装几个self.tokenscapacity self.lasttime.monotonic()self.lockthreading.Lock()deftake(self,tokens1):withself.lock:nowtime.monotonic()# 先按流逝时间补水self.tokensmin(self.capacity,self.tokens(now-self.last)*self.rate)self.lastnowifself.tokenstokens:self.tokens-tokensreturnTruereturnFalse注意两个细节。第一用time.monotonic()而不是time.time()。服务器做NTP时间同步时系统时钟可能往回拨time.time()一回拨令牌计算全乱套。monotonic是单调时钟只跟流逝时间有关专治这种坑。第二加了threading.Lock。线上服务都是多线程的不加锁两个请求同时读到还剩1个令牌都会放行限流就形同虚设。接到问答机器人上一行判断就行bucketTokenBucket(rate5,capacity10)defchat(user_id,question):ifnotbucket.take():return{error:请求太频繁稍后再试,retry_after:1}returnrobot.answer(question)被限流的请求别直接抛500返回429状态码加Retry-After头。前端看到429就知道等一秒再试用户体验差不到哪去。直接500的话调用方以为你服务挂了。效果我实测过。模拟10个用户每人连发20个问题总共200个请求打进rate5、capacity10的桶。结果11个通过189个被拦。0.4秒涌进200个请求正常用户根本感知不到脚本却被掐得死死的。这就是我们要的效果误伤不了正常人放不过刷子。上线前还有四个坑都是血泪换来的。第一个坑桶容量别设太小。capacity2的话用户连问两个问题第二个就被拦正常人也要骂娘。经验值capacity设成rate的2到3倍给正常连问留余量。第二个坑多进程部署。gunicorn开4个worker每个进程内存里各有一个桶限流额度直接翻4倍。单机多进程还能忍分布式部署必须换共享存储Redis版令牌桶是标准答案。第三个坑按用户限流不是按全局。全局rate5意味着全站每秒只能5个请求用户一多全被误杀。正确姿势是每个user_id一个桶恶意脚本只封它自己。内存里搞个dict存user_id到桶的映射定期清理不活跃的就行。第四个坑限流阈值要跟账单对齐。先算一笔账你的API套餐每分钟允许多少token单个请求平均多少token倒推出每秒放几个请求。限流不是拍脑袋定的是拿账单反推出来的。收藏这篇吧。下次你的应用要接大模型API这60行代码直接抄过去今晚就能睡个安稳觉不用半夜爬起来看账单。下期预告第8期我们给机器人加上调用日志和成本统计每个用户烧了多少钱一目了然。这个系列在专栏《Python实战》里连着更新从50行检索引擎一路搭到能扛刷子的线上服务关注后进专栏能从头连着看。讨论时间你的应用被脚本刷过量、或者被API限流坑过吗最后是怎么拦下来的一句话说说。Cyrus