code refactoring, added logic to allow users to save their details in the browser local storage

app.py

@@ -1,41 +1,181 @@
-from prompts.prompts_manager import PromptsManager
-
-from repository.repository import get_repository
-from repository.repository_abc import ModelRoles, Model
+import json
 
 import streamlit as st
+from streamlit import session_state as ss
+from streamlit_local_storage import LocalStorage
 
-from ui_manager import build_ui_for_initial_state, build_ui_for_parsing_answers, build_ui_for_ask_again, \
-    build_ui_for_check_category, build_ui_for_form_created
-from utils.env_utils import build_repo_from_environment
+from form.form import build_form_data_from_answers, write_pdf_form
+from llm_manager.llm_parser import LlmParser
+from local_storage.entities import PersonalDetails, LocationDetails, ContractorDetails
+from prompts.prompts_manager import PromptsManager, Questions as Q
+from repository.repository import build_repo_from_environment, get_repository
+from repository.repository_abc import Model, ModelRoles
+from utils.parsing_utils import check_for_missing_answers
 
 user_msg = "Please describe what you need to do. To get the best results try to answer all the following questions:"
+ls: LocalStorage = LocalStorage()
+class UIManager:
+    def __init__(self):
+        self.pm: PromptsManager = PromptsManager()
+        self.repository = (build_repo_from_environment(self.pm.system_prompt) or
+                           get_repository("testing",
+                                          Model("fakeModel", ModelRoles("a", "b", "c"))))
+
+    @staticmethod
+    def get_current_step():
+        return ss.get("step")
+
+    @staticmethod
+    def _build_base_ui():
+        st.markdown("## Dubai Asset Management red tape cutter")
+
+    def build_ui_for_initial_state(self, user_message):
+        help_ = user_message
+        self._build_base_ui()
+        with st.form("Please describe your request"):
+            user_input = st.text_area("Your input", height=700, label_visibility="hidden", placeholder=help_,
+                                      help=help_)
+            signature = st.file_uploader("Your signature", key="file_upload")
+            ss["signature"] = signature
+            submit_button = st.form_submit_button()
+            if submit_button:
+                ss["user_input"] = user_input
+                ss["step"] = "parsing_answers"
+                st.rerun()
+
+    def build_ui_for_parsing_answers(self):
+        self._build_base_ui()
+        with st.status("initialising LLM"):
+            self.repository.init()
+        with st.status("waiting for LLM"):
+            answer = self.repository.send_prompt(self.pm.verify_user_input_prompt(ss["user_input"]))
+            st.write(f"answers from LLM: {answer['content']}")
+        with st.status("Checking for missing answers"):
+            answers = LlmParser.parse_verification_prompt_answers(answer['content'])
+            ss["answers"] = answers
+            if len(answers) != len(Q):
+                ss["step"] = "parsing_error"
+                st.rerun()
+        ss["missing_answers"] = check_for_missing_answers(ss["answers"])
+        if not ss.get("missing_answers"):
+            ss["step"] = "check_category"
+        else:
+            ss["step"] = "ask_again"
+        st.rerun()
+
+    def build_ui_for_ask_again(self):
+        self._build_base_ui()
+        with st.form("form1"):
+            for ma in ss["missing_answers"]:
+                st.text_input(self.pm.questions[ma].lower(), key=ma)
+            submitted = st.form_submit_button("Submit answers")
+            if submitted:
+                for ma in ss["missing_answers"]:
+                    ss["answers"][ma] = ss[ma]
+                ss["step"] = "check_category"
+                st.rerun()
+
+    def build_ui_for_check_category(self):
+        self._build_base_ui()
+        with st.status("finding the work categories applicable to your work"):
+            answer = self.repository.send_prompt(self.pm.get_work_category(ss["answers"][1]))
+            categories = LlmParser.parse_get_categories_answer(answer['content'])
+
+        with st.status("categories found, creating PDF form"):
+            form_data, filename = build_form_data_from_answers(ss["answers"], categories,
+                                                               ss.get("signature"))
+            pdf_form = write_pdf_form(form_data)
+            pdf_form_filename = filename
+            ss["pdf_form"] = pdf_form
+            ss["pdf_form_filename"] = pdf_form_filename
+            ss["step"] = "form_created"
+            st.rerun()
+
+    def build_ui_for_form_created(self):
+        self._build_base_ui()
+        st.download_button("download form", ss["pdf_form"],
+                           file_name=ss["pdf_form_filename"], mime="application/pdf")
+        start_over_button = st.button("Start over")
+        if start_over_button:
+            del ss["step"]
+            del ss["pdf_form"]
+            del ss["pdf_form_filename"]
+            if "signature" in ss:
+                del ss["signature"]
+            st.rerun()
+
+    def build_ui_for_parsing_error(self):
+        def build_form_fragment(form_, col, title, *questions):
+            form_.text(title)
+            for user_data in questions:
+                with col:
+                    form_.text_input(self.pm.questions_to_field_labels()[user_data], value=ss.get("answers", {})
+                                     .get(user_data), key=f"fq_{user_data.name}")
+            with col:
+                form_.text_input("Save as", key=title.replace(" ", "_"))
+
+        self._build_base_ui()
+        f = st.form("Please check the following information and correct fix any inaccuracies")
+        col1, col2 = f.columns(2)
+        build_form_fragment(f, col1, "your details", Q.FULL_NAME, Q.CONTACT_NUMBER, Q.YOUR_EMAIL)
+        build_form_fragment(f, col2, "work details", Q.WORK_TO_DO, Q.START_DATE, Q.END_DATE)
+        build_form_fragment(f, col1, "location details", Q.COMMUNITY, Q.BUILDING, Q.UNIT_APT_NUMBER,
+                            Q.OWNER_OR_TENANT)
+        build_form_fragment(f, col2, "contractor details", Q.COMPANY_NAME, Q.COMPANY_NUMBER, Q.COMPANY_EMAIL)
+        submit_data = f.form_submit_button()
+        if submit_data:
+            for i in range(len(Q)):
+                ss["answers"][Q(i).name] = ss[f"fq_{Q(i).name}"]
+
+            for details_key, func in [("your_details", self._get_personal_details),
+                                      ("location_details", self._get_location_details),
+                                      ("contractor_details", self._get_contractor_details)]:
+                details = func(details_key)
+                if details:
+                    key = ss[details_key]  # get the name under which this data should be saved
+                    ls.setItem(key, json.dumps(details.__dict__))
+
+    @staticmethod
+    def _get_personal_details(personal_details_key) -> PersonalDetails | None:
+        if ss.get(personal_details_key):
+            return PersonalDetails(ss[f"fq_{Q.FULL_NAME.name}"], ss[f"fq_{Q.YOUR_EMAIL.name}"], ss[f"fq_{Q.CONTACT_NUMBER.name}"])
+        return None
+
+    @staticmethod
+    def _get_location_details(location_details_key) -> LocationDetails | None:
+        if ss.get(location_details_key):
+            return LocationDetails(ss[f"fq_{Q.OWNER_OR_TENANT.name}"], ss[f"fq_{Q.COMMUNITY.name}"],
+                                   ss[f"fq_{Q.BUILDING.name}"], ss[f"fq_{Q.UNIT_APT_NUMBER.name}"])
+        return None
+
+    @staticmethod
+    def _get_contractor_details(contractor_details_key) -> ContractorDetails | None:
+        if ss.get(contractor_details_key):
+            return ContractorDetails(ss[f"fq_{Q.COMPANY_NAME}"], ss[f"fq_{Q.COMPANY_NUMBER}"],
+                                     ss[f"fq_{Q.COMPANY_EMAIL}"])
+        return None
 
 
 def use_streamlit():
