Proof of concept

.gitignore

@@ -1 +1,2 @@
-/.venv
+/.venv
+.env

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()

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()

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()