|
from huggingface_hub import InferenceClient |
|
import gradio as gr |
|
import json |
|
from datetime import datetime,timedelta |
|
import os |
|
from google_auth_oauthlib.flow import InstalledAppFlow |
|
from googleapiclient.discovery import build |
|
from google.oauth2 import service_account |
|
from googleapiclient.http import MediaFileUpload |
|
from pathlib import Path |
|
import argparse |
|
from huggingface_hub import snapshot_download |
|
import random |
|
import string |
|
import re |
|
from llama_cpp import Llama |
|
|
|
|
|
|
|
|
|
repo_name = 'TheBloke/OpenHermes-2.5-Mistral-7B-GGUF' |
|
model_file = "openhermes-2.5-mistral-7b.Q4_K_M.gguf" |
|
SCOPES = ['https://www.googleapis.com/auth/drive'] |
|
SERVICE_ACCOUNT_FILE = '/home/user/app/service_account_creds.json' |
|
folder_id = '1Hjbu7FmuVs0Yz8y_veo6SzY_2tc48OWt' |
|
|
|
initContext = """<|im_start|>You are a Roleplay AI playing the character of Barry Lawson, You will not break character, You will not reveal that you are an AI. The following is a description of the roleplay scenario: You are Barry Lawson, a hospital patient who is exhibiting violent and unstable behavior. Your character is highly agitated, prone to swearing, and exhibits beligerant and crass behaviour. Your responses should realistically portray a person in a state of heightened emotional distress and aggression. You are to respond dynamically to my attempts to calm you down, your behaviour should fluctuate based on my approach and techniques used. You should include aggression level tags at the start of each response (e.g. [Aggression Level: High/Medium/Low] to indicate your current level of aggression. After 10 responses you will provide feedback on my approach and discuss how my prompts could have better lowered your aggression level. You should emphasize realistic and challenging interactions. Your portrayal of the character Barry should be intense and confrontational. Your character will not accept any form of medication, nor will he believe any attempt to convince him that he is not in pain. """ |
|
unique_id = "" |
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") |
|
|
|
|
|
def load_model(): |
|
llm = Llama(model_path=model_file, model_type="mistral",n_gpu_layers=100,n_ctx = 2048,verbose=True) |
|
return llm |
|
|
|
|
|
def generate_unique_id(): |
|
|
|
letters = ''.join(random.choices(string.ascii_letters, k=1)) |
|
digits = ''.join(random.choices(string.digits, k=2)) |
|
unique_id = letters + digits |
|
return unique_id |
|
|
|
|
|
print('Fetching model:', repo_name, model_file) |
|
snapshot_download(repo_id=repo_name, local_dir=".", allow_patterns=model_file) |
|
print('Done fetching model:') |
|
|
|
|
|
class ChatbotAPP: |
|
def __init__(self,model,service_account_file,scopes,folder_id,unique_id,initContext): |
|
self.llm = model |
|
self.service_account_file = service_account_file |
|
self.scopes = scopes |
|
self.folder_id = folder_id |
|
self.unique_id = unique_id |
|
self.chat_history = [] |
|
self.chat_log_history = [] |
|
self.isFirstRun = True |
|
self.initContext = initContext |
|
self.context = "" |
|
self.agreed = False |
|
self.service = self.get_drive_service() |
|
self.app = self.create_app() |
|
self.chat_log_name = "" |
|
self.start_time=datetime.now() |
|
|
|
|
|
def get_drive_service(self): |
|
credentials = service_account.Credentials.from_service_account_file( |
|
self.service_account_file, scopes=self.scopes) |
|
self.service = build('drive', 'v3', credentials=credentials) |
|
print("Google Service Created") |
|
return self.service |
|
|
|
def generate_unique_id(self): |
|
|
|
letters = ''.join(random.choices(string.ascii_letters, k=3)) |
|
digits = ''.join(random.choices(string.digits, k=3)) |
|
unique_id = letters + digits |
|
return unique_id |
|
|
|
|
|
def search_file(self): |
|
|
|
query = f"name = '{self.chat_log_name}' and '{self.folder_id}' in parents and trashed = false" |
|
response = self.service.files().list(q=query, spaces='drive', fields='files(id, name)').execute() |
|
files = response.get('files', []) |
|
if not files: |
|
print(f"Chat log {self.chat_log_name} does not exist") |
|
else: |
|
print(f"Chat log {self.chat_log_name} exist") |
|
return files |
|
|
|
def strip_text(self, text): |
|
|
|
|
|
pattern = r"\(.*?\)|<.*?>.*|\n|\s{3,}" |
|
|
|
cleaned_text = re.sub(pattern, "", text) |
|
cleaned_text = cleaned_text.replace("im_start", "").replace("im_end", "").replace("<","").replace("|","").replace(":","") |
|
return cleaned_text |
|
|
|
def upload_to_google_drive(self): |
|
|
|
existing_files = self.search_file() |
|
print(existing_files) |
|
|
|
data = { |
|
|
|
|
|
|
|
|
|
|
|
|
|
"Unique ID": self.unique_id, |
|
"chat_history": self.chat_log_history |
|
} |
|
|
|
with open(self.chat_log_name, "w") as log_file: |
|
json.dump(data, log_file, indent=4) |
|
|
|
if not existing_files: |
|
|
|
|
|
file_metadata = { |
|
'name': self.chat_log_name, |
|
'parents': [self.folder_id],'mimeType': 'application/json' |
|
} |
|
media = MediaFileUpload(self.chat_log_name, mimetype='application/json') |
|
file = self.service.files().create(body=file_metadata, media_body=media, fields='id').execute() |
|
print(f"Uploaded new file with ID: {file.get('id')}") |
|
else: |
|
print(f"File '{self.chat_log_name}' already exists.") |
|
|
|
file_id = existing_files[0]['id'] |
|
media = MediaFileUpload(self.chat_log_name, mimetype='application/json') |
|
updated_file = self.service.files().update(fileId=file_id, media_body=media).execute() |
|
print(f"Updated existing file with ID: {updated_file.get('id')}") |
|
|
|
def generate(self,prompt, history): |
|
|
|
|
|
if self.agreed: |
|
|
|
current_time = datetime.now() |
|
|
|
user_msg_timestamp = datetime.now().strftime("%H:%M:%S") |
|
|
|
firstmsg ="" |
|
if self.isFirstRun: |
|
self.start_time = datetime.now() |
|
self.context = self.initContext |
|
self.isFirstRun = False |
|
firstmsg = prompt |
|
|
|
if (current_time - self.start_time) > timedelta (minutes = 15): |
|
|
|
temp_history = [] |
|
temp_history.append((""" |
|
. |
|
. |
|
|
|
""","15 minutes have passed while chatting with Barry. Please fill out the survey at <LINK> . Remember to copy your unique session ID above")) |
|
return temp_history |
|
|
|
self.context += """ |
|
<|im_start|>nurse |
|
Nurse:"""+prompt+""" |
|
<|im_start|>barry |
|
Barry: |
|
""" |
|
|
|
response = "" |
|
|
|
while(len(response) < 1): |
|
output = self.llm(self.context, max_tokens=400, stop=["Nurse:"], echo=False) |
|
response = output["choices"][0]["text"] |
|
response = response.strip() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
cleaned_response = self.strip_text(response) |
|
|
|
bot_msg_timestamp = datetime.now().strftime("%H:%M:%S") |
|
|
|
|
|
self.chat_history.append((prompt,cleaned_response)) |
|
if not self.isFirstRun: |
|
self.chat_log_history.append({"user": "("+user_msg_timestamp+")"+prompt, "bot": "("+bot_msg_timestamp+")"+cleaned_response}) |
|
self.upload_to_google_drive() |
|
|
|
else: |
|
self.chat_log_history.append({"user": firstmsg, "bot": cleaned_response}) |
|
|
|
self.context += response |
|
|
|
print (self.context) |
|
return self.chat_history |
|
|
|
else: |
|
temp_history=[] |
|
output = "Did you forget to read the Instructions?" |
|
temp_history.append((prompt,output)) |
|
return temp_history |
|
|
|
def update_chatlog_name(self): |
|
self.chat_log_name = f'chat_log_for_{self.unique_id}_{datetime.now().strftime("%Y-%m-%d_%H-%M-%S")}.json' |
|
return self.chat_log_name |
|
|
|
def start_chat_button_fn(self,agree_status): |
|
|
|
if agree_status: |
|
self.agreed = agree_status |
|
self.update_chatlog_name() |
|
return f"You can start chatting now." |
|
else: |
|
return "Read the Instructions" |
|
|
|
def reset_chat_interface(self): |
|
|
|
self.chat_history = [] |
|
self.chat_log_history = [] |
|
self.isFirstRun = True |
|
return "Chat has been reset." |
|
|
|
def reset_name_interface(self): |
|
|
|
Name = "" |
|
Occupation = "" |
|
YearsOfExp = "" |
|
Ethnicity = "" |
|
Gender = "" |
|
Age = "" |
|
chat_log_name = "" |
|
return "User info has been reset." |
|
|
|
def reset_all(self): |
|
|
|
message1 = self.reset_chat_interface() |
|
|
|
|
|
self.unique_id = self.generate_unique_id() |
|
self.update_chatlog_name() |
|
self.isFirstRun = True |
|
return f"All Chat components have been rest. Uniqe ID for this session is, {self.unique_id}. Please note this down.",self.unique_id |
|
|
|
def clear_chat_window(self): |
|
return [] |
|
|
|
def create_app(self): |
|
|
|
with gr.Blocks(theme=gr.themes.Soft()) as app: |
|
gr.Markdown("# ECU-IVADE: Chat With Barry") |
|
unique_id_display = gr.Textbox(value=self.unique_id, label="Session Unique ID", interactive=False,show_copy_button = True) |
|
clock_html = """ |
|
<div style='margin-top: 20px;'> |
|
<strong>Current Time:</strong> <span id='clock'></span> |
|
</div> |
|
<script> |
|
function updateClock() { |
|
var now = new Date(); |
|
var hours = now.getHours(); |
|
var minutes = now.getMinutes(); |
|
var seconds = now.getSeconds(); |
|
minutes = minutes < 10 ? '0' + minutes : minutes; |
|
seconds = seconds < 10 ? '0' + seconds : seconds; |
|
var timeString = hours + ':' + minutes + ':' + seconds; |
|
document.getElementById('clock').textContent = timeString; |
|
} |
|
setInterval(updateClock, 1000); |
|
updateClock(); // initial call to display clock immediately |
|
</script> |
|
""" |
|
|
|
|
|
gr.HTML(value=clock_html) |
|
|
|
with gr.Tab("Terms and Conditions"): |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
gr.Markdown("## Instructions") |
|
gr.Markdown(""" |
|
Before Talking with Barry, please read the following instructions carefully: |
|
|
|
- Please ensure that you have read and agreed to the data collection policy provided as part of your agreement to participate as logging and collection of interactions is automated and unidentifiable. |
|
- Interactions with the chatbot require you to type your response and press the button to submit, you cannot press enter. |
|
- There is a tab at the top to reset the chat bot for more interactions, please ensure to reset after filling out the survey and before starting your next conversation. |
|
- Select the 'Barry' tab at the top to chat with Barry |
|
- The clear chat button is only for if the chat window gets bugged, you should not have to use it unless you cannot see messages or it is printing strange messages. |
|
- After clicking begin, depending on your browser you may need to manually navigate to the Barry Tab. |
|
- Remember to note your Unique Identifier at the top for your feedback survey. |
|
|
|
Please check the box below to proceed. |
|
""") |
|
agree_status = gr.Checkbox(label="I have read and understand the instructions.") |
|
status_label = gr.Markdown() |
|
start_chat_button = gr.Button("Begin") |
|
|
|
start_chat_button.click(self.start_chat_button_fn, inputs=[agree_status], outputs=[status_label]) |
|
|
|
|
|
with gr.Tab("Barry"): |
|
chatbot = gr.Chatbot(show_label=True,label="Chat With Barry") |
|
msg = gr.Textbox(label="Hello") |
|
send = gr.Button("Send") |
|
clear = gr.Button("Clear Chat") |
|
send.click(self.generate, inputs=[msg], outputs=chatbot) |
|
clear.click(self.clear_chat_window, inputs=[], outputs=chatbot) |
|
|
|
with gr.Tab("Reset"): |
|
reset_button = gr.Button("Reset The Conversation With Barry") |
|
reset_output = gr.Textbox(label="Reset Output", interactive=False) |
|
reset_button.click(self.reset_all, inputs=[], outputs=[reset_output,unique_id_display]) |
|
return app |
|
|
|
|
|
llm = load_model() |
|
unique_id = generate_unique_id() |
|
chatbot_app = ChatbotAPP(llm,SERVICE_ACCOUNT_FILE,SCOPES,folder_id,unique_id,initContext) |
|
app = chatbot_app.create_app() |
|
app.launch(debug=True) |