-    pm = PromptsManager()
-    help_ = f"{user_msg}\n\n" + '\n'.join(pm.questions)
-    repository = (build_repo_from_environment(pm.system_prompt) or
-                  get_repository("testing",
-                                 Model("fakeModel", ModelRoles("a", "b", "c"))))
-    st.markdown("## Dubai Asset Management red tape cutter")
-    if not st.session_state.get("step"):
-        build_ui_for_initial_state(help_)
-        llama3 = "meta-llama/Meta-Llama-3-8B-Instruct"
-        # repository = get_repository("intel_npu", Model(llama3,
-        #                                                ModelRoles("system", "user", "assistant")),
-        #                             pm.system_prompt, Path("llm_log.txt"))
-
-    if st.session_state.get("step") == "parsing_answers":
-        build_ui_for_parsing_answers(repository, pm)
-
-    if st.session_state.get("step") == "ask_again":
-        build_ui_for_ask_again(pm)
-    if st.session_state.get("step") == "check_category":
-        build_ui_for_check_category(repository, pm)
-
-    if st.session_state.get("step") == "form_created":
-        build_ui_for_form_created()
+
+    um = UIManager()
+
+    if not um.get_current_step():
+        um.build_ui_for_initial_state(user_msg)
+    if um.get_current_step() == "parsing_answers":
+        um.build_ui_for_parsing_answers()
+    if um.get_current_step() == "parsing_error":
+        um.build_ui_for_parsing_error()
+    if um.get_current_step() == "ask_again":
+        um.build_ui_for_ask_again()
+    if um.get_current_step() == "check_category":
+        um.build_ui_for_check_category()
+    if um.get_current_step() == "form_created":
+        um.build_ui_for_form_created()
 
 
 use_streamlit()
