final command line implementation

app.py

@@ -1,11 +1,11 @@
 from pathlib import Path
 
+from llm_manager.llm_parser import LlmParser
 from prompts.prompts_manager import PromptsManager
-from repository.intel_npu import IntelNpuRepository
-from repository.ollama import OllamaRepository
-from llm.llm import Model
-from repository.repository import ModelRoles
-from form.form import work_categories, build_form_data_from_answers, write_pdf_form
+
+from repository.repository import get_repository
+from repository.repository_abc import ModelRoles, Model
+from form.form import build_form_data_from_answers, write_pdf_form
 
 
 def check_for_missing_answers(parsed_questions: dict[int, str]):
@@ -14,8 +14,8 @@ def check_for_missing_answers(parsed_questions: dict[int, str]):
 
 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
+ new_answer = input(f"I could not find the answer to this question: {user_questions[id_].lower()}")
+ parsed_questions[id_] = new_answer
 
 
 if __name__ == '__main__':
@@ -23,31 +23,20 @@ if __name__ == '__main__':
  user_prompt = input(f"Please describe what you need to do. To get the best results "
  f"try to answer all the following questions:\n{'\n'.join(prompts_manager.questions)}\n\n>")
 
- # repository = OllamaRepository(Model("llama3.1",
- # ModelRoles("system", "user", "assistant")),
- # prompts_manager.system_prompt,
- # )
- repository = IntelNpuRepository("meta-llama/Meta-Llama-3-8B-Instruct")
+ repository = get_repository("intel_npu", Model("meta-llama/Meta-Llama-3-8B-Instruct",
+ ModelRoles("system", "user", "assistant")),
+ prompts_manager.system_prompt, Path("llm_log.txt"))
  repository.init()
- repository.send_prompt(f"Ingest the following information: {user_prompt}")
- answers = {x:None for x in range(0,11)}
- answer = repository.send_prompt(f"Answer the following questions, if the answer is not present just answer null. Put the answers between curly braces, separate each answer with a comma, keep the answer brief and maintain the order in which the questions are asked. Do not add any preamble: {"\n".join(prompts_manager.verification_prompt)}")
- for idx, a in enumerate(answer['content'].split(",")):
- answers[idx] = None if 'null' in a else a
-
- # for idx, q in enumerate(prompts_manager.verification_prompt):
- # answer = repository.send_prompt(
- # f"Answer the following questions, if the answer is not present just answer null. Keep the answer brief and separate each answer with a comma and maintain the order in which the questions are asked: {q}")
- # answers[idx] = None if 'null' in answer["content"].lower() else answer['content']
+ # repository.send_prompt(prompts_manager.ingest_user_answers(user_prompt))
+ answer = repository.send_prompt(prompts_manager.verify_user_input_prompt(user_prompt))
+ answers = LlmParser.parse_verification_prompt_answers(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 = 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])
+ answer = repository.send_prompt(prompts_manager.get_work_category(answers[1]))
+ categories = LlmParser.parse_get_categories_answer(answer['content'])
 
  form_data = build_form_data_from_answers(answers, categories, f"{Path(__file__, "..", "signature.png")}")
- write_pdf_form(form_data, Path("signed_form1.pdf"))
+ write_pdf_form(form_data, Path("signed_form1.pdf"))

llm/llm.py

@@ -1,7 +0,0 @@
-from repository.repository import ModelRoles
-
-
-class Model:
- def __init__(self, name: str, roles: ModelRoles):
- self.name: str = name
- self.roles: ModelRoles = roles

llm_manager/llm_parser.py

@@ -0,0 +1,45 @@
+import re
+
+from form.form import work_categories
+
+
+class LlmParser:
+ _verification_prompt_answers_regex = re.compile(r"\|\s*([^|]*)\s?", re.MULTILINE)
+
+ @classmethod
+ def parse_verification_prompt_answers(cls, llm_answer) -> dict[int, 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]
+ 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
+ i += 1
+ question_id += 1
+ elif ":" in lines[i]:
+ answers[question_id] = line.split(":")[1]
+ i += 1
+ question_id += 1
+ else:
+ answers[question_id] = line
+ i+=1
+ question_id += 1
+ return answers
+
+
+
+
+ @classmethod
+ def parse_get_categories_answer(cls, category_answer) -> list[str]:
+ categories = []
+ for category in category_answer.split(";"):
+ categories.extend([k for k, v in work_categories.items() if category in v])
+ return categories

prompts/prompts_manager.py

@@ -2,18 +2,34 @@ import datetime
 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 PromptsManager:
