enricorampazzo commited on
Commit
88f9105
·
1 Parent(s): 6508ff2

now taking into account saved details when prompting and parsing the answer of the LLM

Browse files
app.py CHANGED
@@ -9,14 +9,15 @@ from form.form import build_form_data_from_answers, write_pdf_form, work_categor
9
  from llm_manager.llm_parser import LlmParser
10
  from local_storage.entities import PersonalDetails, LocationDetails, ContractorDetails
11
  from local_storage.ls_manager import save_details, get_detail
12
- from prompts.prompts_manager import PromptsManager, Questions as Q
 
13
  from repository.repository import build_repo_from_environment, get_repository
14
  from repository import ModelRoles, Model
15
  from utils.parsing_utils import check_for_missing_answers
16
 
17
  user_msg = "Please describe what you need to do. To get the best results try to answer all the following questions:"
18
 
19
- find_tags_regex = re.compile(r"(@\S)")
20
 
21
  class Steps(Enum):
22
  INITIAL_STATE = 1
@@ -35,6 +36,7 @@ class Steps(Enum):
35
  def __hash__(self):
36
  return hash(self.value)
37
 
 
38
  class UIManager:
39
  def __init__(self):
40
  self.pm: PromptsManager = PromptsManager(work_categories=work_categories)
@@ -76,14 +78,13 @@ class UIManager:
76
  with st.status("parsing user input for tags"):
77
  tags = find_tags_regex.findall(ss["user_input"])
78
  details = [get_detail(t) for t in tags]
79
- ss["details"] = details
80
  with st.status("initialising LLM"):
81
  self.repository.init()
82
  with st.status("waiting for LLM"):
83
  answer = self.repository.send_prompt(self.pm.verify_user_input_prompt(ss["user_input"], details))
84
  st.write(f"answers from LLM: {answer['content']}")
85
  with st.status("Checking for missing answers"):
86
- answers = LlmParser.parse_verification_prompt_answers(answer['content'])
87
  ss["answers"] = answers
88
  if len(answers) != len(Q):
89
  self.set_current_step(Steps.PARSING_ERROR)
 
9
  from llm_manager.llm_parser import LlmParser
10
  from local_storage.entities import PersonalDetails, LocationDetails, ContractorDetails
11
  from local_storage.ls_manager import save_details, get_detail
12
+ from prompts.prompts_manager import PromptsManager
13
+ from enums import Questions as Q
14
  from repository.repository import build_repo_from_environment, get_repository
15
  from repository import ModelRoles, Model
16
  from utils.parsing_utils import check_for_missing_answers
17
 
18
  user_msg = "Please describe what you need to do. To get the best results try to answer all the following questions:"
19
 
20
+ find_tags_regex = re.compile(r"@(\S*)")
21
 
22
  class Steps(Enum):
23
  INITIAL_STATE = 1
 
36
  def __hash__(self):
37
  return hash(self.value)
38
 
39
+
40
  class UIManager:
41
  def __init__(self):
42
  self.pm: PromptsManager = PromptsManager(work_categories=work_categories)
 
78
  with st.status("parsing user input for tags"):
79
  tags = find_tags_regex.findall(ss["user_input"])
80
  details = [get_detail(t) for t in tags]
 
81
  with st.status("initialising LLM"):
82
  self.repository.init()
83
  with st.status("waiting for LLM"):
84
  answer = self.repository.send_prompt(self.pm.verify_user_input_prompt(ss["user_input"], details))
85
  st.write(f"answers from LLM: {answer['content']}")
86
  with st.status("Checking for missing answers"):
87
+ answers = LlmParser.parse_verification_prompt_answers(answer['content'], details)
88
  ss["answers"] = answers
89
  if len(answers) != len(Q):
90
  self.set_current_step(Steps.PARSING_ERROR)
