streamlit ui and ondemand integration

app.py

@@ -1,3 +1,5 @@
+import os
+import uuid
 from pathlib import Path
 
 from llm_manager.llm_parser import LlmParser
@@ -7,6 +9,9 @@ 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
 
+import streamlit as st
+
+user_msg = "Please describe what you need to do. To get the best results try to answer all the following questions:"
 
 def check_for_missing_answers(parsed_questions: dict[int, str]):
     return [k for k in parsed_questions if parsed_questions[k] is None]
@@ -18,10 +23,9 @@ def ask_again(missing_questions: list[int], user_questions: list[str], parsed_qu
         parsed_questions[id_] = new_answer
 
 
-if __name__ == '__main__':
+def use_command_line():
     prompts_manager = PromptsManager()
-    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>")
+    user_prompt = input(f"{user_msg}\n{'\n'.join(prompts_manager.questions)}\n\n>")
 
     repository = get_repository("intel_npu", Model("meta-llama/Meta-Llama-3-8B-Instruct",
                                                    ModelRoles("system", "user", "assistant")),
@@ -40,3 +44,76 @@ if __name__ == '__main__':
 
     form_data = build_form_data_from_answers(answers, categories, f"{Path(__file__, "..", "signature.png")}")
     write_pdf_form(form_data, Path("signed_form1.pdf"))
+
+
+def update_answer (answers, missing_answer):
+    answers[missing_answer] = getattr(st.session_state, f"ma_{missing_answer}")
+
+
+def use_streamlit():
+    pm = PromptsManager()
+    help_ = f"{user_msg}\n\n{'\n'.join(pm.questions)}"
+    repository = get_repository("ondemand", Model("ondemand-gpt-3.5-turbo", ModelRoles("system", "user", "assistant")))
+    if not st.session_state.get("step"):
+        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
+            st.session_state["session_id"] = str(uuid.uuid4())
+            button = st.form_submit_button()
+
+        if button:
+            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"))
+            st.session_state["step"] = "parsing_answers"
+    if st.session_state.get("step") == "parsing_answers":
+        with st.status("initialising LLM"):
+            repository.init()
+        with st.status("waiting for LLM"):
+            repository.send_prompt(pm.ingest_user_answers(user_input))
+            answer = repository.send_prompt(pm.verify_user_input_prompt(user_input))
+        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"
+
+    if st.session_state.get("step") == "ask_again":
+        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:
+                st.session_state["step"] = "check_category"
+                for ma in st.session_state["missing_answers"]:
+                    st.session_state["answers"][ma] = st.session_state[ma]
+
+    if st.session_state.get("step") == "check_category":
+        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_filename = f"{st.session_state["session_id"]}_form.pdf"
+            st.session_state["form_filename"] = form_filename
+            form_data = build_form_data_from_answers(st.session_state["answers"], categories,
+                                                     st.session_state.get("signature"))
+            write_pdf_form(form_data, Path(form_filename))
+            st.session_state["step"] = "form_created"
+    if st.session_state.get("step") == "form_created":
+        with open(Path(st.session_state["form_filename"]), "rb") as form:
+            st.download_button("download form", form.read(), mime="application/pdf")
+        start_over_button = st.button("Start over")
+        if start_over_button:
+            del st.session_state["step"]
+            os.unlink(st.session_state["form_filename"])
+
+
+use_streamlit()
+#use_command_line()

prompts/prompts_manager.py

@@ -22,7 +22,7 @@ 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, if the answer is not present just answer null. "
+        return (f"Using only this information \n {user_prompt} \n answer the following questions, if the answer is not present or you are not sure about the 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}")

repository/intel_npu.py

@@ -42,7 +42,7 @@ class IntelNpuRepository(Repository):
         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)
+        outputs = self.model.generate(input_ids, eos_token_id=self.terminators, do_sample=True, max_new_tokens=2000, cache_position=None)
         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_info().roles.ai_role, "content": generated_tokens}

repository/ondemand.py