+
+
+
+

llm_manager/llm_parser.py

@@ -1,39 +1,34 @@
 from form.form import work_categories
+from prompts.prompts_manager import Questions
 
 
 class LlmParser:
 
     @classmethod
-    def parse_verification_prompt_answers(cls, llm_answer) -> dict[int, str | None]:
+    def parse_verification_prompt_answers(cls, llm_answer) -> dict[Questions, str | None]:
         print(f"llm answer: {llm_answer}")
-        expected_answers_count = 13
         answers = {}
         i = 0
         question_id = 0
-        lines = [l for l in llm_answer.split("\n") if len(l.strip()) > 0]
+        lines = [l for l in llm_answer.split("\n") if len(l.strip()) > 0 and not l.strip().endswith("?")]
         while i < len(lines):
             line = lines[i].strip()
             if len(line) == 0:
                 i += 1
-            elif line.endswith("?") and i+1<len(lines):
-                i+=1
             elif "null" in lines[i]:
-                answers[question_id] = None
+                answers[Questions(question_id)] = None
                 i += 1
                 question_id += 1
             elif ":" in lines[i]:
-                answers[question_id] = line.split(":")[1]
+                answers[Questions(question_id)] = line.split(":")[1]
                 i += 1
                 question_id += 1
             else:
-                answers[question_id] = line
-                i+=1
+                answers[Questions(question_id)] = line
+                i += 1
                 question_id += 1
         return answers
 
-
-
-
     @classmethod
     def parse_get_categories_answer(cls, category_answer) -> list[str]:
         categories = []

local_storage/entities.py

@@ -0,0 +1,53 @@
+import abc
+import json
+
+from streamlit_local_storage import LocalStorage
+
+ls: LocalStorage = LocalStorage()
+
+
+class SavedDetails(abc.ABC):
+    def __init__(self, type_: str):
+        self.type_ = type_
+
+    def save_to_local_storage(self, key: str):
+        ls.setItem(key, json.dumps(self.__dict__))
+
+    @classmethod
+    def load(cls, json_data: str):
+        data = json.loads(json_data)
+        type_ = data.get("type_")
+        if not type_ or type_ != cls.type_:
+            raise ValueError(f"the expected type is {cls.type_} but is actually {type_}")
+        return cls.__init__(**{k: v for k, v in data if k != "type"})
+
+
+class PersonalDetails(SavedDetails):
+    type_ = "personal_details"
+
+    def __init__(self, full_name, email, contact_number):
+        super().__init__(self.type_)
+        self.full_name = full_name
+        self.email = email
+        self.contact_number = contact_number
+
+
+class LocationDetails(SavedDetails):
+    type_ = "location_details"
+
+    def __init__(self, owner_or_tenant: str, community: str, building: str, unit_number: str):
+        super().__init__(self.type_)
+        self.owner_or_tenant = owner_or_tenant
+        self.community = community
+        self.building = building
+        self.unit_number = unit_number
+
+
+class ContractorDetails(SavedDetails):
+    type_ = "contractor_details"
+
+    def __init__(self, contractor_name:str, contractor_contact_number:str, contractor_email:str):
+        super().__init__(self.type_)
+        self.contractor_name = contractor_name
+        self.contractor_contact_number = contractor_contact_number
+        self.contractor_email = contractor_email

