first E2E working implementation

test.json

@@ -1,61 +0,0 @@
-[
- {
- "$schema": "https://json-schema.org/draft/2020-12/schema",
- "$id": "https://www.enricorampazzo.tech/dam_helper.schema.json",
- "title": "Answer",
- "description": "an answer to a question",
- "type": "array",
- "items": {
- "type": "object",
- "properties": {
- "question": {
- "type": "string"
- },
- "answer": {
- "type": ["string", "null"]
- }
- },
- "required": ["question", "answer"]
- }
- },
- {
- "question": "1) What do you need to do?",
- "answer": "Pest control"
- },
- {
- "question": "2) In which community is the work taking place?",
- "answer": "JBR"
- },
- {
- "question": "3) In which building?",
- "answer": "Sadaf 5"
- },
- {
- "question": "4) In which unit/apartment number?",
- "answer": "2408"
- },
- {
- "question": "5) Am I an owner or a tenant?",
- "answer": "unclear"
- },
- {
- "question": "6) In which date is the work taking place?",
- "answer": "7/9/2024"
- },
- {
- "question": "7) In which date will the work finish?",
- "answer": null
- },
- {
- "question": "8) What is my contact number?",
- "answer": null
- },
- {
- "question": "9) What is the name of the contracting company?",
- "answer": "Breathe Maintenance"
- },
- {
- "question": "10) What is the contact number of the contracting company?",
- "answer": null
- }
-]

.gitignore

@@ -1,4 +1,3 @@
 .idea
-form.pdf
 /.idea/
 /.idea/inspectionProfiles/

app.py

@@ -1,78 +1,45 @@
-import json
-import re
-from json import JSONDecodeError
+from pathlib import Path
 
+from prompts.prompts_manager import PromptsManager
 from repository.ollama import OllamaRepository
-from schema import ModelRoles
+from llm.llm import ModelRoles, Model
+from form.form import work_categories, build_form_data_from_answers, write_pdf_form
 
-# this regex is most definitely *not* going to make my system blow up unexpectedly at some point in the future
-get_questions_and_answers_regex = re.compile(r'{\s*"question":\s*"([^"]+)",\s*"answer":\s*(\s*null\s*|"[^"]+"\s*)}')
 
+def check_for_missing_answers(parsed_questions: dict[int, str]):
+ return [k for k in parsed_questions if parsed_questions[k] is None]
 
-def print_current_status(total_success, first_try_successes, second_try_successes, regex_successes):
- print(f"""so far, questions and answers were parsed successfully {total_success} times, of which 
- {first_try_successes} were successful at the first attempt, 
- {second_try_successes} were successful after asking the model to self-correct with a 'JSON expert' agent 
- and {regex_successes} were successful using a regex
- """)
+
+def ask_again(missing_questions: list[int], user_questions: list[str], parsed_questions: dict[int, str]):
+ for id_ in missing_questions:
+ answer = input(f"I could not find the answer to this question: {user_questions[id_].lower()}")
+ parsed_questions[id_] = answer
 
 
 if __name__ == '__main__':