- def __init__(self):
+ def __init__(self, work_categories: dict[str, str] = None):
+
+ if not work_categories:
+ self.work_categories = wc
  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()
+ verification_prompt = verification_prompt_file.read()
  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
+ verification_prompt = verification_prompt.replace("{today}", todays_date)
+ 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, if the answer is not present 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}. Given the following categories {", ".join(self.work_categories.values())} "
+ f"which ones are the most relevant? Only return the categories, separated by a semicolon")
+
+ def ingest_user_answers(self, user_prompt: str) -> str:
+ return f"Ingest the following information: {user_prompt}"

prompts/system_prompt.txt

@@ -1,5 +1,5 @@
 You are a helpful assistant helping tenants and owners when filling minor work permits. Do not answer question outside
 of this domain.
-Keep in mind that 'JBR' is a community, and buildings are named after the following names followed by a number:
+Keep in mind that 'JBR' is a community, and buildings are named after the following names followed by one or more numbers:
 Murjan, Bahar, Shams, Amwaj, Sadaf
-for example "Murjan 1"
+for example "Murjan 1", "Bahar 3", "Sadaf 8"

prompts/verification_prompt2.txt

@@ -2,12 +2,12 @@ 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?
+In which unit/apartment number? Answer only with the unit 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 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 {today}
 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?
+What is my email?

repository/intel_npu.py

@@ -1,47 +1,57 @@
-from intel_npu_acceleration_library import NPUModelForCausalLM, int4, int8
+import json
+from pathlib import Path
+
+from intel_npu_acceleration_library import NPUModelForCausalLM, int4
 from intel_npu_acceleration_library.compiler import CompilerConfig
 from transformers import AutoTokenizer
 
-from repository.repository import Repository, ModelRoles
+from repository.repository_abc import Repository, Model
 
 
 class IntelNpuRepository(Repository):
- def __init__(self, model_name: str):
- self.model_name = model_name
+ def __init__(self, model_info: Model, system_msg: str = None, log_to_file: Path = None):
+ self.model_info: Model = model_info
  self.message_history: list[dict[str, str]] = []
- self.roles = ModelRoles("system", "user", "assistant")
+ self.set_message_for_role(self.model_info.roles.system_role, system_msg)
  self.model = None
  self.tokenizer = None
  self.terminators = None
+ self.log_to_file = log_to_file
 
- def get_model_roles(self) -> ModelRoles:
- return self.roles
-
- def get_model_name(self) -> str:
- return self.model_name
+ def get_model_info(self) -> Model:
+ return self.model_info
 
  def get_message_history(self) -> list[dict[str, str]]:
  return self.message_history
 
- def set_message_for_role(self, message:str, role: str):
- self.get_message_history().append({"role": role, "content": message})
-
  def init(self):
  compiler_conf = CompilerConfig(dtype=int4)
- self.model = NPUModelForCausalLM.from_pretrained(self.get_model_name(), use_cache=True, config=compiler_conf, export=True, temperature=0).eval()
- self.tokenizer = AutoTokenizer.from_pretrained(self.get_model_name())
+ 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|>")]
 
  def send_prompt(self, prompt: str, add_to_history: bool = True) -> dict[str, str]:
