如何用Python返回SSE数据

发表于 更新于 分类于 Waline：

最近在使用python调用gpt的sse接口并通过sse的方式返回给自己的前端。遇到了几个问题，简单记录下，也没什么代码量，但也用了大半天的时间才搞定。

  • python如何返回sse
  • 为什么我返回的sse中的换行符总是会丢失
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import json
import time

from flask import Flask, request
from flask import Response, jsonify

app = Flask(__name__)

import openai
import time
import sys

openai.api_key = ''

@app.route('/stream', methods=["POST"])
def stream():
    messages = request.get_json()['messages']

    print(messages)
    response = openai.ChatCompletion.create(
        model='gpt-3.5-turbo',
        messages=messages,
        temperature=0.7,
        stream=True  # again, we set stream=True
    )
    def eventStream():
        id = 0
        for chunk in response:
            id += 1
            chunk_message = chunk['choices'][0]['delta']  # extract the message
            yield 'id: {}\nevent: message\ndata: {}\n\n'.format(id,json.dumps({'data': chunk_message.get('content', '')}))
        
        yield 'id: {}\nevent: close\ndata: {}\n\n'.format(id,json.dumps({'data': chunk_message.get('content', '')}))

    return Response(eventStream(), mimetype="text/event-stream")


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8889)

这里面有几个关键点，分别对应一开始的两个问题：

  • sse接口返回的数据类型要是"text/event-stream"
  • 返回的数据要被格式化'id: {}\nevent: message\ndata: {}\n\n'.format(id,json.dumps({'data': chunk_message.get('content', '')}))，这是sse的协议规定，一个换行符表示不同的数据，两个换行符表示一条数据，id要不重复。
  • data数据最好用json字符串返回，前端接收以后重新解析，即使你的data原本就是个string，不然你很可能会遇到换行符丢了的问题