- with open("questions.txt") as questions_file:
- questions = questions_file.read()
- with open("system_prompt.txt") as system_prompt_file:
- system_prompt = system_prompt_file.read()
- with open("verification_prompt.txt") as verification_prompt_file:
- verification_prompt = verification_prompt_file.read()
- verification_prompt = verification_prompt.replace("{questions}", questions)
+ prompts_manager = PromptsManager()
  user_prompt = input(f"Please describe what you need to do. To get the best results "
- f"try to answer the following questions:\n{questions}\n\n>")
-
- ollama_repository = OllamaRepository("llama3.1", system_prompt,
- ModelRoles("system", "user", "assistant"))
- successful = 0
- corrected_json_was_ok = 0
- regex_got_all_answers = 0
- successful_at_first_attempt = 0
-
- while True:
- ollama_repository.send_prompt(f"Ingest the following information: {user_prompt}")
- answer = ollama_repository.send_prompt(verification_prompt)
- ollama_repository.system_msg = "You are an expert at JSON format, and can detect and fix errors in JSON strings"
- fixed_json = ollama_repository.send_prompt(f"Verify if this is legal JSON and correct any mistake, output just the fixed json without adding anything else {answer['content']}")
- answers = fixed_json["content"]
- try:
- json_answers = json.loads(answers, strict=False)
- print(answers)
- successful_at_first_attempt +=1
- successful+=1
- print_current_status(successful, successful_at_first_attempt, corrected_json_was_ok, regex_got_all_answers)
- except JSONDecodeError as e:
- print("there was a problem in this json, asking the LLM to fix it passing the exception error message")
- answer = (
- ollama_repository.send_prompt(f"When parsing this json {answers} I got this error {str(e)}. "
- f"Fix this error and return just the corrected json without adding anything else"))
-
- print("trying to parse the corrected json")
- try:
- json_answers = json.loads(answer["content"])
- corrected_json_was_ok+=1
- print_current_status(successful, successful_at_first_attempt, corrected_json_was_ok,
- regex_got_all_answers)
- except JSONDecodeError as e:
- print("still error, going old school")
- regex_parse_result = {}
- for match in get_questions_and_answers_regex.findall(answer["content"]):
- question, answer = match
- regex_parse_result[question] = answer
- if len(regex_parse_result) == 10:
- print("I got all 10 answers apparently")
- successful+=1
- regex_got_all_answers+=1
- print_current_status(successful, successful_at_first_attempt, corrected_json_was_ok,
- regex_got_all_answers)
- else:
- print(f"unable to parse \n {answers}\n giving up")
- break
-
-
+ f"try to answer all the following questions:\n{'\n'.join(prompts_manager.questions)}\n\n>")
+
+ ollama_repository = OllamaRepository(Model("llama3.1",
+ ModelRoles("system", "user", "assistant")),
+ prompts_manager.system_prompt,
+ )
+ ollama_repository.send_prompt(f"Ingest the following information: {user_prompt}")
+ answers = {}
+ for idx, q in enumerate(prompts_manager.verification_prompt):
+ answer = ollama_repository.send_prompt(
+ f"Answer the following question, if the answer is not present just answer null. Keep the answer brief: {q}")
+ answers[idx] = None if 'null' in answer["content"].lower() else answer['content']
+ missing_answers = check_for_missing_answers(answers)
+ while missing_answers:
+ ask_again(missing_answers, prompts_manager.questions, answers)
+ missing_answers = check_for_missing_answers(answers)
+ answer = ollama_repository.send_prompt(
+ f"The work to do is {answers[1]}. Given the following categories {work_categories.values()} which ones are the most relevant? Only return one categories, separated by a semicolon")
+ categories = []
+ for category in answer["content"].split(";"):
+ categories.extend([k for k, v in work_categories.items() if category in v])
+
+ form_data = build_form_data_from_answers(answers, categories, f"{Path(__file__, "..", "signature.png")}")
+ write_pdf_form(form_data, Path("signed_form1.pdf"))

schema.py → form/form.py

@@ -1,5 +1,40 @@
+from pathlib import Path
 from typing import TypedDict
+from PyPDFForm import PdfWrapper
 
