Create functions.py

functions.py

@@ -0,0 +1,552 @@
+# import gradio as gr
+import gradio
+# import lmdb
+# import base64
+# import io
+import random
+import time
+import os
+import re
+import json
+import copy
+# import sqlite3
+import hashlib
+import uuid
+from urllib.parse import urljoin
+import openai
+
+
+def get_random_sleep(base_time, random_range):
+    return (base_time + random.randint(-random_range, random_range))*0.001
+
+
+def js_load(txt):
+    try:
+        return json.loads(txt)
+    except Exception as error:
+        print('')
+        print('js_load:')
+        print(str(error))
+        print('')
+        return None
+
+
+def js_dump(thing):
+    try:
+        return json.dumps(thing)
+    except Exception as error:
+        print('')
+        print('js_dump:')
+        print(str(error))
+        print('')
+        return None
+
+
+def filtered_history(history, num=0):
+    if num > 0:
+        filtered = list(filter(lambda it:(it['type'] in ['request', 'response']), history))
+        return filtered[-num:]
+    return []
+
+
+def filtered_history_messages(history, num=0):
+    filtered = filtered_history(history, num)
+    return list(map(lambda it:{'role': it.get('role'), 'content': it.get('content')}, filtered))
+
+
+def make_md_line(role, content):
+    return f"""\n##### `{role}`\n\n{content}\n"""
+
+
+def make_md_by_history(history):
+    md = ""
+    for item in history:
+        md += make_md_line(item.get('role'), item.get('content'))
+    return md
+
+
+def make_history_file_fn(history):
+
+    uuid4 = str(uuid.uuid4())
+    json_file_path = None
+    md_file_path = None
+
+    try:
+        # 如果目录不存在，则创建目录
+        os.makedirs('temp_files', exist_ok=True)
+
+        json_file_content = json.dumps(history)
+        json_file_path = os.path.join('temp_files', f'history[{uuid4}].json')
+        with open(json_file_path, 'w') as f:
+            f.write(json_file_content)
+
+        md_file_content = make_md_by_history(history)
+        md_file_path = os.path.join('temp_files', f'history[{uuid4}].md')
+        with open(md_file_path, 'w') as f:
+            f.write(md_file_content)
+
+        return json_file_path, md_file_path, gradio.update(visible=True)
+
+    except Exception as error:
+        print(f"\n{error}\n")
+
+    return json_file_path, md_file_path, gradio.update(visible=True)
+
+
+def make_history_file_fn__(history):
+    uuid4 = str(uuid.uuid4())
+    try:
+        json_file_content = json.dumps(history)
+        json_file_path = os.path.join('temp_files', f'history[{uuid4}].json')
+        with open(json_file_path, 'w') as f:
+            f.write(json_file_content)
+    except Exception as error:
+        print(f"\n{error}\n")
+        json_file_path = None
+    try:
+        md_file_content = make_md_by_history(history)
+        md_file_path = os.path.join('temp_files', f'history[{uuid4}].md')
+        with open(md_file_path, 'w') as f:
+            f.write(md_file_content)
+    except Exception as error:
+        print(f"\n{error}\n")
+        md_file_path = None
+
+    return json_file_path, md_file_path, gradio.update(visible=True)
+
+
+def make_user_message_list_fn__(
+        user_message_template,  # 模板，套用到每一条消息上
+        user_message_template_mask,  # 模板中要被替换的部分
+        user_message_template_mask_is_regex,  # 决定如何构造用于替换的正则表达式
+        user_message_list_text,  # 一段文本，包含了每一条用户消息
+        user_message_list_text_splitter,  # 描述了应该以什么为线索来切分 user_message_list_text
+        user_message_list_text_splitter_is_regex,  # 决定如何进行切分
+    ) -> list:
+    # 返回套用了模板的用户信息列表
+    # 这个实现首先根据是否使用正则表达式来切分用户消息列表文本，并将切分后的消息存储在一个列表中。
+    # 然后，针对每个消息，根据user_message_template_mask及user_message_template_mask_is_regex替换模板中的部分内容，
+    # 并将替换后的结果添加到结果列表中。
+    # 最后，返回结果列表。
+
+    # 切分用户消息列表文本
+    if user_message_list_text_splitter_is_regex:
+        user_messages = re.split(user_message_list_text_splitter, user_message_list_text)
+    else:
+        user_messages = user_message_list_text.split(user_message_list_text_splitter)
+
+    # 生成套用模板的用户信息列表
+    user_message_result_list = []
+    for message in user_messages:
+        # 替换模板内容
+        if user_message_template_mask_is_regex:
+            transformed_message = re.sub(user_message_template_mask, message, user_message_template)
+        else:
+            transformed_message = user_message_template.replace(user_message_template_mask, message)
+
+        user_message_result_list.append(transformed_message)
+
+    return user_message_result_list
+
+
+def make_user_message_list_fn(
+        user_message_template,  
+        user_message_template_mask, 
+        user_message_template_mask_is_regex,  
+        user_message_list_text, 
+        user_message_list_text_splitter,  
+        user_message_list_text_splitter_is_regex,  
+    ) -> list:
+
+    # 实际上，只要保证在使用正则表达式进行替换或切分操作之前，已经将其编译为正则表达式对象即可。
+    # 在我的修改中，针对 xxx_is_regex 参数为 True 的情况，将这些参数编译成正则表达式。
+    # 对于替换操作和切分操作，只需检查是否已经编译为正则表达式，并使用相应的方法即可。
+
+    # 编译正则表达式
+    if user_message_template_mask_is_regex:
+        user_message_template_mask = re.compile(user_message_template_mask)
+
+    if user_message_list_text_splitter_is_regex:
+        user_message_list_text_splitter = re.compile(user_message_list_text_splitter)
+
+    # 切分用户消息列表文本
+    if user_message_list_text_splitter_is_regex:
+        user_messages = user_message_list_text_splitter.split(user_message_list_text)
+    else:
+        user_messages = user_message_list_text.split(user_message_list_text_splitter)
+
+    # 生成套用模板的用户信息列表
+    user_message_result_list = []
+    for message in user_messages:
+        # 替换模板内容
+        if user_message_template_mask_is_regex:
+            transformed_message = user_message_template_mask.sub(message, user_message_template)
+        else:
+            transformed_message = user_message_template.replace(user_message_template_mask, message)
+
+        user_message_result_list.append(transformed_message)
+
+    return user_message_result_list
+
+
+def sequential_chat_once_fn(payload, api_key_text, history, history_md_stable, history_md_stream, tips):
+    print("\n\n")
+
+    assistant_message = ""
+    tips = ""
+
+    try:
+        openai.api_key = api_key_text
+        completion = openai.ChatCompletion.create(**payload)
+
+        if payload.get('stream'):
+            is_first=True
+            for chunk in completion:
+                if is_first:
+                    is_first = False
+                    continue
+                if chunk.choices[0].finish_reason is None:
+                    print(chunk.choices[0].delta.content or '')
+                    assistant_message += chunk.choices[0].delta.content or ''
+                    history_md_stream = make_md_line('assistant', assistant_message)
+                    tips = 'streaming'
+                    yield assistant_message, history_md_stream, tips, history
+                else:
+                    pass
+                pass
+            pass
+        else:
+            assistant_message = completion.choices[0].message.content
+            history_md_stream = make_md_line('assistant', assistant_message)
+            tips = 'got'
+            print(assistant_message)
+            yield assistant_message, history_md_stream, tips, history
+        pass
+
+    except Exception as error:
+        tips = str(error)
+        history.append({"role": "app", "content": tips})
+        print(f"\n{tips}\n")
+        yield assistant_message, history_md_stream, tips, history
+        pass
+
+    print("\n\n")
+
+
+def sequential_chat_fn(
+        history,
+
+        system_prompt_enabled,
+        system_prompt,
+        user_message_template,
+        user_message_template_mask,
+        user_message_template_mask_is_regex,
+        user_message_list_text,
+        user_message_list_text_splitter,
+        user_message_list_text_splitter_is_regex,
+        history_prompt_num,
+
+        api_key_text, token_text,
+
+        sleep_base, sleep_rand,
+
+        prop_stream, prop_model, prop_temperature, prop_top_p, prop_choices_num, prop_max_tokens, prop_presence_penalty, prop_frequency_penalty, prop_logit_bias,
+    ):
+    # outputs=[
+    #     history,
+    #     history_md_stable,
+    #     history_md_stream,
+    #     tips,
+    #     file_row,
+    # ],
+
+    history_md_stable = ""
+    history_md_stream = ""
+    tips = ""
+
+    try:
+
+        user_message_list = make_user_message_list_fn(
+            user_message_template,
+            user_message_template_mask,
+            user_message_template_mask_is_regex,
+            user_message_list_text,
+            user_message_list_text_splitter,
+            user_message_list_text_splitter_is_regex,
+        )
+
+        payload = {
+            'model': prop_model,
+            'temperature': prop_temperature,
+            'top_p': prop_top_p,
+            'n': prop_choices_num,
+            'stream': prop_stream,
+            'presence_penalty': prop_presence_penalty,
+            'frequency_penalty': prop_frequency_penalty,
+            'user': token_text,
+        }
+        if prop_max_tokens>0:
+            payload['max_tokens'] = prop_max_tokens
+        # if prop_logit_bias is not None:
+        #     payload['logit_bias'] = prop_logit_bias
+        # headers = {
+        #     "Content-Type": "application/json",
+        #     "Authorization": f"Bearer {api_key_text}"
+        # }
+
+        for user_message in user_message_list:
+            print('')
+            print(user_message)
+            print('')
+            # make the_messages to sent
+            the_messages = []
+            if system_prompt_enabled:
+                the_messages.append({"role": "system", "content": system_prompt})
+            for msg in filtered_history_messages(history, num=history_prompt_num):
+                the_messages.append(msg)
+            the_messages.append({"role": "user", "content": user_message})
+            payload['messages'] = the_messages
+
+            history.append({"role": "user", "content": user_message, "type": "request", "payload": payload})
+
+            history_md_stable = make_md_by_history(history)
+            history_md_stream = ""
+            tips = ""
+            yield history, history_md_stable, history_md_stream, tips, gradio.update(visible=False)
+
+            try:
+                for (assistant_message, history_md_stream, tips, history) in sequential_chat_once_fn(payload, api_key_text, history, history_md_stable, history_md_stream, tips):
+
+                    yield history, history_md_stable, history_md_stream, tips, gradio.update()
+
+                history.append({"role": "assistant", "content": assistant_message, "type": "request"})
+                history_md_stable += history_md_stream
+                history_md_stream = ""
+                tips = "fine"
+                yield history, history_md_stable, history_md_stream, tips, gradio.update(visible=False)
+
+            except Exception as error:
+                tips = f'error: {str(error)}'
+                history.append({"role": "app", "content": tips})
+                print(f"\n{tips}\n")
+                yield history, history_md_stable, history_md_stream, tips, gradio.update(visible=False)
+
+            time.sleep(get_random_sleep(sleep_base, sleep_rand))
+        pass
+
+
+
+    except Exception as error:
+        tips = str(error)
+        history.append({"role": "app", "content": tips})
+        print(f"\n{tips}\n")
+        yield history, history_md_stable, history_md_stream, tips, gradio.update(visible=False)
+        pass
+
+
+
+
+def on_click_send_btn(
+        global_state_json, api_key_text, chat_input_role, chat_input, prompt_table, chat_use_prompt, chat_use_history, chat_log,
+        chat_model, temperature, top_p, choices_num, stream, max_tokens, presence_penalty, frequency_penalty, logit_bias,
+    ):
+
+    old_state = json.loads(global_state_json or "{}")
+
+    print('\n\n\n\n\n')
+    print(prompt_table)
+    prompt_table = prompt_table or []
+
+    chat_log = chat_log or []
+
+    chat_log_md = ''
+    if chat_use_prompt:
+        chat_log_md += '<center>(prompt)</center>\n\n'
+        chat_log_md += "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", prompt_table)])
+        chat_log_md += '\n---\n'
+    if True:
+        chat_log_md += '<center>(history)</center>\n\n' if chat_use_history else '<center>(not used history)</center>\n\n'
+        chat_log_md += "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", chat_log)])
+        chat_log_md += '\n---\n'
+
+    # if chat_input=='':
+    #     return json.dumps(old_state), chat_log, chat_log_md, chat_log_md, None, None, chat_input
+
+    print('\n')
+    print(chat_input)
+    print('')
+
+    try:
+        logit_bias_json = json.dumps(logit_bias) if logit_bias else None
+    except:
+        return json.dumps(old_state), chat_log, chat_log_md, chat_log_md, None, None, chat_input
+
+    new_state = copy.deepcopy(old_state) or {}
+
+
+
+    req_hist = copy.deepcopy(prompt_table) if chat_use_prompt else []
+
+    if chat_use_history:
+        for hh in (chat_log or []):
+            req_hist.append(hh)
+
+    if chat_input and chat_input!="":
+        req_hist.append([(chat_input_role or 'user'), chat_input])
+
+    openai.api_key = api_key_text
+
+    props = {
+        'model': chat_model,
+        'messages': [xx for xx in map(lambda it: {'role':it[0], 'content':it[1]}, req_hist)],
+        'temperature': temperature,
+        'top_p': top_p,
+        'n': choices_num,
+        'stream': stream,
+        'presence_penalty': presence_penalty,
+        'frequency_penalty': frequency_penalty,
+    }
+    if max_tokens>0:
+        props['max_tokens'] = max_tokens
+    if logit_bias_json is not None:
+        props['logit_bias'] = logit_bias_json
+
+    props_json = json.dumps(props)
+
+    try:
+        completion = openai.ChatCompletion.create(**props)
+        print('')
+
+        # print(completion.choices)
+        # the_response_role = completion.choices[0].message.role
+        # the_response = completion.choices[0].message.content
+        # print(the_response)
+        # print('')
+        # chat_last_resp = json.dumps(completion.__dict__)
+        # chat_last_resp_dict = json.loads(chat_last_resp)
+        # chat_last_resp_dict['api_key'] = "hidden by UI"
+        # chat_last_resp_dict['organization'] = "hidden by UI"
+        # chat_last_resp = json.dumps(chat_last_resp_dict)
+
+
+        chat_log_md = ''
+        if chat_use_prompt:
+            chat_log_md += '<center>(prompt)</center>\n\n'
+            chat_log_md += "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", prompt_table)])
+            chat_log_md += '\n---\n'
+        if True:
+            chat_log_md += '<center>(history)</center>\n\n' if chat_use_history else '<center>(not used history)</center>\n\n'
+            chat_log_md += "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", chat_log)])
+            chat_log_md += '\n---\n'
+
+        if chat_input and chat_input!="":
+            chat_log.append([(chat_input_role or 'user'), chat_input])
+            chat_log_md += f"##### `{(chat_input_role or 'user')}`\n\n{chat_input}\n\n"
+
+        partial_words = ""
+        counter=0
+
+        if stream:
+            the_response = ''
+            the_response_role = ''
+            for chunk in completion:
+                #Skipping first chunk
+                if counter == 0:
+                    the_response_role = chunk.choices[0].delta.role
+                    chat_log_md += f"##### `{the_response_role}`\n\n"
+                    counter += 1
+                    continue
+                # print(('chunk', chunk))
+                if chunk.choices[0].finish_reason is None:
+                    the_response_chunk = chunk.choices[0].delta.content
+                    the_response += the_response_chunk
+                    chat_log_md += f"{the_response_chunk}"
+                    yield json.dumps(new_state), chat_log, chat_log_md, chat_log_md, "{}", props_json, ''
+                else:
+                    chat_log.append([the_response_role, the_response])
+                    chat_log_md += f"\n\n"
+                    yield json.dumps(new_state), chat_log, chat_log_md, chat_log_md, '{"msg": "stream模式不支持显示"}', props_json, ''
+                # chat_last_resp = json.dumps(completion.__dict__)
+                # chat_last_resp_dict = json.loads(chat_last_resp)
+                # chat_last_resp_dict['api_key'] = "hidden by UI"
+                # chat_last_resp_dict['organization'] = "hidden by UI"
+                # chat_last_resp = json.dumps(chat_last_resp_dict)
+        else:
+            the_response_role = completion.choices[0].message.role
+            the_response = completion.choices[0].message.content
+            print(the_response)
+            print('')
+
+            chat_log.append([the_response_role, the_response])
+            chat_log_md += f"##### `{the_response_role}`\n\n{the_response}\n\n"
+
+            chat_last_resp = json.dumps(completion.__dict__)
+            chat_last_resp_dict = json.loads(chat_last_resp)
+            chat_last_resp_dict['api_key'] = "hidden by UI"
+            chat_last_resp_dict['organization'] = "hidden by UI"
+            chat_last_resp = json.dumps(chat_last_resp_dict)
+    
+            return json.dumps(new_state), chat_log, chat_log_md, chat_log_md, chat_last_resp, props_json, ''
+
+        # chat_log.append([the_response_role, the_response])
+        # chat_log_md += f"##### `{the_response_role}`\n\n{the_response}\n\n"
+        # return json.dumps(new_state), chat_log, chat_log_md, chat_log_md, chat_last_resp, props_json, ''
+    except Exception as error:
+        print(error)
+        print('error!!!!!!')
+
+        chat_log_md = ''
+        if chat_use_prompt:
+            chat_log_md += '<center>(prompt)</center>\n\n'
+            chat_log_md += "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", prompt_table)])
+            chat_log_md += '\n---\n'
+        if True:
+            chat_log_md += '<center>(history)</center>\n\n' if chat_use_history else '<center>(not used history)</center>\n\n'
+            chat_log_md += "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", chat_log)])
+            chat_log_md += '\n---\n'
+
+        # chat_log_md = ''
+        # chat_log_md = "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", prompt_table)]) if chat_use_prompt else ''
+        # chat_log_md += "\n".join([xx for xx in map(lambda it: f"##### `{it[0]}`\n\n{it[1]}\n\n", hist)])
+
+        chat_log_md += "\n"
+        chat_log_md += str(error)
+        return json.dumps(new_state), chat_log, chat_log_md, chat_log_md, None, props_json, chat_input
+
+
+def clear_history():
+    return [], ""
+
+
+def copy_history(txt):
+    # print('\n\n copying')
+    # print(txt)
+    # print('\n\n')
+    pass
+
+
+def update_saved_prompt_titles(global_state_json, selected_saved_prompt_title):
+    print('')
+    global_state = json.loads(global_state_json or "{}")
+    print(global_state)
+    print(selected_saved_prompt_title)
+    saved_prompts = global_state.get('saved_prompts') or []
+    print(saved_prompts)
+    the_choices = [(it.get('title') or '[untitled]') for it in saved_prompts]
+    print(the_choices)
+    print('')
+    return gradio.Dropdown.update(choices=the_choices)
+
+
+def save_prompt(global_state_json, saved_prompts, prompt_title, prompt_table):
+    the_choices = []
+    global_state = json.loads(global_state_json or "{}")
+    saved_prompts = global_state.get('saved_prompts') or []
+    if len(saved_prompts):
+        the_choices = [it.get('title') or '[untitled]' for it in saved_prompts]
+        pass
+    return global_state_json, gradio.Dropdown.update(choices=the_choices, value=prompt_title), prompt_title, prompt_table
+
+
+def load_saved_prompt(title):
+    pass