一、为什么需要 Streaming非流式传统方式 ┌──────────────────────────────────────────────┐ │ 用户帮我写一封邮件给客户 │ │ │ │ 时间轴 │ │ 0s ──── 发送请求 │ │ 5s ──── 还在等 │ │ 10s ─── 还在等 │ │ 15s ─── 收到完整回复用户已经不耐烦了 │ └──────────────────────────────────────────────┘ 流式Streaming ┌──────────────────────────────────────────────┐ │ 用户帮我写一封邮件给客户 │ │ │ │ 时间轴 │ │ 0s ──── 发送请求 │ │ 1s ──── 看到好的 │ │ 2s ──── 看到我正在为您起草邮件 │ │ 3s ──── 看到尊敬的客户 │ │ ... ──── 逐字显示用户感知几乎无延迟 │ └──────────────────────────────────────────────┘Streaming 的核心价值将 TTFB首字节时间从 10-20s 降到 1-2s大幅提升用户体验。二、SSE 协议基础SSEServer-Sent Events是 Streaming 的标准协议基于 HTTP 长连接。SSE 数据格式 ─────────────────────────────────────────────── data: {choices:[{delta:{content:你好}}]} data: {choices:[{delta:{content:}}]} data: {choices:[{delta:{content:世界}}]} data: [DONE] ─────────────────────────────────────────────── 特点 1. 每个事件以 data: 开头 2. 事件之间用空行分隔 3. 以 [DONE] 标记结束 4. 浏览器原生支持 EventSource API三、服务端 Streaming 实现// streaming/server.go package streaming import ( bufio bytes encoding/json fmt io log net/http sync time ) // StreamEvent SSE 事件 type StreamEvent struct { ID string json:id,omitempty Event string json:event,omitempty Data json.RawMessage json:data Retry int json:retry,omitempty // 重连间隔毫秒 } // StreamWriter SSE 写入器 type StreamWriter struct { writer http.ResponseWriter flusher http.Flusher mu sync.Mutex closed bool } // NewStreamWriter 创建 SSE 写入器 func NewStreamWriter(w http.ResponseWriter) (*StreamWriter, error) { flusher, ok : w.(http.Flusher) if !ok { return nil, fmt.Errorf(streaming not supported) } // 设置 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(Access-Control-Allow-Origin, *) w.Header().Set(X-Accel-Buffering, no) // 禁用 Nginx 缓冲 return StreamWriter{ writer: w, flusher: flusher, }, nil } // WriteEvent 写入一个 SSE 事件 func (sw *StreamWriter) WriteEvent(event StreamEvent) error { sw.mu.Lock() defer sw.mu.Unlock() if sw.closed { return fmt.Errorf(stream already closed) } var buf bytes.Buffer if event.ID ! { buf.WriteString(fmt.Sprintf(id: %s\n, event.ID)) } if event.Event ! { buf.WriteString(fmt.Sprintf(event: %s\n, event.Event)) } if event.Retry 0 { buf.WriteString(fmt.Sprintf(retry: %d\n, event.Retry)) } // data 字段支持多行 dataStr : string(event.Data) for _, line : range bytes.Split([]byte(dataStr), []byte(\n)) { buf.WriteString(fmt.Sprintf(data: %s\n, line)) } buf.WriteString(\n) // 空行分隔事件 _, err : sw.writer.Write(buf.Bytes()) if err ! nil { return err } sw.flusher.Flush() return nil } // WriteData 写入纯数据最常用 func (sw *StreamWriter) WriteData(data []byte) error { return sw.WriteEvent(StreamEvent{ Data: data, }) } // WriteDone 写入结束标记 func (sw *StreamWriter) WriteDone() error { sw.mu.Lock() defer sw.mu.Unlock() if sw.closed { return nil } _, err : sw.writer.Write([]byte(data: [DONE]\n\n)) sw.closed true if err ! nil { return err } sw.flusher.Flush() return nil } // WriteError 写入错误事件 func (sw *StreamWriter) WriteError(code int, message string) error { errData, _ : json.Marshal(map[string]interface{}{ error: map[string]interface{}{ code: code, message: message, }, }) return sw.WriteData(errData) } // Close 关闭流 func (sw *StreamWriter) Close() { sw.mu.Lock() defer sw.mu.Unlock() sw.closed true } // ─── LLM Streaming Adapter ─── // LLMStreamAdapter LLM 流式适配器 type LLMStreamAdapter struct { provider string apiKey string baseURL string client *http.Client } // NewLLMStreamAdapter 创建流式适配器 func NewLLMStreamAdapter(provider, apiKey string) *LLMStreamAdapter { return LLMStreamAdapter{ provider: provider, apiKey: apiKey, baseURL: getBaseURL(provider), client: http.Client{Timeout: 60 * time.Second}, } } // ChatStream 流式聊天 func (a *LLMStreamAdapter) ChatStream( messages []map[string]string, onToken func(token string), onDone func(), onError func(error), ) { reqBody : map[string]interface{}{ model: gpt-4, messages: messages, stream: true, } body, _ : json.Marshal(reqBody) httpReq, _ : http.NewRequest(POST, a.baseURL/chat/completions, bytes.NewReader(body)) httpReq.Header.Set(Authorization, Bearer a.apiKey) httpReq.Header.Set(Content-Type, application/json) resp, err : a.client.Do(httpReq) if err ! nil { onError(fmt.Errorf(request failed: %w, err)) return } defer resp.Body.Close() if resp.StatusCode ! 200 { bodyBytes, _ : io.ReadAll(resp.Body) onError(fmt.Errorf(API error %d: %s, resp.StatusCode, string(bodyBytes))) return } reader : bufio.NewReader(resp.Body) for { line, err : reader.ReadString(\n) if err ! nil { if err io.EOF { onDone() return } onError(fmt.Errorf(read error: %w, err)) return } line trimSSELine(line) if line || line [DONE] { if line [DONE] { onDone() return } continue } // 解析 SSE 数据 var streamResp struct { Choices []struct { Delta struct { Content string json:content } json:delta FinishReason string json:finish_reason } json:choices } if err : json.Unmarshal([]byte(line), streamResp); err ! nil { continue } for _, choice : range streamResp.Choices { if choice.Delta.Content ! { onToken(choice.Delta.Content) } if choice.FinishReason stop { onDone() return } } } } func getBaseURL(provider string) string { switch provider { case openai: return https://api.openai.com/v1 case claude: return https://api.anthropic.com/v1 default: return https://api.openai.com/v1 } } func trimSSELine(line string) string { if len(line) 5 line[:5] data: { line line[5:] } line bytes.TrimSpace([]byte(line)) return string(line) }四、HTTP 端点实现// streaming/handler.go package streaming import ( encoding/json log net/http time ) // StreamChatHandler 流式聊天 HTTP 处理器 type StreamChatHandler struct { adapter *LLMStreamAdapter } // NewStreamChatHandler 创建处理器 func NewStreamChatHandler(adapter *LLMStreamAdapter) *StreamChatHandler { return StreamChatHandler{adapter: adapter} } // ServeHTTP 处理流式聊天请求 func (h *StreamChatHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method ! POST { http.Error(w, Method not allowed, http.StatusMethodNotAllowed) return } // 1. 解析请求 var req struct { Messages []map[string]string json:messages } if err : json.NewDecoder(r.Body).Decode(req); err ! nil { http.Error(w, Invalid request body, http.StatusBadRequest) return } // 2. 创建 SSE 写入器 sw, err : NewStreamWriter(w) if err ! nil { http.Error(w, Streaming not supported, http.StatusInternalServerError) return } // 3. 监听客户端断开 notify : r.Context().Done() go func() { -notify sw.Close() log.Println([Stream] Client disconnected) }() // 4. 发送开始事件 startData, _ : json.Marshal(map[string]string{ type: start, message: 开始生成回复, }) sw.WriteData(startData) // 5. 发起流式调用 tokenCount : 0 startTime : time.Now() h.adapter.ChatStream( req.Messages, func(token string) { // 每个 token 作为一个事件发送 tokenData, _ : json.Marshal(map[string]interface{}{ type: token, token: token, }) if err : sw.WriteData(tokenData); err ! nil { log.Printf([Stream] Write error: %v, err) return } tokenCount }, func() { // 完成事件 elapsed : time.Since(startTime) doneData, _ : json.Marshal(map[string]interface{}{ type: done, token_count: tokenCount, elapsed_ms: elapsed.Milliseconds(), tokens_per_s: float64(tokenCount) / elapsed.Seconds(), }) sw.WriteData(doneData) sw.WriteDone() log.Printf([Stream] Completed: %d tokens in %v (%.1f tok/s), tokenCount, elapsed, float64(tokenCount)/elapsed.Seconds()) }, func(err error) { log.Printf([Stream] Error: %v, err) sw.WriteError(500, err.Error()) sw.WriteDone() }, ) } // ─── 启动 HTTP 服务 ─── func StartStreamingServer(addr string) { adapter : NewLLMStreamAdapter(openai, sk-your-key) handler : NewStreamChatHandler(adapter) http.Handle(/v1/chat/stream, handler) http.Handle(/health, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte({status:ok})) })) log.Printf([Stream] Server starting on %s, addr) log.Fatal(http.ListenAndServe(addr, nil)) }五、客户端实现// streaming/client.go package streaming import ( bufio bytes encoding/json fmt io net/http strings sync ) // StreamClient SSE 客户端 type StreamClient struct { baseURL string client *http.Client } // NewStreamClient 创建客户端 func NewStreamClient(baseURL string) *StreamClient { return StreamClient{ baseURL: baseURL, client: http.Client{}, } } // ChatCompletion 流式聊天回调方式 func (c *StreamClient) ChatCompletion( messages []map[string]string, onToken func(string), onDone func(StreamStats), onError func(error), ) { reqBody, _ : json.Marshal(map[string]interface{}{ messages: messages, }) req, _ : http.NewRequest(POST, c.baseURL/v1/chat/stream, bytes.NewReader(reqBody)) req.Header.Set(Content-Type, application/json) resp, err : c.client.Do(req) if err ! nil { onError(fmt.Errorf(request failed: %w, err)) return } defer resp.Body.Close() if resp.StatusCode ! 200 { body, _ : io.ReadAll(resp.Body) onError(fmt.Errorf(server error %d: %s, resp.StatusCode, string(body))) return } // 检查 Content-Type ct : resp.Header.Get(Content-Type) if !strings.HasPrefix(ct, text/event-stream) { body, _ : io.ReadAll(resp.Body) onError(fmt.Errorf(unexpected content type: %s, body: %s, ct, string(body))) return } reader : bufio.NewReader(resp.Body) stats : StreamStats{StartTime: now()} for { line, err : reader.ReadString(\n) if err ! nil { if err io.EOF { onDone(stats) return } onError(fmt.Errorf(read error: %w, err)) return } line strings.TrimSpace(line) if !strings.HasPrefix(line, data: ) { continue } data : strings.TrimPrefix(line, data: ) if data [DONE] { onDone(stats) return } var event struct { Type string json:type Token string json:token TokenCount int json:token_count ElapsedMs int64 json:elapsed_ms } if err : json.Unmarshal([]byte(data), event); err ! nil { continue } switch event.Type { case token: onToken(event.Token) stats.TokenCount case done: stats.TokenCount event.TokenCount stats.ElapsedMs event.ElapsedMs onDone(stats) return case error: onError(fmt.Errorf(server error: %s, data)) return } } } // StreamStats 流式统计 type StreamStats struct { TokenCount int ElapsedMs int64 StartTime time.Time } // ─── JavaScript 客户端示例 ─── const jsClientCode // 前端 JS 示例 class StreamChat { constructor(url) { this.url url; this.abortController null; } async send(messages, callbacks) { const { onToken, onDone, onError } callbacks; try { const response await fetch(this.url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages }), signal: this.abortController?.signal }); if (!response.ok) { throw new Error(HTTP response.status); } const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n); buffer lines.pop() || ; for (const line of lines) { if (!line.startsWith(data: )) continue; const data line.slice(6); if (data [DONE]) { onDone?.(); return; } try { const event JSON.parse(data); switch (event.type) { case token: onToken?.(event.token); break; case done: onDone?.(event); break; case error: onError?.(new Error(event.message)); break; } } catch (e) { // skip invalid JSON } } } } catch (err) { if (err.name ! AbortError) { onError?.(err); } } } abort() { this.abortController?.abort(); } } // 使用示例 const chat new StreamChat(/v1/chat/stream); chat.send([ { role: user, content: 帮我写一封邮件 } ], { onToken: (token) { document.getElementById(output).textContent token; }, onDone: () { console.log(完成); }, onError: (err) { console.error(错误:, err); } }); 六、流式场景下的工具调用// streaming/fcall.go package streaming import ( encoding/json fmt log strings ) // StreamFCallHandler 流式 Function Calling 处理器 type StreamFCallHandler struct { fcEngine *fcall.Engine adapter *LLMStreamAdapter } // HandleStreamFCall 处理流式场景下的工具调用 func (h *StreamFCallHandler) HandleStreamFCall( messages []map[string]string, sw *StreamWriter, ) { fullResponse : strings.Builder{} pendingToolCalls : make([]*fcall.ToolCall, 0) inToolCall : false toolCallBuffer : strings.Builder{} h.adapter.ChatStream( messages, func(token string) { // 检测是否进入工具调用模式 if strings.Contains(token, tool_call) { inToolCall true toolCallBuffer.Reset() return } if inToolCall { if strings.Contains(token, /tool_call) { // 工具调用结束解析并执行 inToolCall false tc : parseToolCall(toolCallBuffer.String()) if tc ! nil { pendingToolCalls append(pendingToolCalls, tc) // 发送工具调用标记给客户端 tcData, _ : json.Marshal(map[string]interface{}{ type: tool_call, tool: tc.Name, args: string(tc.Args), }) sw.WriteData(tcData) } return } toolCallBuffer.WriteString(token) return } // 普通 token转发给客户端 fullResponse.WriteString(token) tokenData, _ : json.Marshal(map[string]interface{}{ type: token, token: token, }) sw.WriteData(tokenData) }, func() { // 首次生成完成如果有工具调用则执行 if len(pendingToolCalls) 0 { log.Printf([StreamFCall] Executing %d tool calls, len(pendingToolCalls)) // 执行工具 results : h.fcEngine.ExecuteToolCalls(nil, pendingToolCalls) // 将工具结果加入消息 for i, tr : range results { resultJSON, _ : json.Marshal(tr.Result) messages append(messages, map[string]string{ role: tool, content: string(resultJSON), }) // 发送工具结果给客户端 resData, _ : json.Marshal(map[string]interface{}{ type: tool_result, tool: tr.Name, result: tr.Result, }) sw.WriteData(resData) } // 第二次流式生成 h.adapter.ChatStream( messages, func(token string) { fullResponse.WriteString(token) tokenData, _ : json.Marshal(map[string]interface{}{ type: token, token: token, }) sw.WriteData(tokenData) }, func() { // 最终完成 doneData, _ : json.Marshal(map[string]interface{}{ type: done, full_response: fullResponse.String(), }) sw.WriteData(doneData) sw.WriteDone() }, func(err error) { sw.WriteError(500, err.Error()) sw.WriteDone() }, ) return } // 没有工具调用直接完成 doneData, _ : json.Marshal(map[string]interface{}{ type: done, full_response: fullResponse.String(), }) sw.WriteData(doneData) sw.WriteDone() }, func(err error) { sw.WriteError(500, err.Error()) sw.WriteDone() }, ) } // parseToolCall 解析工具调用标记 func parseToolCall(raw string) *fcall.ToolCall { raw strings.TrimSpace(raw) // 期望格式: {name: xxx, args: {...}} var tc struct { Name string json:name Args json.RawMessage json:args } if err : json.Unmarshal([]byte(raw), tc); err ! nil { log.Printf([StreamFCall] Failed to parse tool call: %v, err) return nil } return fcall.ToolCall{ ID: fmt.Sprintf(call_stream_%d, time.Now().UnixNano()), Name: tc.Name, Args: tc.Args, } }七、性能优化// streaming/optimize.go package streaming import ( sync time ) // TokenBucket 令牌桶限流发送频率控制 type TokenBucket struct { rate float64 burst int tokens float64 lastRefill time.Time mu sync.Mutex } func NewTokenBucket(rate float64, burst int) *TokenBucket { return TokenBucket{ rate: rate, burst: burst, tokens: float64(burst), lastRefill: time.Now(), } } func (tb *TokenBucket) Allow() bool { tb.mu.Lock() defer tb.mu.Unlock() now : time.Now() elapsed : now.Sub(tb.lastRefill).Seconds() tb.tokens min(float64(tb.burst), tb.tokenselapsed*tb.rate) tb.lastRefill now if tb.tokens 1 { tb.tokens-- return true } return false } // BatchSender 批量发送器合并小 token 减少 I/O type BatchSender struct { sw *StreamWriter buffer []byte maxBatch int timer *time.Timer mu sync.Mutex } func NewBatchSender(sw *StreamWriter, maxBatch int, flushInterval time.Duration) *BatchSender { bs : BatchSender{ sw: sw, maxBatch: maxBatch, } bs.timer time.AfterFunc(flushInterval, bs.flush) return bs } func (bs *BatchSender) Add(token string) { bs.mu.Lock() defer bs.mu.Unlock() bs.buffer append(bs.buffer, []byte(token)...) if len(bs.buffer) bs.maxBatch { bs.flushNow() } } func (bs *BatchSender) flush() { bs.mu.Lock() defer bs.mu.Unlock() bs.flushNow() } func (bs *BatchSender) flushNow() { if len(bs.buffer) 0 { return } data, _ : json.Marshal(map[string]interface{}{ type: batch, tokens: string(bs.buffer), }) bs.sw.WriteData(data) bs.buffer bs.buffer[:0] } func (bs *BatchSender) Close() { bs.timer.Stop() bs.flush() } // 心跳机制防止代理/负载均衡器断开连接 func startHeartbeat(sw *StreamWriter, interval time.Duration, stop chan struct{}) { ticker : time.NewTicker(interval) defer ticker.Stop() for { select { case -ticker.C: sw.WriteEvent(StreamEvent{ Event: heartbeat, Data: json.RawMessage({}), }) case -stop: return } } }八、生产坑位#坑位现象解决方案1代理缓冲Nginx/反向代理缓存 SSE 数据客户端收不到流式效果设置X-Accel-Buffering: no头2连接断开用户刷新页面或网络波动导致连接中断客户端自动重连 服务端心跳保活3并发写入多个 goroutine 同时写入 ResponseWriter 导致 panic用 mutex 保护写入操作4内存泄漏客户端断开后 goroutine 仍在运行监听 Context.Done() 及时清理资源5Token 碎片每个 token 单独发送导致大量小包 I/O合并批次发送100ms 窗口或累积 50 字符6工具调用时序流式过程中插入工具调用打乱前端展示逻辑使用事件类型区分 token/tool_call/tool_result九、本讲小结┌─────────────────────────────────────────────────────────────┐ │ 第6讲 核心收获 │ │ │ │ ✅ 理解了 SSE 协议原理和数据格式 │ │ ✅ 实现了服务端 StreamWriter写入/心跳/关闭 │ │ ✅ 搭建了 LLM 流式适配器OpenAI 兼容 │ │ ✅ 实现了 HTTP 端点 客户端Go/JavaScript │ │ ✅ 实现了流式场景下的 Function Calling 集成 │ │ ✅ 完成了性能优化令牌桶限流、批量发送、心跳保活 │ │ ✅ 了解了 6 个常见坑位及其解决方案 │ │ │ │ 核心理念Streaming 将用户等待时间从转圈圈变为逐字阅读 │ └─────────────────────────────────────────────────────────────┘下一讲预告《可观测性与 AI 监控——你的 Agent 到底在干什么》你会学到LLM 调用链路追踪、Token 用量监控、响应质量评估、告警体系搭建、日志审计与合规。准备好了告诉我继续。 
长河君 · 2026/10/31 2:05:05 · 阅读 1.2k 第6讲：Streaming 优化——告别“转圈圈“的用户体验
一、为什么需要 Streaming非流式传统方式 ┌──────────────────────────────────────────────┐ │ 用户帮我写一封邮件给客户 │ │ │ │ 时间轴 │ │ 0s ──── 发送请求 │ │ 5s ──── 还在等 │ │ 10s ─── 还在等 │ │ 15s ─── 收到完整回复用户已经不耐烦了 │ └──────────────────────────────────────────────┘ 流式Streaming ┌──────────────────────────────────────────────┐ │ 用户帮我写一封邮件给客户 │ │ │ │ 时间轴 │ │ 0s ──── 发送请求 │ │ 1s ──── 看到好的 │ │ 2s ──── 看到我正在为您起草邮件 │ │ 3s ──── 看到尊敬的客户 │ │ ... ──── 逐字显示用户感知几乎无延迟 │ └──────────────────────────────────────────────┘Streaming 的核心价值将 TTFB首字节时间从 10-20s 降到 1-2s大幅提升用户体验。二、SSE 协议基础SSEServer-Sent Events是 Streaming 的标准协议基于 HTTP 长连接。SSE 数据格式 ─────────────────────────────────────────────── data: {choices:[{delta:{content:你好}}]} data: {choices:[{delta:{content:}}]} data: {choices:[{delta:{content:世界}}]} data: [DONE] ─────────────────────────────────────────────── 特点 1. 每个事件以 data: 开头 2. 事件之间用空行分隔 3. 以 [DONE] 标记结束 4. 浏览器原生支持 EventSource API三、服务端 Streaming 实现// streaming/server.go package streaming import ( bufio bytes encoding/json fmt io log net/http sync time ) // StreamEvent SSE 事件 type StreamEvent struct { ID string json:id,omitempty Event string json:event,omitempty Data json.RawMessage json:data Retry int json:retry,omitempty // 重连间隔毫秒 } // StreamWriter SSE 写入器 type StreamWriter struct { writer http.ResponseWriter flusher http.Flusher mu sync.Mutex closed bool } // NewStreamWriter 创建 SSE 写入器 func NewStreamWriter(w http.ResponseWriter) (*StreamWriter, error) { flusher, ok : w.(http.Flusher) if !ok { return nil, fmt.Errorf(streaming not supported) } // 设置 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(Access-Control-Allow-Origin, *) w.Header().Set(X-Accel-Buffering, no) // 禁用 Nginx 缓冲 return StreamWriter{ writer: w, flusher: flusher, }, nil } // WriteEvent 写入一个 SSE 事件 func (sw *StreamWriter) WriteEvent(event StreamEvent) error { sw.mu.Lock() defer sw.mu.Unlock() if sw.closed { return fmt.Errorf(stream already closed) } var buf bytes.Buffer if event.ID ! { buf.WriteString(fmt.Sprintf(id: %s\n, event.ID)) } if event.Event ! { buf.WriteString(fmt.Sprintf(event: %s\n, event.Event)) } if event.Retry 0 { buf.WriteString(fmt.Sprintf(retry: %d\n, event.Retry)) } // data 字段支持多行 dataStr : string(event.Data) for _, line : range bytes.Split([]byte(dataStr), []byte(\n)) { buf.WriteString(fmt.Sprintf(data: %s\n, line)) } buf.WriteString(\n) // 空行分隔事件 _, err : sw.writer.Write(buf.Bytes()) if err ! nil { return err } sw.flusher.Flush() return nil } // WriteData 写入纯数据最常用 func (sw *StreamWriter) WriteData(data []byte) error { return sw.WriteEvent(StreamEvent{ Data: data, }) } // WriteDone 写入结束标记 func (sw *StreamWriter) WriteDone() error { sw.mu.Lock() defer sw.mu.Unlock() if sw.closed { return nil } _, err : sw.writer.Write([]byte(data: [DONE]\n\n)) sw.closed true if err ! nil { return err } sw.flusher.Flush() return nil } // WriteError 写入错误事件 func (sw *StreamWriter) WriteError(code int, message string) error { errData, _ : json.Marshal(map[string]interface{}{ error: map[string]interface{}{ code: code, message: message, }, }) return sw.WriteData(errData) } // Close 关闭流 func (sw *StreamWriter) Close() { sw.mu.Lock() defer sw.mu.Unlock() sw.closed true } // ─── LLM Streaming Adapter ─── // LLMStreamAdapter LLM 流式适配器 type LLMStreamAdapter struct { provider string apiKey string baseURL string client *http.Client } // NewLLMStreamAdapter 创建流式适配器 func NewLLMStreamAdapter(provider, apiKey string) *LLMStreamAdapter { return LLMStreamAdapter{ provider: provider, apiKey: apiKey, baseURL: getBaseURL(provider), client: http.Client{Timeout: 60 * time.Second}, } } // ChatStream 流式聊天 func (a *LLMStreamAdapter) ChatStream( messages []map[string]string, onToken func(token string), onDone func(), onError func(error), ) { reqBody : map[string]interface{}{ model: gpt-4, messages: messages, stream: true, } body, _ : json.Marshal(reqBody) httpReq, _ : http.NewRequest(POST, a.baseURL/chat/completions, bytes.NewReader(body)) httpReq.Header.Set(Authorization, Bearer a.apiKey) httpReq.Header.Set(Content-Type, application/json) resp, err : a.client.Do(httpReq) if err ! nil { onError(fmt.Errorf(request failed: %w, err)) return } defer resp.Body.Close() if resp.StatusCode ! 200 { bodyBytes, _ : io.ReadAll(resp.Body) onError(fmt.Errorf(API error %d: %s, resp.StatusCode, string(bodyBytes))) return } reader : bufio.NewReader(resp.Body) for { line, err : reader.ReadString(\n) if err ! nil { if err io.EOF { onDone() return } onError(fmt.Errorf(read error: %w, err)) return } line trimSSELine(line) if line || line [DONE] { if line [DONE] { onDone() return } continue } // 解析 SSE 数据 var streamResp struct { Choices []struct { Delta struct { Content string json:content } json:delta FinishReason string json:finish_reason } json:choices } if err : json.Unmarshal([]byte(line), streamResp); err ! nil { continue } for _, choice : range streamResp.Choices { if choice.Delta.Content ! { onToken(choice.Delta.Content) } if choice.FinishReason stop { onDone() return } } } } func getBaseURL(provider string) string { switch provider { case openai: return https://api.openai.com/v1 case claude: return https://api.anthropic.com/v1 default: return https://api.openai.com/v1 } } func trimSSELine(line string) string { if len(line) 5 line[:5] data: { line line[5:] } line bytes.TrimSpace([]byte(line)) return string(line) }四、HTTP 端点实现// streaming/handler.go package streaming import ( encoding/json log net/http time ) // StreamChatHandler 流式聊天 HTTP 处理器 type StreamChatHandler struct { adapter *LLMStreamAdapter } // NewStreamChatHandler 创建处理器 func NewStreamChatHandler(adapter *LLMStreamAdapter) *StreamChatHandler { return StreamChatHandler{adapter: adapter} } // ServeHTTP 处理流式聊天请求 func (h *StreamChatHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method ! POST { http.Error(w, Method not allowed, http.StatusMethodNotAllowed) return } // 1. 解析请求 var req struct { Messages []map[string]string json:messages } if err : json.NewDecoder(r.Body).Decode(req); err ! nil { http.Error(w, Invalid request body, http.StatusBadRequest) return } // 2. 创建 SSE 写入器 sw, err : NewStreamWriter(w) if err ! nil { http.Error(w, Streaming not supported, http.StatusInternalServerError) return } // 3. 监听客户端断开 notify : r.Context().Done() go func() { -notify sw.Close() log.Println([Stream] Client disconnected) }() // 4. 发送开始事件 startData, _ : json.Marshal(map[string]string{ type: start, message: 开始生成回复, }) sw.WriteData(startData) // 5. 发起流式调用 tokenCount : 0 startTime : time.Now() h.adapter.ChatStream( req.Messages, func(token string) { // 每个 token 作为一个事件发送 tokenData, _ : json.Marshal(map[string]interface{}{ type: token, token: token, }) if err : sw.WriteData(tokenData); err ! nil { log.Printf([Stream] Write error: %v, err) return } tokenCount }, func() { // 完成事件 elapsed : time.Since(startTime) doneData, _ : json.Marshal(map[string]interface{}{ type: done, token_count: tokenCount, elapsed_ms: elapsed.Milliseconds(), tokens_per_s: float64(tokenCount) / elapsed.Seconds(), }) sw.WriteData(doneData) sw.WriteDone() log.Printf([Stream] Completed: %d tokens in %v (%.1f tok/s), tokenCount, elapsed, float64(tokenCount)/elapsed.Seconds()) }, func(err error) { log.Printf([Stream] Error: %v, err) sw.WriteError(500, err.Error()) sw.WriteDone() }, ) } // ─── 启动 HTTP 服务 ─── func StartStreamingServer(addr string) { adapter : NewLLMStreamAdapter(openai, sk-your-key) handler : NewStreamChatHandler(adapter) http.Handle(/v1/chat/stream, handler) http.Handle(/health, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Write([]byte({status:ok})) })) log.Printf([Stream] Server starting on %s, addr) log.Fatal(http.ListenAndServe(addr, nil)) }五、客户端实现// streaming/client.go package streaming import ( bufio bytes encoding/json fmt io net/http strings sync ) // StreamClient SSE 客户端 type StreamClient struct { baseURL string client *http.Client } // NewStreamClient 创建客户端 func NewStreamClient(baseURL string) *StreamClient { return StreamClient{ baseURL: baseURL, client: http.Client{}, } } // ChatCompletion 流式聊天回调方式 func (c *StreamClient) ChatCompletion( messages []map[string]string, onToken func(string), onDone func(StreamStats), onError func(error), ) { reqBody, _ : json.Marshal(map[string]interface{}{ messages: messages, }) req, _ : http.NewRequest(POST, c.baseURL/v1/chat/stream, bytes.NewReader(reqBody)) req.Header.Set(Content-Type, application/json) resp, err : c.client.Do(req) if err ! nil { onError(fmt.Errorf(request failed: %w, err)) return } defer resp.Body.Close() if resp.StatusCode ! 200 { body, _ : io.ReadAll(resp.Body) onError(fmt.Errorf(server error %d: %s, resp.StatusCode, string(body))) return } // 检查 Content-Type ct : resp.Header.Get(Content-Type) if !strings.HasPrefix(ct, text/event-stream) { body, _ : io.ReadAll(resp.Body) onError(fmt.Errorf(unexpected content type: %s, body: %s, ct, string(body))) return } reader : bufio.NewReader(resp.Body) stats : StreamStats{StartTime: now()} for { line, err : reader.ReadString(\n) if err ! nil { if err io.EOF { onDone(stats) return } onError(fmt.Errorf(read error: %w, err)) return } line strings.TrimSpace(line) if !strings.HasPrefix(line, data: ) { continue } data : strings.TrimPrefix(line, data: ) if data [DONE] { onDone(stats) return } var event struct { Type string json:type Token string json:token TokenCount int json:token_count ElapsedMs int64 json:elapsed_ms } if err : json.Unmarshal([]byte(data), event); err ! nil { continue } switch event.Type { case token: onToken(event.Token) stats.TokenCount case done: stats.TokenCount event.TokenCount stats.ElapsedMs event.ElapsedMs onDone(stats) return case error: onError(fmt.Errorf(server error: %s, data)) return } } } // StreamStats 流式统计 type StreamStats struct { TokenCount int ElapsedMs int64 StartTime time.Time } // ─── JavaScript 客户端示例 ─── const jsClientCode // 前端 JS 示例 class StreamChat { constructor(url) { this.url url; this.abortController null; } async send(messages, callbacks) { const { onToken, onDone, onError } callbacks; try { const response await fetch(this.url, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ messages }), signal: this.abortController?.signal }); if (!response.ok) { throw new Error(HTTP response.status); } const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n); buffer lines.pop() || ; for (const line of lines) { if (!line.startsWith(data: )) continue; const data line.slice(6); if (data [DONE]) { onDone?.(); return; } try { const event JSON.parse(data); switch (event.type) { case token: onToken?.(event.token); break; case done: onDone?.(event); break; case error: onError?.(new Error(event.message)); break; } } catch (e) { // skip invalid JSON } } } } catch (err) { if (err.name ! AbortError) { onError?.(err); } } } abort() { this.abortController?.abort(); } } // 使用示例 const chat new StreamChat(/v1/chat/stream); chat.send([ { role: user, content: 帮我写一封邮件 } ], { onToken: (token) { document.getElementById(output).textContent token; }, onDone: () { console.log(完成); }, onError: (err) { console.error(错误:, err); } }); 六、流式场景下的工具调用// streaming/fcall.go package streaming import ( encoding/json fmt log strings ) // StreamFCallHandler 流式 Function Calling 处理器 type StreamFCallHandler struct { fcEngine *fcall.Engine adapter *LLMStreamAdapter } // HandleStreamFCall 处理流式场景下的工具调用 func (h *StreamFCallHandler) HandleStreamFCall( messages []map[string]string, sw *StreamWriter, ) { fullResponse : strings.Builder{} pendingToolCalls : make([]*fcall.ToolCall, 0) inToolCall : false toolCallBuffer : strings.Builder{} h.adapter.ChatStream( messages, func(token string) { // 检测是否进入工具调用模式 if strings.Contains(token, tool_call) { inToolCall true toolCallBuffer.Reset() return } if inToolCall { if strings.Contains(token, /tool_call) { // 工具调用结束解析并执行 inToolCall false tc : parseToolCall(toolCallBuffer.String()) if tc ! nil { pendingToolCalls append(pendingToolCalls, tc) // 发送工具调用标记给客户端 tcData, _ : json.Marshal(map[string]interface{}{ type: tool_call, tool: tc.Name, args: string(tc.Args), }) sw.WriteData(tcData) } return } toolCallBuffer.WriteString(token) return } // 普通 token转发给客户端 fullResponse.WriteString(token) tokenData, _ : json.Marshal(map[string]interface{}{ type: token, token: token, }) sw.WriteData(tokenData) }, func() { // 首次生成完成如果有工具调用则执行 if len(pendingToolCalls) 0 { log.Printf([StreamFCall] Executing %d tool calls, len(pendingToolCalls)) // 执行工具 results : h.fcEngine.ExecuteToolCalls(nil, pendingToolCalls) // 将工具结果加入消息 for i, tr : range results { resultJSON, _ : json.Marshal(tr.Result) messages append(messages, map[string]string{ role: tool, content: string(resultJSON), }) // 发送工具结果给客户端 resData, _ : json.Marshal(map[string]interface{}{ type: tool_result, tool: tr.Name, result: tr.Result, }) sw.WriteData(resData) } // 第二次流式生成 h.adapter.ChatStream( messages, func(token string) { fullResponse.WriteString(token) tokenData, _ : json.Marshal(map[string]interface{}{ type: token, token: token, }) sw.WriteData(tokenData) }, func() { // 最终完成 doneData, _ : json.Marshal(map[string]interface{}{ type: done, full_response: fullResponse.String(), }) sw.WriteData(doneData) sw.WriteDone() }, func(err error) { sw.WriteError(500, err.Error()) sw.WriteDone() }, ) return } // 没有工具调用直接完成 doneData, _ : json.Marshal(map[string]interface{}{ type: done, full_response: fullResponse.String(), }) sw.WriteData(doneData) sw.WriteDone() }, func(err error) { sw.WriteError(500, err.Error()) sw.WriteDone() }, ) } // parseToolCall 解析工具调用标记 func parseToolCall(raw string) *fcall.ToolCall { raw strings.TrimSpace(raw) // 期望格式: {name: xxx, args: {...}} var tc struct { Name string json:name Args json.RawMessage json:args } if err : json.Unmarshal([]byte(raw), tc); err ! nil { log.Printf([StreamFCall] Failed to parse tool call: %v, err) return nil } return fcall.ToolCall{ ID: fmt.Sprintf(call_stream_%d, time.Now().UnixNano()), Name: tc.Name, Args: tc.Args, } }七、性能优化// streaming/optimize.go package streaming import ( sync time ) // TokenBucket 令牌桶限流发送频率控制 type TokenBucket struct { rate float64 burst int tokens float64 lastRefill time.Time mu sync.Mutex } func NewTokenBucket(rate float64, burst int) *TokenBucket { return TokenBucket{ rate: rate, burst: burst, tokens: float64(burst), lastRefill: time.Now(), } } func (tb *TokenBucket) Allow() bool { tb.mu.Lock() defer tb.mu.Unlock() now : time.Now() elapsed : now.Sub(tb.lastRefill).Seconds() tb.tokens min(float64(tb.burst), tb.tokenselapsed*tb.rate) tb.lastRefill now if tb.tokens 1 { tb.tokens-- return true } return false } // BatchSender 批量发送器合并小 token 减少 I/O type BatchSender struct { sw *StreamWriter buffer []byte maxBatch int timer *time.Timer mu sync.Mutex } func NewBatchSender(sw *StreamWriter, maxBatch int, flushInterval time.Duration) *BatchSender { bs : BatchSender{ sw: sw, maxBatch: maxBatch, } bs.timer time.AfterFunc(flushInterval, bs.flush) return bs } func (bs *BatchSender) Add(token string) { bs.mu.Lock() defer bs.mu.Unlock() bs.buffer append(bs.buffer, []byte(token)...) if len(bs.buffer) bs.maxBatch { bs.flushNow() } } func (bs *BatchSender) flush() { bs.mu.Lock() defer bs.mu.Unlock() bs.flushNow() } func (bs *BatchSender) flushNow() { if len(bs.buffer) 0 { return } data, _ : json.Marshal(map[string]interface{}{ type: batch, tokens: string(bs.buffer), }) bs.sw.WriteData(data) bs.buffer bs.buffer[:0] } func (bs *BatchSender) Close() { bs.timer.Stop() bs.flush() } // 心跳机制防止代理/负载均衡器断开连接 func startHeartbeat(sw *StreamWriter, interval time.Duration, stop chan struct{}) { ticker : time.NewTicker(interval) defer ticker.Stop() for { select { case -ticker.C: sw.WriteEvent(StreamEvent{ Event: heartbeat, Data: json.RawMessage({}), }) case -stop: return } } }八、生产坑位#坑位现象解决方案1代理缓冲Nginx/反向代理缓存 SSE 数据客户端收不到流式效果设置X-Accel-Buffering: no头2连接断开用户刷新页面或网络波动导致连接中断客户端自动重连 服务端心跳保活3并发写入多个 goroutine 同时写入 ResponseWriter 导致 panic用 mutex 保护写入操作4内存泄漏客户端断开后 goroutine 仍在运行监听 Context.Done() 及时清理资源5Token 碎片每个 token 单独发送导致大量小包 I/O合并批次发送100ms 窗口或累积 50 字符6工具调用时序流式过程中插入工具调用打乱前端展示逻辑使用事件类型区分 token/tool_call/tool_result九、本讲小结┌─────────────────────────────────────────────────────────────┐ │ 第6讲 核心收获 │ │ │ │ ✅ 理解了 SSE 协议原理和数据格式 │ │ ✅ 实现了服务端 StreamWriter写入/心跳/关闭 │ │ ✅ 搭建了 LLM 流式适配器OpenAI 兼容 │ │ ✅ 实现了 HTTP 端点 客户端Go/JavaScript │ │ ✅ 实现了流式场景下的 Function Calling 集成 │ │ ✅ 完成了性能优化令牌桶限流、批量发送、心跳保活 │ │ ✅ 了解了 6 个常见坑位及其解决方案 │ │ │ │ 核心理念Streaming 将用户等待时间从转圈圈变为逐字阅读 │ └─────────────────────────────────────────────────────────────┘下一讲预告《可观测性与 AI 监控——你的 Agent 到底在干什么》你会学到LLM 调用链路追踪、Token 用量监控、响应质量评估、告警体系搭建、日志审计与合规。准备好了告诉我继续。