+from utils.date_utils import get_today_date_as_dd_mm_yyyy
+from utils.parsing_utils import find_and_parse_date, find_and_parse_phone_number
+
+
+def build_form_data_from_answers(answers: dict[int, str], categories: list[str], signature: str | None = None):
+ form_data = {}
+ for category in categories:
+ form_data[form_fields_map[category]] = True
+ form_data[form_fields_map["minor_work"]] = True
+ form_data[form_fields_map["signed_by"]] = answers[0]
+ form_data[form_fields_map["community"]] = answers[2]
+ form_data[form_fields_map["building_name"]] = answers[3]
+ form_data[form_fields_map["unit_no"]] = answers[4]
+ form_data[form_fields_map["owner" if "owner" in answers[5].lower() else "tenant"]] = True
+ form_data[form_fields_map["start_date"]] = find_and_parse_date(answers[6])
+ form_data[form_fields_map["end_date"]] = find_and_parse_date(answers[7])
+ form_data[form_fields_map["contact_number"]] = find_and_parse_phone_number(answers[8])
+ form_data[form_fields_map["contractor_name"]] = answers[9]
+ form_data[form_fields_map["contractor_contact_number"]] = answers[10]
+ form_data[form_fields_map["contractor_email"]] = answers[11]
+ form_data[form_fields_map["email"]] = answers[12]
+ form_data[form_fields_map["occupant_name"]] = answers[0]
+ form_data[form_fields_map["signature_date"]] = get_today_date_as_dd_mm_yyyy()
+ if signature:
+ form_data[form_fields_map["signature"]] = signature
+ return form_data
+
+def write_pdf_form(form_data: dict[str, str | bool | int], output_path:Path):
+ base_form_path = Path(__file__, "..", "form.pdf")
+ with open(base_form_path, "rb") as blank_form:
+ filled_form = PdfWrapper(blank_form).fill(form_data)
+ with open(output_path, "wb+") as filled_form_file:
+ filled_form_file.write(filled_form.read())
 
 class FormFields(TypedDict, total=True):
  start_date: str
@@ -44,55 +79,6 @@ class FormFields(TypedDict, total=True):
  minor_work: str
 
 
-class MinorWorkForm:
- def __init__(self):
- self.start_date: str
- self.end_date: str
- self.signed_by: str
- self.signature_date: str
- self.community: str
- self.email: str
- self.unit_no: str
- self.contact_number: str
- self.building_name: str
- self.contractor_name: str
- self.contractor_email: str
- self.contractor_contact_number: str
- self.occupant_name: str
- self.civil: bool
- self.electrical_AC: bool
- self.furniture_delivery: bool
- self.handyman: bool
- self.pest_control: bool
- self.sail_shade: bool
- self.wardrobes: bool
- self.wooden_flooring: bool
- self.plumbing_sanitary: bool
- self.safety_systems: bool
- self.soft_landscaping: bool
- self.balcony_tiles: bool
- self.bathroom_toilet_refurbish: bool
- self.staircase_railing: bool
- self.storage_box: bool
- self.tiling_works: bool
- self.doors_door_frames: bool
- self.gypsum_ceiling: bool
- self.kitchen_refurbish: bool
- self.lights_and_sockets: bool
- self.painting_wallpapering: bool
- self.seating_area_barbecuing_fountain: bool
- self.other_work_checkbox: bool
- self.owner: bool
- self.tenant: bool
- self.signature: str
-
-class ModelRoles:
- def __init__(self, system_role, user_role, ai_role):
- self.system_role = system_role
- self.user_role = user_role
- self.ai_role = ai_role
-
-
 form_fields_map: FormFields = {"start_date": "From", "end_date": "to", "signed_by": "name-10",
  "signature_date": "From--1", "community": "Name-2", "email": "Name-3",
  "unit_no": "Name-4", "contact_number": "Name-5", "building_name": "Name-6",
@@ -101,11 +87,28 @@ form_fields_map: FormFields = {"start_date": "From", "end_date": "to", "signed_b
  "other_work": "name-19", "civil": "3", "electrical_AC": "10", "furniture_delivery": "14",
  "handyman": "18", "pest_control": "21", "sail_shade": "12", "wardrobes": "16",
  "wooden_flooring": "24",
- "plumbing_sanitary": "4", "safety systems": "4", "soft_landscaping": "9",
+ "plumbing_sanitary": "4", "safety_systems": "4", "soft_landscaping": "9",
  "balcony_tiles": "13", "bathroom_toilet_refurbish": "20",
  "staircase_railing": "17", "storage_box": "11", "tiling_works": "15",
  "doors_door_frames": "6", "gypsum_ceiling": "22", "kitchen_refurbish": "7",
  "lights_and_sockets": "23", "painting_wallpapering": "19",
  "seating_area_barbecuing_fountain": "8",
  "other_work_checkbox": "69", "owner": "Check Box4", "tenant": "Check Box3",
- "signature": "Signature2", "minor_work": "1"}
+ "signature": "Signature2", "minor_work": "2"}
+work_categories = {"other_work": "other work", "civil": "Civil work, masonry",
+ "electrical_AC": "electrical / AC / air conditioning", "furniture_delivery": "furniture delivery",
+ "handyman": "handyman work, repairs, furniture assembly, furniture installation", "pest_control": "pest control",
+ "sail_shade": "sail shade, curtains", "wardrobes": "wardrobes",
+ "wooden_flooring": "wooden flooring",
+ "plumbing_sanitary": "plumbing / sanitary / bathroom works such as repairing a sink, tap, toilet, shower or bathtub",
+ "safety_systems": "safety systems, surveillance systems installation",
+ "soft_landscaping": "soft landscaping", "balcony_tiles": "installation and fixing of balcony tiles",
+ "bathroom_toilet_refurbish": "bathroom or toilet refurbishment or renovation",
+ "staircase_railing": "staircase rails, handrails", "storage_box": "storage box",
+ "tiling_works": "tiles installation or repair",
+ "doors_door_frames": "work related to doors and door frames",
+ "gypsum_ceiling": "gypsum ceiling installation or maintenance",
+ "kitchen_refurbish": "kitchen renovation or refurbishment",
+ "lights_and_sockets": "installation of lights, ceiling lights, power sockets",
+ "painting_wallpapering": "painting, wallpaper installation",
+ "seating_area_barbecuing_fountain": "installation or renovation of outdoor structures such as seating area, barbecuing area or fountains"}