enums.py ADDED
@@ -0,0 +1,33 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from enum import Enum
2
+
3
+
4
+ class DetailsType(Enum):
5
+ PERSONAL_DETAILS = 1
6
+ LOCATION_DETAILS = 2
7
+ CONTRACTOR_DETAILS = 3
8
+
9
+ @classmethod
10
+ def values(cls):
11
+ return [cls.PERSONAL_DETAILS, cls.LOCATION_DETAILS, cls.CONTRACTOR_DETAILS]
12
+
13
+
14
+ class Questions(Enum):
15
+ FULL_NAME = 0
16
+ WORK_TO_DO = 1
17
+ COMMUNITY = 2
18
+ BUILDING = 3
19
+ UNIT_APT_NUMBER = 4
20
+ OWNER_OR_TENANT = 5
21
+ START_DATE = 6
22
+ END_DATE = 7
23
+ CONTACT_NUMBER = 8
24
+ COMPANY_NAME = 9
25
+ COMPANY_EMAIL = 10
26
+ COMPANY_NUMBER = 11
27
+ YOUR_EMAIL = 12
28
+
29
+ @classmethod
30
+ def values(cls):
31
+ return [Questions.FULL_NAME, Questions.WORK_TO_DO, Questions.BUILDING, Questions.UNIT_APT_NUMBER,
32
+ Questions.OWNER_OR_TENANT, Questions.START_DATE, Questions.END_DATE, Questions.CONTACT_NUMBER,
33
+ Questions.COMPANY_NAME, Questions.COMPANY_EMAIL, Questions.COMPANY_NUMBER, Questions.YOUR_EMAIL]
form/form.py CHANGED
@@ -2,7 +2,7 @@ from pathlib import Path
2
  from typing import TypedDict, Tuple
3
  from PyPDFForm import PdfWrapper
4
 
5
- from prompts.prompts_manager import Questions
6
  from utils.date_utils import get_today_date_as_dd_mm_yyyy
7
  from utils.parsing_utils import find_and_parse_date, find_and_parse_phone_number
8
 
 
2
  from typing import TypedDict, Tuple
3
  from PyPDFForm import PdfWrapper
4
 
5
+ from enums import Questions
6
  from utils.date_utils import get_today_date_as_dd_mm_yyyy
7
  from utils.parsing_utils import find_and_parse_date, find_and_parse_phone_number
8
 
llm_manager/llm_parser.py CHANGED
@@ -1,32 +1,41 @@
1
  from form.form import work_categories
2
- from prompts.prompts_manager import Questions
 
3
 
4
 
5
  class LlmParser:
6
 
7
  @classmethod
8
- def parse_verification_prompt_answers(cls, llm_answer) -> dict[Questions, str | None]:
 
9
  print(f"llm answer: {llm_answer}")
10
  answers = {}
 
 
 
 
 
11
  i = 0
12
- question_id = 0
13
  lines = [l for l in llm_answer.split("\n") if len(l.strip()) > 0 and not l.strip().endswith("?")]
14
  while i < len(lines):
 
 
15
  line = lines[i].strip()
16
  if len(line) == 0:
17
  i += 1
18
  elif "null" in lines[i]:
19
  answers[Questions(question_id)] = None
20
  i += 1
21
- question_id += 1
22
  elif ":" in lines[i]:
23
  answers[Questions(question_id)] = line.split(":")[1]
24
  i += 1
25
- question_id += 1
26
  else:
27
  answers[Questions(question_id)] = line
28
  i += 1
29
- question_id += 1
30
  return answers
31
 
32
  @classmethod
@@ -35,3 +44,4 @@ class LlmParser:
35
  for category in category_answer.split(";"):
36
  categories.extend([k for k, v in work_categories.items() if category.lower() in v.lower()])
37
  return categories
 
 
1
  from form.form import work_categories
2
+ from local_storage.entities import SavedDetails
3
+ from enums import Questions
4
 
5
 
6
  class LlmParser:
7
 
8
  @classmethod
9
+ def parse_verification_prompt_answers(cls, llm_answer:str, details: list[SavedDetails] | None)\
10
+ -> dict[Questions, str | None]:
11
  print(f"llm answer: {llm_answer}")
12
  answers = {}
13
+ for d in details:
14
+ answers.update(d.to_answers())
15
+
16
+ unanswered_questions = sorted([k.value for k in Questions.values() if k not in answers])
17
+ unanswered_questions = iter(unanswered_questions)
18
  i = 0
19
+ question_id = next(unanswered_questions)
20
  lines = [l for l in llm_answer.split("\n") if len(l.strip()) > 0 and not l.strip().endswith("?")]
21
  while i < len(lines):
22
+ if question_id is None:
23
+ break
24
  line = lines[i].strip()
25
  if len(line) == 0:
26
  i += 1
27
  elif "null" in lines[i]:
28
  answers[Questions(question_id)] = None
29
  i += 1
30
+ question_id = next(unanswered_questions, None)
31
  elif ":" in lines[i]:
