分别使用了flask框架和sanic框架进行openai接口的流式传输，结果flask可以，但是sanic不行（具体表现为flask可以出现打字机效果而sanic同样是流式输出却只能一次性全部输出）代码均已调试成功，目前
找不出sanic框架无法出现打字机效果的原因。以下是没有办法实现打字机效果的sanic-21.3.1框架代码：
import openai, os
from sanic import Sanic, request
app = Sanic(name='gpt_sanic')
from sanic.views import stream
from sanic_cors import CORS
from sanic import HTTPResponse
# from sanic.views import stream
from sanic.response import stream
# app.config['SECRET_KEY'] = os.urandom(24)
CORS(app, sources=r'/*')
def text_handle(text_0, parame_dict):
import re
parame_list = parame_dict.keys()
z = re.findall(r'[{](.*?)[}]', text_0)
for i in z:
if i in parame_list:
text_0 = text_0.replace("{" + i + "}", parame_dict[i])
else:
text_0 = text_0.replace("{" + i + "}", '')
print(f'{i}字段未定义')
return text_0
@app.route('/', methods=['POST', "GET"])
async def get_gpt_response(request):
application4 = request.json["text_"]
parame_dict = request.json
print(parame_dict, type(parame_dict))
question = text_handle(application4, parame_dict)
temperature = int(request.json['temperature']) / 10
question = str(question).strip()
openai.api_key = ""
messages = []
user_message_dict = {"role": "user", "content": question}
messages.append(user_message_dict)
response0 = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=messages, stream=True,temperature=temperature)
async def streaming_fn0(response):
# response=request.respond(content_type="text/csc")
for i, text in enumerate(response0, start=1):
if text["choices"][0]['finish_reason'] is not None:
data = '[DONE]'
# response.write(data)
else:
data = text["choices"][0].get('delta', '').get('content', '')
print(data)
await response.write("%s\n\n" % data.replace('\n', '<br>'))
# yield "%s\n\n" % data.replace('\n', '<br>')
return stream(streaming_fn0,content_type="text/csc")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)
期望解惑！
涛哥 我爱你😍
import openai, os
from flask import Flask, jsonify, json, request
import flask
from flask_cors import CORS
import json, re
app = Flask(__name__)
app.config['SECRET_KEY'] = os.urandom(24)
CORS(app, sources=r'/*')
def key_value(x):
xx = x[1:-1]#str
z = re.findall(r'[{](.*?)[}]', xx)
dic_ = dict()
if z:
for i in z:
str_ = "{" + i + "}"
json_ = eval(str_)
dic_[json_['title']] = json_['value']
return dic_
def text_handle(text_0, parame_list):
print(parame_list)
parame_dict = key_value(parame_list)
z = re.findall(r'[{](.*?)[}]', text_0)
if z:
for i in z:
if i in parame_list:
text_0 = text_0.replace("{" + i + "}", parame_dict[i])
else:
text_0 = text_0.replace("{" + i + "}", '')
print(f'警告：{i}字段未定义')
print(text_0)
return text_0
@app.route('/test/response', methods=['POST', "GET"])
def get_gpt_response():
"""
[
{title: "品牌名称", value: "", key: "a0"},
{title: "产品名称", value: "", key: "a1"},
{title: "产品类型", value: "", key: "a2"},
{title: "优势", value: "", key: "a3"}
]
:return:null
"""
application4 = json.loads(request.get_data())["text_"]
parame_list = json.loads(request.get_data())["input_list"]
question = text_handle(application4, parame_list)
temperature = int(json.loads(request.get_data())['temperature']) / 10
question = str(question).strip()
openai.api_key = ""
messages = []
user_message_dict = {"role": "user", "content": question}
messages.append(user_message_dict)
response0 = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=messages, stream=True,
temperature=temperature)
def streaming_fn0():
for i, text in enumerate(response0, start=1):
if text["choices"][0]['finish_reason'] is not None:
data = ''
else:
if i==2:
data='  ' + text["choices"][0].get('delta', '').get('content', '')
else:
data = text["choices"][0].get('delta', '').get('content', '')
yield "%s\n" % data.replace('\n', '<br/>  ')
return flask.Response(streaming_fn0(), mimetype="text/event-stream")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)
flask可以出现打字机效果而sanic却只能一次性输出的原因，应该是你的使用方法的问题，sanic是支持流的形式接收并响应由客户端发送来的数据。
但首先你要在路由上启用流式传输，之后你就可以使用 await request.stream.read() 方法来获取请求数据流。在方法上要加注解设置stream=true:
@app.post("/stream", stream=True)
在您提供的代码中，使用sanic.response.stream来构建响应对象并返回流式数据。这是sanic框架的正确用法。但是，可能会出现一些问题，导致您无法看到打字机效果。以下是一些可能导致这种情况的原因：
浏览器缓存问题：您可能已经尝试了在浏览器中打开sanic应用程序，然后刷新浏览器，但是无法看到打字机效果。这是因为浏览器可能会缓存响应，从而使您无法看到新的响应。您可以尝试在浏览器中禁用缓存，或者使用一个不同的浏览器来测试您的应用程序。
响应的MIME类型：确保您的响应的MIME类型设置为"text/plain"或"text/html"，而不是"text/csc"。如果您的响应MIME类型不正确，浏览器可能会无法正确解析响应并显示打字机效果。
流式响应的超时设置：确保您的流式响应的超时设置足够长。有时，如果响应的超时时间过短，则会导致浏览器无法正确接收到流式数据，从而无法显示打字机效果。您可以尝试将响应的超时时间设置得更长一些，以确保浏览器有足够的时间来接收流式数据并显示打字机效果。
浏览器的JavaScript设置：打字机效果通常是通过JavaScript脚本来实现的。如果您的浏览器禁用了JavaScript，或者您的浏览器中有一些插件或扩展程序禁用了JavaScript，那么您可能无法看到打字机效果。确保您的浏览器中启用了JavaScript，或者尝试在不同的浏览器中测试您的应用程序。
可能是因为Sanic框架的异步机制与Flask框架的同步机制不同所导致的。在Flask中，您可以使用yield语句逐个返回响应。而在Sanic中，您需要使用异步生成器和async for语句来实现流式响应。以下是一个基本的Sanic示例，它演示了如何使用异步生成器和async for语句来实现流式响应：
from sanic import Sanic, response
import asyncio
async def generate():
for i in range(10):
yield f"{i + 1}\n"
await asyncio.sleep(1)
app = Sanic(__name__)
@app.route("/")
async def index(request):
headers = {"Content-Type": "text/plain; charset=UTF-8"}
return response.stream(generate(), headers=headers)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000, debug=True)
在这个示例中，我们定义了一个异步生成器函数generate()，它将每个数字作为一个字符串返回给客户端，并在每个数字之间暂停1秒。然后，我们使用response.stream()方法来创建一个异步的流式响应，它将逐步返回数字。对于每个数字，我们使用yield语句来返回它，然后使用await asyncio.sleep(1)语句暂停1秒钟。返回带有Content-Type标头的响应和迭代器。这样，当客户端请求页面时，它们将获得一个逐渐生成响应的流。
这个问题的原因可能是因为Sanic框架和Flask框架在处理HTTP请求和响应时的差异引起的。
首先，需要理解一下HTTP请求和响应的工作原理。当客户端向服务端发起一个HTTP请求时，服务端会将请求头中的 Transfer-Encoding: chunked 设置为响应头，告诉客户端要启用HTTP分块编码。这样，客户端就可以使用流的方式，将HTTP请求分块发送到服务端。服务端在接收到每个分块数据后，就可以立即处理并返回响应。这个过程就称为流式传输。
在Flask框架中，使用 flask.Response 对象可以处理HTTP分块编码和流式传输。具体来说， Response 对象有一个 iter_chunks 方法，可以让你逐步生成要发送到客户端的分块数据。
而在Sanic框架中，它的HTTP响应是由 sanic.response.HTTPResponse 对象表示的，它没有 iter_chunks 方法来处理HTTP分块编码和流式传输。因此，在使用Sanic框架实现流式传输的时候，你需要自己手动处理分块数据，并将它们作为文本或字节串发送到客户端。具体的实现方式可以参考Sanic框架的官方文档中的示例代码。
另外，需要注意的是，浏览器对HTTP分块编码的支持也是有限的。虽然现代浏览器都支持HTTP分块编码，但是不同浏览器对分块编码的支持程度是不同的，可能会出现一些兼容性问题。因此，在实现流式传输的时候，最好进行充分的测试，确保在不同的浏览器和操作系统中都能正常工作。
综上所述，如果您想在Sanic框架中实现流式传输，需要手动处理HTTP分块编码，并将分块数据逐步发送到客户端。同时，需要注意兼容性问题，并进行充分的测试，确保在不同的浏览器和操作系统中都能正常工作。