llm/llm.py

@@ -0,0 +1,11 @@
+class ModelRoles:
+ def __init__(self, system_role: str, user_role: str, ai_role: str):
+ self.system_role: str = system_role
+ self.user_role: str = user_role
+ self.ai_role: str = ai_role
+
+
+class Model:
+ def __init__(self, name: str, roles: ModelRoles):
+ self.name: str = name
+ self.roles: ModelRoles = roles

prompts/prompts_manager.py

@@ -0,0 +1,19 @@
+import datetime
+from pathlib import Path
+
+from utils.date_utils import get_today_date_as_dd_mm_yyyy
+
+
+class PromptsManager:
+ def __init__(self):
+ base_path = Path(__file__, "..")
+ with open(Path(base_path, "system_prompt.txt")) as sysprompt_file:
+ self.system_prompt: str = sysprompt_file.read()
+ with open(Path(base_path, "questions.txt")) as questions_file:
+ self.questions: list[str] = questions_file.readlines()
+ with open(Path(base_path, "verification_prompt2.txt")) as verification_prompt_file:
+ verification_prompt = verification_prompt_file.readlines()
+ todays_date = get_today_date_as_dd_mm_yyyy()
+ for line in verification_prompt:
+ line.replace("{today}", todays_date)
+ self.verification_prompt: list[str] = verification_prompt

prompts/questions.txt

@@ -0,0 +1,13 @@
+What is your full name?
+What is the nature of the work you need to do?
+In which community is the work taking place?
+In which building?
+In which unit/apartment number?
+Are you an owner or a tenant?
+In which date is the work taking place?
+In which date will the work finish?
+What is your contact number?
+What is the name of the contracting company?
+What is the contact number of the contracting company?
+What is the email of the contracting company?
+What is your email?

prompts/verification_prompt2.txt

@@ -0,0 +1,13 @@
+What is my full name?
+What is the nature of the work I need to do?
+In which community is the work taking place?
+In which building?
+In which unit/apartment number?
+Am I the owner or the tenant?
+In which date is the work taking place? Please answer with just a date formatted as dd/mm/yyyy. In case I used expressions like today, tomorrow, in two days, ecc, know that today it is 08/09/2024
+In which date will the work finish? Please answer with just a date formatted as dd/mm/yyyy. In case I used expressions like today, tomorrow, in two days, ecc, know that today it is {today}. If no date is provided, consider that it will finish on the same day as the start date
+What is my contact number?
+What is the name of the contracting company?
+What is the contact number of the contracting company?
+What is the email of the contracting company?
+What is your email?

