流式输出的深度剖析
之前一直在调用openai的key，只是照着文档进行流式调用，也只知其确是流式与api有所不同，而未成体系深究其实现原理。
就以openai的官方流式输出为切入。
概述
流式输出（Streaming Output）是 HTTP 响应中的一种模式，服务器可以在生成部分内容时立即将这些内容发送给客户端，而无需等待整个响应内容生成完成。这种方式常用于实时交互、高延迟操作或长时间任务中，比如 OpenAI 的 GPT 模型生成流式对话。

1. package main
   
2. 
   
3. import (
   
4.         "bufio"
   
5.         "bytes"
   
6.         "encoding/json"
   
7.         "fmt"
   
8.         "net/http"
   
9.         "strings"
   
10.         "time"
    
11. )
    
12. 
    
13. // 定义必要的数据结构
    
14. type Message struct {
    
15.         Role    string `json:"role"`
    
16.         Content string `json:"content"`
    
17. }
    
18. 
    
19. type RequestBody struct {
    
20.         Model       string    `json:"model"`
    
21.         Messages    []Message `json:"messages"`
    
22.         Temperature float64   `json:"temperature"`
    
23.         Stream      bool      `json:"stream"`
    
24. }
    
25. 
    
26. type Choice struct {
    
27.         Delta struct {
    
28.                 Content string `json:"content"`
    
29.         } `json:"delta"`
    
30. }
    
31. 
    
32. type ResponseBody struct {
    
33.         Choices []Choice `json:"choices"`
    
34. }
    
35. 
    
36. const (
    
37.         apiURL      = "https://api.example.com/v1/chat/completions" // 替换为实际的 API 地址
    
38.         authToken   = "your-auth-token"                             // 替换为实际的 Token
    
39.         model       = "gpt-3.5-turbo"
    
40.         temperature = 0.7
    
41. )
    
42. 
    
43. func StreamHandler(w http.ResponseWriter, r *http.Request) {
    
44.         // 从查询参数获取输入内容
    
45.         content := r.URL.Query().Get("content")
    
46.         if content == "" {
    
47.                 http.Error(w, "Missing 'content' parameter", http.StatusBadRequest)
    
48.                 return
    
49.         }
    
50. 
    
51.         // 构造请求体
    
52.         message := Message{
    
53.                 Role:    "user",
    
54.                 Content: content,
    
55.         }
    
56.         requestBody := RequestBody{
    
57.                 Model:       model,
    
58.                 Messages:    []Message{message},
    
59.                 Temperature: temperature,
    
60.                 Stream:      true,
    
61.         }
    
62.         jsonData, err := json.Marshal(requestBody)
    
63.         if err != nil {
    
64.                 http.Error(w, "Failed to marshal request body", http.StatusInternalServerError)
    
65.                 return
    
66.         }
    
67. 
    
68.         // 创建 HTTP 请求
    
69.         req, err := http.NewRequest("POST", apiURL, bytes.NewBuffer(jsonData))
    
70.         if err != nil {
    
71.                 http.Error(w, "Failed to create request", http.StatusInternalServerError)
    
72.                 return
    
73.         }
    
74.         req.Header.Set("Content-Type", "application/json")
    
75.         req.Header.Set("Authorization", "Bearer "+authToken)
    
76. 
    
77.         // 设置 HTTP 客户端
    
78.         client := &http.Client{Timeout: time.Second * 50}
    
79.         resp, err := client.Do(req)
    
80.         if err != nil {
    
81.                 http.Error(w, "Failed to get response", http.StatusInternalServerError)
    
82.                 return
    
83.         }
    
84.         defer resp.Body.Close()
    
85. 
    
86.         // 设置响应头，开启流式输出
    
87.         w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
    
88.         w.Header().Set("Cache-Control", "no-cache")
    
89.         w.Header().Set("Connection", "keep-alive")
    
90. 
    
91.         // 确保 ResponseWriter 支持 Flusher
    
92.         flusher, ok := w.(http.Flusher)
    
93.         if !ok {
    
94.                 http.Error(w, "Streaming unsupported", http.StatusInternalServerError)
    
95.                 return
    
96.         }
    
97. 
    
98.         // 处理流式响应
    
99.         scanner := bufio.NewScanner(resp.Body)
    
100.         for scanner.Scan() {
     
101.                 line := scanner.Text()
     
102. 
     
103.                 // 处理以 "data: " 开头的行
     
104.                 if strings.HasPrefix(line, "data: ") {
     
105.                         line = strings.TrimPrefix(line, "data: ")
     
106.                 }
     
107.                 if line == "[DONE]" {
     
108.                         break
     
109.                 }
     
110.                 if line == "" {
     
111.                         continue
     
112.                 }
     
113. 
     
114.                 // 解析响应内容
     
115.                 var chunk ResponseBody
     
116.                 if err := json.Unmarshal([]byte(line), &chunk); err != nil {
     
117.                         continue
     
118.                 }
     
119. 
     
120.                 // 将响应数据逐步发送给客户端
     
121.                 for _, choice := range chunk.Choices {
     
122.                         content := choice.Delta.Content
     
123.                         _, err := w.Write([]byte(content))
     
124.                         if err != nil {
     
125.                                 http.Error(w, "Failed to write response", http.StatusInternalServerError)
     
126.                                 return
     
127.                         }
     
128.                         flusher.Flush() // 刷新缓冲区
     
129.                 }
     
130.         }
     
131. 
     
132.         if err := scanner.Err(); err != nil {
     
133.                 http.Error(w, "Scanner error", http.StatusInternalServerError)
     
134.                 return
     
135.         }
     
136. }
     
