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 = """

🔥覃秉丰的ChatGPT🔥

""" description = """Language models can be conditioned to act like dialogue agents through a conversational prompt that typically takes the form: ``` User: Assistant: User: Assistant: ... ``` 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( # '''
Duplicate Space点击图标复制App
''') 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)