int/gpt_academic
1
0
派生 0
镜像自地址 https://github.com/binary-husky/gpt_academic.git 已同步 2025-12-06 06:26:47 +00:00
代码 工单 软件包 项目 版本发布 百科 动态

format source code

浏览代码
这个提交包含在：
binary-husky
2024-01-13 18:00:52 +08:00
父节点 1714116a89
当前提交 7ab379688e
共有 12 个文件被更改，包括 1049 次插入 和 595 次删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件

417
toolbox.py
查看文件

@@ -11,6 +11,7 @@ from functools import wraps
from shared_utils.config_loader import get_conf
from shared_utils.config_loader import set_conf
from shared_utils.advanced_markdown_format import format_io
from shared_utils.advanced_markdown_format import markdown_convertion
from shared_utils.key_pattern_manager import select_api_key
from shared_utils.key_pattern_manager import is_any_api_key
from shared_utils.key_pattern_manager import what_keys
@@ -20,10 +21,10 @@ from shared_utils.connect_void_terminal import get_plugin_default_kwargs
from shared_utils.connect_void_terminal import get_chat_default_kwargs
pj = os.path.join
default_user_name = 'default_user'
default_user_name = "default_user"
"""
========================================================================
=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
第一部分
函数插件输入输出接驳区
- ChatBotWithCookies: 带Cookies的Chatbot类，为实现更多强大的功能做基础
@@ -32,7 +33,7 @@ default_user_name = 'default_user'
- CatchException: 将插件中出的所有问题显示在界面上
- HotReload: 实现插件的热更新
- trimmed_format_exc: 打印traceback，为了安全而隐藏绝对地址
========================================================================
=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
"""
@@ -120,22 +121,30 @@ def ArgsGeneralWrapper(f):
return decorated
def update_ui(chatbot, history, msg='正常', **kwargs): # 刷新界面
def update_ui(chatbot, history, msg="正常", **kwargs): # 刷新界面
"""
刷新用户界面
"""
assert isinstance(chatbot, ChatBotWithCookies), "在传递chatbot的过程中不要将其丢弃。必要时, 可用clear将其清空, 然后用for+append循环重新赋值。"
assert isinstance(
chatbot, ChatBotWithCookies
), "在传递chatbot的过程中不要将其丢弃。必要时, 可用clear将其清空, 然后用for+append循环重新赋值。"
cookies = chatbot.get_cookies()
# 备份一份History作为记录
cookies.update({'history': history})
cookies.update({"history": history})
# 解决插件锁定时的界面显示问题
if cookies.get('lock_plugin', None):
label = cookies.get('llm_model', "") + " | " + "正在锁定插件" + cookies.get('lock_plugin', None)
if cookies.get("lock_plugin", None):
label = (
cookies.get("llm_model", "")
+ " | "
+ "正在锁定插件"
+ cookies.get("lock_plugin", None)
)
chatbot_gr = gradio.update(value=chatbot, label=label)
if cookies.get('label', "") != label: cookies['label'] = label # 记住当前的label
elif cookies.get('label', None):
chatbot_gr = gradio.update(value=chatbot, label=cookies.get('llm_model', ""))
cookies['label'] = None # 清空label
if cookies.get("label", "") != label:
cookies["label"] = label # 记住当前的label
elif cookies.get("label", None):
chatbot_gr = gradio.update(value=chatbot, label=cookies.get("llm_model", ""))
cookies["label"] = None # 清空label
else:
chatbot_gr = chatbot
@@ -146,7 +155,8 @@ def update_ui_lastest_msg(lastmsg, chatbot, history, delay=1): # 刷新界面
"""
刷新用户界面
"""
if len(chatbot) == 0: chatbot.append(["update_ui_last_msg", lastmsg])
if len(chatbot) == 0:
chatbot.append(["update_ui_last_msg", lastmsg])
chatbot[-1] = list(chatbot[-1])
chatbot[-1][-1] = lastmsg
yield from update_ui(chatbot=chatbot, history=history)
@@ -155,6 +165,7 @@ def update_ui_lastest_msg(lastmsg, chatbot, history, delay=1): # 刷新界面
def trimmed_format_exc():
import os, traceback
str = traceback.format_exc()
current_path = os.getcwd()
replace_path = "."
@@ -194,19 +205,21 @@ def HotReload(f):
最后，使用yield from语句返回重新加载过的函数，并在被装饰的函数上执行。
最终，装饰器函数返回内部函数。这个内部函数可以将函数的原始定义更新为最新版本，并执行函数的新版本。
"""
if get_conf('PLUGIN_HOT_RELOAD'):
if get_conf("PLUGIN_HOT_RELOAD"):
@wraps(f)
def decorated(*args, **kwargs):
fn_name = f.__name__
f_hot_reload = getattr(importlib.reload(inspect.getmodule(f)), fn_name)
yield from f_hot_reload(*args, **kwargs)
return decorated
else:
return f
"""
========================================================================
=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
第二部分
其他小工具:
- write_history_to_file: 将结果写入markdown文件中
@@ -220,13 +233,13 @@ def HotReload(f):
- clip_history: 当历史上下文过长时，自动截断
- get_conf: 获取设置
- select_api_key: 根据当前的模型类别，抽取可用的api-key
========================================================================
=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
"""
def get_reduce_token_percent(text):
"""
* 此函数未来将被弃用
* 此函数未来将被弃用
"""
try:
# text = "maximum context length is 4097 tokens. However, your messages resulted in 4870 tokens"
@@ -239,36 +252,40 @@ def get_reduce_token_percent(text):
assert ratio > 0 and ratio < 1
return ratio, str(int(current_tokens - max_limit))
except:
return 0.5, '不详'
return 0.5, "不详"
def write_history_to_file(history, file_basename=None, file_fullname=None, auto_caption=True):
def write_history_to_file(
history, file_basename=None, file_fullname=None, auto_caption=True
):
"""
将对话记录history以Markdown格式写入文件中。如果没有指定文件名，则使用当前时间生成文件名。
"""
import os
import time
if file_fullname is None:
if file_basename is not None:
file_fullname = pj(get_log_folder(), file_basename)
else:
file_fullname = pj(get_log_folder(), f'GPT-Academic-{gen_time_str()}.md')
file_fullname = pj(get_log_folder(), f"GPT-Academic-{gen_time_str()}.md")
os.makedirs(os.path.dirname(file_fullname), exist_ok=True)
with open(file_fullname, 'w', encoding='utf8') as f:
f.write('# GPT-Academic Report\n')
with open(file_fullname, "w", encoding="utf8") as f:
f.write("# GPT-Academic Report\n")
for i, content in enumerate(history):
try:
if type(content) != str: content = str(content)
if type(content) != str:
content = str(content)
except:
continue
if i % 2 == 0 and auto_caption:
f.write('## ')
f.write("## ")
try:
f.write(content)
except:
# remove everything that cannot be handled by utf8
f.write(content.encode('utf-8', 'ignore').decode())
f.write('\n\n')
f.write(content.encode("utf-8", "ignore").decode())
f.write("\n\n")
res = os.path.abspath(file_fullname)
return res
@@ -277,9 +294,9 @@ def regular_txt_to_markdown(text):
"""
将普通文本转换为Markdown格式的文本。
"""
text = text.replace('\n', '\n\n')
text = text.replace('\n\n\n', '\n\n')
text = text.replace('\n\n\n', '\n\n')
text = text.replace("\n", "\n\n")
text = text.replace("\n\n\n", "\n\n")
text = text.replace("\n\n\n", "\n\n")
return text
@@ -297,8 +314,9 @@ def find_free_port():
"""
import socket
from contextlib import closing
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
s.bind(('', 0))
s.bind(("", 0))
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
return s.getsockname()[1]
@@ -307,54 +325,58 @@ def extract_archive(file_path, dest_dir):
import zipfile
import tarfile
import os
# Get the file extension of the input file
file_extension = os.path.splitext(file_path)[1]
# Extract the archive based on its extension
if file_extension == '.zip':
with zipfile.ZipFile(file_path, 'r') as zipobj:
if file_extension == ".zip":
with zipfile.ZipFile(file_path, "r") as zipobj:
zipobj.extractall(path=dest_dir)
print("Successfully extracted zip archive to {}".format(dest_dir))
elif file_extension in ['.tar', '.gz', '.bz2']:
with tarfile.open(file_path, 'r:*') as tarobj:
elif file_extension in [".tar", ".gz", ".bz2"]:
with tarfile.open(file_path, "r:*") as tarobj:
tarobj.extractall(path=dest_dir)
print("Successfully extracted tar archive to {}".format(dest_dir))
# 第三方库，需要预先pip install rarfile
# 此外，Windows上还需要安装winrar软件，配置其Path环境变量，如"C:\Program Files\WinRAR"才可以
elif file_extension == '.rar':
elif file_extension == ".rar":
try:
import rarfile
with rarfile.RarFile(file_path) as rf:
rf.extractall(path=dest_dir)
print("Successfully extracted rar archive to {}".format(dest_dir))
except:
print("Rar format requires additional dependencies to install")
return '\n\n解压失败! 需要安装pip install rarfile来解压rar文件。建议：使用zip压缩格式。'
return "\n\n解压失败! 需要安装pip install rarfile来解压rar文件。建议：使用zip压缩格式。"
# 第三方库，需要预先pip install py7zr
elif file_extension == '.7z':
elif file_extension == ".7z":
try:
import py7zr
with py7zr.SevenZipFile(file_path, mode='r') as f:
with py7zr.SevenZipFile(file_path, mode="r") as f:
f.extractall(path=dest_dir)
print("Successfully extracted 7z archive to {}".format(dest_dir))
except:
print("7z format requires additional dependencies to install")
return '\n\n解压失败! 需要安装pip install py7zr来解压7z文件'
return "\n\n解压失败! 需要安装pip install py7zr来解压7z文件"
else:
return ''
return ''
return ""
return ""
def find_recent_files(directory):
"""
me: find files that is created with in one minutes under a directory with python, write a function
gpt: here it is!
me: find files that is created with in one minutes under a directory with python, write a function
gpt: here it is!
"""
import os
import time
current_time = time.time()
one_minute_ago = current_time - 60
recent_files = []
@@ -362,7 +384,7 @@ def find_recent_files(directory):
os.makedirs(directory, exist_ok=True)
for filename in os.listdir(directory):
file_path = pj(directory, filename)
if file_path.endswith('.log'):
if file_path.endswith(".log"):
continue
created_time = os.path.getmtime(file_path)
if created_time >= one_minute_ago:
@@ -388,49 +410,53 @@ def file_already_in_downloadzone(file, user_path):
def promote_file_to_downloadzone(file, rename_file=None, chatbot=None):
# 将文件复制一份到下载区
import shutil
if chatbot is not None:
user_name = get_user(chatbot)
else:
user_name = default_user_name
if not os.path.exists(file):
raise FileNotFoundError(f'文件{file}不存在')
raise FileNotFoundError(f"文件{file}不存在")
user_path = get_log_folder(user_name, plugin_name=None)
if file_already_in_downloadzone(file, user_path):
new_path = file
else:
user_path = get_log_folder(user_name, plugin_name='downloadzone')
if rename_file is None: rename_file = f'{gen_time_str()}-{os.path.basename(file)}'
user_path = get_log_folder(user_name, plugin_name="downloadzone")
if rename_file is None:
rename_file = f"{gen_time_str()}-{os.path.basename(file)}"
new_path = pj(user_path, rename_file)
# 如果已经存在，先删除
if os.path.exists(new_path) and not os.path.samefile(new_path, file): os.remove(new_path)
if os.path.exists(new_path) and not os.path.samefile(new_path, file):
os.remove(new_path)
# 把文件复制过去
if not os.path.exists(new_path): shutil.copyfile(file, new_path)
if not os.path.exists(new_path):
shutil.copyfile(file, new_path)
# 将文件添加到chatbot cookie中
if chatbot is not None:
if 'files_to_promote' in chatbot._cookies:
current = chatbot._cookies['files_to_promote']
if "files_to_promote" in chatbot._cookies:
current = chatbot._cookies["files_to_promote"]
else:
current = []
if new_path not in current: # 避免把同一个文件添加多次
chatbot._cookies.update({'files_to_promote': [new_path] + current})
if new_path not in current: # 避免把同一个文件添加多次
chatbot._cookies.update({"files_to_promote": [new_path] + current})
return new_path
def disable_auto_promotion(chatbot):
chatbot._cookies.update({'files_to_promote': []})
chatbot._cookies.update({"files_to_promote": []})
return
def del_outdated_uploads(outdate_time_seconds, target_path_base=None):
if target_path_base is None:
user_upload_dir = get_conf('PATH_PRIVATE_UPLOAD')
user_upload_dir = get_conf("PATH_PRIVATE_UPLOAD")
else:
user_upload_dir = target_path_base
current_time = time.time()
one_hour_ago = current_time - outdate_time_seconds
# Get a list of all subdirectories in the user_upload_dir folder
# Remove subdirectories that are older than one hour
for subdirectory in glob.glob(f'{user_upload_dir}/*'):
for subdirectory in glob.glob(f"{user_upload_dir}/*"):
subdirectory_time = os.path.getmtime(subdirectory)
if subdirectory_time < one_hour_ago:
try:
@@ -447,8 +473,8 @@ def html_local_file(file):
return file
def html_local_img(__file, layout='left', max_width=None, max_height=None, md=True):
style = ''
def html_local_img(__file, layout="left", max_width=None, max_height=None, md=True):
style = ""
if max_width is not None:
style += f"max-width: {max_width};"
if max_height is not None:
@@ -456,20 +482,23 @@ def html_local_img(__file, layout='left', max_width=None, max_height=None, md=Tr
__file = html_local_file(__file)
a = f'<div align="{layout}"><img src="{__file}" style="{style}"></div>'
if md:
a = f'![{__file}]({__file})'
a = f"![{__file}]({__file})"
return a
def file_manifest_filter_type(file_list, filter_: list = None):
new_list = []
if not filter_: filter_ = ['png', 'jpg', 'jpeg']
if not filter_:
filter_ = ["png", "jpg", "jpeg"]
for file in file_list:
if str(os.path.basename(file)).split('.')[-1] in filter_:
if str(os.path.basename(file)).split(".")[-1] in filter_:
new_list.append(html_local_img(file, md=False))
else:
new_list.append(file)
return new_list
def to_markdown_tabs(head: list, tabs: list, alignment=':---:', column=False):
def to_markdown_tabs(head: list, tabs: list, alignment=":---:", column=False):
"""
Args:
head: 表头：[]
@@ -487,17 +516,20 @@ def to_markdown_tabs(head: list, tabs: list, alignment=':---:', column=False):
max_len = max(len(column) for column in transposed_tabs)
tab_format = "| %s "
tabs_list = "".join([tab_format % i for i in head]) + '|\n'
tabs_list += "".join([tab_format % alignment for i in head]) + '|\n'
tabs_list = "".join([tab_format % i for i in head]) + "|\n"
tabs_list += "".join([tab_format % alignment for i in head]) + "|\n"
for i in range(max_len):
row_data = [tab[i] if i < len(tab) else '' for tab in transposed_tabs]
row_data = [tab[i] if i < len(tab) else "" for tab in transposed_tabs]
row_data = file_manifest_filter_type(row_data, filter_=None)
tabs_list += "".join([tab_format % i for i in row_data]) + '|\n'
tabs_list += "".join([tab_format % i for i in row_data]) + "|\n"
return tabs_list
def on_file_uploaded(request: gradio.Request, files, chatbot, txt, txt2, checkboxes, cookies):
def on_file_uploaded(
request: gradio.Request, files, chatbot, txt, txt2, checkboxes, cookies
):
"""
当文件被上传时的回调函数
"""
@@ -515,94 +547,118 @@ def on_file_uploaded(request: gradio.Request, files, chatbot, txt, txt2, checkbo
del_outdated_uploads(outdate_time_seconds, get_upload_folder(user_name))
# 逐个文件转移到目标路径
upload_msg = ''
upload_msg = ""
for file in files:
file_origin_name = os.path.basename(file.orig_name)
this_file_path = pj(target_path_base, file_origin_name)
shutil.move(file.name, this_file_path)
upload_msg += extract_archive(file_path=this_file_path, dest_dir=this_file_path + '.extract')
upload_msg += extract_archive(
file_path=this_file_path, dest_dir=this_file_path + ".extract"
)
# 整理文件集合 输出消息
moved_files = [fp for fp in glob.glob(f'{target_path_base}/**/*', recursive=True)]
moved_files_str = to_markdown_tabs(head=['文件'], tabs=[moved_files])
chatbot.append(['我上传了文件，请查收',
f'[Local Message] 收到以下文件: \n\n{moved_files_str}' +
f'\n\n调用路径参数已自动修正到: \n\n{txt}' +
f'\n\n现在您点击任意函数插件时，以上文件将被作为输入参数' + upload_msg])
moved_files = [fp for fp in glob.glob(f"{target_path_base}/**/*", recursive=True)]
moved_files_str = to_markdown_tabs(head=["文件"], tabs=[moved_files])
chatbot.append(
[
"我上传了文件，请查收",
f"[Local Message] 收到以下文件: \n\n{moved_files_str}"
+ f"\n\n调用路径参数已自动修正到: \n\n{txt}"
+ f"\n\n现在您点击任意函数插件时，以上文件将被作为输入参数"
+ upload_msg,
]
)
txt, txt2 = target_path_base, ""
if "浮动输入区" in checkboxes:
txt, txt2 = txt2, txt
# 记录近期文件
cookies.update({
'most_recent_uploaded': {
'path': target_path_base,
'time': time.time(),
'time_str': time_tag
}})
cookies.update(
{
"most_recent_uploaded": {
"path": target_path_base,
"time": time.time(),
"time_str": time_tag,
}
}
)
return chatbot, txt, txt2, cookies
def on_report_generated(cookies, files, chatbot):
# from toolbox import find_recent_files
# PATH_LOGGING = get_conf('PATH_LOGGING')
if 'files_to_promote' in cookies:
report_files = cookies['files_to_promote']
cookies.pop('files_to_promote')
if "files_to_promote" in cookies:
report_files = cookies["files_to_promote"]
cookies.pop("files_to_promote")
else:
report_files = []
# report_files = find_recent_files(PATH_LOGGING)
if len(report_files) == 0:
return cookies, None, chatbot
# files.extend(report_files)
file_links = ''
for f in report_files: file_links += f'<br/><a href="file={os.path.abspath(f)}" target="_blank">{f}</a>'
chatbot.append(['报告如何远程获取？', f'报告已经添加到右侧“文件上传区”（可能处于折叠状态），请查收。{file_links}'])
file_links = ""
for f in report_files:
file_links += (
f'<br/><a href="file={os.path.abspath(f)}" target="_blank">{f}</a>'
)
chatbot.append(["报告如何远程获取？", f"报告已经添加到右侧“文件上传区”（可能处于折叠状态），请查收。{file_links}"])
return cookies, report_files, chatbot
def load_chat_cookies():
API_KEY, LLM_MODEL, AZURE_API_KEY = get_conf('API_KEY', 'LLM_MODEL', 'AZURE_API_KEY')
AZURE_CFG_ARRAY, NUM_CUSTOM_BASIC_BTN = get_conf('AZURE_CFG_ARRAY', 'NUM_CUSTOM_BASIC_BTN')
API_KEY, LLM_MODEL, AZURE_API_KEY = get_conf(
"API_KEY", "LLM_MODEL", "AZURE_API_KEY"
)
AZURE_CFG_ARRAY, NUM_CUSTOM_BASIC_BTN = get_conf(
"AZURE_CFG_ARRAY", "NUM_CUSTOM_BASIC_BTN"
)
# deal with azure openai key
if is_any_api_key(AZURE_API_KEY):
if is_any_api_key(API_KEY):
API_KEY = API_KEY + ',' + AZURE_API_KEY
API_KEY = API_KEY + "," + AZURE_API_KEY
else:
API_KEY = AZURE_API_KEY
if len(AZURE_CFG_ARRAY) > 0:
for azure_model_name, azure_cfg_dict in AZURE_CFG_ARRAY.items():
if not azure_model_name.startswith('azure'):
if not azure_model_name.startswith("azure"):
raise ValueError("AZURE_CFG_ARRAY中配置的模型必须以azure开头")
AZURE_API_KEY_ = azure_cfg_dict["AZURE_API_KEY"]
if is_any_api_key(AZURE_API_KEY_):
if is_any_api_key(API_KEY):
API_KEY = API_KEY + ',' + AZURE_API_KEY_
API_KEY = API_KEY + "," + AZURE_API_KEY_
else:
API_KEY = AZURE_API_KEY_
customize_fn_overwrite_ = {}
for k in range(NUM_CUSTOM_BASIC_BTN):
customize_fn_overwrite_.update({
"自定义按钮" + str(k+1):{
"Title": r"",
"Prefix": r"请在自定义菜单中定义提示词前缀.",
"Suffix": r"请在自定义菜单中定义提示词后缀",
customize_fn_overwrite_.update(
{
"自定义按钮"
+ str(k + 1): {
"Title": r"",
"Prefix": r"请在自定义菜单中定义提示词前缀.",
"Suffix": r"请在自定义菜单中定义提示词后缀",
}
}
})
return {'api_key': API_KEY, 'llm_model': LLM_MODEL, 'customize_fn_overwrite': customize_fn_overwrite_}
)
return {
"api_key": API_KEY,
"llm_model": LLM_MODEL,
"customize_fn_overwrite": customize_fn_overwrite_,
}
def clear_line_break(txt):
txt = txt.replace('\n', ' ')
txt = txt.replace(' ', ' ')
txt = txt.replace(' ', ' ')
txt = txt.replace("\n", " ")
txt = txt.replace(" ", " ")
txt = txt.replace(" ", " ")
return txt
class DummyWith():
class DummyWith:
"""
这段代码定义了一个名为DummyWith的空上下文管理器，
它的作用是……额……就是不起作用，即在代码结构不变得情况下取代其他的上下文管理器。
@@ -626,34 +682,47 @@ def run_gradio_in_subpath(demo, auth, port, custom_path):
"""
def is_path_legal(path: str) -> bool:
'''
"""
check path for sub url
path: path to check
return value: do sub url wrap
'''
if path == "/": return True
"""
if path == "/":
return True
if len(path) == 0:
print("ilegal custom path: {}\npath must not be empty\ndeploy on root url".format(path))
print(
"ilegal custom path: {}\npath must not be empty\ndeploy on root url".format(
path
)
)
return False
if path[0] == '/':
if path[1] != '/':
if path[0] == "/":
if path[1] != "/":
print("deploy on sub-path {}".format(path))
return True
return False
print("ilegal custom path: {}\npath should begin with \'/\'\ndeploy on root url".format(path))
print(
"ilegal custom path: {}\npath should begin with '/'\ndeploy on root url".format(
path
)
)
return False
if not is_path_legal(custom_path): raise RuntimeError('Ilegal custom path')
if not is_path_legal(custom_path):
raise RuntimeError("Ilegal custom path")
import uvicorn
import gradio as gr
from fastapi import FastAPI
app = FastAPI()
if custom_path != "/":
@app.get("/")
def read_main():
return {"message": f"Gradio is running at: {custom_path}"}
app = gr.mount_gradio_app(app, demo, path=custom_path)
uvicorn.run(app, host="0.0.0.0", port=port) # , auth=auth
uvicorn.run(app, host="0.0.0.0", port=port) # , auth=auth
def clip_history(inputs, history, tokenizer, max_token_limit):
@@ -667,13 +736,18 @@ def clip_history(inputs, history, tokenizer, max_token_limit):
"""
import numpy as np
from request_llms.bridge_all import model_info
def get_token_num(txt):
return len(tokenizer.encode(txt, disallowed_special=()))
input_token_num = get_token_num(inputs)
if max_token_limit < 5000: output_token_expect = 256 # 4k & 2k models
elif max_token_limit < 9000: output_token_expect = 512 # 8k models
else: output_token_expect = 1024 # 16k & 32k models
if max_token_limit < 5000:
output_token_expect = 256 # 4k & 2k models
elif max_token_limit < 9000:
output_token_expect = 512 # 8k models
else:
output_token_expect = 1024 # 16k & 32k models
if input_token_num < max_token_limit * 3 / 4:
# 当输入部分的token占比小于限制的3/4时，裁剪时
@@ -690,9 +764,9 @@ def clip_history(inputs, history, tokenizer, max_token_limit):
history = []
return history
everything = ['']
everything = [""]
everything.extend(history)
n_token = get_token_num('\n'.join(everything))
n_token = get_token_num("\n".join(everything))
everything_token = [get_token_num(e) for e in everything]
# 截断时的颗粒度
@@ -701,30 +775,33 @@ def clip_history(inputs, history, tokenizer, max_token_limit):
while n_token > max_token_limit:
where = np.argmax(everything_token)
encoded = tokenizer.encode(everything[where], disallowed_special=())
clipped_encoded = encoded[:len(encoded) - delta]
everything[where] = tokenizer.decode(clipped_encoded)[:-1] # -1 to remove the may-be illegal char
clipped_encoded = encoded[: len(encoded) - delta]
everything[where] = tokenizer.decode(clipped_encoded)[
:-1
] # -1 to remove the may-be illegal char
everything_token[where] = get_token_num(everything[where])
n_token = get_token_num('\n'.join(everything))
n_token = get_token_num("\n".join(everything))
history = everything[1:]
return history
"""
========================================================================
=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
第三部分
其他小工具:
- zip_folder: 把某个路径下所有文件压缩，然后转移到指定的另一个路径中（gpt写的）
- gen_time_str: 生成时间戳
- ProxyNetworkActivate: 临时地启动代理网络（如果有）
- objdump/objload: 快捷的调试函数
========================================================================
=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
"""
def zip_folder(source_folder, dest_folder, zip_name):
import zipfile
import os
# Make sure the source folder exists
if not os.path.exists(source_folder):
print(f"{source_folder} does not exist")
@@ -739,7 +816,7 @@ def zip_folder(source_folder, dest_folder, zip_name):
zip_file = pj(dest_folder, zip_name)
# Create a ZipFile object
with zipfile.ZipFile(zip_file, 'w', zipfile.ZIP_DEFLATED) as zipf:
with zipfile.ZipFile(zip_file, "w", zipfile.ZIP_DEFLATED) as zipf:
# Walk through the source folder and add files to the zip file
for foldername, subfolders, filenames in os.walk(source_folder):
for filename in filenames:
@@ -756,29 +833,33 @@ def zip_folder(source_folder, dest_folder, zip_name):
def zip_result(folder):
t = gen_time_str()
zip_folder(folder, get_log_folder(), f'{t}-result.zip')
return pj(get_log_folder(), f'{t}-result.zip')
zip_folder(folder, get_log_folder(), f"{t}-result.zip")
return pj(get_log_folder(), f"{t}-result.zip")
def gen_time_str():
import time
return time.strftime("%Y-%m-%d-%H-%M-%S", time.localtime())
def get_log_folder(user=default_user_name, plugin_name='shared'):
if user is None: user = default_user_name
PATH_LOGGING = get_conf('PATH_LOGGING')
def get_log_folder(user=default_user_name, plugin_name="shared"):
if user is None:
user = default_user_name
PATH_LOGGING = get_conf("PATH_LOGGING")
if plugin_name is None:
_dir = pj(PATH_LOGGING, user)
else:
_dir = pj(PATH_LOGGING, user, plugin_name)
if not os.path.exists(_dir): os.makedirs(_dir)
if not os.path.exists(_dir):
os.makedirs(_dir)
return _dir
def get_upload_folder(user=default_user_name, tag=None):
PATH_PRIVATE_UPLOAD = get_conf('PATH_PRIVATE_UPLOAD')
if user is None: user = default_user_name
PATH_PRIVATE_UPLOAD = get_conf("PATH_PRIVATE_UPLOAD")
if user is None:
user = default_user_name
if tag is None or len(tag) == 0:
target_path_base = pj(PATH_PRIVATE_UPLOAD, user)
else:
@@ -787,9 +868,9 @@ def get_upload_folder(user=default_user_name, tag=None):
def is_the_upload_folder(string):
PATH_PRIVATE_UPLOAD = get_conf('PATH_PRIVATE_UPLOAD')
pattern = r'^PATH_PRIVATE_UPLOAD[\\/][A-Za-z0-9_-]+[\\/]\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-\d{2}$'
pattern = pattern.replace('PATH_PRIVATE_UPLOAD', PATH_PRIVATE_UPLOAD)
PATH_PRIVATE_UPLOAD = get_conf("PATH_PRIVATE_UPLOAD")
pattern = r"^PATH_PRIVATE_UPLOAD[\\/][A-Za-z0-9_-]+[\\/]\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-\d{2}$"
pattern = pattern.replace("PATH_PRIVATE_UPLOAD", PATH_PRIVATE_UPLOAD)
if re.match(pattern, string):
return True
else:
@@ -797,10 +878,10 @@ def is_the_upload_folder(string):
def get_user(chatbotwithcookies):
return chatbotwithcookies._cookies.get('user_name', default_user_name)
return chatbotwithcookies._cookies.get("user_name", default_user_name)
class ProxyNetworkActivate():
class ProxyNetworkActivate:
"""
这段代码定义了一个名为ProxyNetworkActivate的空上下文管理器, 用于给一小段代码上代理
"""
@@ -813,38 +894,48 @@ class ProxyNetworkActivate():
else:
# 给定了task, 我们检查一下
from toolbox import get_conf
WHEN_TO_USE_PROXY = get_conf('WHEN_TO_USE_PROXY')
self.valid = (task in WHEN_TO_USE_PROXY)
WHEN_TO_USE_PROXY = get_conf("WHEN_TO_USE_PROXY")
self.valid = task in WHEN_TO_USE_PROXY
def __enter__(self):
if not self.valid: return self
if not self.valid:
return self
from toolbox import get_conf
proxies = get_conf('proxies')
if 'no_proxy' in os.environ: os.environ.pop('no_proxy')
proxies = get_conf("proxies")
if "no_proxy" in os.environ:
os.environ.pop("no_proxy")
if proxies is not None:
if 'http' in proxies: os.environ['HTTP_PROXY'] = proxies['http']
if 'https' in proxies: os.environ['HTTPS_PROXY'] = proxies['https']
if "http" in proxies:
os.environ["HTTP_PROXY"] = proxies["http"]
if "https" in proxies:
os.environ["HTTPS_PROXY"] = proxies["https"]
return self
def __exit__(self, exc_type, exc_value, traceback):
os.environ['no_proxy'] = '*'
if 'HTTP_PROXY' in os.environ: os.environ.pop('HTTP_PROXY')
if 'HTTPS_PROXY' in os.environ: os.environ.pop('HTTPS_PROXY')
os.environ["no_proxy"] = "*"
if "HTTP_PROXY" in os.environ:
os.environ.pop("HTTP_PROXY")
if "HTTPS_PROXY" in os.environ:
os.environ.pop("HTTPS_PROXY")
return
def objdump(obj, file='objdump.tmp'):
def objdump(obj, file="objdump.tmp"):
import pickle
with open(file, 'wb+') as f:
with open(file, "wb+") as f:
pickle.dump(obj, f)
return
def objload(file='objdump.tmp'):
def objload(file="objdump.tmp"):
import pickle, os
if not os.path.exists(file):
return
with open(file, 'rb') as f:
with open(file, "rb") as f:
return pickle.load(f)
@@ -863,22 +954,25 @@ def Singleton(cls):
def get_pictures_list(path):
file_manifest = [f for f in glob.glob(f'{path}/**/*.jpg', recursive=True)]
file_manifest += [f for f in glob.glob(f'{path}/**/*.jpeg', recursive=True)]
file_manifest += [f for f in glob.glob(f'{path}/**/*.png', recursive=True)]
file_manifest = [f for f in glob.glob(f"{path}/**/*.jpg", recursive=True)]
file_manifest += [f for f in glob.glob(f"{path}/**/*.jpeg", recursive=True)]
file_manifest += [f for f in glob.glob(f"{path}/**/*.png", recursive=True)]
return file_manifest
def have_any_recent_upload_image_files(chatbot):
_5min = 5 * 60
if chatbot is None: return False, None # chatbot is None
if chatbot is None:
return False, None # chatbot is None
most_recent_uploaded = chatbot._cookies.get("most_recent_uploaded", None)
if not most_recent_uploaded: return False, None # most_recent_uploaded is None
if not most_recent_uploaded:
return False, None # most_recent_uploaded is None
if time.time() - most_recent_uploaded["time"] < _5min:
most_recent_uploaded = chatbot._cookies.get("most_recent_uploaded", None)
path = most_recent_uploaded['path']
path = most_recent_uploaded["path"]
file_manifest = get_pictures_list(path)
if len(file_manifest) == 0: return False, None
if len(file_manifest) == 0:
return False, None
return True, file_manifest # most_recent_uploaded is new
else:
return False, None # most_recent_uploaded is too old
@@ -887,16 +981,19 @@ def have_any_recent_upload_image_files(chatbot):
# Function to encode the image
def encode_image(image_path):
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
return base64.b64encode(image_file.read()).decode("utf-8")
def get_max_token(llm_kwargs):
from request_llms.bridge_all import model_info
return model_info[llm_kwargs['llm_model']]['max_token']
return model_info[llm_kwargs["llm_model"]]["max_token"]
def check_packages(packages=[]):
import importlib.util
for p in packages:
spam_spec = importlib.util.find_spec(p)
if spam_spec is None: raise ModuleNotFoundError
if spam_spec is None:
raise ModuleNotFoundError