Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| import os | |
| import json | |
| import requests | |
| prompt_templates = {"默认ChatGPT": ""} | |
| # Streaming endpoint | |
| # API_URL = "https://api.openai.com/v1/chat/completions" # os.getenv("API_URL") + "/generate_stream" | |
| OPENAI_URL = "https://api.openai.com/v1/chat/completions" # os.getenv("API_URL") + "/generate_stream" | |
| API2D_URL = "https://openai.api2d.net/v1/chat/completions" # os.getenv("API_URL") + "/generate_stream" | |
| convo_id = 'default' | |
| #5c72c157a8fd54357bd13112cd71952a | |
| def on_prompt_template_change(prompt_template): | |
| if not isinstance(prompt_template, str): return | |
| if prompt_template: | |
| return prompt_templates[prompt_template] | |
| else: | |
| '' | |
| def get_empty_state(): | |
| return {"total_tokens": 0, "messages": []} | |
| def get_prompt_templates(): | |
| with open('./prompts_zh.json','r',encoding='utf8') as fp: | |
| json_data = json.load(fp) | |
| for data in json_data: | |
| act = data['act'] | |
| prompt = data['prompt'] | |
| prompt_templates[act] = prompt | |
| # reader = csv.reader(csv_file) | |
| # next(reader) # skip the header row | |
| # for row in reader: | |
| # if len(row) >= 2: | |
| # act = row[0].strip('"') | |
| # prompt = row[1].strip('"') | |
| # prompt_templates[act] = prompt | |
| choices = list(prompt_templates.keys()) | |
| choices = choices[:1] + sorted(choices[1:]) | |
| return gr.update(value=choices[0], choices=choices) | |
| # Testing with my Open AI Key | |
| # OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") | |
| def predict(inputs, prompt_template, temperature, openai_api_key, chat_counter, context_length, chatbot=[], | |
| history=[]): # repetition_penalty, top_k | |
| print(openai_api_key) | |
| if openai_api_key.startswith('sk-'): | |
| API_URL = OPENAI_URL | |
| print(1) | |
| elif openai_api_key.startswith('fk'): | |
| API_URL = API2D_URL | |
| print(2) | |
| else: | |
| API_URL = OPENAI_URL | |
| print(3) | |
| if inputs==None: | |
| inputs = '' | |
| if prompt_template: | |
| prompt_template = prompt_templates[prompt_template] | |
| else: | |
| prompt_template = "" | |
| # system_prompt = [] | |
| # if prompt_template: | |
| # history = [{"role": "system", "content": prompt_template}] | |
| payload = { | |
| "model": "gpt-3.5-turbo", | |
| "messages": [{"role": "system", "content": prompt_template},{"role": "user", "content": f"{inputs}"}], | |
| "temperature": 1.0, | |
| "top_p": 1.0, | |
| "n": 1, | |
| "stream": True, | |
| "presence_penalty": 0, | |
| "frequency_penalty": 0, | |
| } | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Authorization": f"Bearer {openai_api_key}" | |
| } | |
| # print(f"chat_counter - {chat_counter}") | |
| if chat_counter != 0: | |
| messages = [] | |
| # print(chatbot) | |
| # print(chatbot[-context_length:]) | |
| # print(context_length) | |
| for data in chatbot[-context_length:]: | |
| temp1 = {} | |
| temp1["role"] = "user" | |
| temp1["content"] = data[0] | |
| temp2 = {} | |
| temp2["role"] = "assistant" | |
| temp2["content"] = data[1] | |
| messages.append(temp1) | |
| messages.append(temp2) | |
| temp3 = {} | |
| temp3["role"] = "user" | |
| temp3["content"] = inputs | |
| messages.append(temp3) | |
| # print(messages) | |
| # messages | |
| payload = { | |
| "model": "gpt-3.5-turbo", | |
| "messages": [{"role": "system", "content": prompt_template}]+messages, # [{"role": "user", "content": f"{inputs}"}], | |
| "temperature": temperature, # 1.0, | |
| "n": 1, | |
| "stream": True, | |
| "presence_penalty": 0, | |
| "frequency_penalty": 0, | |
| } | |
| history.append(inputs) | |
| # print(f"payload is - {payload}") | |
| # make a POST request to the API endpoint using the requests.post method, passing in stream=True | |
| # print('payload',payload) | |
| response = requests.post(API_URL, headers=headers, json=payload, stream=True) | |
| # print('response', response) | |
| # print('content',response.content) | |
| # print('text', response.text) | |
| if response.status_code != 200: | |
| try: | |
| payload['id'] = response.content['id'] | |
| response = requests.post(API_URL, headers=headers, json=payload, stream=True) | |
| if response.status_code != 200: | |
| payload['id'] = response.content['id'] | |
| response = requests.post(API_URL, headers=headers, json=payload, stream=True) | |
| except: | |
| pass | |
| # print('status_code', response.status_code) | |
| # response = requests.post(API_URL, headers=headers, json=payload, stream=True) | |
| token_counter = 0 | |
| partial_words = "" | |
| counter = 0 | |
| if response.status_code==200: | |
| chat_counter += 1 | |
| # print('chunk') | |
| for chunk in response.iter_lines(): | |
| # Skipping first chunk | |
| if counter == 0: | |
| counter += 1 | |
| continue | |
| # check whether each line is non-empty | |
| chunk = chunk.decode("utf-8")[6:] | |
| if chunk: | |
| # print(chunk) | |
| if chunk=='[DONE]': | |
| break | |
| resp: dict = json.loads(chunk) | |
| choices = resp.get("choices") | |
| if not choices: | |
| continue | |
| delta = choices[0].get("delta") | |
| if not delta: | |
| continue | |
| # decode each line as response data is in bytes | |
| if len(chunk) > 12 and "content" in resp['choices'][0]['delta']: | |
| # if len(json.loads(chunk.decode()[6:])['choices'][0]["delta"]) == 0: | |
| # break | |
| partial_words = partial_words + resp['choices'][0]["delta"]["content"] | |
| # print(partial_words) | |
| if token_counter == 0: | |
| history.append(" " + partial_words) | |
| else: | |
| history[-1] = partial_words | |
| chat = [(history[i], history[i + 1]) for i in | |
| range(0, len(history) - 1, 2)] # convert to tuples of list | |
| # print(chat) | |
| token_counter += 1 | |
| yield chat, history, chat_counter # resembles {chatbot: chat, state: history} | |
| else: | |
| chat = [(history[i], history[i + 1]) for i in | |
| range(0, len(history) - 1, 2)] # convert to tuples of list | |
| chat.append((inputs, "OpenAI服务器网络出现错误，请重试，或重启对话")) | |
| token_counter += 1 | |
| yield chat, history, chat_counter # resembles {chatbot: chat, state: history} | |
| # yield ['OpenAI服务器网络出现错误'], ['OpenAI服务器网络出现错误'], gr.update(value=0) | |
| def reset_textbox(): | |
| return gr.update(value='') | |
| def clear_conversation(chatbot): | |
| return gr.update(value=None, visible=True), [], [], gr.update(value=0) | |
| title = """<h1 align="center">🔥覃秉丰的ChatGPT🔥</h1>""" | |
| description = """Language models can be conditioned to act like dialogue agents through a conversational prompt that typically takes the form: | |
| ``` | |
| User: <utterance> | |
| Assistant: <utterance> | |
| User: <utterance> | |
| Assistant: <utterance> | |
| ... | |
| ``` | |
| In this app, you can explore the outputs of a gpt-3.5-turbo LLM. | |
| """ | |
| with gr.Blocks(css="""#col_container {width: 800px; margin-left: auto; margin-right: auto;} | |
| #chatbot {height: 500px; overflow: auto;} | |
| #inputs {font-size: 20px;} | |
| #prompt_template_preview {padding: 1em; border-width: 1px; border-style: solid; border-color: #e0e0e0; border-radius: 4px;}""") as demo: | |
| gr.HTML(title) | |
| # gr.HTML( | |
| # '''<center><a href="https://huggingface.co/spaces/QinBingFeng/chatgpt?duplicate=true"><img src="https://bit.ly/3gLdBN6" alt="Duplicate Space"></a>点击图标复制App</center>''') | |
| with gr.Column(elem_id="col_container"): | |
| openai_api_key = gr.Textbox(type='password', label="输入你的API Key",placeholder="OpenAI API Key 或者 API2D") | |
| chatbot = gr.Chatbot(elem_id='chatbot') # c | |
| inputs = gr.Textbox(show_label=False, placeholder="在这里输入内容",elem_id="inputs",value='') # t | |
| state = gr.State([]) # s | |
| # state = gr.State(get_empty_state()) | |
| b1 = gr.Button("提交") | |
| btn_clear_conversation = gr.Button("🔃 开启新的对话") | |
| # inputs, top_p, temperature, top_k, repetition_penalty | |
| with gr.Accordion("高级设置", open=False): | |
| context_length = gr.Slider(minimum=1, maximum=6, value=2, step=1, label="对话长度", | |
| info="关联之前的几轮对话，数值越高tokens消耗越多") | |
| temperature = gr.Slider(minimum=0, maximum=2.0, value=0.7, step=0.1, label="Temperature", | |
| info="数值越高创造性越强") | |
| prompt_template = gr.Dropdown(label="选择机器人类型", | |
| choices=list(prompt_templates.keys())) | |
| prompt_template_preview = gr.Markdown(elem_id="prompt_template_preview") | |
| # top_k = gr.Slider( minimum=1, maximum=50, value=4, step=1, interactive=True, label="Top-k",) | |
| # repetition_penalty = gr.Slider( minimum=0.1, maximum=3.0, value=1.03, step=0.01, interactive=True, label="Repetition Penalty", ) | |
| chat_counter = gr.Number(value=0, visible=False, precision=0) | |
| inputs.submit(predict, [inputs, prompt_template, temperature, openai_api_key, chat_counter, context_length, chatbot, state], | |
| [chatbot, state, chat_counter], ) | |
| b1.click(predict, [inputs, prompt_template, temperature, openai_api_key, chat_counter, context_length, chatbot, state], | |
| [chatbot, state, chat_counter], ) | |
| b1.click(reset_textbox, [], [inputs]) | |
| btn_clear_conversation.click(clear_conversation, [], [inputs, chatbot, state, chat_counter]) | |
| inputs.submit(reset_textbox, [], [inputs]) | |
| prompt_template.change(on_prompt_template_change, inputs=[prompt_template], outputs=[prompt_template_preview]) | |
| demo.load(get_prompt_templates, inputs=None, outputs=[prompt_template], queur=False) | |
| # gr.Markdown(description) | |
| demo.queue(concurrency_count=10) | |
| demo.launch(debug=True) |