prompts/prompts_manager.py

@@ -1,13 +1,28 @@
 import datetime
+from enum import Enum
 from pathlib import Path
 
 from utils.date_utils import get_today_date_as_dd_mm_yyyy
 from form.form import work_categories as wc
 
 
+class Questions(Enum):
+    FULL_NAME = 0
+    WORK_TO_DO = 1
+    COMMUNITY = 2
+    BUILDING = 3
+    UNIT_APT_NUMBER = 4
+    OWNER_OR_TENANT = 5
+    START_DATE = 6
+    END_DATE = 7
+    CONTACT_NUMBER = 8
+    COMPANY_NAME = 9
+    COMPANY_EMAIL = 10
+    COMPANY_NUMBER = 11
+    YOUR_EMAIL = 12
+
 class PromptsManager:
     def __init__(self, work_categories: dict[str, str] = None):
-
         if not work_categories:
             self.work_categories = wc
         base_path = Path(__file__).parent
@@ -22,14 +37,25 @@ class PromptsManager:
             self.verification_prompt: str = verification_prompt
 
     def verify_user_input_prompt(self, user_prompt) -> str:
-        return (f"Using only this information \n {user_prompt} \n answer the following questions, for each question that you cannot answer just answer 'null'. "
-                f"Put each answer in a new line, keep the answer brief "
-                f"and maintain the order in which the questions are asked. Do not add any preamble: "
-                f"{self.verification_prompt}")
+        return (
+            f"Using only this information \n {user_prompt} \n answer the following questions, for each question that you cannot answer just answer 'null'. "
+            f"Put each answer in a new line, keep the answer brief "
+            f"and maintain the order in which the questions are asked. Do not add any preamble: "
+            f"{self.verification_prompt}")
 
     def get_work_category(self, work_description: str) -> str:
-        return (f"The work to do is {work_description}: choose the most accurate categories among the following {', '.join(self.work_categories.values())} "
-                f"Only return the categories, separated by a semicolon")
+        return (
+            f"The work to do is {work_description}: choose the most accurate categories among the following {', '.join(self.work_categories.values())} "
+            f"Only return the categories, separated by a semicolon")
+
+    @staticmethod
+    def questions_to_field_labels():
+        return {
+            Questions.FULL_NAME: "Full name", Questions.WORK_TO_DO: "Work to do", Questions.COMMUNITY: "Community",
+            Questions.BUILDING:"Building name", Questions.UNIT_APT_NUMBER: "Unit/apartment number",
+            Questions.OWNER_OR_TENANT: "Owner/Tenant", Questions.START_DATE: "Start date",
+            Questions.END_DATE: "End date", Questions.CONTACT_NUMBER: "Your contact number",
+            Questions.COMPANY_NAME: "Contractor company name", Questions.COMPANY_EMAIL: "Contracting company email",
+            Questions.COMPANY_NUMBER: "Contracting company contact number", Questions.YOUR_EMAIL: "Your email"
+        }
 
-    def ingest_user_answers(self, user_prompt: str) -> str:
-        return f"Ingest the following information: {user_prompt}"

prompts/verification_prompt2.txt

@@ -8,6 +8,6 @@ In which date is the work taking place? Please answer with just a date formatted
 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 contact number of the contracting company? 
 What is the email of the contracting company?
 What is my email?

