Spaces:
Sleeping
Sleeping
# chat_bot.py | |
import streamlit as st | |
from streamlit_chat import message | |
import _thread as thread | |
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 # 使用websocket_client | |
import sys, io | |
answer = "" | |
class Ws_Param(object): | |
# 初始化 | |
def __init__(self, APPID, APIKey, APISecret, Spark_url): | |
self.APPID = APPID | |
self.APIKey = APIKey | |
self.APISecret = APISecret | |
self.host = urlparse(Spark_url).netloc | |
self.path = urlparse(Spark_url).path | |
self.Spark_url = Spark_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.Spark_url + '?' + urlencode(v) | |
# 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释,比对相同参数时生成的url与自己代码生成的url是否一致 | |
return url | |
# 收到websocket错误的处理 | |
def on_error(ws, error): | |
print("### error:", error) | |
# 收到websocket关闭的处理 | |
def on_close(ws, one, two): | |
print(" ") | |
# 收到websocket连接建立的处理 | |
def on_open(ws): | |
thread.start_new_thread(run, (ws,)) | |
def run(ws, *args): | |
data = json.dumps(gen_params(appid=ws.appid, domain=ws.domain, 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="") | |
global answer | |
answer += content | |
# print(1) | |
if status == 2: | |
ws.close() | |
def gen_params(appid, domain, question): | |
""" | |
通过appid和用户的提问来生成请参数 | |
""" | |
data = { | |
"header": { | |
"app_id": appid, | |
"uid": "1234" | |
}, | |
"parameter": { | |
"chat": { | |
"domain": domain, | |
"random_threshold": 0.5, | |
"max_tokens": 2048, | |
"auditing": "default" | |
} | |
}, | |
"payload": { | |
"message": { | |
"text": question | |
} | |
} | |
} | |
return data | |
def main(appid, api_key, api_secret, Spark_url, domain, question): | |
# print("星火:") | |
wsParam = Ws_Param(appid, api_key, api_secret, Spark_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.domain = domain | |
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) | |
# 以下密钥信息从控制台获取 | |
appid = "7e203b6c" # 填写控制台中获取的 APPID 信息 | |
api_secret = "MjU0MGUwZDI1MDk4MDVjZTg1ZjU0YjZi" # 填写控制台中获取的 APISecret 信息 | |
api_key = "ef969b3800d8d3a367c06729844f6ab4" # 填写控制台中获取的 APIKey 信息 | |
# 用于配置大模型版本,默认“general/generalv2” | |
#domain = "general" # v1.5版本 | |
domain = "generalv2" # v2.0版本 | |
# 云端环境的服务地址 | |
#Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat" # v1.5环境的地址 | |
Spark_url = "ws://spark-api.xf-yun.com/v2.1/chat" # v2.0环境的地址 | |
text = [] | |
# length = 0 | |
def getText(role, content): | |
jsoncon = {} | |
jsoncon["role"] = role | |
jsoncon["content"] = content | |
text.append(jsoncon) | |
return text | |
def getlength(text): | |
length = 0 | |
for content in text: | |
temp = content["content"] | |
leng = len(temp) | |
length += leng | |
return length | |
def checklen(text): | |
while (getlength(text) > 8000): | |
del text[0] | |
return text | |
def getanswerByXunFei(t): | |
st.session_state['prompts'].append(t) | |
text.clear() | |
question = checklen(getText("user", t)) | |
ans = "" | |
main(appid, api_key, api_secret, Spark_url, domain, question) | |
getText("assistant", ans) | |
return answer | |
if 'prompts' not in st.session_state: | |
st.session_state['prompts'] = [] | |
if 'generated' not in st.session_state: | |
st.session_state['generated'] = [] | |
if 'past' not in st.session_state: | |
st.session_state['past'] = [] | |
def end_click(): | |
st.session_state['prompts'] = [] | |
st.session_state['past'] = [] | |
st.session_state['generated'] = [] | |
st.session_state['user'] = "" | |
def chat_click(): | |
if st.session_state['user'] != '': | |
chat_input = st.session_state['user'] | |
output = getanswerByXunFei(chat_input) | |
st.session_state['past'].append(chat_input) | |
st.session_state['generated'].append(output) | |
st.session_state['prompts'].append(output) | |
st.session_state['user'] = "" | |
st.image("./logo.jpg", use_column_width='auto') | |
st.title("珠宝贝专属聊天机器人") | |
user_input = st.text_input("输入:", key="user") | |
chat_button = st.button("发送", on_click=chat_click) | |
end_button = st.button("新聊天", on_click=end_click) | |
if st.session_state['generated']: | |
for i in range(0, len(st.session_state['generated']), 1): | |
message(st.session_state['past'][i], is_user=True) | |
message(st.session_state['generated'][i], key=str(i)) |