diff --git a/.gitignore b/.gitignore index fb9df04..9702d2a 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ -/.venv \ No newline at end of file +/.venv +.env \ No newline at end of file diff --git a/__pycache__/bot.cpython-310.pyc b/__pycache__/bot.cpython-310.pyc new file mode 100644 index 0000000..27dcee8 Binary files /dev/null and b/__pycache__/bot.cpython-310.pyc differ diff --git a/__pycache__/bot_strings.cpython-310.pyc b/__pycache__/bot_strings.cpython-310.pyc new file mode 100644 index 0000000..b877f30 Binary files /dev/null and b/__pycache__/bot_strings.cpython-310.pyc differ diff --git a/bot.py b/bot.py index e69de29..fe13e57 100644 --- a/bot.py +++ b/bot.py @@ -0,0 +1,396 @@ +import re +import json +import os +import openai +from dotenv import load_dotenv +import requests +from bot_strings import botstrings + +# Call the load_dotenv function +load_dotenv() + +from pprint import pprint + + +class Report: + def __init__(self): + self.description = None + self.incident = None + self.name = None + self.role = None + self.location = None + self.image = None + self.size = None + self.shape = None + self.color = None + self.organization = None + self.type = None + self.marked: bool = None + self.blockage: bool = None + self.area_description = None + self.contact_person_phone_number = False + self.contact_person_name = False + + self.info_groups = [ + ["description", "color", "size", "shape"], + ["image"], + ["location", "area_description"], + ["incident"], + ["blockage", "marked"], + ["contact_person_name", "contact_person_phone_number"], + ["organization", "name", "role"], + ] + + self.descriptions = { + "description": "A general description of the object.", + "incident": "Has there been an incident related to the object?", + "name": "The name of the person reporting.", + "role": "The role or occupation of the person reporting.", + "location": "The location of the object.", # TODO Add more details + "image": "A picture of the object.", + "size": "The size of the object.", + "shape": "The shape of the object.", + "color": "The color of the object.", + "organization": "The organization of the person reporting.", + "type": "The type of the object.", + "marked": "Is the object marked?", + "blockage": "Is the object blocking anything?", + "area_description": "A description of the area where the object is located.", + "contact_person_phone_number": "The phone number of the contact person.", + "contact_person_name": "The name of the contact person.", + } + + self.looking_for: list = self.info_groups[0] + + +class BaseBot: + def __init__(self) -> None: + # Fetch the OpenAI key from the environment variables + self.OPEN_AI_KEY = os.getenv("OPEN_AI_KEY") + # # Set the OpenAI key + # openai.api_key = self.OPEN_AI_KEY + self.url = "https://api.openai.com/v1/chat/completions" + self.headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.OPEN_AI_KEY}", + } + + self.memory: list[dict] = None + + def generate(self, message): + + message = re.sub(" +", " ", message) + + if self.memory: + messages = self.memory + if messages[-1]["content"] != message: + messages.append({"role": "user", "content": message}) + else: + messages = [ + {"role": "system", "content": self.system_prompt}, + {"role": "user", "content": message}, + ] + + data = {"model": "gpt-3.5-turbo", "messages": messages} + # Print the last message in yellow + last_message = messages[-1]["content"] + print("\n\033[94m" + last_message + "\033[0m\n") + + response = requests.post(self.url, headers=self.headers, json=data).json() + answer = response["choices"][0]["message"] + + print("\033[95m" + answer["content"] + "\033[0m") + + return answer + + +class Chatbot(BaseBot): + def __init__(self) -> None: + system_prompt = botstrings.chatbot_system_prompt + super().__init__() # Inherit from BaseBot + # TODO Check if there is an older conversation with the same number. + # Set the system prompt and the first user message. + self.first_instructions_sent = False + self.informations_requested = False + self.memory = [{"role": "system", "content": system_prompt}] + + def ask_for_info(self, report: Report, chatbot: "Chatbot"): + # Ask for information + looking_for = [] + group_found = False + for group in report.info_groups: + if not group_found: + for info in group: + if getattr(report, info) is None: + group_found = True + looking_for.append(info) + + # Check so that the set of info is not already asked for without an answer + if looking_for == report.looking_for: + for info in looking_for: + setattr(report, info, "Unanswered") + looking_for = [] + group_found = False + + if looking_for != []: + report.looking_for = looking_for + # Ask for the information + general_bot.memory = [ + {"role": "user", "content": 'Formulate a question asking for the following information: ["name", "age"]'}, + {"role": "assistant", "content": "Can you tell me about the name and age of what you've found?"}, + ] + if looking_for == ["image"]: + question = 'image' + else: + prompt = f"Formulate a question asking for the following information: {looking_for}" + general_bot.memory.append({"role": "user", "content": prompt}) + question = general_bot.generate(prompt)["content"] + return question + + else: + return False, False + + +class CheckerBot(BaseBot): + def __init__(self) -> None: + self.system_prompt = botstrings.checker_bot_system_prompt + super().__init__() # Inherit from BaseBot + self.history = [] + + def check_message_type(self, message): + prompt = f"""What kind of message is this? + '''{message}''' + Don't care about the content, only the type of message. \ + Answer with any of the following: ["greeting", "question", "information", "statement"]\ + """ + + result = self.generate(prompt)["content"] + if "greeting" in result.lower(): + message_type = "greeting" + elif "question" in result.lower(): + message_type = "question" + elif "information" in result.lower() or "statement" in result.lower(): #TODO should "statement" be here? + message_type = "information" + + return message_type + + def check_answer(self, answer, question, chatbot: Chatbot): + return True # TODO We need to fins a good way to check if the answer is an answer to the question. + question = question.replace("\n", " ") + answer = answer.replace("\n", " ") + if question == botstrings.first_instructions: + question = "What have you found?" + + prompt = f''' + Have a look at this conversation:\n + """ + {chat2string(chatbot.memory[-4:]).strip()} + """\n + Is the last message a resonable answer to the last question ({question})? + Answer ONLY with any of the following: "True", "False", "Unclear" + '''.strip() + result = self.generate(prompt)["content"] + + if "unclear" in result.lower(): + prompt = f""" + A user is having a conversation with an assistant. This is the conversation:\n'''{chatbot.memory}'''\ + Is the last message ('''{answer}''') an answer to the question ('{question}')? + Answer ONLY with any of the following: "True", "False", "Unclear"\ + """ + self.memory.append({"role": "user", "content": prompt}) + + result = self.generate(prompt) + + if "true" in result.lower(): + answered = True + + elif "false" in result.lower() or "unclear" in result.lower(): + answered = False + + return answered + + def check_for_info(self, user_message, report: Report, looking_for: list, n_try=0): + info_dict = {} + for info in looking_for: + if getattr(report, info) is None: + info_dict[info] = report.descriptions[info] + prompt = f"""" + This is a message from a user: '''{user_message}'''\n + This is a dict describing what info I want:\n\n{info_dict} \n\n\ + Take a look at the message and create a dictionary with the information that is requested.\ + There might not be information available in the mesasge for all fields. \ + If you can't find information for a certain field, fill that with a python null value ("None" or "null"). + Do not include any explanations, only provide a RFC8259 compliant JSON response following this format without deviation.\ + """ + + n_try += 1 + result = self.generate(prompt) + + try: + json_content = json.loads( + result["content"] + ) # Parse the string back into a dictionary + for key, value in json_content.items(): + if value not in ["null", "None", "", "Unknown"]: + setattr(report, key, value) + except: + try: + result_content = result_content[ + result_content.find("{") : result_content.rfind("}") + 1 + ] + json_content = json.loads( + result["content"] + ) # Parse the string back into a dictionary + for key, value in json_content.items(): + if value not in ["null", "None", "", "Unknown"]: + setattr(report, key, value) + except: + if n_try > 3: + return False + self.check_for_info(user_message, report, looking_for, n_try=n_try) + + return json_content + + def check_for_tips(self, bot_message): + # Check if the message is a tip + prompt = f""" + An assistant wants to send this message to a user: '''{bot_message}'''\ + Is the message a tip or recommendation on how to handle a suspicious object? + Answer ONLY with any of the following: "True", "False"\ + """ + result = self.generate(prompt)["content"] + print(result) + if "true" in result.lower(): + is_tip = True + elif "false" in result.lower(): + is_tip = False + + return is_tip + + +class GeneralBot(BaseBot): + def __init__(self) -> None: + self.system_prompt = botstrings.general_bot_system_prompt + super().__init__() + self.memory = [] + + +def send(message, check=True): + + # Check if the message is a tip. + if check: + tip = checker_bot.check_for_tips(message) + else: + tip = False + + if tip: + return False + else: + print(message) # TODO Send the message to the user + return True + + +def chat2string(chat: list): + chat_string = "" + for message in chat: + chat_string += f"{message['role']}: {message['content']}\n" + return chat_string + + +if __name__ == "__main__": + + #! For testing + user_input = "I have found a strange looking thing on my lawn." + + # Initiate the chatbot. + chatbot = Chatbot( + first_message=user_input, + system_prompt="You are an assistant chatting with a user.", + ) # TODO ? Add phone number, etc. + checker_bot = CheckerBot() + general_bot = GeneralBot() + report = Report() + + while True: + # Check the message type + message_type = checker_bot.check_message_type(message=user_input) + print("Message type:", message_type) + + if "greeting" in message_type.lower(): + # Answer the greeting + bot_answer = chatbot.generate() + if not chatbot.first_instructions_sent: + # Give instructions for how to use the bot + chatbot.instructions_sent = True + chatbot.informations_requested = True + chatbot.memory.append( + {"role": "system", "content": botstrings.first_instructions} + ) + else: + chatbot.memory.append(bot_answer) + + elif "question" in message_type.lower(): + if not chatbot.first_instructions_sent: + # Give instructions for how to use the bot + chatbot.first_instructions_sent = True + chatbot.memory.append( + {"role": "system", "content": botstrings.first_instructions} + ) + + bot_answer = chatbot.generate() # TODO How to handle questions? + + # Check if the answer is an answer to the question. + answered = checker_bot.check_answer( + bot_answer["content"], user_input, chatbot + ) + if not answered: + bot_answer = "I could not understand your question. Please ask again." + send(bot_answer) + chatbot.memory.append({"role": "system", "content": bot_answer}) + + elif "information" in message_type.lower(): + if not chatbot.first_instructions_sent: + # Give instructions for how to use the bot + chatbot.first_instructions_sent = True + chatbot.informations_requested = True + chatbot.memory.append( + {"role": "system", "content": botstrings.first_instructions} + ) + send(botstrings.first_instructions, check=False) + else: + if chatbot.informations_requested: + answered = checker_bot.check_answer( + user_input, chatbot.memory[-1]["content"], chatbot + ) + if answered: + # Ask for information and send that message to the report + result = checker_bot.check_for_info( + user_input, report, report.looking_for + ) + if result: + pprint(result) + else: + send( + "I could not understand your message. Please try again.", + check=False, + ) + looking_for = chatbot.ask_for_info(report) + + else: + send( + "I could not understand your message. Please try again.", + check=False, + ) + + else: + print("Unknown message type") + + user_input = input(">>> ") + + if os.path.isfile(user_input): + + report.image = user_input + else: + print("User input is not a valid image file.") + +general_bot = GeneralBot() diff --git a/bot_strings.py b/bot_strings.py new file mode 100644 index 0000000..50db7ab --- /dev/null +++ b/bot_strings.py @@ -0,0 +1,12 @@ +class BotStrings(): + def __init__(self): + self.first_instructions = """ + Hi! I'm a chatbot from UNMAS (United Nations Mine Action Service). Use me to report a suspicious object that you think might me a landmine or an explosive remnant of war. + """.strip() + + self.chatbot_system_prompt = 'You are an assistant chatting with a user.' #TODO Add instructions for how to answer and what not to answer. + self.checker_bot_system_prompt='A user is chatting with an assistant. You are checking the messages.' + self.general_bot_system_prompt='You are a bot used for finding information.' + + +botstrings = BotStrings() \ No newline at end of file diff --git a/interface.py b/interface.py deleted file mode 100644 index e69de29..0000000 diff --git a/streamlit_interface.py b/streamlit_interface.py new file mode 100644 index 0000000..33d43a8 --- /dev/null +++ b/streamlit_interface.py @@ -0,0 +1,162 @@ +import streamlit as st +import numpy as np +from PIL import Image +from bot import Chatbot, Report, CheckerBot, GeneralBot, botstrings +from time import sleep +from pprint import pprint +def send(message, check=True, add_to_memory=True): + store_state() + # Check if the message is a tip. + if check: + tip = checker_bot.check_for_tips(message) + else: + tip = False + + if tip: + print("☠️ I'TS A TIP ☠️") + return False + else: + with st.chat_message("assistant"): + st.markdown(message) + # Add assistant response to chat history + st.session_state.messages.append({"role": "assistant", "content": message}) + if add_to_memory: + chatbot.memory.append({"role":"assistant", "content": message}) + return True + +def store_state(): + st.session_state.chatbot = chatbot + st.session_state.report = report + + +st.title("UNMAS Bot") + +# Initialize chat history +if "messages" not in st.session_state: + st.session_state.messages = [] + +if 'upload_image' not in st.session_state: + st.session_state.upload_image = False + +# Load chatbot from session state or create a new one +if "chatbot" not in st.session_state: + chatbot = Chatbot() + st.session_state.chatbot = chatbot +else: + chatbot = st.session_state.chatbot + +# Load report from session state or create a new one +if "report" not in st.session_state: + report = Report() + st.session_state.report = report +else: + report = st.session_state.report + +# Load checker and general bot from session state or create a new one +if "checker_bot" not in st.session_state: + checker_bot = CheckerBot() + st.session_state.checker_bot = checker_bot +else: + checker_bot = st.session_state.checker_bot + + +# Display chat messages from history on app rerun +for message in st.session_state.messages: + with st.chat_message(message["role"]): + st.markdown(message["content"]) + + +if st.session_state.upload_image: + user_input = None + img_file = st.file_uploader('Upload an image', type=['png', 'jpg', 'jpeg']) + + if img_file is not None: + image = Image.open(img_file) + img_array = np.array(image) + report.image = img_array + send('Thanks!', check=False, add_to_memory=False) + question = chatbot.ask_for_info(report, chatbot=chatbot) + send(question, check=False) + st.session_state.upload_image = False + store_state() + + + # user_input = st.chat_input('') + # if user_input: + # st.chat_message("user").markdown(user_input) + # chatbot.memory.append({"role":"user", "content": user_input}) + # st.session_state.messages.append({"role": "user", "content": user_input}) + +else: + user_input = st.chat_input('') + + +if user_input and not st.session_state.upload_image: + + st.chat_message("user").markdown(user_input) + # Add user message to chat history + st.session_state.messages.append({"role": "user", "content": user_input}) + chatbot.memory.append({"role":"user", "content": user_input}) + + # Check the message type + print('Check message type') + message_type = checker_bot.check_message_type(message=user_input) + print('Message type:', message_type) + + if not chatbot.first_instructions_sent: + # Give instructions for how to use the bot + chatbot.first_instructions_sent = True + chatbot.informations_requested = True + send(botstrings.first_instructions, check=False, add_to_memory=False) + sleep(1.2) + send('So first, tell me what you found?', check=False) + + else: + if 'greeting' in message_type.lower(): + # Answer the greeting + bot_answer = chatbot.generate(user_input) + send(bot_answer['content']) + + + elif 'question' in message_type.lower(): + + bot_answer = chatbot.generate(user_input) # TODO How to handle questions? + + # Check if the answer is an answer to the question. + answered = checker_bot.check_answer(bot_answer['content'], user_input, chatbot) + if not answered: + bot_answer = 'I could not understand your question. Please ask again.' + send(bot_answer) + chatbot.memory.append({"role":"assistant", "content": bot_answer}) + + elif 'information' in message_type.lower(): + + if chatbot.informations_requested: + answered = checker_bot.check_answer(user_input, chatbot.memory[-1]['content'], chatbot) + if answered: + # Ask for information and send that message to the report + result = checker_bot.check_for_info(user_input, report, report.looking_for) + if result: + print(result) + else: + send('I could not understand your message. Please try again.', check=False) + question = chatbot.ask_for_info(report, chatbot=chatbot) + + if question == 'image': + question = "Can you upload a picture of what you have found?" + st.session_state.upload_image = True + print(st.session_state.upload_image) + send(question, check=False, add_to_memory=False) + st.rerun() + else: + send(question, check=False) + + + else: + send('I could not understand your message. Please try again.', check=False) + + + else: + send('I could not understand your message. Please try again.', check=False) + + store_state()