Python自建chatgpt服务器：使用Flask实现类似Chat服务器流式API接口 ...

弁言

使用Flask框架可以轻松实现一个类似chatgpt流式响应的接口，该接口通过POST请求接收prompt和history参数，分别对应为谈天系统的提示词和对话历史，server则给予server-sent event (SSE)返回谈天系统的响应。
服务端

开发前的思考

我们梳理一下,为了实现流式API接口，Server端需要完成的主要工作
1、创建一个flask app， 查抄传入的POST请求是否包罗JSON，提取必要的数据，并举行验证。
2、响应为SSE： 设置得当的响应头以顺应服务器发送事件（text/event-stream），并保持毗连活动状态。
3、生成函数： 提供了一个占位符generate()函数。这个函数应包罗根据prompt和history生成响应的逻辑。在这个示例中，它简朴地流回输入数据。现实开发应当采用真正的LLM大模型，此处从简。
4、流式传输数据： generate()函数计划为持续流式传输数据。在现实应用中，你需要用现实的事件生成逻辑更换循环和time.sleep()。
其中，1比力简朴，因为很容易想象处理json数据是Flask的主要工作。3也比力简朴，如果使用过大语言模型的产物，你多半访问过打字机式的的UI结果。现实上的大语言模型输出不见得是严格一个个字输出，大概是几个字为单元输出，这是由于其模型输出采用分词器的缘故，即所谓Tokenizer。这不是本文主题，以是我们仅采用一个循环打印来模拟这种返回结果。2和4相对要生疏一些，不外我们可以明白它们就是实现流式接口的必备技能基础，纵然没有深入明白也可以实现。正所谓自顶向下的明白一个题目，我们先从最表层的行动：实现（Implementation），开始。
代码实现

严谨起见，首先安装 Flask：

1. pip install Flask

复制代码

server部分代码如下

1. from flask import Flask, request, jsonify, Response
3. app = Flask(__name__)
5. @app.route('/api/stream-chat', methods=['POST'])
6. def stream_chat():
7.     # Check if the request contains JSON data
8.     if request.is_json:
9.         # Get JSON data
10.         data = request.get_json()
12.         # Extract 'prompt' field; return an error if it's missing
13.         prompt = data.get('prompt')
14.         if prompt is None:
15.             return jsonify({"error": "Missing 'prompt' field"}), 400
17.         # Extract 'history' field; return an error if it's missing
18.         history = data.get('history')
19.         if history is None:
20.             return jsonify({"error": "Missing 'history' field"}), 400
22.         # Ensure 'history' is a list
23.         if not isinstance(history, list):
24.             return jsonify({"error": "'history' field must be a list"}), 400
26.         # Call a generate function to process the prompt and history into a response
27.         response = Response(generate(message=prompt, chat_history=history), mimetype='application/json')
28.         response.headers['Content-Type'] = 'text/event-stream'
29.         response.headers['Cache-Control'] = 'no-cache'
30.         response.destroy_headers['Connection'] = 'keep-alive'
31.         return response
32.     else:
33.         # If the request does not contain JSON data, return an error
34.         return jsonify({"error": "Request body must be JSON"}), 400
36. def generate(message, chat_history):
37.     # This function should generate the events. Here is a simple example that just echoes the input.
38.     # Normally, you might have more complex logic here.
39.     import json
40.     import time
41.     while True:
42.         data = {
43.             "prompt": message,
44.             "history": chat_history,
45.             "response": "Generated response based on input"
46.         }
47.         yield f"data:{json.dumps(data)}\n\n"
48.         time.sleep(1)  # Delay for demonstration; remove or adjust in production
50. if __name__ == '__main__':
51.     app.run(debug=True, threaded=True)

复制代码

接下来我们可以实现一个客户端以测试该接口。测试前请运行刚刚实现的server脚本并保持服务器开启状态。Flask默认使用的端口是5000。
为了与上述 Flask 服务器举行交互，并处理服务器端事件流（SSE），我们使用 Python 的 requests 库来创建一个客户端。这个客户端将发送 POST 请求到 Flask 应用，并监听返回的事件流。
客户端

客户端工作：

• 发送 POST 请求：使用 requests.post 发送一个含有 JSON 数据的 POST 请求到服务器。stream=True 参数允许我们处理持续的响应数据流。
  
• 处理响应：循环读取服务器的响应。当检测到以 data: 开头的行时，解析该行并打印出数据。这里使用了 JSON 解析来处理数据。
  

测试客户端和服务器：

• 确保你的 Flask 服务器正在运行。
  
• 运行上述客户端代码。
  

首先安装 requests 库

1. pip install requests

复制代码

下面是客户端代码的示例：

1. import requests
2. import json
4. def send_post_request(url, payload):
5.     # 将字典转换为JSON格式的字符串
6.     headers = {'Content-Type': 'application/json'}
7.     response = requests.post(url, data=json.dumps(payload), headers=headers, stream=True)
8.    
9.     # 检查请求是否成功
10.     if response.status_code == 200:
11.         try:
12.             print("Connected to server, listening for events...")
13.             for line in response.iter_lines():
14.                 if line:
15.                     decoded_line = line.decode('utf-8')
16.                     if decoded_line.startswith('data:'):
17.                         # 提取数据部分
18.                         json_data = json.loads(decoded_line.split(':', 1)[1].strip())
19.                         print("Received data:", json_data)
20.         except Exception as e:
21.             print("Error:", e)
22.     else:
23.         print("Failed to connect:", response.status_code)
25. # URL指向你的Flask应用
26. url = 'http://127.0.0.1:5000/api/stream-chat'
28. # 构造请求的数据
29. payload = {
30.     'prompt': 'Example prompt',
31.     'history': ['First action', 'Second action']
32. }
34. # 发送POST请求并处理SSE
35. send_post_request(url, payload)

复制代码

这样，你的客户端会发送一个 POST 请求到 Flask 服务器，并及时接收并打印从服务器发送的数据。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。