企业官网建设流程全解析

前六期下来，问答机器人已经像个正经产品了：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.rate=rate# 每秒补充几个令牌self.capacity=capacity# 桶最多装几个self.tokens=capacity self.last=time.monotonic()self.lock=threading.Lock()deftake(self,tokens=1):withself.lock:now=time.monotonic()# 先按流逝时间补水self.tokens=min(self.capacity,self.tokens+(now-self.last)*self.rate)self.last=nowifself.tokens>=tokens:self.tokens-=tokensreturnTruereturnFalse

注意两个细节。第一，用time.monotonic()而不是time.time()。服务器做NTP时间同步时，系统时钟可能往回拨，time.time()一回拨，令牌计算全乱套。monotonic是单调时钟，只跟流逝时间有关，专治这种坑。第二，加了threading.Lock。线上服务都是多线程的，不加锁，两个请求同时读到"还剩1个令牌"，都会放行，限流就形同虚设。

接到问答机器人上，一行判断就行：

bucket=TokenBucket(rate=5,capacity=10)defchat(user_id,question):ifnotbucket.take():return{"error":"请求太频繁，稍后再试","retry_after":1}returnrobot.answer(question)

被限流的请求别直接抛500，返回429状态码加Retry-After头。前端看到429就知道等一秒再试，用户体验差不到哪去。直接500的话，调用方以为你服务挂了。

效果我实测过。模拟10个用户，每人连发20个问题，总共200个请求打进rate=5、capacity=10的桶。结果：11个通过，189个被拦。0.4秒涌进200个请求，正常用户根本感知不到，脚本却被掐得死死的。这就是我们要的效果：误伤不了正常人，放不过刷子。

上线前还有四个坑，都是血泪换来的。

第一个坑：桶容量别设太小。capacity=2的话，用户连问两个问题，第二个就被拦，正常人也要骂娘。经验值：capacity设成rate的2到3倍，给正常连问留余量。

第二个坑：多进程部署。gunicorn开4个worker，每个进程内存里各有一个桶，限流额度直接翻4倍。单机多进程还能忍，分布式部署必须换共享存储，Redis版令牌桶是标准答案。

第三个坑：按用户限流，不是按全局。全局rate=5，意味着全站每秒只能5个请求，用户一多全被误杀。正确姿势是每个user_id一个桶，恶意脚本只封它自己。内存里搞个dict存user_id到桶的映射，定期清理不活跃的就行。

第四个坑：限流阈值要跟账单对齐。先算一笔账：你的API套餐每分钟允许多少token，单个请求平均多少token，倒推出每秒放几个请求。限流不是拍脑袋定的，是拿账单反推出来的。

收藏这篇吧。下次你的应用要接大模型API，这60行代码直接抄过去，今晚就能睡个安稳觉，不用半夜爬起来看账单。

下期预告：第8期我们给机器人加上调用日志和成本统计，每个用户烧了多少钱一目了然。这个系列在专栏《Python实战》里连着更新，从50行检索引擎一路搭到能扛刷子的线上服务，关注后进专栏能从头连着看。

讨论时间：你的应用被脚本刷过量、或者被API限流坑过吗？最后是怎么拦下来的？一句话说说。

Cyrus