python星火大模型

Question

from flask import Flask, request, render_template

import time
import base64

import base64
import datetime
import hashlib
import hmac
import json
from urllib.parse import urlparse
import ssl
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time

import websocket
import openpyxl
from concurrent.futures import ThreadPoolExecutor, as_completed, thread
import os

app = Flask(__name__)

@app.route('/')
def hello_world():  # put application's code here

    text = 'ABC'
    return render_template("index.html", text=text)


@app.route('/submit', methods=['POST'])
def submit():
    data = request.form['data']
    print(data)




class Ws_Param(object):
    # 初始化
    def __init__(self, APPID, APIKey, APISecret, gpt_url):
        self.APPID = APPID
        self.APIKey = APIKey
        self.APISecret = APISecret
        self.host = urlparse(gpt_url).netloc
        self.path = urlparse(gpt_url).path
        self.gpt_url = gpt_url

    # 生成url
    def create_url(self):
        # 生成RFC1123格式的时间戳
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))

        # 拼接字符串
        signature_origin = "host: " + self.host + "\n"
        signature_origin += "date: " + date + "\n"
        signature_origin += "GET " + self.path + " HTTP/1.1"

        # 进行hmac-sha256进行加密
        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
                                 digestmod=hashlib.sha256).digest()

        signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')

        authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'

        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')

        # 将请求的鉴权参数组合为字典
        v = {
            "authorization": authorization,
            "date": date,
            "host": self.host
        }
        # 拼接鉴权参数，生成url
        url = self.gpt_url + '?' + urlencode(v)
        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
        return url


# 收到websocket错误的处理
def on_error(ws, error):
    print("### error:", error)


# 收到websocket关闭的处理
def on_close(ws):
    print("### closed ###")


# 收到websocket连接建立的处理
def on_open(ws):
    thread.start_new_thread(run, (ws,))


def run(ws, *args):
    data = json.dumps(gen_params(appid=ws.appid, query=ws.query, domain=ws.domain))
    ws.send(data)


# 收到websocket消息的处理
def on_message(ws, message):
    # print(message)
    data = json.loads(message)
    code = data['header']['code']
    if code != 0:
        print(f'请求错误: {code}, {data}')
        ws.close()
    else:
        choices = data["payload"]["choices"]
        status = choices["status"]
        content = choices["text"][0]["content"]
        print(content,end='')
        if status == 2:
            print("#### 关闭会话")
            ws.close()


def gen_params(appid, query, domain):
    """
    通过appid和用户的提问来生成请参数
    """

    data = {
        "header": {
            "app_id": appid,
            "uid": "1234",
            # "patch_id": []    #接入微调模型，对应服务发布后的resourceid
        },
        "parameter": {
            "chat": {
                "domain": domain,
                "temperature": 0.5,
                "max_tokens": 4096,
                "auditing": "default",
            }
        },
        "payload": {
            "message": {
                "text": [{"role": "user", "content": query}]
            }
        }
    }
    return data


def answer(appid, api_secret, api_key, gpt_url, domain, query):
    wsParam = Ws_Param(appid, api_key, api_secret, gpt_url)
    websocket.enableTrace(False)
    wsUrl = wsParam.create_url()

    ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)
    ws.appid = appid
    ws.query = query
    ws.domain = domain
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})


if __name__ == '__main__':
    app.run()

在python何处正确调用answer方法

Answer 1

引自免费微信小程序：皆我百晓生

要在Python程序中正确调用answer方法，你需要先确保已经定义了所需的变量，包括appid, api_secret, api_key, gpt_url, domain, 和 query。然后，在主程序运行的适当位置调用这个函数。

根据你的代码结构，一个合适的调用位置可能是hello_world视图函数返回HTML模板之后，或者在用户提交表单后（即submit视图函数）。这里，我将提供一个例子，展示如何在用户提交表单后调用answer方法。

首先，我们需要在submit视图函数中获取用户输入的数据，并传递给answer方法。由于这是一个示例，我将简单地从URL中获取这些值作为演示。实际应用中，你应该从表单数据中提取这些值。

@app.route('/', methods=['GET', 'POST'])
def hello_world():  # put application's code here

    text = 'ABC'
    
    if request.method == 'POST':
        appid = request.args.get('appid')
        api_secret = request.args.get('api_secret')
        api_key = request.args.get('api_key')
        gpt_url = request.args.get('gpt_url')
        domain = request.args.get('domain')
        query = request.args.get('query')

        answer(appid, api_secret, api_key, gpt_url, domain, query)

    return render_template("index.html", text=text)

在这个例子中，我假设你有一个包含appid, api_secret, api_key, gpt_url, domain, 和 query字段的HTML表单。当用户提交表单时，这些值将作为URL查询参数传递到服务器。然后，我们从request.args中获取这些值并传递给answer函数。

为了使这个示例工作，你需要在你的HTML模板（"index.html"）中创建一个表单，允许用户输入这些参数。例如：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>StarFire GPT</title>
</head>
<body>
    <form action="/" method="post">
        <label for="appid">App ID:</label><br>
        <input type="text" id="appid" name="appid"><br>
        
        <!-- Add similar input fields for other parameters like api_secret, api_key, gpt_url, domain, and query -->

        <input type="submit" value="Submit">
    </form>
</body>
</html>

这样，当用户填写表单并点击提交按钮时，就会触发上述的hello_world视图函数，并调用answer方法来发送请求并接收GPT的响应。注意，这只是一个基本示例，实际应用中可能需要添加更多的错误检查和验证步骤。