questions.txt

@@ -1,10 +0,0 @@
-1) What do you need to do?
-2) In which community is the work taking place?
-3) In which building?
-4) In which unit/apartment number?
-5) Am I an owner or a tenant?
-6) In which date is the work taking place?
-7) In which date will the work finish?
-8) What is my contact number?
-9) What is the name of the contracting company?
-10) What is the contact number of the contracting company?

repository/ollama.py

@@ -1,21 +1,22 @@
 import ollama
 from ollama import Options
 
-from schema import ModelRoles
+from llm.llm import ModelRoles, Model
 
 
 class OllamaRepository:
- def __init__(self, model, system_msg, roles: ModelRoles):
- self.model = model
- self.system_msg = system_msg
- self.roles = roles
- self.message_history: list[dict[str, str]] = [{"role": self.roles.system_role, "content": system_msg}]
+ def __init__(self, model:Model, system_msg):
+ self.model: Model = model
+ self.system_msg: str = system_msg
+ self.message_history: list[dict[str, str]] = [{"role": self.model.roles.system_role, "content": system_msg}]
 
- def send_prompt(self, prompt:str, add_to_history:bool = False) -> dict[str, str]:
- options: Options = Options(temperature=0.1)
- self.message_history.append({"role": self.roles.user_role, "content":prompt})
- response = ollama.chat(self.model, self.message_history, options=options)
- answer = {"role": self.roles.ai_role, "content": response["message"]["content"]}
+ def send_prompt(self, prompt:str, add_to_history:bool = True) -> dict[str, str]:
+ options: Options = Options(temperature=0)
+ self.message_history.append({"role": self.model.roles.user_role, "content":prompt})
+ response = ollama.chat(self.model.name, self.message_history, options=options)
+ answer = {"role": self.model.roles.ai_role, "content": response["message"]["content"]}
  if add_to_history:
  self.message_history.append(answer)
+ else:
+ self.message_history.pop()
  return answer

utils/date_utils.py

@@ -0,0 +1,3 @@
+import datetime
+def get_today_date_as_dd_mm_yyyy():
+ return datetime.datetime.now().strftime("%d/%m/%Y")

utils/parsing_utils.py

@@ -0,0 +1,17 @@
+import re
+parse_date_regex = re.compile(r"([0-9]{1,2}/[0-9]{1,2}/[0-9]{4})")
+parse_phone_number_regex = re.compile(r"\+?[0-9]*")
+
+
+def _find_and_parse(llm_answer, regex):
+ match = regex.findall(llm_answer)
+ return next(iter(match), None) if match else None
+
+
+def find_and_parse_date(llm_answer: str) -> str | None:
+ return _find_and_parse(llm_answer, parse_date_regex)
+
+
+def find_and_parse_phone_number(llm_answer: str):
+ return _find_and_parse(llm_answer, parse_phone_number_regex)
+

verification_prompt.txt

@@ -1,27 +0,0 @@
-Please tell if I answered the following questions:
-
-{questions}
-
-The answer should be a list of json objects that follow this schema and be legal json.
-{
- "$schema": "https://json-schema.org/draft/2020-12/schema",
- "$id": "https://www.enricorampazzo.tech/dam_helper.schema.json",
- "title": "Answer",
- "description": "an answer to a question",
- "type": "object",
- "properties": {
- "question": {
- "description": "the question being answered",
- "type": "string"
- },
- "answer": {
- "description": "the answer to the question",
- "type": "string"
-
- }
- }
-}
-if the answer is a date format it as day/month/year.
-If the answer contains temporal references such as 'tomorrow', 'in two days' etc. consider that today it is the
-6th of september 2024.
-If the answer is not provided just the answer string should be null