@@ -0,0 +1,39 @@
+import os
+from pathlib import Path
+
+import requests
+
+from repository.repository_abc import Repository, Model, ModelRoles
+
+
+class OndemandRepository(Repository):
+    session_url = "https://api.on-demand.io/chat/v1/sessions"
+    def __init__(self, model_info: Model, system_message: str = None, log_to_file:Path=None):
+        self.model_info = model_info
+        self.system_message = system_message
+        self.log_to_file = log_to_file
+        self.session_id = None
+
+    def init(self):
+        if not self.session_id:
+            headers = {"apiKey": os.getenv("API_KEY")}
+            session_body = {"pluginIds": [], "externalUserId": "virtualDAM"}
+            response = requests.post(self.session_url, headers=headers, json=session_body)
+            response_data = response.json()
+            self.session_id = response_data["data"]["id"]
+
+    def get_model_roles(self) -> ModelRoles:
+        return self.model_info.roles
+
+    def get_model_info(self) -> Model:
+        return self.model_info
+
+    def send_prompt(self, prompt: str, add_to_history: bool = None) -> dict[str, str]:
+        headers = {"apiKey": os.getenv("API_KEY")}
+        body = {'endpointId': 'predefined-openai-gpt3.5turbo', 'query': prompt, 'pluginIds': [], 'responseMode': 'sync'}
+        url = f'https://api.on-demand.io/chat/v1/sessions/{self.session_id}/query'
+        response = requests.post(url, headers=headers, json=body)
+        return {"content": response.json()["data"]["answer"]}
+
+    def get_message_history(self) -> list[dict[str, str]]:
+        return []

repository/repository.py

@@ -2,14 +2,33 @@ from pathlib import Path
 
 from repository.intel_npu import IntelNpuRepository
 from repository.ollama import OllamaRepository
+from repository.ondemand import OndemandRepository
 from repository.repository_abc import Model
+from repository.testing_repo import TestingRepository
 
 
 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"]:
+    if not implementation or implementation.lower() not in ["ollama", "intel_npu", "testing", "ondemand"]:
         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)
+    if "ondemand" == implementation:
+        return OndemandRepository(model, system_msg, log_to_file)
+    if "testing" == implementation:
+        return TestingRepository(prompts_answers=[
+            {
+                "role": "assistant",
+                "content": "OK"
+            },
+            {
+                "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"
+            },
+            {
+                "role":"assistant",
+                "content":"pest_control"
+            }
+        ])

repository/repository_abc.py

@@ -28,7 +28,7 @@ class Repository(abc.ABC):
     def send_prompt(self, prompt: str, add_to_history: bool) -> dict[str, str]:
         pass
 
-    def set_message_for_role(self, message: str, role: str):
+    def set_message_for_role(self, role: str, message: str):
         self.get_message_history().append({"role": role, "content": message})
 
     def init(self):

repository/testing_repo.py

@@ -0,0 +1,31 @@
+from typing import Tuple
+
+
+from repository.repository_abc import Repository, Model, ModelRoles
+
+
+class TestingRepository(Repository):
+
+    def __init__(self, prompts_answers: list[dict[str, str]], model_info:Model=None):
+        self.prompt_answers = prompts_answers
+        self.next_answer = iter(self.prompt_answers)
+        self.message_history = []
+        self.model_info = model_info or Model("fake_model",
+                                              ModelRoles("system", "user", "assistant"))
+    def init(self):
+        pass
+
+    def send_prompt(self, prompt: str, add_to_history: bool = True) -> dict[str, str]:
+        response = next(self.next_answer)
+        if add_to_history:
+            self.get_message_history().append(response)
+        return response
+
+    def get_message_history(self) -> list[dict[str, str]]:
+        return self.message_history
+
+    def get_model_info(self) -> Model:
+        return self.model_info
+
+    def get_model_roles(self) -> ModelRoles:
+        return self.model_info.roles

utils/parsing_utils.py

@@ -15,3 +15,6 @@ def find_and_parse_date(llm_answer: str) -> str | None:
 def find_and_parse_phone_number(llm_answer: str):
     return _find_and_parse(llm_answer, parse_phone_number_regex)
 
+
+def check_for_missing_answers(parsed_questions: dict[int, str]):
+    return [k for k in parsed_questions if parsed_questions[k] is None]