repository/intel_npu.py

@@ -1,8 +1,9 @@
 import json
 from pathlib import Path
 
-from intel_npu_acceleration_library import NPUModelForCausalLM, int4
+from intel_npu_acceleration_library import NPUModelForCausalLM, int4, float16
 from intel_npu_acceleration_library.compiler import CompilerConfig
+from intel_npu_acceleration_library.dtypes import float32, float64, bfloat16
 from transformers import AutoTokenizer
 
 from repository.repository_abc import Repository, Model
@@ -25,14 +26,10 @@ class IntelNpuRepository(Repository):
         return self.message_history
 
     def init(self):
-        compiler_conf = CompilerConfig(dtype=int4)
-        self.model = NPUModelForCausalLM.from_pretrained(self.model_info.name, use_cache=True, config=compiler_conf,
-                                                          export=True, temperature=0).eval()
-        self.tokenizer = AutoTokenizer.from_pretrained(self.model_info.name)
-        self.terminators = [self.tokenizer.eos_token_id, self.tokenizer.convert_tokens_to_ids("<|eot_id|>")]
+        self._init_model()
+        self._init_tokenizer()
 
     def send_prompt(self, prompt: str, add_to_history: bool = True) -> dict[str, str]:
-        pass
         print("prompt to be sent: " + prompt)
         user_prompt = {"role": self.model_info.roles.user_role, "content": prompt}
         if self.log_to_file:
@@ -40,6 +37,10 @@ class IntelNpuRepository(Repository):
                 log_file.write(json.dumps(user_prompt, indent=2))
                 log_file.write("\n")
         self.get_message_history().append(user_prompt)
+        if self.model is None:
+            self._init_model()
+        if self.tokenizer is None:
+            self._init_tokenizer()
         input_ids = (self.tokenizer.apply_chat_template(self.get_message_history(), add_generation_prompt=True,
                                                         return_tensors="pt")
                      .to(self.model.device))
@@ -57,3 +58,12 @@ class IntelNpuRepository(Repository):
         else:
             self.message_history.pop()
         return answer
+
+    def _init_tokenizer(self):
+        self.tokenizer = AutoTokenizer.from_pretrained(self.model_info.name)
+        self.terminators = [self.tokenizer.eos_token_id, self.tokenizer.convert_tokens_to_ids("<|eot_id|>")]
+
+    def _init_model(self):
+        compiler_conf = CompilerConfig(dtype=float16)
+        self.model = NPUModelForCausalLM.from_pretrained(self.model_info.name, use_cache=True, config=compiler_conf,
+                                                         export=True, temperature=0.1).eval()

repository/repository.py

@@ -1,4 +1,6 @@
+import os
 from pathlib import Path
+
 from utils.env_utils import in_hf
 
 if not in_hf():
@@ -8,6 +10,17 @@ from repository.ondemand import OndemandRepository
 from repository.repository_abc import Model
 from repository.testing_repo import TestingRepository
 
+def build_repo_from_environment(system_prompt: str):
+    implementation = os.getenv("implementation")
+    model_name = os.getenv("model_name")
+
+    if implementation:
+        return get_repository(implementation, Model(model_name, ModelRoles("system",
+                                                                                                 "user",
+                                                                                                 "assistant")),
+                                                    system_prompt)
+    else:
+        return None
 
 def get_repository(implementation: str, model: Model, system_msg: str = None, log_to_file: Path = None):
     known_implementations = ["ollama", "intel_npu", "testing", "ondemand"]
