LLM 流式输出架构大模型后端的 SSE 推送与背压控制从同步等待到实时响应一、同步请求的体验瓶颈用户等待 30 秒的不可承受之重大模型推理的延迟是后端架构的核心挑战。一个 2000 Token 的回复在 GPT-4 级别模型上的生成时间可能达到 20—40 秒。如果采用传统的同步请求模式用户需要等待全部内容生成完毕后才能看到任何输出——这种黑盒等待的体验在产品层面是不可接受的。流式输出Streaming是解决这一问题的标准方案模型逐 Token 生成后端逐块推送给前端用户看到的是打字机效果而非空白等待。但流式输出引入了新的工程复杂性——SSE 连接管理、背压控制、错误恢复、Token 计量等这些在同步模式下不需要考虑的问题在流式场景下变得至关重要。二、流式输出的技术架构与数据流sequenceDiagram participant C as 客户端 participant G as API 网关 participant S as 流式调度器 participant M as 模型推理服务 participant R as Redis C-G: POST /chat (SSE 请求) G-S: 建立流式连接 S-R: 检查令牌配额 R--S: 配额充足 S-M: 发起推理请求 loop 逐 Token 生成 M--S: Token 片段 S-R: 更新 Token 消耗 S--G: SSE: data: {token} G--C: SSE: data: {token} end M--S: [DONE] S--G: SSE: data: [DONE] G--C: SSE: data: [DONE]架构的核心是流式调度器——它负责管理模型推理连接、控制推送速率背压、计量 Token 消耗、处理异常中断。网关层仅做 SSE 协议转换和连接保活不参与业务逻辑。三、生产级实现Go 语言的流式推理网关// stream_gateway.go — LLM 流式推理网关 package gateway import ( context encoding/json fmt io net/http sync/atomic time github.com/redis/go-redis/v9 ) // StreamScheduler 流式调度器 // 设计意图统一管理模型推理连接和 SSE 推送 // 处理背压、超时、配额控制等横切关注点 type StreamScheduler struct { redisClient *redis.Client modelClient ModelClient maxTokenRate int64 // 每秒最大推送 Token 数背压控制 } // ChatRequest 聊天请求 type ChatRequest struct { Model string json:model Messages []ChatMessage json:messages Stream bool json:stream MaxTokens int json:max_tokens,omitempty } // ChatMessage 聊天消息 type ChatMessage struct { Role string json:role Content string json:content } // StreamEvent SSE 事件 type StreamEvent struct { ID string json:id Object string json:object Choices []struct { Delta struct { Content string json:content } json:delta FinishReason *string json:finish_reason } json:choices Usage *TokenUsage json:usage,omitempty } // TokenUsage Token 用量 type TokenUsage struct { PromptTokens int json:prompt_tokens CompletionTokens int json:completion_tokens TotalTokens int json:total_tokens } // HandleStreamChat 处理流式聊天请求 // 设计意图将模型推理的流式响应逐块转发给客户端 // 同时处理配额检查、背压控制和异常恢复 func (s *StreamScheduler) HandleStreamChat(w http.ResponseWriter, r *http.Request) { var req ChatRequest if err : json.NewDecoder(r.Body).Decode(req); err ! nil { http.Error(w, 无效请求, http.StatusBadRequest) return } // 配额检查 userID : r.Header.Get(X-User-ID) if err : s.checkQuota(r.Context(), userID); err ! nil { http.Error(w, err.Error(), http.StatusTooManyRequests) return } // 设置 SSE 响应头 w.Header().Set(Content-Type, text/event-stream) w.Header().Set(Cache-Control, no-cache) w.Header().Set(Connection, keep-alive) w.Header().Set(X-Accel-Buffering, no) // 禁用 Nginx 缓冲 flusher, ok : w.(http.Flusher) if !ok { http.Error(w, 不支持流式响应, http.StatusInternalServerError) return } // 建立模型推理连接 stream, err : s.modelClient.StreamChat(r.Context(), req) if err ! nil { s.sendSSE(w, flusher, error, err.Error()) return } defer stream.Close() // Token 计量与背压控制 var totalTokens int64 tokenRateLimiter : time.NewTicker(time.Second / time.Duration(s.maxTokenRate)) defer tokenRateLimiter.Stop() // 超时控制单次请求最大 120 秒 ctx, cancel : context.WithTimeout(r.Context(), 120*time.Second) defer cancel() for { select { case -ctx.Done(): // 超时中断发送部分结果 s.sendSSE(w, flusher, timeout, 请求超时) return case -tokenRateLimiter.C: // 背压控制限制推送速率防止客户端消费不过来 event, err : stream.Recv() if err io.EOF { // 流结束 s.sendSSE(w, flusher, done, [DONE]) s.recordUsage(r.Context(), userID, totalTokens) return } if err ! nil { s.sendSSE(w, flusher, error, err.Error()) return } // 累计 Token 消耗 if event.Usage ! nil { atomic.AddInt64(totalTokens, int64(event.Usage.CompletionTokens)) } // 序列化并推送 data, _ : json.Marshal(event) s.sendSSE(w, flusher, message, string(data)) } } } // sendSSE 发送 SSE 事件 func (s *StreamScheduler) sendSSE(w http.ResponseWriter, flusher http.Flusher, eventType, data string) { fmt.Fprintf(w, event: %s\ndata: %s\n\n, eventType, data) flusher.Flush() } // checkQuota 检查用户 Token 配额 func (s *StreamScheduler) checkQuota(ctx context.Context, userID string) error { key : fmt.Sprintf(quota:%s:%s, userID, time.Now().Format(20060102)) used, err : s.redisClient.Get(ctx, key).Int64() if err ! nil err ! redis.Nil { return fmt.Errorf(配额查询失败) } if used 100000 { // 每日 10 万 Token 配额 return fmt.Errorf(今日配额已用尽) } return nil } // recordUsage 记录 Token 消耗 func (s *StreamScheduler) recordUsage(ctx context.Context, userID string, tokens int64) { key : fmt.Sprintf(quota:%s:%s, userID, time.Now().Format(20060102)) s.redisClient.IncrBy(ctx, key, tokens) s.redisClient.Expire(ctx, key, 48*time.Hour) }四、Trade-offs流式输出的工程代价与适用场景连接资源消耗。SSE 是长连接每个活跃用户占用一个 HTTP 连接。在高并发场景下如同时 1000 个用户对话连接数可能成为瓶颈。优化手段使用 HTTP/2 多路复用减少连接开销、设置合理的空闲超时自动断开、对非活跃连接降级为轮询模式。背压控制的必要性。如果模型生成速度远快于客户端消费速度如客户端网络慢服务端缓冲区会持续增长。Go 的 HTTP ResponseWriter 默认有缓冲但不会无限增长——当缓冲满时写入会阻塞进而阻塞模型推理连接。必须实现显式的速率控制在客户端消费不过来时主动降速。错误恢复的复杂性。流式传输中途发生网络中断时已发送的 Token 无法回退客户端需要自行拼接已接收的内容并决定是否重试。建议在每次推送中携带序列号客户端据此检测是否有遗漏片段。Token 计量的精度。流式场景下的 Token 计量依赖模型返回的 Usage 字段但部分模型仅在最后一个事件中返回累计用量。如果连接中途断开可能无法获取准确的消耗数据。建议在调度器侧维护独立的 Token 计数器以实际接收的 Token 数为准。五、总结流式输出是大模型产品的体验基石但其工程复杂度远超同步请求。落地路径第一步实现基本的 SSE 推送链路确保模型输出能实时到达客户端第二步加入背压控制和超时机制防止慢客户端拖垮服务端第三步实现 Token 配额和计量支撑商业化计费第四步建立流式传输的可观测性监控连接数、推送速率、中断率等指标。核心原则流式输出的核心挑战不在推送本身而在连接管理、资源控制和异常恢复。