Compare commits

...

5 Commits

Author SHA1 Message Date
Lasse Studion 657e08648e Day 2 2 years ago
Lasse Studion 4b9284f100 Fix bug in login functionality*** 2 years ago
Lasse Server 1d8ebbb25b Fix memory initialization and add image conversion to Base64 2 years ago
Lasse Server 49df3b653e Fix GeneralBot memory error 2 years ago
Lasse Studion 463d0c57e8 Add chatbot documentation 2 years ago
  1. 111
      bot.py
  2. 2
      bot_strings.py
  3. 1
      notes.md
  4. 2
      report_template.html
  5. 75
      streamlit_interface.py
  6. 0
      whatsapp.py

111
bot.py

@ -26,39 +26,56 @@ class Report:
self.color = None
self.organization = None
self.type = None
self.marked: bool = None
self.blockage: bool = None
self.marked = None
self.blockage = None
self.area_description = None
self.contact_person_phone_number = False
self.contact_person_name = False
self.contact_person_phone_number = None
self.contact_person_name = None
self.at_the_location: bool = None
self.coordinates = None
self.woreda = None
self.info_groups = [
["description", "color", "size", "shape"],
["description"],
["color", "size", "shape"],
["at_the_location"],
#['coordinates'],
["image"],
["location", "area_description"],
["woreda"],
["location"],
["area_description"],
["incident"],
["blockage", "marked"],
["contact_person_name", "contact_person_phone_number"],
["organization", "name", "role"],
]
self.info_explicit = [
"name",
"contact_person_name",
"contact_person_phone_number",
]
self.info_professional = ["organization", "role", "type"]
self.descriptions = {
"description": "A general description of the object.",
"at_the_location": "If the person reporting is still where the object was found. (True or False)",
"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
"woreda": "The woreda where the object was found.",
"location": "The closest city or village to where the object was found.", # TODO Add more details
"coordinates": "The coordinates of the person reporting.",
"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.",
"type": "The type of the object.",
}
self.looking_for: list = self.info_groups[0]
@ -93,11 +110,13 @@ class BaseBot:
]
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()
pprint(response)
answer = response["choices"][0]["message"]
print("\033[95m" + answer["content"] + "\033[0m")
@ -122,6 +141,7 @@ class Chatbot(BaseBot):
for group in report.info_groups:
if not group_found:
for info in group:
print("#", info, getattr(report, info))
if getattr(report, info) is None:
group_found = True
looking_for.append(info)
@ -139,20 +159,25 @@ class Chatbot(BaseBot):
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"]
general_bot.memory = None
description_dict = {}
for i in looking_for:
description_dict[i] = report.descriptions[i]
# Give the bot examles of how to ask for the information
general_bot.memory = [
{
"role": "user",
"content": 'Formulate a question asking for the following information: ["name", "age"]',
"content": """I'm looking for the information in this disctionary:
{'description': 'If the person reporting is still where the object was found?'}
Formulate a question asking for the information in the dictionary.""",
},
{
"role": "assistant",
"content": "Can you tell me about the name and age of what you've found?",
"content": f"Are you still at the place where object was found?",
},
]
# If we know what the object is, we can ask about it.
if report.object is not None:
general_bot.memory.insert(
0,
@ -162,12 +187,19 @@ class Chatbot(BaseBot):
+ f" The object found might be {report.object}.",
},
)
general_bot.memory[2]["content"] = f"Can you tell me about the name and age of the {report.object}?"
general_bot.memory[2]["content"] = f"Are you still at the place where the {report.object} was found?"
prompt = f"Formulate a question asking for the following information: {looking_for}"
general_bot.memory.append({"role": "user", "content": prompt})
object_string = ''
if report.object is not None:
object_string = f" The object found is {report.object}."
prompt = f"""I'm looking for the information about an object. {object_string}. The information needed is described in this disctionary:
{description_dict}
Formulate a question asking for the information in the dictionary."""
question = general_bot.generate(prompt)["content"]
general_bot.memory = None
general_bot.memory = []
return question
else:
@ -234,7 +266,7 @@ class CheckerBot(BaseBot):
return answered
def check_for_info(self, user_message, report: Report, looking_for: list, n_try=0):
def check_for_info(self, user_message, report: Report, question: str, looking_for: list, n_try=0) -> dict:
if report.object is None:
@ -264,7 +296,7 @@ class CheckerBot(BaseBot):
]
self.memory = messages
result = self.generate(user_message)["content"]
self.memory = None
self.memory = []
if result not in ["null", "None", "", "Unknown"]:
print("Object:", result)
@ -274,10 +306,17 @@ class CheckerBot(BaseBot):
for info in looking_for:
if getattr(report, info) is None:
info_dict[info] = report.descriptions[info]
# If the bot asks for a description, it should also check for color, size and shape.
if looking_for == ["description"]:
for i in ["color", "size", "shape"]:
info_dict[i] = report.descriptions[i]
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.\
A user was asked '''{question}'''
This is the answer from the user: '''{user_message}'''\n
This is a dictionary describing what info I want:\n\n{info_dict} \n\n\
Take a look at the message along with the question 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.\
@ -291,23 +330,32 @@ class CheckerBot(BaseBot):
result["content"]
) # Parse the string back into a dictionary
for key, value in json_content.items():
if value not in ["null", "None", "", "Unknown"]:
if value not in ["null", "None", "", "Unknown", None]:
setattr(report, key, value)
else: # Don't ask for these again.
if (
key in ["shape", "size", "color", "marked", "blockage"]
and key in looking_for
):
setattr(report, key, "Unknown")
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)
self.check_for_info(user_message, report, looking_for = looking_for, question=question, n_try=n_try)
return json_content
@ -435,13 +483,20 @@ if __name__ == "__main__":
send(botstrings.first_instructions, check=False)
else:
if chatbot.informations_requested:
question = ''
for i in chatbot.memory:
if i["role"] == "assistant":
question = i["content"]
print('QUESTION:', question)
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
user_input, report, looking_for=report.looking_for, question=question
)
if result:
pprint(result)