@@ -24,7 +37,7 @@ def get_repository(implementation: str, model: Model, system_msg: str = None, lo
         return TestingRepository(prompts_answers=[
             {
                 "role": "assistant",
-                "content": "What is my full name?\n\nnull\n\nWhat is the nature of the work I need to do?\n\nPest control\n\nIn which community is the work taking place?\n\nJBR\n\nIn which building?\n\nnull\n\nIn which unit/apartment number?\n\nnull\n\nAm I the owner or the tenant?\n\nTenant\n\nIn which date is the work taking place?\n\n12/09/2024\n\nIn which date will the work finish?\n\n12/09/2024\n\nWhat is my contact number?\n\nnull\n\nWhat is the name of the contracting company?\n\nnull\n\nWhat is the contact number of the contracting company?\n\nnull\n\nWhat is the email of the contracting company?\n\nnull\n\nWhat is my email?\n\nnull"
+                "content": "What is my full name?\n\nnull\n\nWhat is the nature of the work I need to do?\n\nPest control\n\nIn which community is the work taking place?\n\nJBR\n\nIn which building?\n\nnull\n\nIn which unit/apartment number?\n\nnull\n\nAm I the owner or the tenant?\n\nTenant\n\nIn which date is the work taking place?\n\n12/09/2024\n\nIn which date will the work finish?\n\n12/09/2024\n\nWhat is my contact number?\n\nnull\n\nWhat is the name of the contracting company?\n\nnull\n\nWhat is the contact number of the contracting company?\n\nnull\n\nWhat is the email of the contracting company?\n\nnull"
             },
             {
                 "role":"assistant",

requirements-base.txt

@@ -1,3 +1,4 @@
 transformers
 streamlit
-PyPDFForm
+PyPDFForm
+streamlit-local-storage

ui_manager.py

@@ -1,67 +1,162 @@
+import json
+import os
 import streamlit as st
+from streamlit import session_state as ss
+from streamlit_local_storage import LocalStorage
 
 from form.form import build_form_data_from_answers, write_pdf_form
 from llm_manager.llm_parser import LlmParser
+from local_storage.entities import PersonalDetails, LocationDetails, ContractorDetails
+from prompts.prompts_manager import PromptsManager, Questions as Q
+from repository.repository import get_repository
+from repository.repository_abc import Model, ModelRoles
 from utils.parsing_utils import check_for_missing_answers
 
+ls: LocalStorage = LocalStorage()
 
-def build_ui_for_initial_state(help_):
-    with st.form("Please describe your request"):
-        user_input = st.text_area("Your input", height=700, label_visibility="hidden", placeholder=help_, help=help_)
-        signature = st.file_uploader("Your signature", key="file_upload")
-        st.session_state["signature"] = signature
-        submit_button = st.form_submit_button()
-        if submit_button:
-            st.session_state["user_input"] = user_input
-            st.session_state["step"] = "parsing_answers"
+def in_hf() -> bool:
+    return os.getenv("env") == "hf"
+
+
+
+
+class UIManager:
+    def __init__(self):
+        self.pm: PromptsManager = PromptsManager()
+        self.repository = (build_repo_from_environment(self.pm.system_prompt) or
+                           get_repository("testing",
+                                          Model("fakeModel", ModelRoles("a", "b", "c"))))
+
+    @staticmethod
+    def get_current_step():
+        return ss.get("step")
+
+    @staticmethod
+    def _build_base_ui():
+        st.markdown("## Dubai Asset Management red tape cutter")
+
+    def build_ui_for_initial_state(self, user_message):
+        help_ = user_message
+        self._build_base_ui()
+        with st.form("Please describe your request"):
+            user_input = st.text_area("Your input", height=700, label_visibility="hidden", placeholder=help_,
+                                      help=help_)
+            signature = st.file_uploader("Your signature", key="file_upload")
+            ss["signature"] = signature
+            submit_button = st.form_submit_button()
+            if submit_button:
+                ss["user_input"] = user_input
+                ss["step"] = "parsing_answers"
+                st.rerun()
+
+    def build_ui_for_parsing_answers(self):
+        self._build_base_ui()
+        with st.status("initialising LLM"):
+            self.repository.init()
+        with st.status("waiting for LLM"):
+            answer = self.repository.send_prompt(self.pm.verify_user_input_prompt(ss["user_input"]))
+            st.write(f"answers from LLM: {answer['content']}")
+        with st.status("Checking for missing answers"):
+            answers = LlmParser.parse_verification_prompt_answers(answer['content'])
+            ss["answers"] = answers
+            if len(answers) != len(Q):
+                ss["step"] = "parsing_error"
+                st.rerun()
+        ss["missing_answers"] = check_for_missing_answers(ss["answers"])
+        if not ss.get("missing_answers"):
+            ss["step"] = "check_category"
+        else:
+            ss["step"] = "ask_again"
+        st.rerun()
+
+    def build_ui_for_ask_again(self):
+        self._build_base_ui()
+        with st.form("form1"):
+            for ma in ss["missing_answers"]:
+                st.text_input(self.pm.questions[ma].lower(), key=ma)
+            submitted = st.form_submit_button("Submit answers")
+            if submitted:
+                for ma in ss["missing_answers"]:
+                    ss["answers"][ma] = ss[ma]
+                ss["step"] = "check_category"
+                st.rerun()
+
+    def build_ui_for_check_category(self):
+        self._build_base_ui()
+        with st.status("finding the work categories applicable to your work"):
+            answer = self.repository.send_prompt(self.pm.get_work_category(ss["answers"][1]))
+            categories = LlmParser.parse_get_categories_answer(answer['content'])
+
+        with st.status("categories found, creating PDF form"):
+            form_data, filename = build_form_data_from_answers(ss["answers"], categories,
+                                                               ss.get("signature"))
+            pdf_form = write_pdf_form(form_data)
+            pdf_form_filename = filename
+            ss["pdf_form"] = pdf_form
+            ss["pdf_form_filename"] = pdf_form_filename
+            ss["step"] = "form_created"
             st.rerun()
-def build_ui_for_parsing_answers(repository, pm):
-    with st.status("initialising LLM"):
-        repository.init()
-    with st.status("waiting for LLM"):
-        answer = repository.send_prompt(pm.verify_user_input_prompt(st.session_state["user_input"]))
-        st.write(f"answers from LLM: {answer['content']}")
-    with st.status("Checking for missing answers"):
-        st.session_state["answers"] = LlmParser.parse_verification_prompt_answers(answer['content'])
-    st.session_state["missing_answers"] = check_for_missing_answers(st.session_state["answers"])
-    if not st.session_state.get("missing_answers"):
-        st.session_state["step"] = "check_category"
-    else:
-        st.session_state["step"] = "ask_again"
-    st.rerun()
-
-def build_ui_for_ask_again(pm):
-    with st.form("form1"):
-        for ma in st.session_state["missing_answers"]:
-            st.text_input(pm.questions[ma].lower(), key=ma)
-        submitted = st.form_submit_button("Submit answers")
-        if submitted:
-            for ma in st.session_state["missing_answers"]:
-                st.session_state["answers"][ma] = st.session_state[ma]
-            st.session_state["step"] = "check_category"
+
+    def build_ui_for_form_created(self):
+        self._build_base_ui()
+        st.download_button("download form", ss["pdf_form"],
+                           file_name=ss["pdf_form_filename"], mime="application/pdf")
+        start_over_button = st.button("Start over")
+        if start_over_button:
+            del ss["step"]
+            del ss["pdf_form"]
+            del ss["pdf_form_filename"]
+            if "signature" in ss:
+                del ss["signature"]
             st.rerun()
-def build_ui_for_check_category(repository, pm):
-    with st.status("finding the work categories applicable to your work"):
-        answer = repository.send_prompt(pm.get_work_category(st.session_state["answers"][1]))
-        categories = LlmParser.parse_get_categories_answer(answer['content'])
-
-    with st.status("categories found, creating PDF form"):
-        form_data, filename = build_form_data_from_answers(st.session_state["answers"], categories,
-                                                           st.session_state.get("signature"))
-        pdf_form = write_pdf_form(form_data)
-        pdf_form_filename = filename
-        st.session_state["pdf_form"] = pdf_form
-        st.session_state["pdf_form_filename"] = pdf_form_filename
-        st.session_state["step"] = "form_created"
-        st.rerun()
-def build_ui_for_form_created():
-    st.download_button("download form", st.session_state["pdf_form"],
-                       file_name=st.session_state["pdf_form_filename"], mime="application/pdf")
-    start_over_button = st.button("Start over")
-    if start_over_button:
-        del st.session_state["step"]
-        del st.session_state["pdf_form"]
-        del st.session_state["pdf_form_filename"]
-        if "signature" in st.session_state:
-            del st.session_state["signature"]
-        st.rerun()
+
+    def build_ui_for_parsing_error(self):
+        def build_form_fragment(form_, col, title, *questions):
+            form_.text(title)
+            for user_data in questions:
+                with col:
+                    form_.text_input(self.pm.questions_to_field_labels()[user_data], value=ss.get("answers", {})
+                                     .get(user_data), key=f"fq_{user_data.value}")
+            with col:
+                form_.text_input("Save as", key=title.replace(" ", "_"))
+
+        self._build_base_ui()
+        f = st.form("Please check the following information and correct fix any inaccuracies")
+        col1, col2 = f.columns(2)
+        build_form_fragment(f, col1, "your details", Q.FULL_NAME, Q.CONTACT_NUMBER, Q.YOUR_EMAIL)
+        build_form_fragment(f, col2, "work details", Q.WORK_TO_DO, Q.START_DATE, Q.END_DATE)
+        build_form_fragment(f, col1, "location details", Q.COMMUNITY, Q.BUILDING, Q.UNIT_APT_NUMBER,
+                            Q.OWNER_OR_TENANT)
+        build_form_fragment(f, col2, "contractor details", Q.COMPANY_NAME, Q.COMPANY_NUMBER, Q.COMPANY_EMAIL)
+        submit_data = f.form_submit_button()
+        if submit_data:
+            for i in range(len(Q)):
+                ss["answers"][Q(i)] = ss[f"fq_{i}"]
+
+            for details_key, func in [("your_details", self._get_personal_details),
+                                      ("location_details", self._get_location_details),
+                                      ("contractor_details", self._get_contractor_details)]:
+                details = func(details_key)
+                if details:
+                    key = ss[details_key]  # get the name under which this data should be saved
+                    ls.setItem(key, json.dumps(details))
+
+    @staticmethod
+    def _get_personal_details(personal_details_key) -> PersonalDetails | None:
+        if ss.get(personal_details_key):
+            return PersonalDetails(ss[f"fq_{Q.FULL_NAME}"], ss[f"fq_{Q.FULL_NAME}"], ss[f"fq_{Q.CONTACT_NUMBER}"])
+        return None
+
+    @staticmethod
+    def _get_location_details(location_details_key) -> LocationDetails | None:
+        if ss.get(location_details_key):
+            return LocationDetails(ss[f"fq_{Q.OWNER_OR_TENANT}"], ss[f"fq_{Q.COMMUNITY}"], ss[f"fq_{Q.BUILDING}"],
+                                   ss[f"fq_{Q.UNIT_APT_NUMBER}"])
+        return None
+
+    @staticmethod
+    def _get_contractor_details(contractor_details_key) -> ContractorDetails | None:
+        if ss.get(contractor_details_key):
+            return ContractorDetails(ss[f"fq_{Q.COMPANY_NAME}"], ss[f"fq_{Q.COMPANY_NUMBER}"],
+                                     ss[f"fq_{Q.COMPANY_EMAIL}"])
+        return None

utils/env_utils.py

@@ -1,22 +1,5 @@
 import os
 
-import repository.repository
-from repository.repository_abc import Model, ModelRoles
-
 
 def in_hf() -> bool:
     return os.getenv("env") == "hf"
-
-
-def build_repo_from_environment(system_prompt: str):
-    implementation = os.getenv("implementation")
-    model_name = os.getenv("model_name")
-
-    if implementation:
-        return repository.repository.get_repository(implementation, Model(model_name, ModelRoles("system",
-                                                                                                 "user",
-                                                                                                 "assistant")),
-                                                    system_prompt)
-    else:
-        return None
-