一、为什么需要 Streaming

非流式（传统方式）： ┌──────────────────────────────────────────────┐ │ 用户：帮我写一封邮件给客户 │ │ │ │ 时间轴： │ │ 0s ──── 发送请求 │ │ 5s ──── 还在等 │ │ 10s ─── 还在等 │ │ 15s ─── 收到完整回复（用户已经不耐烦了） │ └──────────────────────────────────────────────┘ 流式（Streaming）： ┌──────────────────────────────────────────────┐ │ 用户：帮我写一封邮件给客户 │ │ │ │ 时间轴： │ │ 0s ──── 发送请求 │ │ 1s ──── 看到"好的，" │ │ 2s ──── 看到"我正在为您起草邮件" │ │ 3s ──── 看到"尊敬的客户，" │ │ ... ──── 逐字显示，用户感知几乎无延迟 │ └──────────────────────────────────────────────┘

Streaming 的核心价值：将 TTFB（首字节时间）从 10-20s 降到 1-2s，大幅提升用户体验。

二、SSE 协议基础

SSE（Server-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.tokens+elapsed*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 } } }

八、生产坑位

九、本讲小结

┌─────────────────────────────────────────────────────────────┐ │ 第6讲 核心收获 │ │ │ │ ✅ 理解了 SSE 协议原理和数据格式 │ │ ✅ 实现了服务端 StreamWriter（写入/心跳/关闭） │ │ ✅ 搭建了 LLM 流式适配器（OpenAI 兼容） │ │ ✅ 实现了 HTTP 端点 + 客户端（Go/JavaScript） │ │ ✅ 实现了流式场景下的 Function Calling 集成 │ │ ✅ 完成了性能优化（令牌桶限流、批量发送、心跳保活） │ │ ✅ 了解了 6 个常见坑位及其解决方案 │ │ │ │ 核心理念：Streaming 将用户等待时间从"转圈圈"变为"逐字阅读" │ └─────────────────────────────────────────────────────────────┘

下一讲预告：《可观测性与 AI 监控——你的 Agent 到底在干什么》

你会学到：LLM 调用链路追踪、Token 用量监控、响应质量评估、告警体系搭建、日志审计与合规。

准备好了告诉我"继续"。