Streaming

百科 · 模型与对话 · 边生成边吐

前面几次都是 "stream": false：服务器把全文攒完，一次性丢回一大团 JSON。改成 true，连接不关，按 SSE（Server-Sent Events）一块块往外推。还是同一支 /chat/completions，不是新接口。

一次流式调用

curl https://api.deepseek.com/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $DEEPSEEK_API_KEY" \
  -N \
  -d '{
    "model": "deepseek-chat",
    "messages": [
      {"role": "user", "content": "你是谁"}
    ],
    "stream": true
  }'

-N 关掉缓冲，否则 curl 可能把碎块攒成一团再给你看。响应头是 Content-Type: text/event-stream。正文长这样：

data: {"id":"chatcmpl-a1","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"我"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{"content":"是"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{"content":" DeepSeek"},"finish_reason":null}]}

data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

翻译成人话：

  • 每一行 data: 后面是一小块 JSON，字段是 delta，不是整段 message
  • 把各块 delta.content 按顺序拼起来，才是「我是 DeepSeek……」
  • 最后一块带 finish_reason。为 stop 是它自己停的；为 length 仍是撞上 max_tokens
  • data: [DONE] 表示这条流结束。这是 DeepSeek / OpenAI 的接口约定，不是 SSE 规范里的事件

choices 仍然通常只有一项。流式不会变成多个回答，只是把一个回答切碎了送。

SSE 是什么

SSE 是一种单向推送：服务器往客户端写，客户端只读。不是 WebSocket，你不会在这条连接上把下一句用户话塞回去。

约定很短：HTTP 长连接，Content-Type: text/event-stream。一个事件是若干行字段，后面跟一个空行。客户端按空行切开。

几种事件类型

一条事件由这几类行组成。没写 event: 时，类型就是默认的 message。

  • event：事件名。省略则是 message。同名的才进同一个回调
  • data：正文。可以连写多行，客户端用换行拼成一整段，再解析
  • id：这条事件的编号。断线重连时，客户端用请求头 Last-Event-ID 带回去
  • retry：建议的重连间隔，单位毫秒

以 : 开头的是注释，不算事件。

DeepSeek 的流几乎只写 data:，所以类型都是 message。[DONE] 是 DeepSeek / OpenAI 放在 data 里的结束标记，不是另一种 event，SSE 规范里也没有它。

用 net/http 收

标准库没有 SSE 客户端。自己按行读 resp.Body 即可：

body := bytes.NewReader([]byte(`{
  "model": "deepseek-chat",
  "messages": [{"role": "user", "content": "你是谁"}],
  "stream": true
}`))
req, err := http.NewRequest(http.MethodPost, "https://api.deepseek.com/chat/completions", body)
if err != nil {
	return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+os.Getenv("DEEPSEEK_API_KEY"))

resp, err := http.DefaultClient.Do(req)
if err != nil {
	return err
}
defer resp.Body.Close()

sc := bufio.NewScanner(resp.Body)
for sc.Scan() {
	line := sc.Text()
	if !strings.HasPrefix(line, "data:") {
		continue
	}
	data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
	if data == "[DONE]" {
		break
	}
	fmt.Println(data)
}
return sc.Err()

空行、注释行直接跳过。只认 data:。拼 delta.content 仍是你自己的事，这里只把每一块 JSON 打出来。

这段只处理了 data。生产里还要断线重连、用 id 做断点续传、按自定义 event 分发、遵守 retry。这些不必手写。r3labs/sse 的 master 不能指定请求方法，而这支接口是 POST，所以用 resty v3 的 SSE 客户端。

用 resty 收

NewSSESource 默认是 GET。SetMethod 改成 POST，再带上 JSON body。收到 [DONE] 要 Close()，否则它会按重连再打一次，等于又开一轮补全。

sse := resty.NewSSESource()
return sse.
	SetURL("https://api.deepseek.com/chat/completions").
	SetMethod(resty.MethodPost).
	SetHeader("Content-Type", "application/json").
	SetHeader("Authorization", "Bearer "+os.Getenv("DEEPSEEK_API_KEY")).
	SetBody(strings.NewReader(`{
  "model": "deepseek-chat",
  "messages": [{"role": "user", "content": "你是谁"}],
  "stream": true
}`)).
	OnMessage(func(e any) {
		ev := e.(*resty.SSE)
		if ev.Data == "[DONE]" {
			sse.Close()
			return
		}
		fmt.Println(ev.Data)
	}, nil).
	Get()

ev.Name 是事件名（没写 event 时是 message），ev.Data 是去掉 data: 之后的正文，ev.ID 对应 id。断线时库会带上 Last-Event-ID，并遵守服务端的 retry。库替你切块和重连，不替你拼 delta。

总结

  • Streaming = 同一支补全接口，改用 SSE 边生成边吐
  • 事件行主要是 event / data / id / retry。没写 event 就是 message
  • 看 delta，客户端负责拼接；[DONE] 是 DeepSeek / OpenAI 用来表示流结束的约定
  • 标准库按行读 data:；resty v3 用 SetMethod(POST) 收同一条流
  • 还是一问一答，答完就停。只是字先到，JSON 整包不到

参见