基于星火和Gradio的聊天机器人-Toy模板网

这篇具有很好参考价值的文章主要介绍了基于星火和Gradio的聊天机器人。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

星火大模型官网：https://xinghuo.xfyun.cn/

1 创建虚拟环境（windows）

conda create -n Gradio python=3.8
pip install gradio

中间遇到os报错，解决方案：

pip install aiofiles==23.2.1

2 代码

SparkDesk.py：
下面代码是一个用于与基于WebSocket的聊天机器人API进行交互的Python实现。下面逐个解释代码的每个部分：

代码导入了几个模块，用于各种功能，例如线程处理（_thread），base64编码，哈希算法，JSON操作，与时间相关的操作，URL解析，SSL处理，CSV处理以及websocket库。
接下来，代码定义了一个名为Ws_Param的类，该类用于初始化建立WebSocket连接和生成URL所需的参数。
Ws_Param类中有一个名为create_url()的方法，用于生成建立WebSocket连接的URL。它使用HMAC-SHA256加密构建必要的身份验证头。
代码定义了几个WebSocket事件处理函数：
a. 5.on_error：处理WebSocket连接期间发生的错误。
b. on_close：处理WebSocket连接关闭事件。
c. on_open：处理WebSocket连接打开事件。
d. on_message：处理WebSocket连接收到的传入消息。
run函数设计为在单独的线程中运行，并向聊天机器人API发送初始请求载荷。
gen_params函数是一个辅助函数，用于生成聊天机器人API请求的JSON载荷。
main函数是启动WebSocket连接的入口点。它创建Ws_Param类的实例，生成WebSocket URL，并使用websocket库中的WebSocketApp类来初始化WebSocket连接。
ask_question函数是对main函数的封装，提供了一个便捷的方法来向聊天机器人API提问问题。
最后，startSparkOne函数是ask_question的封装函数，可以通过将问题作为参数传递给它来启动与聊天机器人API的交互。
要使用此代码，在调用startSparkOne函数时，您需要为appid，api_key，api_secret和gpt_url参数提供适当的值。

import _thread as thread
import base64
import datetime
import hashlib
import hmac
import json
import pickle
import time
from urllib.parse import urlparse
import ssl
import sys
import io
import csv
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time
# websocket-client
import websocket


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, status_code, reason):
    print("")


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


def run(ws, *args):
    data = json.dumps(gen_params(appid=ws.appid, question=ws.question))
    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:
            ws.close()


def gen_params(appid, question):
    """
    通过appid和用户的提问来生成请参数
    """
    data = {
        "header": {
            "app_id": appid,
            "uid": "1234"
        },
        "parameter": {
            "chat": {
                "domain": "general",
                "random_threshold": 0.5,
                "max_tokens": 2048,
                "auditing": "default"
            }
        },
        "payload": {
            "message": {
                "text": [
                    {"role": "user", "content": question}
                ]
            }
        }
    }
    return data


def main(appid, api_key, api_secret, gpt_url, question):
    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.question = question
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})


def ask_question(question):
    main(appid="",
         api_secret="",
         api_key="",
         gpt_url="ws://spark-api.xf-yun.com/v1.1/chat",
         question=question)


def startSparkOne(question):
    ask_question(question)

app.py:
这段代码实现了一个简单的聊天界面，用户可以输入消息与SparkDesk Chatbot进行交互。下面对每部分代码进行详细解释：

导入了random模块，用于随机生成响应消息，gradio模块用于创建交互界面，startSparkOne来自SparkDesk模块，用于与聊天机器人进行交互。另外还导入了sys和io模块，用于处理标准输出和字符串流对象。
random_response函数是一个用于生成随机响应的函数。它接收两个参数，message表示用户输入的消息，history表示之前的对话历史。在函数内部，它创建了一个字符串流对象，并将标准输出重定向到该流对象。然后调用startSparkOne函数与聊天机器人交互，并将机器人的回复输出到标准输出流。最后，它将标准输出重定向回原来的标准输出，并获取标准输出流的内容，即机器人的回复。
demo是一个gr.ChatInterface对象，用于创建聊天界面。它接受两个参数，第一个参数是一个函数，用于处理用户输入和生成响应，这里使用了random_response函数。第二个参数是界面的标题和描述。
最后，调用demo.launch()启动聊天界面。
通过这段代码，可以在一个简单的界面中输入消息与SparkDesk Chatbot进行交互，并获得机器人的随机响应。

import random
import gradio as gr
from SparkDesk import startSparkOne
import sys, io

def random_response(message, history):
    # 创建一个字符串流对象，并将其设置为标准输出
    old_stdout = sys.stdout
    new_stdout = io.StringIO()
    sys.stdout = new_stdout

    startSparkOne(message)

    # 将标准输出重新设置为原来的标准输出
    sys.stdout = old_stdout

    # 获取标准输出的内容
    response = new_stdout.getvalue()

    return response


demo = gr.ChatInterface(
    random_response,
    title="SparkDesk Chatbot",
    description="输入消息来与 SparkDesk Chatbot 进行交互。",
)

demo.launch()