32
  answers[Questions(question_id)] = line.split(":")[1]
33
  i += 1
34
+ question_id = next(unanswered_questions, None)
35
  else:
36
  answers[Questions(question_id)] = line
37
  i += 1
38
+ question_id = next(unanswered_questions, None)
39
  return answers
40
 
41
  @classmethod
 
44
  for category in category_answer.split(";"):
45
  categories.extend([k for k, v in work_categories.items() if category.lower() in v.lower()])
46
  return categories
47
+
local_storage/entities.py CHANGED
@@ -1,15 +1,6 @@
1
  import abc
2
- from enum import Enum
3
 
4
-
5
- class DetailsType(Enum):
6
- PERSONAL_DETAILS = 1
7
- LOCATION_DETAILS = 2
8
- CONTRACTOR_DETAILS = 3
9
-
10
- @classmethod
11
- def values(cls):
12
- return [cls.PERSONAL_DETAILS, cls.LOCATION_DETAILS, cls.CONTRACTOR_DETAILS]
13
 
14
 
15
  class SavedDetails(abc.ABC):
@@ -28,6 +19,9 @@ class SavedDetails(abc.ABC):
28
  def to_json(self):
29
  return {k:v for k,v in self.__dict__.items() if k not in self.excluded_fields}
30
 
 
 
 
31
  class PersonalDetails(SavedDetails):
32
  type_ = DetailsType.PERSONAL_DETAILS
33
 
@@ -37,6 +31,15 @@ class PersonalDetails(SavedDetails):
37
  self.email = email
38
  self.contact_number = contact_number
39
 
 
 
 
 
 
 
 
 
 
40
 
41
  class LocationDetails(SavedDetails):
42
  type_ = DetailsType.LOCATION_DETAILS
@@ -48,6 +51,17 @@ class LocationDetails(SavedDetails):
48
  self.building = building
49
  self.unit_number = unit_number
50
 
 
 
 
 
 
 
 
 
 
 
 
51
 
52
  class ContractorDetails(SavedDetails):
53
  type_ = DetailsType.CONTRACTOR_DETAILS
@@ -57,3 +71,13 @@ class ContractorDetails(SavedDetails):
57
  self.contractor_name = contractor_name
58
  self.contractor_contact_number = contractor_contact_number
59
  self.contractor_email = contractor_email
 
 
 
 
 
 
 
 
 
 
 
1
  import abc
 
2
 
3
+ from enums import DetailsType, Questions
 
 
 
 
 
 
 
 
4
 
5
 
6
  class SavedDetails(abc.ABC):
 
19
  def to_json(self):
20
  return {k:v for k,v in self.__dict__.items() if k not in self.excluded_fields}
21
 
22
+ def to_answers(self):
23
+ pass
24
+
25
  class PersonalDetails(SavedDetails):
26
  type_ = DetailsType.PERSONAL_DETAILS
27
 
 
31
  self.email = email
32
  self.contact_number = contact_number
33
 
34
+ def to_answers(self) -> dict[Questions, str]:
35
+ answers = {}
36
+ if self.full_name is not None:
37
+ answers[Questions.FULL_NAME] = self.full_name
38
+ if self.email is not None:
39
+ answers[Questions.YOUR_EMAIL] = self.email
40
+ if self.contact_number is not None:
41
+ answers[Questions.CONTACT_NUMBER] = self.contact_number
42
+ return answers
43
 
44
  class LocationDetails(SavedDetails):
45
  type_ = DetailsType.LOCATION_DETAILS
 
51
  self.building = building
52
  self.unit_number = unit_number
53
 
54
+ def to_answers(self) -> dict[Questions, str]:
55
+ answers = {}
56
+ if self.owner_or_tenant is not None:
57
+ answers[Questions.OWNER_OR_TENANT] = self.owner_or_tenant
58
+ if self.building is not None:
59
+ answers[Questions.BUILDING] = self.building
60
+ if self.community is not None:
61
+ answers[Questions.COMMUNITY] = self.community
62
+ if self.unit_number is not None:
63
+ answers[Questions.UNIT_APT_NUMBER] = self.unit_number
64
+ return answers
65
 
66
  class ContractorDetails(SavedDetails):
67
  type_ = DetailsType.CONTRACTOR_DETAILS
 
71
  self.contractor_name = contractor_name
72
  self.contractor_contact_number = contractor_contact_number
