核心内容摘要
51破解版下载安装最新版本：解锁无限可能，畅享前沿科技
背景痛点Demo 变“卡死”的三道坎做 Chatbot Demo 时我们往往只跑一条请求效果惊艳一旦并发上来现场立刻翻车。
我最早用 FlaskThreading 模型每来一个用户就开一条线程去调 LLM结果线程阻塞LLM 平均响应 2 s线程池 200 条瞬间被占满新用户直接 502。
状态同步困难线程间共享一个 dict 记录对话历史高并发下出现“串台”A 用户收到 B 用户的回复。
横向扩展尴尬把服务 Docker 化后多副本之间状态不同步Nginx 轮询导致用户“跳房间”。
一句话玩具代码扛不住生产流量。
要让它真正“可用”得从架构到代码全链路异步化。
技术选型WebSocket、SSE、长轮询实测对比我把三种实时推送方案放在同一台 4C8G 机器上压测客户端 1 k 并发每 200 ms 发一条“你好”持续 5 min结果如下方案99th 延迟CPU 占用内存峰值断线率长轮询
2 s85 %
4 GB12 %SSE380 ms70 %
9 GB5 %WebSocket95 ms45 %
5 GB1 %结论WebSocket 在实时性与资源消耗上全面胜出SSE 只能服务端→客户端单向若做语音对话还得再开一条上行通道长轮询在 1 k 并发时 epoll 句柄飙到 5 w内核直接报错。
最终敲定 WebSocket 做全双工通道。
核心实现FastAPIWebSocketRedis Stream
异步入口# main.py from fastapi import FastAPI, WebSocket, WebSocketDisconnect import aioredis, asyncio, json, logging, os from pool import llm_pool from redis_stream import RedisStream app FastAPI() redis aioredis.from_url(redis://localhost:6379, decode_responsesTrue) stream RedisStream(redis, streamchat, groupbot, consumerdemo) app.websocket(/ws) async def ws_chat(websocket: WebSocket): await websocket.accept() sid websocket.headers.get(x-session-id, default) try: async for msg in stream.consume(sid): await websocket.send_text(msg) except WebSocketDisconnect: logging.info(client %s disconnected, sid) except Exception as e: # 兜底异常防止协程退出 logging.exception(e) await websocket.close(code1011, reasonstr(e))
Redis Stream 封装持久化重试# redis_stream.py import aioredis, json, uuid, asyncio from typing import Optional class RedisStream: def __init__(self, redis: aioredis.Redis, stream: str, group: str, consumer: str): self.r redis self.stream stream self.group group self.consumer consumer async def init(self): try: await self.r.xgroup_create(self.stream, self.group, id0, mkstreamTrue) except aioredis.ResponseError: pass # 组已存在 async def add(self, payload: dict, max_len
- str: return await self.r.xadd(self.stream, payload, maxlenmax_len, approximateTrue) async def consume(self, session_id: str): await self.init() while True: msgs await self.r.xreadgroup( self.group, self.consumer, {self.stream: }, count1, block5000 ) for _, fields in msgs: if fields.get(sid) session_id: yield fields[text] await self.r.xack(self.stream, self.group, _)
LLM 连接池背压控制# pool.py import aiohttp, asyncio from typing import List from asyncio import BoundedSemaphore class LLMPool: def __init__(self, base_url: str, max_conn: int
: self.base base_url.rstrip(/) self.sem BoundedSemaphore(max_conn) self.session aiohttp.ClientSession( connectoraiohttp.TCPConnector(limit0, limit_per_hostmax_conn), timeoutaiohttp.ClientTimeout(total
, ) async def ask(self, prompt: str) - str: async with self.sem: # 令牌信号量做背压 async with self.session.post( f{self.base}/chat, json{prompt: prompt, max_tokens: 512}, ) as r: r.raise_for_status() data await r.json() return data[text] llm_pool LLMPool(http://llm-gateway:
性能优化让 QPS 翻三倍
批处理策略LLM 支持批量推理一次 4 条比 4×1 条快
3 倍。
我在网关层加了一个“微批窗口”窗口时间 50 ms 或 batch_size4先到先走。
用 asyncio.Queue 收集请求超时或满批一次性发。
返回时按顺序拆包通过 Redis Stream 写回各自 WebSocket。
压测结果单副本 QPS 从 120 → 380提升 217 %加上横向扩容 3 副本整体 QPS 破千。
令牌桶限流防止突刺把 LLM 打挂用 RedisLua 脚本实现分布式令牌桶# ratelimit.py import aioredis, time async def acquire(redis: aioredis.Redis, key: str, rate: int, burst: int) - bool: script local key KEYS[1] local now tonumber(ARGV[1]) local rate tonumber(ARGV[2]) local burst tonumber(ARGV[3]) local fill_time 1/rate local last redis.call(HMGET, key, ts, tokens) local last_ts, tokens tonumber(last[1]) or 0, tonumber(last[2]) or burst local delta math.max(0, now - last_ts) tokens math.min(burst, tokens delta * rate) if tokens 1 then redis.call(HMSET, key, ts, now, tokens, tokens) return 0 else redis.call(HMSET, key, ts, now, tokens, tokens-
return 1 end return await redis.eval(script, 1, key, int(time.time()*
, rate, burst)网关层统一拦截超过限流直接返回 429保护后端。
避坑指南那些半夜报警的坑
WebSocket 保活NAT 设备默认 60 s 无数据就踢连接。
我踩过的坑只发业务心跳结果 90 s 后大量“Broken pipe”。
解决每 30 s 双向 ping/pong同时把 TCP keepalive 调到 40 s。
await websocket.ping() await asyncio.wait_for(websocket.pong(), timeout
10)
分布式会话一致性多副本时WebSocket 落在 Pod-ALLM 回调落在 Pod-B如何找到同一会话用 Redis Hash 存储sid - pod-ip回调时先读 Hash再转发到对应 Pod。
或者干脆把会话状态全放 Redis网关层无状态任意 Pod 都能回包。
开放问题延迟与吞吐的天平批处理能把吞吐拉高却会把首包延迟从 200 ms 推到 400 ms限流保障稳定性却会在峰值时“劝退”用户。
你的业务更看重哪一边是否愿意牺牲部分吞吐换取 P99 延迟 300 ms欢迎留言聊聊你的权衡思路。
动手试试把 Demo 变成“产品级”如果你也想从零搭一套低延迟、可扩展的语音/文字 Chatbot不妨直接跑一遍从0打造个人豆包实时通话AI动手实验。
实验把 ASR→LLM→TTS 整条链路都封装好了WebSocket 部分与我上面讲的思路一致代码开箱即用。
我跟着做完发现本地 30 分钟就能跑通再花一下午把批处理和限流加上QPS 直接翻三倍。
小白也能顺利体验建议边敲代码边对照本文的优化点相信你会更快把自己的 Chatbot Demo 推向生产。