-
Notifications
You must be signed in to change notification settings - Fork 33
/
dialog.py
81 lines (65 loc) · 2.41 KB
/
dialog.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import os
from ..config import config_path, current_dir_path, load_api_keys
class start_dialog:
def __init__(self):
self.start = True
# 生成一个hash值作为id
self.id = hash(str(self))
# temp文件夹不存在就创建
if not os.path.exists(os.path.join(current_dir_path, "temp")):
os.makedirs(os.path.join(current_dir_path, "temp"))
# 构建prompt.txt的绝对路径
self.prompt_path = os.path.join(current_dir_path, "temp", str(self.id) + ".txt")
# 如果文件不存在,创建prompt.txt文件,存在就覆盖文件
if not os.path.exists(self.prompt_path):
with open(self.prompt_path, "w", encoding="utf-8") as f:
f.write("")
@classmethod
def INPUT_TYPES(s):
return {"required": {"start_dialog": ("STRING", {})}}
RETURN_TYPES = (
"STRING",
"STRING",
)
RETURN_NAMES = (
"user_prompt",
"dialog_id",
)
FUNCTION = "dialog"
# OUTPUT_NODE = False
CATEGORY = "大模型派对(llm_party)/工作流(workflow)"
def dialog(self, start_dialog):
if self.start == False:
# 读取prompt.txt文件内容
with open(self.prompt_path, "r", encoding="utf-8") as f:
prompt = f.read()
else:
prompt = start_dialog
self.start = False
dialog_id = self.id
return (
prompt,
dialog_id,
)
class end_dialog:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"dialog_id": ("STRING", {"forceInput": True}),
"assistant_response": ("STRING", {"forceInput": True}),
}
}
RETURN_TYPES = ()
RETURN_NAMES = ()
FUNCTION = "dialog"
OUTPUT_NODE = False
CATEGORY = "大模型派对(llm_party)/工作流(workflow)"
def dialog(self, dialog_id, assistant_response):
# 构建prompt.txt的绝对路径
self.prompt_path = os.path.join(current_dir_path, "temp", str(dialog_id) + ".txt")
# 如果文件不存在,创建prompt.txt文件,存在就覆盖文件
if not os.path.exists(self.prompt_path):
with open(self.prompt_path, "w", encoding="utf-8") as f:
f.write(assistant_response)
return ()