- self.get_message_history().append({"role":self.get_model_roles().user_role, "content":prompt})
- input_ids = (self.tokenizer.apply_chat_template(self.get_message_history(), add_generation_prompt=True, return_tensors="pt")
+ print("prompt to be sent: " + prompt)
+ user_prompt = {"role": self.model_info.roles.user_role, "content": prompt}
+ if self.log_to_file:
+ with open(self.log_to_file, "a+") as log_file:
+ log_file.write(json.dumps(user_prompt, indent=2))
+ log_file.write("\n")
+ self.get_message_history().append(user_prompt)
+ input_ids = (self.tokenizer.apply_chat_template(self.get_message_history(), add_generation_prompt=True,
+ return_tensors="pt")
  .to(self.model.device))
  outputs = self.model.generate(input_ids, eos_token_id=self.terminators, do_sample=True, max_new_tokens=2000)
  generated_token_array = outputs[0][len(input_ids[0]):]
  generated_tokens = "".join(self.tokenizer.batch_decode(generated_token_array, skip_special_tokens=True))
- answer = {"role": self.get_model_roles().ai_role, "content":generated_tokens}
+ answer = {"role": self.get_model_info().roles.ai_role, "content": generated_tokens}
+ if self.log_to_file:
+ with open(self.log_to_file, "a+") as log_file:
+ log_file.write(json.dumps(answer, indent=2))
+ log_file.write("\n")
  if add_to_history:
  self.message_history.append(answer)
  else:
  self.message_history.pop()
- return answer
+ return answer

repository/ollama.py

@@ -1,29 +1,27 @@
 import ollama
 from ollama import Options
 
-from llm.llm import Model
-from repository.repository import Repository, ModelRoles
+from repository.repository_abc import Repository, Model
 
 
 class OllamaRepository(Repository):
- def __init__(self, model: Model, system_msg):
- self.model: Model = model
+ def __init__(self, model_info: Model, system_msg):
+ self.model_info: Model = model_info
  self.system_msg: str = system_msg
- self.message_history: list[dict[str, str]] = [{"role": self.model.roles.system_role, "content": system_msg}]
+ self.message_history: list[dict[str, str]] = [
+ {"role": self.model_info.roles.system_role, "content": system_msg}]
 
  def send_prompt(self, prompt: str, add_to_history: bool = True) -> dict[str, str]:
+ print("Prompt to be sent:" + prompt)
  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"]}
+ self.message_history.append({"role": self.model_info.roles.user_role, "content": prompt})
+ response = ollama.chat(self.model_info.name, self.message_history, options=options)
+ answer = {"role": self.model_info.roles.ai_role, "content": response["message"]["content"]}
  if add_to_history:
  self.message_history.append(answer)
  else:
  self.message_history.pop()
  return answer
 
- def get_model_name(self) -> str:
- return self.model.name
-
- def get_model_roles(self) -> ModelRoles:
- return self.model.roles
+ def get_model_info(self) -> Model:
+ return self.model_info

repository/repository.py

@@ -1,29 +1,15 @@
-import abc
+from pathlib import Path
 
+from repository.intel_npu import IntelNpuRepository
+from repository.ollama import OllamaRepository
+from repository.repository_abc import Model
 
-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 Repository(abc.ABC):
-
- def get_model_name(self) -> str:
- pass
-
- def get_model_roles(self) -> ModelRoles:
- pass
-
- def get_message_history(self) -> list[dict[str, str]]:
- pass
-
- def send_prompt(self, prompt: str, add_to_history: bool) -> dict[str, str]:
- pass
-
- def set_message_for_role(self, message:str, role: ModelRoles):
- pass
-
- def init(self):
- pass
+def get_repository(implementation: str, model: Model, system_msg: str = None, log_to_file: Path = None):
+ known_implementations = ["ollama", "intel_npu"]
+ if not implementation or implementation.lower() not in ["ollama", "intel_npu"]:
+ raise ValueError(f"Unknown implementation {implementation}. Known implementations: {known_implementations}")
+ if "ollama" == implementation:
+ return OllamaRepository(model, system_msg)
+ if "intel_npu" == implementation:
+ return IntelNpuRepository(model, system_msg, log_to_file)

repository/repository_abc.py

@@ -0,0 +1,35 @@
+import abc
+
+
+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, model_name: str, model_roles: ModelRoles):
+ self.name = model_name
+ self.roles = model_roles
+
+
+class Repository(abc.ABC):
+
+ def get_model_info(self) -> Model:
+ pass
+
+ def get_model_roles(self) -> ModelRoles:
+ pass
+
+ def get_message_history(self) -> list[dict[str, str]]:
+ pass
+
+ def send_prompt(self, prompt: str, add_to_history: bool) -> dict[str, str]:
+ pass
+
+ def set_message_for_role(self, message: str, role: str):
+ self.get_message_history().append({"role": role, "content": message})
+
+ def init(self):
+ pass

requirements.txt

@@ -1,3 +1,4 @@
 PyPDFForm
 ollama
+transformers
 intel-npu-acceleration-library

test.py

@@ -0,0 +1,56 @@
+llm_answer = """What is my full name?
+
+Enrico Rampazzo
+
+What is the nature of the work I need to do?
+
+A/C clean
+
+In which community is the work taking place?
+
+JBR
+
+In which building?
+
+Amwaj
+
+In which unit/apartment number?
+
+99999
+
+Am I the owner or the tenant?
+
+Owner
+
+In which date is the work taking place?
+
+17/09/2024
+
+In which date will the work finish?
+
+17/09/2024
+
+What is my contact number?
+
+989764
+
+What is the name of the contracting company?
+
+Breathe Maintenance
+
+What is the contact number of the contracting company?
+
+1234567
+
+What is the email of the contracting company?
+
+[email protected]
+
+What is my email?
+
+[email protected]
+"""
+from llm_manager.llm_parser import LlmParser
+
+llm_parser = LlmParser
+llm_parser.parse_verification_prompt_answers(llm_answer)