-
Notifications
You must be signed in to change notification settings - Fork 3
/
app.py
123 lines (99 loc) · 3.86 KB
/
app.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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import os
from typing import Optional, Tuple
import gradio as gr
import pickle
from query_data import get_chain
from ingest_data import embed_chat
from threading import Lock
def click_embed(file):
"""Embed Telegram chat.
"""
embed_chat(file.name)
with open("vectorstore.pkl", "rb") as f:
vectorstore = pickle.load(f)
chain = get_chain(vectorstore)
return chain, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(
visible=True), gr.update(visible=True), gr.update(
visible=True), gr.update(visible=True)
def set_openai_api_key(api_key: str):
"""Set the api key.
"""
if api_key:
os.environ["OPENAI_API_KEY"] = api_key
def upload_file(file_obj):
return file_obj
class ChatWrapper:
def __init__(self):
self.lock = Lock()
def __call__(
self, api_key: str, inp: str, history: Optional[Tuple[str, str]], chain
):
"""Execute the chat functionality."""
self.lock.acquire()
try:
history = history or []
# If chain is None, that is because no API key was provided.
if chain is None:
history.append((inp, "Please paste your OpenAI key to use"))
return history, history
# Set OpenAI key
import openai
openai.api_key = api_key
# Run chain and append input.
output = chain({"question": inp, "chat_history": history})["answer"]
history.append((inp, output))
except Exception as e:
raise e
finally:
self.lock.release()
return history, history
chat = ChatWrapper()
block = gr.Blocks(css=".gradio-container {background-color: lightgray}")
with block:
with gr.Row():
gr.Markdown("<h3><center>Telegram Chat Chat</center></h3>")
openai_api_key_textbox = gr.Textbox(
placeholder="Paste your OpenAI API key (sk-...)",
show_label=False,
lines=1,
type="password",
visible=True
)
telegram_chat_file = gr.File(file_count="single", file_types=["json"], interactive=True, show_label=True,
visible=True,
label="Telegram chat history exported .json file")
embed_button = gr.Button("Create embeddings", visible=True)
chatbot = gr.Chatbot(visible=False)
with gr.Row():
message = gr.Textbox(
label="What's your question?",
placeholder="Ask questions about your Telegram conversation",
lines=1,
visible=False
)
submit = gr.Button(value="Send", variant="secondary", visible=False).style(full_width=False)
with gr.Row(visible=False) as examples_row:
gr.Examples(
examples=[
"When was [your name] happy and how does it show?",
"How could [your name] have been more compassionate?",
"What triggers [your name] instantly?",
],
inputs=message,
)
gr.HTML(
"<center>Powered by <a href='https://github.com/hwchase17/langchain'>LangChain 🦜️🔗</a></center>"
)
state = gr.State()
agent_state = gr.State()
embed_button.click(click_embed, inputs=[telegram_chat_file],
outputs=[agent_state, openai_api_key_textbox, telegram_chat_file, embed_button, chatbot, message,
submit, examples_row])
submit.click(chat, inputs=[openai_api_key_textbox, message, state, agent_state], outputs=[chatbot, state])
print(agent_state)
message.submit(chat, inputs=[openai_api_key_textbox, message, state, agent_state], outputs=[chatbot, state])
openai_api_key_textbox.change(
set_openai_api_key,
inputs=[openai_api_key_textbox],
)
block.launch(debug=True)