背景痛点：Demo 变“卡死”的三道坎
做 Chatbot Demo 时，我们往往只跑一条请求，效果惊艳；一旦并发上来，现场立刻翻车。我最早用 Flask+Threading 模型，每来一个用户就开一条线程去调 LLM，结果：
- 线程阻塞：LLM 平均响应 2 s，线程池 200 条瞬间被占满，新用户直接 502。
- 状态同步困难：线程间共享一个 dict 记录对话历史，高并发下出现“串台”，A 用户收到 B 用户的回复。
- 横向扩展尴尬：把服务 Docker 化后，多副本之间状态不同步，Nginx 轮询导致用户“跳房间”。
一句话：玩具代码扛不住生产流量。要让它真正“可用”，得从架构到代码全链路异步化。
技术选型：WebSocket、SSE、长轮询实测对比
我把三种实时推送方案放在同一台 4C8G 机器上压测，客户端 1 k 并发，每 200 ms 发一条“你好”，持续 5 min，结果如下：
| 方案 | 99th 延迟 | CPU 占用 | 内存峰值 | 断线率 |
|---|---|---|---|---|
| 长轮询 | 1.2 s | 85 % | 1.4 GB | 12 % |
| SSE | 380 ms | 70 % | 0.9 GB | 5 % |
| WebSocket | 95 ms | 45 % | 0.5 GB | 1 % |
结论：WebSocket 在实时性与资源消耗上全面胜出；SSE 只能服务端→客户端单向，若做语音对话还得再开一条上行通道；长轮询在 1 k 并发时 epoll 句柄飙到 5 w，内核直接报错。最终敲定 WebSocket 做全双工通道。
核心实现：FastAPI+WebSocket+Redis Stream
1. 异步入口
# 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_responses=True) stream = RedisStream(redis, stream="chat", group="bot", consumer="demo") @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(code=1011, reason=str(e))2. 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, id="0", mkstream=True) except aioredis.ResponseError: pass # 组已存在 async def add(self, payload: dict, max_len=1000) -> str: return await self.r.xadd(self.stream, payload, maxlen=max_len, approximate=True) async def consume(self, session_id: str): await self.init() while True: msgs = await self.r.xreadgroup( self.group, self.consumer, {self.stream: ">"}, count=1, block=5000 ) for _, fields in msgs: if fields.get("sid") == session_id: yield fields["text"] await self.r.xack(self.stream, self.group, _)3. 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 = 20): self.base = base_url.rstrip("/") self.sem = BoundedSemaphore(max_conn) self.session = aiohttp.ClientSession( connector=aiohttp.TCPConnector(limit=0, limit_per_host=max_conn), timeout=aiohttp.ClientTimeout(total=10), ) 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:8000")性能优化：让 QPS 翻三倍
1. 批处理策略
LLM 支持批量推理，一次 4 条比 4×1 条快 2.3 倍。我在网关层加了一个“微批窗口”：
- 窗口时间 50 ms 或 batch_size=4，先到先走。
- 用 asyncio.Queue 收集请求，超时或满批一次性发。
- 返回时按顺序拆包，通过 Redis Stream 写回各自 WebSocket。
压测结果：单副本 QPS 从 120 → 380，提升 217 %；加上横向扩容 3 副本，整体 QPS 破千。
2. 令牌桶限流
防止突刺把 LLM 打挂，用 Redis+Lua 脚本实现分布式令牌桶：
# 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-1) return 1 end """ return await redis.eval(script, 1, key, int(time.time()*1000), rate, burst)网关层统一拦截，超过限流直接返回 429，保护后端。
避坑指南：那些半夜报警的坑
1. 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)2. 分布式会话一致性
多副本时，WebSocket 落在 Pod-A，LLM 回调落在 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 推向生产。