@ -6,7 +6,7 @@ class BotStrings():
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. Keep to the information provided and never make any assumptions.'
self.general_bot_system_prompt='You are a bot used for findin information about things a user have found. You are not to make any assumptions, only provide information based on the information given.'
self.general_bot_system_prompt='You are chatting with a person that has found something. You are not to make any assumptions, only provide information based on the information given. When you ask questions, make sure to direct them to the person you are chatting with.'
self.first_question = 'So first, tell me what you found?'
botstrings = BotStrings()

@ -0,0 +1 @@
WhatsAPP server: https://dev.to/koladev/building-a-web-service-whatsapp-cloud-api-flask-sending-template-messages-part-1-249g

@ -13,7 +13,7 @@
<p><strong>Incident:</strong> {{ report.incident }}</p>
<p><strong>Role:</strong> {{ report.role }}</p>
<p><strong>Location:</strong> {{ report.location }}</p>
<p><strong>Image:</strong> {{ report.image }}</p>
<p><strong>Image:</strong> <img src="data:image/jpeg;base64,{{ report.image }}" alt="Report Image"></p>
<p><strong>Size:</strong> {{ report.size }}</p>
<p><strong>Shape:</strong> {{ report.shape }}</p>
<p><strong>Color:</strong> {{ report.color }}</p>