73
  self.contractor_email = contractor_email
74
+
75
+ def to_answers(self) -> dict[Questions, str]:
76
+ answers = {}
77
+ if self.contractor_email is not None:
78
+ answers[Questions.COMPANY_EMAIL] = self.contractor_email
79
+ if self.contractor_contact_number is not None:
80
+ answers[Questions.COMPANY_NUMBER] = self.contractor_contact_number
81
+ if self.contractor_name is not None:
82
+ answers[Questions.COMPANY_NAME] = self.contractor_name
83
+ return answers
local_storage/ls_manager.py CHANGED
@@ -2,21 +2,19 @@ import json
2
 
3
  from streamlit_local_storage import LocalStorage
4
 
5
- from local_storage.entities import DetailsType, SavedDetails, PersonalDetails, LocationDetails, ContractorDetails
6
-
7
- ls = LocalStorage()
8
 
9
 
10
  def get_detail(key: str):
11
  for detail_type in DetailsType.values():
12
- detail = (ls.getItem(detail_type.name) or {}).get(key)
13
  if detail:
14
- type_ = detail["type"]
15
- if type_ == DetailsType.PERSONAL_DETAILS.name:
16
  return PersonalDetails(**detail)
17
- elif type_ == DetailsType.LOCATION_DETAILS.name:
18
  return LocationDetails(**detail)
19
- elif type_ == DetailsType.CONTRACTOR_DETAILS.name:
20
  return ContractorDetails(**detail)
21
  return None
22
  return None
 
2
 
3
  from streamlit_local_storage import LocalStorage
4
 
5
+ from local_storage.entities import SavedDetails, PersonalDetails, LocationDetails, ContractorDetails
6
+ from enums import DetailsType
 
7
 
8
 
9
  def get_detail(key: str):
10
  for detail_type in DetailsType.values():
11
+ detail = LocalStorage(key=detail_type.name).getItem(detail_type.name).get(key)
12
  if detail:
13
+ if detail_type.value == DetailsType.PERSONAL_DETAILS.value:
 
14
  return PersonalDetails(**detail)
15
+ elif detail_type.value == DetailsType.LOCATION_DETAILS.value:
16
  return LocationDetails(**detail)
17
+ elif detail_type.value == DetailsType.CONTRACTOR_DETAILS.value:
18
  return ContractorDetails(**detail)
19
  return None
20
  return None
prompts/prompts_manager.py CHANGED
@@ -1,26 +1,10 @@
1
- import datetime
2
- from enum import Enum
3
  from pathlib import Path
4
 
5
- from local_storage.entities import DetailsType, SavedDetails
 
6
  from utils.date_utils import get_today_date_as_dd_mm_yyyy
7
 
8
 
9
- class Questions(Enum):
10
- FULL_NAME = 0
11
- WORK_TO_DO = 1
12
- COMMUNITY = 2
13
- BUILDING = 3
14
- UNIT_APT_NUMBER = 4
15
- OWNER_OR_TENANT = 5
16
- START_DATE = 6
17
- END_DATE = 7
18
- CONTACT_NUMBER = 8
19
- COMPANY_NAME = 9
20
- COMPANY_EMAIL = 10
21
- COMPANY_NUMBER = 11
22
- YOUR_EMAIL = 12
23
-
24
  class PromptsManager:
25
  def __init__(self, work_categories: dict[str, str] = None):
26
  self.work_categories = work_categories
@@ -43,7 +27,7 @@ class PromptsManager:
43
  and maintain the order in which the questions are asked. Do not add any preamble:
