API最佳实践第三篇-3 流式输出 + 并发控制

接着前两篇，今天一次讲清流式输出 + 并发控制，

流式输出不改实际耗时，但用户体感差几倍——第一个字出来就觉得在响应了。 并发控制方面，429不是服务不稳定，是限流保护在工作，客户端加个令牌桶就能解决。这篇讲流式输出的正确用法、中断重连的写法，以及生产环境怎么控制并发。 流式输出不改实际耗时，但用户体感差几倍——第一个字出来就觉得在响应了

并发控制方面，429不是服务不稳定，是限流保护在工作，客户端加个令牌桶就能解决。

这篇讲流式输出的正确用法、中断重连的写法，以及生产环境怎么控制并发。

流式 vs 非流式

| 指标 | 非流式 | 流 式 |

| 总耗时 | 5.2秒 | 5.2秒 |

| 首Token时间 | 0.8秒 | 0.8秒 |

| 用户感知 | 等5秒才看到内容 | 0.8秒开始出字 |

实际耗时一样。流式让用户不焦虑。

流式中断重连

流式最大的坑：输出一半连接断了。

python

def stream_with_reconnect(messages, max_retries=2):

python 代码解读复制代码collected = ""

for attempt in range(max_retries + 1):

try:
    stream = client.chat.completions.create(
        model="glm-5.2", messages=messages, stream=True
    )
    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，是限流保护。

python

import time

import asyncio

class TokenBucket:

python 代码解读复制代码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)

批处理控制并发数：

python

async def batch_process(tasks, max_concurrency=5):

python 代码解读复制代码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的机制、各套餐的并发建议、手写令牌桶限流器、批处理任务怎么控制并发上限。