接着前两篇今天一次讲清流式输出 并发控制流式输出不改实际耗时但用户体感差几倍——第一个字出来就觉得在响应了。并发控制方面429不是服务不稳定是限流保护在工作客户端加个令牌桶就能解决。这篇讲流式输出的正确用法、中断重连的写法以及生产环境怎么控制并发。流式 vs 非流式| 指标 | 非流式 | 流 式 || 总耗时 | 5.2秒 | 5.2秒 || 首Token时间 | 0.8秒 | 0.8秒 || 用户感知 | 等5秒才看到内容 | 0.8秒开始出字 |实际耗时一样。流式让用户不焦虑。流式中断重连流式最大的坑输出一半连接断了。pythondef stream_with_reconnect(messages, max_retries2):collected for attempt in range(max_retries 1): try: stream client.chat.completions.create( modelglm-5.2, messagesmessages, streamTrue ) for chunk in stream: if chunk.choices[0].delta.content: collected chunk.choices[0].delta.content yield chunk.choices[0].delta.content return except Exception: if attempt max_retries: messages.append({role: assistant, content: collected}) messages.append({role: user, content: 继续}) else: raise并发控制429不是bug是限流保护。pythonimport timeimport asyncioclass TokenBucket:def __init__(self, rate, capacity): self.rate rate self.capacity capacity self.tokens capacity self.last_time time.monotonic() async def acquire(self): while True: now time.monotonic() self.tokens min(self.capacity, self.tokens (now - self.last_time) * self.rate) self.last_time now if self.tokens 1: self.tokens - 1 return await asyncio.sleep((1 - self.tokens) / self.rate)批处理控制并发数pythonasync def batch_process(tasks, max_concurrency5):sem asyncio.Semaphore(max_concurrency) async def bounded(task): async with sem: return await process(task) return await asyncio.gather(*[bounded(t) for t in tasks])##这篇解决两个常见问题用户觉得慢、并发一高就报429。流式输出部分对比stream开关的实际效果给Python和Node.js的代码示例以及流式中断后怎么自动重连。并发控制部分解释429的机制、各套餐的并发建议、手写令牌桶限流器、批处理任务怎么控制并发上限。