@ -7,6 +7,17 @@ from time import sleep
def send(message, check=True, add_to_memory=True):
"""
Sends a message to the chat interface.
Parameters:
- message (str): The message to be sent.
- check (bool): Whether to check the message for tips. Default is True.
- add_to_memory (bool): Whether to add the message to the chatbot's memory. Default is True.
Returns:
- bool: True if the message was sent successfully, False if it was a tip.
"""
store_state()
# Check if the message is a tip.
if check:
@ -27,11 +38,17 @@ def send(message, check=True, add_to_memory=True):
return True
def store_state():
"""
Stores the current state of the chatbot and report in the session state.
"""
st.session_state.chatbot = chatbot
st.session_state.report = report
def generate_report():
"""
Generates a report using a Jinja2 template and stores it in the session state.
"""
from jinja2 import Environment, FileSystemLoader
# Load the Jinja2 template
@ -42,95 +59,117 @@ def generate_report():
st.session_state.report_html = report_html
st.rerun()
st.title("UNMAS Bot")
# Check if 'report_html', 'messages', 'upload_image',are already in the session state, if not, initialize them
if 'report_html' not in st.session_state:
st.session_state.report_html = ''
# 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 report from session state or create a new one
# Check if 'report' and the bots are already in the session state, if not, initialize them
if "report" not in st.session_state:
report = Report()
st.session_state.report = report
else:
report = st.session_state.report
# 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:
# If 'chatbot' is already in the session state, retrieve it
chatbot = st.session_state.chatbot
# If the report object is not None, update the chatbot's memory with a new system prompt
if report.object is not None:
new_system_prompt = botstrings.checker_bot_system_prompt + f' The user seems to have found {report.object}.'
chatbot.memory[0] = {"role":"system", "content": new_system_prompt}
print(chatbot.memory)
# 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"])
# Write the report to screen
# If 'report_html' in the session state is not an empty string, write the report to the screen and reset 'report_html' to an empty string
if st.session_state.report_html != '':
st.markdown(st.session_state.report_html, unsafe_allow_html=True)
st.session_state.report_html = ''
# Check if 'upload_image' is set to True in the session state
if st.session_state.upload_image:
import base64
from io import BytesIO
# Initialize user_input to None
user_input = None
# Create a file uploader widget that accepts png, jpg, and jpeg files
img_file = st.file_uploader('Upload an image', type=['png', 'jpg', 'jpeg'])
# Create a chat input widget and store the user's input in user_input
user_input = st.chat_input('')
# If an image file was uploaded
if img_file is not None:
# Open the image file and convert it to a numpy array and store the image array in the report object
image = Image.open(img_file)
img_array = np.array(image)
report.image = img_array
# Convert the image to Base64
buffered = BytesIO()
# Convert image to RGB if it's RGBA
if image.mode == 'RGBA':
image = image.convert('RGB')
image.save(buffered, format="JPEG")
img_str = base64.b64encode(buffered.getvalue()).decode()
report.image = img_str
send('Thanks!', check=False, add_to_memory=False)
# Ask the chatbot for more information and store the question
question = chatbot.ask_for_info(report, chatbot=chatbot)
send(question, check=False)
# Set 'upload_image' to False in the session state
st.session_state.upload_image = False
store_state()
st.rerun()
# If the user entered something in the chat input widget
elif user_input:
# Check the user's answer to the image question
answer = checker_bot.check_image_answer(user_input, report)
if answer == 'no' or answer == 'help':
# If the user answered 'no'
if answer == 'no':
# Send a message to the user and set 'upload_image' to False in the session state
send('No problem, we will continue without the image.', check=False, add_to_memory=False)
st.session_state.upload_image = False
elif answer == 'help':
# Send a help message to the user and set 'upload_image' to False in the session state
send('We will provide help for sending a picture in WhatsAPP', check=False) #TODO
st.session_state.upload_image = False
# Ask the chatbot for more information and store the question
question = chatbot.ask_for_info(report, chatbot=chatbot)
send(question, check=False)
# If the user answered 'yes'
elif answer == 'yes':
send('Great! Please upload the image.', check=False, add_to_memory=False)
st.session_state.upload_image = True
store_state()
st.rerun()
else:
user_input = st.chat_input('')
@ -181,9 +220,15 @@ if user_input and not st.session_state.upload_image:
if chatbot.informations_requested:
answered = checker_bot.check_answer(user_input, chatbot.memory[-1]['content'], chatbot)
question = ''
# Make the last bot message the question
for i in chatbot.memory:
if i['role'] == 'assistant':
question = i['content']
if answered:
# Ask for information and send that message to the report
result = checker_bot.check_for_info(user_input, report, report.looking_for)
result = checker_bot.check_for_info(user_input, report, looking_for=report.looking_for, question=question)
if result:
print(result)
else:

Loading…
Cancel
Save