137. 
     
138. func main() {
     
139.         http.HandleFunc("/stream", StreamHandler)
     
140.         fmt.Println("Server started at :8080")
     
141.         http.ListenAndServe(":8080", nil)
     
142. }

复制代码

核心流程
接收到用户输入后，将其作为 content 参数发送给目标 API。
开启流式输出模式，设置 Stream: true。
使用 http.Flusher 将从远程接口接收到的内容逐步发送给客户端。
关键点
1.流式响应头设置

1. w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
   
2. w.Header().Set("Cache-Control", "no-cache")
   
3. w.Header().Set("Connection", "keep-alive")

复制代码

实时输出： 通过 w.Write 输出内容后调用 flusher.Flush() 确保数据实时发送。
启动服务后，通过浏览器访问类似以下 URL：

1. http://localhost:8080/stream?content=Hello%20world

复制代码

客户端会逐步接收内容，类似命令行实时打印。
1. HTTP 协议中的流式响应
流式输出利用 HTTP 协议的特性，不关闭连接，逐步将数据发送给客户端。典型流式响应会设置如下 HTTP Header：
Content-Type: text/event-stream表示这是一个事件流（Event Stream），用于向客户端连续发送数据片段。
Cache-Control: no-cache防止响应被缓存，以确保客户端接收到实时内容。
Connection: keep-alive 保持连接处于活跃状态，支持多次数据传输。
2. 流式输出的工作原理
客户端发起请求，服务器在接收到请求后开始响应。
服务器不一次性生成完整的响应内容，而是将生成的部分数据逐段发送。
客户端收到数据后立即处理，而无需等待完整响应结束。
在数据发送完成后，服务器可以选择关闭连接或保持连接以发送后续数据。
流式输出的常见应用场景
实时聊天：聊天模型逐词/逐句生成时，可以实时传输数据。
日志监控：将服务器的实时日志逐行推送到前端。
流式文件传输：如大文件或视频流传输。
实时进度更新：如任务进度条更新。
到此这篇关于使用Golang实现流式输出的文章就介绍到这了,更多相关Golang流式输出内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

来源：互联网
免责声明：如果侵犯了您的权益，请联系站长（1277306191@qq.com），我们会及时删除侵权内容，谢谢合作！