核心内容摘要
3步搞定Mac录屏：QuickRecorder让系统声音与麦克风完美同步
背景痛点高并发下的“慢”才是真的慢过去一年我把 ChatGPT 接进了公司自己写的低代码平台初衷是让业务同事在画流程图时随时能用自然语言生成 SQL、脚本和单元测试。
上线第一周就翻车了早高峰 200 并发接口 P99 延迟飙到
2 s前端按钮一直转圈。
长提示4 k token场景下首 token 时间TTFB经常 5 s 起步开发同学等得直接开 Slack 吐槽。
速率限制 被触发RPMRequests Per Minute掉到 3 字头大量 429 报错日志一片红。
一句话“AI 辅助开发”一旦慢了反而拖垮开发效率。
于是我把压测脚本、火焰图、OpenAI 账单全拉出来开始死磕“ChatGPT 加速”。
技术方案三板斧流式、缓存、异步
流式响应 vs 批量请求维度流式streamTrue批量batchTTFB低毫秒级高等全量生成用户体验边打字边出结果白屏转圈代码复杂度需要回调/队列简单 for 循环网络开销多段 chunk小包多一次往返适合场景交互式问答、IDE 插件离线代码扫描、批量注释结论面向“人”用流式面向“机器”用批量。
实际部署里我把两者混着来用户输入用流式后台批量跑单测用批量同一条链路根据调用方动态切换后面代码会体现。
多级缓存内存 RedisChatGPT 不保证幂等但业务里“生成 MySQL 建表语句”这种提示 80 % 重复。
第一级进程内 LRU容量 1 k 条TTL 300 s命中
1 ms。
第二级Redis 集群TTL 3 600 skey 用提示的模糊哈希去掉空格、统一大小写。
缓存只挡“只读”场景带用户私有变量的请求自动跳过。
异步 IO 连接池httpx.AsyncClient复用 TCP最大连接数 200比官方同步 SDK 的 1 条连接打天下高到不知哪里。
配合asyncio.Semaphore(
做软限防止把 OpenAI 的 600 RPM 额度瞬间打满。
代码实战Python
11 可运行片段以下代码全部在 0 依赖外部框架装好openai
1.
redis、cachetools即可跑通。
带指数退避的重试import asyncio, random, openai from openai import AsyncOpenAI client AsyncOpenAI( api_keysk-xxx, max_retries7, # 官方库已支持指数退避 timeout30 ) async def chat_with_backoff(messages, **kw): 包装一层触发 429 时自旋等待最多 7 次 for attempt in range(1,
: try: return await client.chat.completions.create( modelgpt-
5-turbo, messagesmessages, **kw ) except openai.RateLimitError: wait (2 ** attempt) random.uniform(0,
await asyncio.sleep(wait) raise RuntimeError(Rate limit still hit after 7 retries)
请求批处理批量→聚合→拆包import asyncio, collections BATCH_SIZE 50 # 实验下来 50 条性价比最高 queue collections.deque() sem asyncio.Semaphore(
async def batch_worker(): while True: await asyncio.sleep(
0.
# 小窗口攒请求 if not queue: continue batch [] while queue and len(batch) BATH_SIZE: batch.append(queue.popleft()) if not batch: continue # 合并成一条多对话请求 tasks [ chat_with_backoff( [{role: user, content: item[prompt]}], max_tokens400 ) for item in batch ] results await asyncio.gather(*tasks) # 回写结果 for fut, res in zip(batch, results): fut[future].set_result(res.choices[0].message.content) async def ask_batch(prompt: str) - str: loop asyncio.get_event_loop() fut loop.create_future() queue.append({prompt: prompt, future: fut}) return await fut
缓存装饰器内存 Redisimport hashlib, pickle, redis, cachetools.func rds redis.Redis(host
127.
0.
1, port6379, decode_responsesFalse) def cache_key(prompt: str) - str: return gpt: hashlib.md5(prompt.strip().lower().encode()).hexdigest() def cached_chat(ttl
: def decorator(func): cachetools.func.ttl_cache(maxsize1024, ttl
def mem_cached(key): return func(key) async def wrapper(prompt: str): key cache_key(prompt) # 先查内存 if mem_cached.cache_info().currsize: hit mem_cached(key) if hit: return hit # 再查 Redis raw rds.get(key) if raw: return pickle.loads(raw) # 回源 resp await func(prompt) rds.setex(key, ttl, pickle.dumps(resp)) return resp return wrapper return decorator把ask_batch再包一层cached_chat(ttl
相同提示第二次直接
1 ms 返回压测 QPS 从 60 提到 720。
性能考量数字说话指标优化前优化后P99 延迟
2 s
4 s平均 TTFB
8 s
6 s429 错误占比
3 %
02 %Token 重复消耗100 %28 %缓存命中月度账单$820$510成本与体验双赢老板终于点头继续扩量。
避坑指南官方文档没写的细节速率限制分 RPM、TPMToken per Minute两级先超 TPM 也会抛 429别只盯请求数。
上下文窗口 ≠ 提示越长越好gpt-
5-turbo 最大 16 k但输入超 4 k 后价格翻倍、延迟线性上涨把系统提示做模板化运行时只拼变量。
流式响应的finish_reasonlength容易被忽略前端要给出“内容被截断”提示否则用户以为 AI 胡言乱语。
异步池别设置太大OpenAI 账号级限频不是单台 IP并发 200 跟 2 000 没区别该 429 还是 429。
延伸思考再往后怎么玩模型蒸馏用 GPT-4 生成 10 w 条指令,输出蒸馏到 7 B 本地模型延迟降到 300 ms成本 ≈ 0。
Function Calling 缓存把工具返回结果也做版本化哈希重复函数调用直接短路。
边缘侧推理火山引擎豆包实时语音大模型已经支持端侧 ASRTTS如果能把文本模型也下放就能做到“本地优先、云端补偿”体验更丝滑。
写在最后把思路再套到“实时通话 AI”上面这套加速套路最初只是为文本场景服务但当我把同样思路迁移到语音实时对话时发现链路更长ASR→LLM→TTS每一步都有延迟叠加如果 LLM 环节卡 3 秒用户就会感到“对面反应慢”。
所以缓存、流式、异步在语音场景更是生命线。
如果你也想亲手搭一个能“秒回”的 AI 伙伴不妨试试这个动手实验——从0打造个人豆包实时通话AI。
实验把火山引擎的豆包语音大模型、WebRTC、FastAPI 全部打包好本地 Docker 一键启动我这种前端都顺利跑通相信你也可以。