44
  """
45
 
46
- skip_questions = [self.get_questions(d) for d in exclude_questions_group]
47
  questions = [q for idx, q in enumerate(self.verification_prompt_questions) if idx not in skip_questions]
48
  return prompt + "\n".join(questions)
49
 
@@ -57,17 +41,21 @@ class PromptsManager:
57
  def questions_to_field_labels():
58
  return {
59
  Questions.FULL_NAME: "Full name", Questions.WORK_TO_DO: "Work to do", Questions.COMMUNITY: "Community",
60
- Questions.BUILDING:"Building name", Questions.UNIT_APT_NUMBER: "Unit/apartment number",
61
  Questions.OWNER_OR_TENANT: "Owner/Tenant", Questions.START_DATE: "Start date",
62
  Questions.END_DATE: "End date", Questions.CONTACT_NUMBER: "Your contact number",
63
  Questions.COMPANY_NAME: "Contractor company name", Questions.COMPANY_EMAIL: "Contracting company email",
64
  Questions.COMPANY_NUMBER: "Contracting company contact number", Questions.YOUR_EMAIL: "Your email"
65
  }
66
- def get_questions(self, type_:DetailsType):
67
- if type_ == DetailsType.PERSONAL_DETAILS:
68
- return [Questions.FULL_NAME, Questions.CONTACT_NUMBER, Questions.YOUR_EMAIL]
69
- if type_ == DetailsType.CONTRACTOR_DETAILS:
70
- return [Questions.COMPANY_NAME, Questions.COMPANY_NUMBER, Questions.COMPANY_EMAIL]
71
- if type_ == DetailsType.LOCATION_DETAILS:
72
- return [Questions.OWNER_OR_TENANT, Questions.COMMUNITY, Questions.BUILDING, Questions.UNIT_APT_NUMBER]
73
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
  from pathlib import Path
2
 
3
+ from enums import Questions
4
+ from local_storage.entities import SavedDetails, PersonalDetails, ContractorDetails, LocationDetails
5
  from utils.date_utils import get_today_date_as_dd_mm_yyyy
6
 
7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8
  class PromptsManager:
9
  def __init__(self, work_categories: dict[str, str] = None):
10
  self.work_categories = work_categories
 
27
  and maintain the order in which the questions are asked. Do not add any preamble:
28
  """
29
 
30
+ skip_questions = self.get_questions(exclude_questions_group)
31
  questions = [q for idx, q in enumerate(self.verification_prompt_questions) if idx not in skip_questions]
32
  return prompt + "\n".join(questions)
33
 
 
41
  def questions_to_field_labels():
42
  return {
43
  Questions.FULL_NAME: "Full name", Questions.WORK_TO_DO: "Work to do", Questions.COMMUNITY: "Community",
44
+ Questions.BUILDING: "Building name", Questions.UNIT_APT_NUMBER: "Unit/apartment number",
45
  Questions.OWNER_OR_TENANT: "Owner/Tenant", Questions.START_DATE: "Start date",
46
  Questions.END_DATE: "End date", Questions.CONTACT_NUMBER: "Your contact number",
47
  Questions.COMPANY_NAME: "Contractor company name", Questions.COMPANY_EMAIL: "Contracting company email",
48
  Questions.COMPANY_NUMBER: "Contracting company contact number", Questions.YOUR_EMAIL: "Your email"
49
  }
 
 
 
 
 
 
 
50
 
51
+ @staticmethod
52
+ def get_questions(details: list[int]):
53
+ to_skip: list[int] = []
54
+ for d in details:
55
+ if isinstance(d, PersonalDetails):
56
+ to_skip.extend([Questions.FULL_NAME.value, Questions.CONTACT_NUMBER.value, Questions.YOUR_EMAIL.value])
57
+ if isinstance(d, ContractorDetails):
58
+ to_skip.extend([Questions.COMPANY_NAME.value, Questions.COMPANY_NUMBER.value, Questions.COMPANY_EMAIL.value])
59
+ if isinstance(d, LocationDetails):
60
+ to_skip.extend([Questions.OWNER_OR_TENANT.value, Questions.COMMUNITY.value, Questions.BUILDING.value, Questions.UNIT_APT_NUMBER.value])
61
+ return to_skip
ui_manager.py CHANGED
@@ -7,7 +7,8 @@ from streamlit_local_storage import LocalStorage
7
  from form.form import build_form_data_from_answers, write_pdf_form
8
  from llm_manager.llm_parser import LlmParser
9
  from local_storage.entities import PersonalDetails, LocationDetails, ContractorDetails
10
- from prompts.prompts_manager import PromptsManager, Questions as Q
 
11
  from repository.repository import get_repository
12
  from repository import ModelRoles, Model
13
  from utils.parsing_utils import check_for_missing_answers
 
7
  from form.form import build_form_data_from_answers, write_pdf_form
8
  from llm_manager.llm_parser import LlmParser
9
  from local_storage.entities import PersonalDetails, LocationDetails, ContractorDetails
10
+ from prompts.prompts_manager import PromptsManager
11
+ from enums import Questions as Q
12
  from repository.repository import get_repository
13
  from repository import ModelRoles, Model
14
  from utils.parsing_utils import check_for_missing_answers