chore: reformat

chatbot/bot/client/lama_cpp_client.py

@@ -1,6 +1,6 @@
 import os
 from pathlib import Path
-from typing import Any, Iterator, Union
+from typing import Any, Iterator
 
 import requests
 from llama_cpp import CreateCompletionResponse, CreateCompletionStreamResponse, Llama
@@ -158,7 +158,7 @@ def stream_answer(self, prompt: str, max_new_tokens: int = 512) -> str:
 
     def start_answer_iterator_streamer(
         self, prompt: str, max_new_tokens: int = 512
-    ) -> Union[CreateCompletionResponse, Iterator[CreateCompletionStreamResponse]]:
+    ) -> CreateCompletionResponse | Iterator[CreateCompletionStreamResponse]:
         """
         Abstract method to start an answer iterator streamer for a given prompt.
 
@@ -181,7 +181,7 @@ def start_answer_iterator_streamer(
 
     async def async_start_answer_iterator_streamer(
         self, prompt: str, max_new_tokens: int = 512
-    ) -> Union[CreateCompletionResponse, Iterator[CreateCompletionStreamResponse]]:
+    ) -> CreateCompletionResponse | Iterator[CreateCompletionStreamResponse]:
         """
         This abstract method should be implemented to asynchronously start an answer iterator streamer,
         providing a flexible way to generate answers in a streaming fashion based on the given prompt.

chatbot/bot/conversation/conversation_retrieval.py

@@ -1,5 +1,5 @@
 from asyncio import get_event_loop
-from typing import Any, List, Tuple
+from typing import Any
 
 from entities.document import Document
 from helpers.log import get_logger
@@ -30,7 +30,7 @@ def __init__(self, llm: LamaCppClient) -> None:
         self.llm = llm
         self.chat_history = []
 
-    def get_chat_history(self) -> List[Tuple[str, str]]:
+    def get_chat_history(self) -> list[tuple[str, str]]:
         """
         Gets the chat history.
 
@@ -40,7 +40,7 @@ def get_chat_history(self) -> List[Tuple[str, str]]:
         """
         return self.chat_history
 
-    def update_chat_history(self, question: str, answer: str) -> List[Tuple[str, str]]:
+    def update_chat_history(self, question: str, answer: str) -> list[tuple[str, str]]:
         """
         Updates the chat history.
 
@@ -57,7 +57,7 @@ def update_chat_history(self, question: str, answer: str) -> List[Tuple[str, str
 
         return self.chat_history
 
-    def keep_chat_history_size(self, max_size: int = 2) -> List[Tuple[str, str]]:
+    def keep_chat_history_size(self, max_size: int = 2) -> list[tuple[str, str]]:
         """
         Keeps the list of chat history at the specified maximum size by popping out the oldest elements.
 
@@ -160,7 +160,7 @@ def answer(self, question: str, max_new_tokens: int = 512) -> Any:
     def context_aware_answer(
         ctx_synthesis_strategy: BaseSynthesisStrategy,
         question: str,
-        retrieved_contents: List[Document],
+        retrieved_contents: list[Document],
         max_new_tokens: int = 512,
     ):
         if isinstance(ctx_synthesis_strategy, AsyncTreeSummarizationStrategy):

chatbot/bot/conversation/ctx_strategy.py

@@ -1,6 +1,6 @@
 import asyncio
 from enum import Enum
-from typing import Any, List, Union
+from typing import Any
 
 import nest_asyncio
 from entities.document import Document
@@ -35,7 +35,7 @@ def __init__(self, llm: LamaCppClient) -> None:
         """
         self.llm = llm
 
-    def generate_response(self, retrieved_contents: List[Document], question: str, max_new_tokens: int = 512):
+    def generate_response(self, retrieved_contents: list[Document], question: str, max_new_tokens: int = 512):
         """
         Generate a response using the synthesis strategy.
 
@@ -61,8 +61,8 @@ def __init__(self, llm: LamaCppClient):
         super().__init__(llm)
 
     def generate_response(
-        self, retrieved_contents: List[Document], question: str, max_new_tokens: int = 512
-    ) -> Union[str, Any]:
+        self, retrieved_contents: list[Document], question: str, max_new_tokens: int = 512
+    ) -> str | Any:
         """
         Generate a response using create and refine strategy.
 
@@ -126,7 +126,7 @@ def __init__(self, llm: LamaCppClient):
         super().__init__(llm)
 
     def generate_response(
-        self, retrieved_contents: List[Document], question: str, max_new_tokens: int = 512, num_children: int = 2
+        self, retrieved_contents: list[Document], question: str, max_new_tokens: int = 512, num_children: int = 2
     ) -> Any:
         """
         Generate a response using hierarchical summarization strategy.
@@ -170,9 +170,9 @@ def generate_response(
 
     def combine_results(
         self,
-        texts: List[str],
+        texts: list[str],
         question: str,
-        cur_prompt_list: List[str],
+        cur_prompt_list: list[str],
         max_new_tokens: int = 512,
         num_children: int = 2,
     ) -> Any:
@@ -227,7 +227,7 @@ def __init__(self, llm: LamaCppClient):
 
     async def generate_response(
         self,
-        retrieved_contents: List[Document],
+        retrieved_contents: list[Document],
         question: str,
         max_new_tokens: int = 512,
         num_children: int = 2,
@@ -278,9 +278,9 @@ async def generate_response(
 
     async def combine_results(
         self,
-        texts: List[str],
+        texts: list[str],
         question: str,
-        cur_prompt_list: List[str],
+        cur_prompt_list: list[str],
         max_new_tokens: int = 512,
         num_children: int = 2,
     ):

chatbot/bot/memory/vector_database/chroma.py

@@ -92,8 +92,8 @@ def add_texts(
 
         Args:
             texts (Iterable[str]): Texts to add to the vectorstore.
-            metadatas (Optional[List[dict]], optional): Optional list of metadatas.
-            ids (Optional[List[str]], optional): Optional list of IDs.
+            metadatas (list[dict] | None): Optional list of metadatas.
+            ids (list[dict] | None): Optional list of IDs.
 
         Returns:
             List[str]: List of IDs of the added texts.
@@ -257,7 +257,7 @@ def similarity_search(self, query: str, k: int = 4, filter: dict[str, str] | Non
         Args:
             query (str): Query text to search for.
             k (int): Number of results to return. Defaults to 4.
-            filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.
+            filter (dict[str, str]|None): Filter by metadata. Defaults to None.
 
         Returns:
             List[Document]: List of documents most similar to the query text.
@@ -278,8 +278,8 @@ def similarity_search_with_score(
         Args:
             query (str): Query text to search for.
             k (int): Number of results to return. Defaults to 4.
-            filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.
-            where_document (Optional[Dict[str, str]]): Filter by document content. Defaults to None.
+            filter (dict[str, str]|None): Filter by metadata. Defaults to None.
+            where_document (dict[str, str]|None): Filter by document content. Defaults to None.
             **kwargs (Any): Additional keyword arguments.
 
         Returns:

chatbot/bot/model/model.py

@@ -1,9 +1,9 @@
 from abc import ABC
-from typing import Any, Dict, Optional
+from typing import Any
 
 
 class Model(ABC):
     url: str
     file_name: str
-    config: Dict[str, Any]
-    config_answer: Optional[Dict[str, Any]]
+    config: dict[str, Any]
+    config_answer: dict[str, Any] | None