update the deployable ctm (#22)

ctm/configs/__init__.py

@@ -0,0 +1,5 @@
+from .ctm_config_base import BaseConsciousnessTuringMachineConfig
+
+__all__ = [
+    "BaseConsciousnessTuringMachineConfig",
+]

ctm/configs/ctm_config_base.py

@@ -1,44 +1,48 @@
 import json
+from typing import Any, Dict, Optional
 
 
-class BaseConsciousnessTuringMachineConfig(object):
-    # Initialize with default values or those passed to the constructor
+class BaseConsciousnessTuringMachineConfig:
     def __init__(
         self,
-        ctm_name=None,
-        max_iter_num=3,
-        output_threshold=0.5,
-        groups_of_processors={},
-        supervisor="gpt4_supervisor",
-        **kwargs,
-    ):
-        self.ctm_name = ctm_name
-        self.max_iter_num = max_iter_num
-        self.output_threshold = output_threshold
-        self.groups_of_processors = groups_of_processors
-        self.supervisor = supervisor
-        # This allows for handling additional, possibly unknown configuration parameters
+        ctm_name: Optional[str] = None,
+        max_iter_num: int = 3,
+        output_threshold: float = 0.5,
+        groups_of_processors: Dict[
+            str, Any
+        ] = {},  # Better to avoid mutable default arguments
+        supervisor: str = "gpt4_supervisor",
+        **kwargs: Any,
+    ) -> None:
+        self.ctm_name: Optional[str] = ctm_name
+        self.max_iter_num: int = max_iter_num
+        self.output_threshold: float = output_threshold
+        self.groups_of_processors: Dict[str, Any] = groups_of_processors
+        self.supervisor: str = supervisor
+        # Handle additional, possibly unknown configuration parameters
         for key, value in kwargs.items():
             setattr(self, key, value)
 
-    def to_json_string(self):
+    def to_json_string(self) -> str:
         """Serializes this instance to a JSON string."""
         return json.dumps(self.__dict__, indent=2) + "\n"
 
     @classmethod
-    def from_json_file(cls, json_file):
+    def from_json_file(
+        cls, json_file: str
+    ) -> "BaseConsciousnessTuringMachineConfig":
         """Creates an instance from a JSON file."""
         with open(json_file, "r", encoding="utf-8") as reader:
             text = reader.read()
         return cls(**json.loads(text))
 
     @classmethod
-    def from_ctm(cls, ctm_name):
+    def from_ctm(cls, ctm_name: str) -> "BaseConsciousnessTuringMachineConfig":
         """
         Simulate fetching a model configuration from a ctm model repository.
         This example assumes the configuration is already downloaded and saved locally.
         """
         # This path would be generated dynamically based on `model_name_or_path`
         # For simplicity, we're directly using it as a path to a local file
-        config_file = f"../ctm/configs/{ctm_name}_config.json"
+        config_file = f"../ctm_conf/{ctm_name}_config.json"
         return cls.from_json_file(config_file)

ctm/configs/sarcasm_ctm_config.json

@@ -1,18 +0,0 @@
-{
-    "ctm_name": "sarcasm_ctm",
-    "max_iter_num": 3,
-    "output_threshold": 0.5,
-    "groups_of_processors": {
-        "group_1": [
-            "gpt4v_scene_location_processor",
-            "gpt4v_cloth_fashion_processor"
-        ],
-        "group_2": [
-            "gpt4v_posture_processor"
-        ],
-        "group_3": [
-            "gpt4v_ocr_processor"
-        ]
-    },
-    "supervisor": "gpt4_supervisor"
-}

ctm/ctms/ctm_base.py

@@ -1,33 +1,41 @@
 import concurrent.futures
 from collections import defaultdict
+from typing import Any, Dict, List, Optional, Set, Tuple
 
 from sklearn.feature_extraction.text import TfidfVectorizer
 from sklearn.metrics.pairwise import cosine_similarity
 
-from ctm.configs.ctm_config_base import (
-    BaseConsciousnessTuringMachineConfig,
-)
-from ctm.processors.processor_base import BaseProcessor
-from ctm.supervisors.supervisor_base import BaseSupervisor
+from ctm.configs import BaseConsciousnessTuringMachineConfig
+from ctm.processors import BaseProcessor
+from ctm.supervisors import BaseSupervisor
 
 
 class BaseConsciousnessTuringMachine(object):
-    def __call__(self, *args, **kwargs):
-        return self.forward(*args, **kwargs)
-
-    def __init__(self, ctm_name=None, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+    def __init__(self, ctm_name: Optional[str] = None) -> None:
+        super().__init__()
         if ctm_name:
             self.config = BaseConsciousnessTuringMachineConfig.from_ctm(
                 ctm_name
             )
         else:
             self.config = BaseConsciousnessTuringMachineConfig()
-        self.processor_list = []
-        self.processor_group_map = defaultdict(list)
+        self.processor_list: List[Dict[str, Any]] = []
+        self.processor_group_map: Dict[str, str] = defaultdict(str)
         self.load_ctm()
 
-    def add_processor(self, processor_name, group_name=None):
+    def __call__(
+        self,
+        query: str,
+        text: Optional[str] = None,
+        image: Optional[Any] = None,
+        audio: Optional[Any] = None,
+        video_frames: Optional[Any] = None,
+    ) -> Tuple[str, float]:
+        return self.forward(query, text, image, audio, video_frames)
+
+    def add_processor(
+        self, processor_name: str, group_name: Optional[str] = 'default_group'
+    ) -> None:
         processor_instance = BaseProcessor(processor_name)
         self.processor_list.append(
             {
@@ -38,37 +46,52 @@ class BaseConsciousnessTuringMachine(object):
         if group_name:
             self.processor_group_map[processor_name] = group_name
 
-    def add_supervisor(self, supervisor_name):
+    def add_supervisor(self, supervisor_name: str) -> None:
         supervisor_instance = BaseSupervisor(supervisor_name)
-        self.supervisor = {
+        self.supervisor: Dict[str, Any] = {
             "supervisor_name": supervisor_name,
             "supervisor_instance": supervisor_instance,
         }
 
     @staticmethod
     def ask_processor(
-        processor, question, context, image_path, audio_path, video_path
-    ):
+        processor: Dict[str, Any],
+        query: str,
+        text: Optional[str] = None,
+        image: Optional[Any] = None,
+        audio: Optional[Any] = None,
+        video_frames: Optional[Any] = None,
+    ) -> Dict[str, Any]:
         processor_instance = processor["processor_instance"]
         processor_name = processor["processor_name"]
+        print(processor_name)
         gist, score = processor_instance.ask(
-            question, context, image_path, audio_path, video_path
+            query=query,
+            text=text,
+            image=image,
+            audio=audio,
+            video_frames=video_frames,
         )
         return {"name": processor_name, "gist": gist, "score": score}
 
     def ask_processors(
-        self, question, context, image_path, audio_path, video_path
-    ):
+        self,
+        query: str,
+        text: Optional[str] = None,
+        image: Optional[Any] = None,
+        audio: Optional[Any] = None,
+        video_frames: Optional[Any] = None,
+    ) -> Dict[str, Dict[str, Any]]:
         with concurrent.futures.ThreadPoolExecutor() as executor:
             futures = [
                 executor.submit(
                     self.ask_processor,
                     processor,
-                    question,
-                    context,
-                    image_path,
-                    audio_path,
-                    video_path,
+                    query,
+                    text,
+                    image,
+                    audio,
+                    video_frames,
                 )
                 for processor in self.processor_list
             ]
@@ -77,7 +100,7 @@ class BaseConsciousnessTuringMachine(object):
                 for future in concurrent.futures.as_completed(futures)
             ]
 
-        output = {}
+        output: Dict[str, Dict[str, Any]] = {}
         for result in results:
             output[result["name"]] = {
                 "gist": result["gist"],
@@ -87,49 +110,68 @@ class BaseConsciousnessTuringMachine(object):
         assert len(output) == len(self.processor_list)
         return output
 
-    def uptree_competition(self, processor_output):
+    def uptree_competition(
+        self, processor_output: Dict[str, Dict[str, Any]]
+    ) -> Dict[str, Any]:
         # Unpack processor outputs into lists for easier processing
-        gists, scores, names = [], [], []
+        gists: List[str] = []
+        scores: List[float] = []
+        names: List[str] = []
+
         for name, info in processor_output.items():
             gists.append(info["gist"])
             scores.append(info["score"])
             names.append(name)
 
         # Determine the unique group for each processor
-        unique_groups = set(self.processor_group_map.values())
+        unique_groups: Set[str] = set(self.processor_group_map.values())
 
         # Prepare to track the best processor by group
-        best_processor_by_group = {
-            group: (None, -1) for group in unique_groups
-        }  # (processor_name, score)
+        best_processor_by_group: Dict[str, Tuple[Optional[str], float]] = {
+            group: (
+                None,
+                float("-inf"),
+            )  # Use negative infinity as the initial lowest score
+            for group in unique_groups
+        }
 
         # Iterate through processors to find the best in each group
         for name, score in zip(names, scores):
-            group = self.processor_group_map[name]
+            group = self.processor_group_map.get(name, "")
             if score > best_processor_by_group[group][1]:
                 best_processor_by_group[group] = (name, score)
 
         # Select the overall best across groups
-        best_overall = max(
+        best_overall: Tuple[Optional[str], float] = max(
             best_processor_by_group.values(), key=lambda x: x[1]
         )
-        best_name = best_overall[0]
-        index = names.index(best_name)
+        best_name: Optional[str] = best_overall[0]
+
+        if best_name is None:
+            raise ValueError(
+                "No valid processor found."
+            )  # Ensure best_name is not None
 
-        winning_info = {
+        index: int = names.index(
+            best_name
+        )  # Now best_name is guaranteed to be not None
+
+        winning_info: Dict[str, Any] = {
             "name": best_name,
             "gist": gists[index],
             "score": scores[index],
         }
         return winning_info
 
-    def ask_supervisor(self, question, processor_info):
+    def ask_supervisor(
+        self, query: str, processor_info: Dict[str, Any]
+    ) -> Tuple[str, float]:
         final_answer, score = self.supervisor["supervisor_instance"].ask(
-            question, processor_info["gist"]
+            query, processor_info["gist"]
         )
         return final_answer, score
 
-    def downtree_broadcast(self, winning_output):
+    def downtree_broadcast(self, winning_output: Dict[str, str]) -> None:
         winning_processor_name = winning_output["name"]
         winning_processor_gist = winning_output["gist"]
         for processor in self.processor_list:
@@ -139,14 +181,16 @@ class BaseConsciousnessTuringMachine(object):
                 )
         return
 
-    def calc_processor_sim(self, processor_output):
+    def calc_processor_sim(
+        self, processor_output: Dict[str, Dict[str, str]]
+    ) -> Any:
         processor_gists = [info["gist"] for info in processor_output.values()]
         tfidf_vectorizer = TfidfVectorizer()
         tfidf_matrix = tfidf_vectorizer.fit_transform(processor_gists)
         cosine_sim = cosine_similarity(tfidf_matrix, tfidf_matrix)
         return cosine_sim
 
-    def link_form(self, processor_output):
+    def link_form(self, processor_output: Dict[str, Dict[str, str]]) -> None:
         sim = self.calc_processor_sim(processor_output)
         print(sim)
         # iterate on each sim pair
@@ -184,34 +228,33 @@ class BaseConsciousnessTuringMachine(object):
                     self.processor_group_map[processor2_name] = group_name
         return
 
-    def processor_fuse(self, infos, scores):
+    def processor_fuse(
+        self, infos: List[str], scores: List[float]
+    ) -> Tuple[List[str], List[float]]:
         return infos, scores
 
     def forward(
         self,
-        question=None,
-        context=None,
-        image_path=None,
-        audio_path=None,
-        video_path=None,
-    ):
+        query: str,
+        text: Optional[str] = None,
+        image: Optional[Any] = None,
+        audio: Optional[Any] = None,
+        video_frames: Optional[Any] = None,
+    ) -> Tuple[str, float]:
         answer_threshold = 0.5
         max_iter = 3
 
         for i in range(max_iter):
             print("start the {}-th iteration".format(i + 1))
             processor_output = self.ask_processors(
-                question=question,
-                context=context,
-                image_path=image_path,
-                audio_path=audio_path,
-                video_path=video_path,
+                query=query,
+                text=text,
+                image=image,
+                audio=audio,
+                video_frames=video_frames,
             )
-            import pdb
-
-            pdb.set_trace()
             winning_output = self.uptree_competition(processor_output)
-            answer, score = self.ask_supervisor(question, winning_output)
+            answer, score = self.ask_supervisor(query, winning_output)
             if score > answer_threshold:
                 break
             else:
@@ -219,7 +262,7 @@ class BaseConsciousnessTuringMachine(object):
                 self.link_form(processor_output)
         return answer, score
 
-    def load_ctm(self):
+    def load_ctm(self) -> None:
         for (
             group_name,
             processor_list,

ctm/messengers/messenger_bart_text_summ.py

@@ -1,22 +1,35 @@
-from typing import Dict, List, Union
+from typing import Any, Dict, List, Optional, TypeVar, Union
 
-from ctm.messengers.messenger_base import BaseMessenger
+from .messenger_base import BaseMessenger
 
+T = TypeVar("T", bound="BaseMessenger")
 
-@BaseMessenger.register_messenger("bart_text_summ_messenger")  # type: ignore[no-untyped-call] # FIX ME
+
+@BaseMessenger.register_messenger("bart_text_summ_messenger")
 class BartTextSummarizationMessenger(BaseMessenger):
-    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+    def __init__(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
         self.init_messenger(role, content)
 
-    def init_messenger(  # type: ignore[no-untyped-def] # FIX ME
-        self, role: str = None, content: Union[str, Dict, List] = None  # type: ignore[assignment, type-arg] # FIX ME
-    ):
-        self.messages = ""
-        if content and role:
-            self.update_messages(role, content)  # type: ignore[attr-defined] # FIX ME
+    def init_messenger(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+    ) -> None:
+        self.messages: str = ""
+        if role and content:
+            self.update_message(role, content)
 
-    def update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
-        self.messages += content  # type: ignore[operator] # FIX ME
+    def update_message(
+        self, role: str, content: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
+        self.messages += content
 
-    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
-        return 1 if len(self.messages) > 0 else 0
+    def check_iter_round_num(self) -> int:
+        return len(self.messages)

ctm/messengers/messenger_base.py

@@ -1,18 +1,39 @@
-from typing import Dict, List, Union
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Type,
+    TypeVar,
+    Union,
+)
 
+# This TypeVar is used for methods that might need to return or work with instances of subclasses of BaseMessenger.
+T = TypeVar("T")
 
-class BaseMessenger(object):
-    _messenger_registry = {}  # type: ignore[var-annotated] # FIX ME
+
+class BaseMessenger:
+    _messenger_registry: Dict[str, Type["BaseMessenger"]] = {}
 
     @classmethod
-    def register_messenger(cls, messenger_name):  # type: ignore[no-untyped-def] # FIX ME
-        def decorator(subclass):  # type: ignore[no-untyped-def] # FIX ME
+    def register_messenger(
+        cls, messenger_name: str
+    ) -> Callable[[Type["BaseMessenger"]], Type["BaseMessenger"]]:
+        def decorator(
+            subclass: Type["BaseMessenger"],
+        ) -> Type["BaseMessenger"]:
             cls._messenger_registry[messenger_name] = subclass
             return subclass
 
         return decorator
 
-    def __new__(cls, messenger_name, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+    def __new__(
+        cls: Type["BaseMessenger"],
+        messenger_name: str,
+        *args: Any,
+        **kwargs: Any,
+    ) -> "BaseMessenger":
         if messenger_name not in cls._messenger_registry:
             raise ValueError(
                 f"No messenger registered with name '{messenger_name}'"
@@ -21,43 +42,63 @@ class BaseMessenger(object):
             cls._messenger_registry[messenger_name]
         )
 
-    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+    def __init__(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+    ) -> None:
         self.init_messenger(role, content)
 
-    def init_messenger(  # type: ignore[no-untyped-def] # FIX ME
-        self, role: str = None, content: Union[str, Dict, List] = None  # type: ignore[assignment, type-arg] # FIX ME
-    ):
-        pass
+    def init_messenger(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+    ) -> None:
+        raise NotImplementedError(
+            "The 'init_messenger' method must be implemented in derived classes."
+        )
 
-    def update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
-        pass
+    def update_message(
+        self, role: str, content: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
+        self.messages.append({"role": role, "content": content})
 
-    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
-        pass
+    def check_iter_round_num(self) -> int:
+        return len(self.messages)
 
-    def add_system_message(self, message: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+    def add_system_message(
+        self, message: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
         self.update_message("system", message)
 
-    def add_assistant_message(self, message: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+    def add_assistant_message(
+        self, message: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
         self.update_message("assistant", message)
 
-    def add_user_message(self, message: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+    def add_user_message(
+        self, message: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
         self.update_message("user", message)
 
-    def add_user_image(self, image_base64: str):  # type: ignore[no-untyped-def] # FIX ME
-        self.add_message(  # type: ignore[attr-defined] # FIX ME
+    def add_user_image(self, image_base64: str) -> None:
+        self.update_message(
             "user",
             {
-                "type": "image_url",
+                "type": "image",
                 "image_url": f"data:image/jpeg;base64,{image_base64}",
             },
         )
 
-    def add_feedback(self, feedback: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
-        self.add_message("system", feedback)  # type: ignore[attr-defined] # FIX ME
+    def add_feedback(
+        self, feedback: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
+        self.update_message("system", feedback)
 
-    def clear(self):  # type: ignore[no-untyped-def] # FIX ME
-        self.messages.clear()  # type: ignore[attr-defined] # FIX ME
+    def clear(self) -> None:
+        self.messages.clear()
 
-    def get_messages(self):  # type: ignore[no-untyped-def] # FIX ME
-        return self.messages  # type: ignore[attr-defined] # FIX ME
+    def get_messages(
+        self,
+    ) -> List[Dict[str, Union[str, Dict[str, Any], List[Any]]]]:
+        return self.messages

ctm/messengers/messenger_gpt4.py

@@ -1,22 +1,39 @@
-from typing import Dict, List, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 
-from ctm.messengers.messenger_base import BaseMessenger
+from .messenger_base import BaseMessenger
 
 
-@BaseMessenger.register_messenger("gpt4_messenger")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming BaseMessenger has a correctly typed decorator:
+@BaseMessenger.register_messenger("gpt4_messenger")
 class GPT4Messenger(BaseMessenger):
-    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+    def __init__(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+        *args: Any,
+        **kwargs: Any
+    ) -> None:
+        super().__init__(*args, **kwargs)
         self.init_messenger(role, content)
 
-    def init_messenger(  # type: ignore[no-untyped-def] # FIX ME
-        self, role: str = None, content: Union[str, Dict, List] = None  # type: ignore[assignment, type-arg] # FIX ME
-    ):
-        self.messages = []  # type: ignore[var-annotated] # FIX ME
-        if content and role:
-            self.update_messages(role, content)  # type: ignore[attr-defined] # FIX ME
+    def init_messenger(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+    ) -> None:
+        # Define messages as a list of dictionaries with specific types
+        self.messages: List[
+            Dict[str, Union[str, Dict[str, Any], List[Any]]]
+        ] = []
+        if role is not None and content is not None:
+            self.update_message(role, content)
 
-    def update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+    def update_message(
+        self, role: str, content: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
+        # Append a new message to the list with a specified structure
         self.messages.append({"role": role, "content": content})
 
-    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
+    def check_iter_round_num(self) -> int:
+        # Return the number of iterations, which is the length of the messages list
         return len(self.messages)

ctm/messengers/messenger_gpt4v.py

@@ -1,22 +1,41 @@
-from typing import Dict, List, Union
+from typing import Any, Dict, List, Optional, TypeVar, Union
 
-from ctm.messengers.messenger_base import BaseMessenger
+from .messenger_base import BaseMessenger
 
+T = TypeVar("T", bound="BaseMessenger")
 
-@BaseMessenger.register_messenger("gpt4v_messenger")  # type: ignore[no-untyped-call] # FIX ME
+
+# If the BaseMessenger has a register_messenger method that is not typed to accept a generic class,
+# you might need to define it properly in BaseMessenger or ensure that the typing is correct there.
+@BaseMessenger.register_messenger("gpt4v_messenger")
 class GPT4VMessenger(BaseMessenger):
-    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+    def __init__(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
         self.init_messenger(role, content)
 
-    def init_messenger(  # type: ignore[no-untyped-def] # FIX ME
-        self, role: str = None, content: Union[str, Dict, List] = None  # type: ignore[assignment, type-arg] # FIX ME
-    ):
-        self.messages = []  # type: ignore[var-annotated] # FIX ME
-        if content and role:
-            self.update_messages(role, content)  # type: ignore[attr-defined] # FIX ME
+    def init_messenger(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+    ) -> None:
+        self.messages: List[
+            Dict[str, Union[str, Dict[str, Any], List[Any]]]
+        ] = []
+        if role is not None and content is not None:
+            self.update_message(role, content)
 
-    def update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+    def update_message(
+        self, role: str, content: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
+        # Ensuring that 'messages' is defined and typed properly in the base class
         self.messages.append({"role": role, "content": content})
 
-    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
+    def check_iter_round_num(self) -> int:
+        # Count the number of entries in the messages list
         return len(self.messages)

ctm/messengers/messenger_roberta_text_sentiment.py

@@ -1,23 +1,34 @@
-from typing import Dict, List, Union
+from typing import Any, Dict, List, Optional, TypeVar, Union
 
-from ctm.messengers.messenger_base import BaseMessenger
+from .messenger_base import BaseMessenger
 
+T = TypeVar("T", bound="BaseMessenger")
 
-@BaseMessenger.register_messenger("roberta_text_sentiment_messenger")  # type: ignore[no-untyped-call] # FIX ME
+
+@BaseMessenger.register_messenger("roberta_text_sentiment_messenger")
 class RobertaTextSentimentMessenger(BaseMessenger):
-    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_messenger(role, content)
+    def __init__(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(role, content, *args, **kwargs)
 
-    def init_messenger(  # type: ignore[no-untyped-def] # FIX ME
-        self, role: str = None, content: Union[str, Dict, List] = None  # type: ignore[assignment, type-arg] # FIX ME
-    ):
-        self.messages = ""
-        if content and role:
-            self.update_messages(role, content)  # type: ignore[attr-defined] # FIX ME
+    def init_messenger(
+        self,
+        role: Optional[str] = None,
+        content: Optional[Union[str, Dict[str, Any], List[Any]]] = None,
+    ) -> None:
+        self.messages: str = ""
+        if role and content:
+            self.update_message(role, content)
 
-    def update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
-        # should replace with updated message
-        self.messages = content  # type: ignore[assignment] # FIX ME
+    def update_message(
+        self, role: str, content: Union[str, Dict[str, Any], List[Any]]
+    ) -> None:
+        self.messages += content
 
-    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
-        return 1 if len(self.messages) > 0 else 0
+    def check_iter_round_num(self) -> int:
+        return len(self.messages)

ctm/processors/processor_bart_text_summary.py

@@ -1,55 +1,61 @@
+import json
 import os
+from typing import Any, Dict, Optional
 
-from huggingface_hub.inference_api import (
-    InferenceApi,  # type: ignore[import] # FIX ME
-)
+from huggingface_hub import InferenceClient
 
 from ctm.messengers.messenger_base import BaseMessenger
 from ctm.processors.processor_base import BaseProcessor
 
 
-@BaseProcessor.register_processor("bart_text_summary_processor")  # type: ignore[no-untyped-call] # FIX ME
+@BaseProcessor.register_processor("bart_text_summary_processor")
 class BartTextSummaryProcessor(BaseProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(
+            *args, **kwargs
+        )  # Ensure base class is properly initialized
 
-    def init_processor(self):  # type: ignore[no-untyped-def] # FIX ME
-        self.model = InferenceApi(
-            token=os.environ["HF_TOKEN"], repo_id="facebook/bart-large-cnn"
-        )
-        self.messenger = BaseMessenger("bart_text_summ_messenger")  # type: ignore[no-untyped-call] # FIX ME
-        return
+    def init_executor(self) -> None:
+        hf_token = os.getenv("HF_TOKEN")
+        if not hf_token:
+            raise ValueError("HF_TOKEN environment variable is not set")
+        self.executor = InferenceClient(token=hf_token)
 
-    def update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+    def init_messenger(self) -> None:
+        self.messenger = BaseMessenger("bart_text_summ_messenger")
+
+    def init_task_info(self) -> None:
+        pass
+
+    def update_info(self, feedback: str) -> None:
         self.messenger.add_assistant_message(feedback)
 
-    def ask_info(  # type: ignore[override] # FIX ME
-        self,
-        query: str,
-        context: str = None,  # type: ignore[assignment] # FIX ME
-        image_path: str = None,  # type: ignore[assignment] # FIX ME
-        audio_path: str = None,  # type: ignore[assignment] # FIX ME
-        video_path: str = None,  # type: ignore[assignment] # FIX ME
-    ) -> str:
-        if self.messenger.check_iter_round_num() == 0:  # type: ignore[no-untyped-call] # FIX ME
-            self.messenger.add_user_message(context)
+    def ask_info(
+        self, text: Optional[str] = None, *args: Any, **kwargs: Any
+    ) -> str | Any:
+        if text is None:
+            raise ValueError("Context must not be None")
+        if self.messenger.check_iter_round_num() == 0:
+            self.messenger.add_user_message(text)
 
-        response = self.model(self.messenger.get_messages())  # type: ignore[no-untyped-call] # FIX ME
-        summary = response[0]["summary_text"]
-        return summary  # type: ignore[no-any-return] # FIX ME
+        response: Dict[str, Any] = json.loads(
+            self.executor.post(
+                json={"inputs": self.messenger.get_messages()},
+                model="facebook/bart-large-cnn",
+            )
+        )[0]
+        return response["summary_text"]
 
 
 if __name__ == "__main__":
-    processor = BaseProcessor("bart_text_summ_processor")  # type: ignore[no-untyped-call] # FIX ME
+    processor = BartTextSummaryProcessor()
     image_path = "../ctmai-test1.png"
-    text: str = (
+    text = (
         "In a shocking turn of events, Hugging Face has released a new version of Transformers "
         "that brings several enhancements and bug fixes. Users are thrilled with the improvements "
         "and are finding the new version to be significantly better than the previous one. "
         "The Hugging Face team is thankful for the community's support and continues to work "
         "towards making the library the best it can be."
     )
-    summary: str = processor.ask_info(  # type: ignore[no-untyped-call] # FIX ME
-        query=None, context=text, image_path=image_path
-    )
+    summary = processor.ask_info(context=text, image_path=image_path)
     print(summary)

ctm/processors/processor_base.py

@@ -1,20 +1,28 @@
-import base64
+from typing import Any, Callable, Dict, Optional, Tuple, Type
 
-from ctm.utils.exponential_backoff import exponential_backoff
+from openai import OpenAI
+
+from ..utils.decorator import score_exponential_backoff
 
 
 class BaseProcessor(object):
-    _processor_registry = {}  # type: ignore[var-annotated] # FIX ME
+    _processor_registry: Dict[str, Type["BaseProcessor"]] = {}
 
     @classmethod
-    def register_processor(cls, processor_name):  # type: ignore[no-untyped-def] # FIX ME
-        def decorator(subclass):  # type: ignore[no-untyped-def] # FIX ME
+    def register_processor(
+        cls, processor_name: str
+    ) -> Callable[[Type["BaseProcessor"]], Type["BaseProcessor"]]:
+        def decorator(
+            subclass: Type["BaseProcessor"],
+        ) -> Type["BaseProcessor"]:
             cls._processor_registry[processor_name] = subclass
             return subclass
 
         return decorator
 
-    def __new__(cls, processor_name, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+    def __new__(
+        cls, processor_name: str, *args: Any, **kwargs: Any
+    ) -> "BaseProcessor":
         if processor_name not in cls._processor_registry:
             raise ValueError(
                 f"No processor registered with name '{processor_name}'"
@@ -23,85 +31,110 @@ class BaseProcessor(object):
             cls._processor_registry[processor_name]
         )
 
-    def set_model(self):  # type: ignore[no-untyped-def] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.init_scorer()
+        self.init_executor()
+        self.init_messenger()
+        self.init_task_info()
+
+    def init_executor(self) -> None:
         raise NotImplementedError(
-            "The 'set_model' method must be implemented in derived classes."
+            "The 'init_executor' method must be implemented in derived classes."
         )
 
-    @staticmethod
-    def process_image(image_path):  # type: ignore[no-untyped-def] # FIX ME
-        with open(image_path, "rb") as image_file:
-            return base64.b64encode(image_file.read()).decode("utf-8")
-
-    @staticmethod
-    def process_audio(audio_path):  # type: ignore[no-untyped-def] # FIX ME
-        return None
+    def init_messenger(self) -> None:
+        raise NotImplementedError(
+            "The 'init_messenger' method must be implemented in derived classes."
+        )
 
-    @staticmethod
-    def process_video(video_path):  # type: ignore[no-untyped-def] # FIX ME
-        return None
+    def init_task_info(self) -> None:
+        raise NotImplementedError(
+            "The 'init_task_info' method must be implemented in derived classes."
+        )
 
-    def ask(self, query, context, image_path, audio_path, video_path):  # type: ignore[no-untyped-def] # FIX ME
-        gist = self.ask_info(  # type: ignore[no-untyped-call] # FIX ME
-            query, context, image_path, audio_path, video_path
+    def init_scorer(self) -> None:
+        self.scorer = OpenAI()
+
+    def ask(
+        self, query: str, text: str, image: str, audio: str, video_frames: str
+    ) -> Tuple[str, float]:
+        gist = self.ask_info(
+            query=query,
+            text=text,
+            image=image,
+            audio=audio,
+            video_frames=video_frames,
         )
-        score = self.ask_score(query, gist, verbose=True)  # type: ignore[no-untyped-call] # FIX ME
+        score = self.ask_score(query, gist, verbose=True)
         return gist, score
 
-    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[misc, no-untyped-call] # FIX ME
+    @score_exponential_backoff(retries=5, base_wait_time=1)
     def ask_relevance(self, query: str, gist: str) -> float:
-        response = self.model.chat.completions.create(  # type: ignore[attr-defined] # FIX ME
+        response = self.scorer.chat.completions.create(
             model="gpt-4-0125-preview",
             messages=[
                 {
                     "role": "user",
-                    "content": "How related is the information ({}) with the query ({})? Answer with a number from 0 to 5 and do not add any other thing.".format(
-                        gist, query
-                    ),
-                },
+                    "content": f"How related is the information ({gist}) with the query ({query})? Answer with a number from 0 to 5 and do not add any other thing.",
+                }
             ],
             max_tokens=50,
         )
-        score = int(response.choices[0].message.content.strip()) / 5
+        score = (
+            float(response.choices[0].message.content.strip()) / 5
+            if response.choices[0].message.content
+            else 0.0
+        )
         return score
 
-    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[misc, no-untyped-call] # FIX ME
+    @score_exponential_backoff(retries=5, base_wait_time=1)
     def ask_confidence(self, query: str, gist: str) -> float:
-        response = self.model.chat.completions.create(  # type: ignore[attr-defined] # FIX ME
+        response = self.scorer.chat.completions.create(
             model="gpt-4-0125-preview",
             messages=[
                 {
                     "role": "user",
-                    "content": "How confidence do you think the information ({}) is a mustk? Answer with a number from 0 to 5 and do not add any other thing.".format(  # type: ignore[str-format] # FIX ME
-                        gist, query
-                    ),
-                },
+                    "content": f"How confident do you think the information ({gist}) is a must-know? Answer with a number from 0 to 5 and do not add any other thing.",
+                }
             ],
             max_tokens=50,
         )
-        score = int(response.choices[0].message.content.strip()) / 5
+        score = (
+            float(response.choices[0].message.content.strip()) / 5
+            if response.choices[0].message.content
+            else 0.0
+        )
         return score
 
-    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[misc, no-untyped-call] # FIX ME
+    @score_exponential_backoff(retries=5, base_wait_time=1)
     def ask_surprise(
-        self, query: str, gist: str, history_gists: str = None  # type: ignore[assignment] # FIX ME
+        self, query: str, gist: str, history_gists: Optional[str] = None
     ) -> float:
-        response = self.model.chat.completions.create(  # type: ignore[attr-defined] # FIX ME
+        response = self.scorer.chat.completions.create(
             model="gpt-4-0125-preview",
             messages=[
                 {
                     "role": "user",
-                    "content": "How surprise do you think the information ({}) is as an output of the processor? Answer with a number from 0 to 5 and do not add any other thing.".format(  # type: ignore[str-format] # FIX ME
-                        gist, query
-                    ),
-                },
+                    "content": f"How surprising do you think the information ({gist}) is as an output of the processor? Answer with a number from 0 to 5 and do not add any other thing.",
+                }
             ],
             max_tokens=50,
         )
-        score = int(response.choices[0].message.content.strip()) / 5
+        score = (
+            float(response.choices[0].message.content.strip()) / 5
+            if response.choices[0].message.content
+            else 0.0
+        )
         return score
 
-    def ask_score(self, query, gist, verbose=False, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+    def ask_score(
+        self,
+        query: str,
+        gist: str,
+        verbose: bool = False,
+        *args: Any,
+        **kwargs: Any,
+    ) -> float:
         relevance = self.ask_relevance(query, gist, *args, **kwargs)
         confidence = self.ask_confidence(query, gist, *args, **kwargs)
         surprise = self.ask_surprise(query, gist, *args, **kwargs)
@@ -109,9 +142,11 @@ class BaseProcessor(object):
             print(
                 f"Relevance: {relevance}, Confidence: {confidence}, Surprise: {surprise}"
             )
-        return relevance * confidence * surprise
 
-    def ask_info(self, query, image_path, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        final_score = relevance * confidence * surprise
+        return final_score
+
+    def ask_info(self, *args: Any, **kwargs: Any) -> str:
         raise NotImplementedError(
-            "The 'ask_information' method must be implemented in derived classes."
+            "The 'ask_info' method must be implemented in derived classes."
         )

ctm/processors/processor_gpt4.py

@@ -1,59 +1,67 @@
+from typing import Any, Dict, Optional
+
 from openai import OpenAI
 
 from ctm.messengers.messenger_base import BaseMessenger
 from ctm.processors.processor_base import BaseProcessor
-from ctm.utils.exponential_backoff import exponential_backoff
+from ctm.utils.decorator import info_exponential_backoff
 
 
-@BaseProcessor.register_processor("gpt4_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming the `register_processor` method has been updated to be properly typed:
+@BaseProcessor.register_processor("gpt4_processor")
 class GPT4Processor(BaseProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
-        self.task_instruction = None
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+
+    def init_task_info(self) -> None:
+        raise NotImplementedError(
+            "The 'init_task_info' method must be implemented in derived classes."
+        )
+
+    def init_executor(self) -> None:
+        self.executor = OpenAI()
 
-    def init_processor(self):  # type: ignore[no-untyped-def] # FIX ME
-        self.model = OpenAI()
-        self.messenger = BaseMessenger("gpt4_messenger")  # type: ignore[no-untyped-call] # FIX ME
-        return
+    def init_messenger(self) -> None:
+        self.messenger = BaseMessenger("gpt4_messenger")
 
-    def process(self, payload: dict) -> dict:  # type: ignore[type-arg] # FIX ME
-        return  # type: ignore[return-value] # FIX ME
+    def process(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        # Assume process should do something and return a dictionary
+        return {}
 
-    def update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+    def update_info(self, feedback: str) -> None:
         self.messenger.add_assistant_message(feedback)
 
-    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[no-untyped-call] # FIX ME
-    def gpt4_requst(self):  # type: ignore[no-untyped-def] # FIX ME
-        response = self.model.chat.completions.create(
+    @info_exponential_backoff(retries=5, base_wait_time=1)
+    def gpt4_request(self) -> Any:
+        response = self.executor.chat.completions.create(
             model="gpt-4-turbo-preview",
-            messages=self.messenger.get_messages(),  # type: ignore[no-untyped-call] # FIX ME
+            messages=self.messenger.get_messages(),
             max_tokens=300,
         )
-        return response
-
-    def ask_info(  # type: ignore[override] # FIX ME
-        self,
-        query: str,
-        context: str = None,  # type: ignore[assignment] # FIX ME
-        image_path: str = None,  # type: ignore[assignment] # FIX ME
-        audio_path: str = None,  # type: ignore[assignment] # FIX ME
-        video_path: str = None,  # type: ignore[assignment] # FIX ME
+        description = response.choices[0].message.content
+        return description
+
+    def ask_info(
+        self, query: str, text: Optional[str] = None, *args: Any, **kwargs: Any
     ) -> str:
-        if self.messenger.check_iter_round_num() == 0:  # type: ignore[no-untyped-call] # FIX ME
-            self.messenger.add_user_message(
-                "The text information for the previously described task is as follows: "
-                + context
-                + "Here is what you should do: "
-                + self.task_instruction  # type: ignore[operator] # FIX ME
+        if self.messenger.check_iter_round_num() == 0:
+            initial_message = "The text information for the previously described task is as follows: "
+            initial_message += (
+                text if text is not None else "No text provided."
             )
+            initial_message += (
+                " Here is what you should do: " + self.task_instruction
+            )
+            self.messenger.add_user_message(initial_message)
 
-        response = self.gpt4_requst()
-        description = response.choices[0].message.content
-        return description  # type: ignore[no-any-return] # FIX ME
+        description = self.gpt4_request()
+        return description
 
 
 if __name__ == "__main__":
-    processor = BaseProcessor("ocr_processor")  # type: ignore[no-untyped-call] # FIX ME
-    image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[no-untyped-call] # FIX ME
+    processor = GPT4Processor()
+    text = "Hugging Face has released a new version of Transformers that brings several enhancements."
+    summary: str = processor.ask_info(
+        query="Summarize the changes.", text=text
+    )
     print(summary)

ctm/processors/processor_gpt4_speaker_intent.py

@@ -1,15 +1,25 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4 import GPT4Processor
 
 
-@GPT4Processor.register_processor("gpt4_speaker_intent_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming GPT4Processor has a properly typed `register_processor` method
+@GPT4Processor.register_processor("gpt4_speaker_intent_processor")
 class GPT4SpeakerIntentProcessor(GPT4Processor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(
+            *args, **kwargs
+        )  # Ensure the parent constructor is called properly
+
+    def init_task_info(self) -> None:
         self.task_instruction = "You are a speaker intent predictor. You can understand the intent of the speaker and describe what is the speaker's intent for saying that. If there is no speaker detected, please answer with None."
 
 
 if __name__ == "__main__":
-    processor = GPT4Processor("close_fashion_processor")  # type: ignore[no-untyped-call] # FIX ME
-    image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    # Instantiate the specific subclass for speaker intent processing
+    processor = GPT4SpeakerIntentProcessor()
+    text = "I can't wait to see the results of the new project. We've put so much effort into it!"
+    summary: str = processor.ask_info(
+        query="What is the intent behind the speaker's statement?", text=text
+    )
     print(summary)

ctm/processors/processor_gpt4_text_emotion.py

@@ -1,15 +1,21 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4 import GPT4Processor
 
 
-@GPT4Processor.register_processor("gpt4_text_emotion_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming GPT4Processor has a properly typed `register_processor` method
+@GPT4Processor.register_processor("gpt4_text_emotion_processor")
 class GPT4TextEmotionProcessor(GPT4Processor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)  # Call to parent class constructor
+
+    def init_task_info(self) -> None:
         self.task_instruction = "You are a text emotion classifier. You can understand the emotion within the text and generate the emotion label. If there is no text detected, please answer with None."
 
 
 if __name__ == "__main__":
-    processor = GPT4Processor("close_fashion_processor")  # type: ignore[no-untyped-call] # FIX ME
-    image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    # Instantiate the specific subclass for text emotion processing
+    processor = GPT4TextEmotionProcessor()
+    text = "I am feeling great today! The sun is shining and I've got a lot of work done."
+    summary: str = processor.ask_info(query="Identify the emotion.", text=text)
     print(summary)

ctm/processors/processor_gpt4_text_summary.py

@@ -1,15 +1,23 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4 import GPT4Processor
 
 
-@GPT4Processor.register_processor("gpt4_text_summary_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming GPT4Processor has a properly typed `register_processor` method
+@GPT4Processor.register_processor("gpt4_text_summary_processor")
 class GPT4TextSummaryProcessor(GPT4Processor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(
+            *args, **kwargs
+        )  # Properly initialize the parent class
+
+    def init_task_info(self) -> None:
         self.task_instruction = "You are a text summarizer. You can understand the meaning of the text and generate the summary."
 
 
 if __name__ == "__main__":
-    processor = GPT4Processor("close_fashion_processor")  # type: ignore[no-untyped-call] # FIX ME
-    image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    # Instantiate the specific subclass for the text summarization task
+    processor = GPT4TextSummaryProcessor()
+    text = "The quick brown fox jumps over the lazy dog. This sentence contains every letter of the alphabet."
+    summary: str = processor.ask_info(query="Summarize the text.", text=text)
     print(summary)

ctm/processors/processor_gpt4v.py

@@ -1,64 +1,79 @@
+from typing import Any, Dict, List, Optional, Union
+
 from openai import OpenAI
 
 from ctm.messengers.messenger_base import BaseMessenger
 from ctm.processors.processor_base import BaseProcessor
-from ctm.utils.exponential_backoff import exponential_backoff
+from ctm.utils.decorator import info_exponential_backoff
 
 
-@BaseProcessor.register_processor("gpt4v_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Ensure that BaseProcessor has a properly typed register_processor method:
+@BaseProcessor.register_processor("gpt4v_processor")
 class GPT4VProcessor(BaseProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
-        self.task_instruction = None
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)  # Properly initialize the base class
+
+    def init_executor(self) -> None:
+        self.executor = OpenAI()
+
+    def init_messenger(self) -> None:
+        self.messenger = BaseMessenger("gpt4v_messenger")
 
-    def init_processor(self):  # type: ignore[no-untyped-def] # FIX ME
-        self.model = OpenAI()
-        self.messenger = BaseMessenger("gpt4v_messenger")  # type: ignore[no-untyped-call] # FIX ME
-        return
+    def init_task_info(self) -> None:
+        raise NotImplementedError(
+            "The 'init_task_info' method must be implemented in derived classes."
+        )
 
-    def process(self, payload: dict) -> dict:  # type: ignore[type-arg] # FIX ME
-        return  # type: ignore[return-value] # FIX ME
+    def process(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        return {}  # Return an empty dict or a meaningful response as required
 
-    def update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+    def update_info(self, feedback: str) -> None:
         self.messenger.add_assistant_message(feedback)
 
-    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[no-untyped-call] # FIX ME
-    def gpt4v_requst(self):  # type: ignore[no-untyped-def] # FIX ME
-        response = self.model.chat.completions.create(
+    @info_exponential_backoff(retries=5, base_wait_time=1)
+    def gpt4v_request(self) -> str | Any:
+        response = self.executor.chat.completions.create(
             model="gpt-4-vision-preview",
-            messages=self.messenger.get_messages(),  # type: ignore[no-untyped-call] # FIX ME
+            messages=self.messenger.get_messages(),
             max_tokens=300,
         )
-        return response
+        description = response.choices[0].message.content
+        return description
 
-    def ask_info(  # type: ignore[override] # FIX ME
+    def ask_info(
         self,
         query: str,
-        context: str = None,  # type: ignore[assignment] # FIX ME
-        image_path: str = None,  # type: ignore[assignment] # FIX ME
-        audio_path: str = None,  # type: ignore[assignment] # FIX ME
-        video_path: str = None,  # type: ignore[assignment] # FIX ME
+        text: Optional[str] = None,
+        image: Optional[str] = None,
+        video_frames: Optional[str] = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> str:
-        if self.messenger.check_iter_round_num() == 0:  # type: ignore[no-untyped-call] # FIX ME
-            image = self.process_image(image_path)  # type: ignore[no-untyped-call] # FIX ME
-            # image = '0'
-            self.messenger.add_user_message(
-                [
-                    {"type": "text", "text": self.task_instruction},
+        if self.messenger.check_iter_round_num() == 0:
+            messages: List[Dict[str, Union[str, Dict[str, str]]]] = [
+                {
+                    "type": "text",
+                    "text": self.task_instruction
+                    or "No instruction provided.",
+                },
+            ]
+            if image:
+                messages.append(
                     {
                         "type": "image_url",
                         "image_url": f"data:image/jpeg;base64,{image}",
-                    },
-                ]
-            )
+                    }
+                )
+            self.messenger.add_user_message(messages)
 
-        response = self.gpt4v_requst()
-        description = response.choices[0].message.content
-        return description  # type: ignore[no-any-return] # FIX ME
+        description = self.gpt4v_request()
+        return description
 
 
 if __name__ == "__main__":
-    processor = BaseProcessor("ocr_processor")  # type: ignore[no-untyped-call] # FIX ME
+    processor = GPT4VProcessor()
     image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[no-untyped-call] # FIX ME
+    summary: str = processor.ask_info(
+        query="Describe the image.", image=image_path
+    )
     print(summary)

ctm/processors/processor_gpt4v_cloth_fashion.py

@@ -1,15 +1,24 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4v import GPT4VProcessor
 
 
-@GPT4VProcessor.register_processor("gpt4v_cloth_fashion_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming GPT4VProcessor has a properly typed `register_processor` method
+@GPT4VProcessor.register_processor("gpt4v_cloth_fashion_processor")
 class GPT4VClothFashionProcessor(GPT4VProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)  # Call to parent class constructor
+
+    def init_task_info(self) -> None:
         self.task_instruction = "Focus on the cloth of people in the image, describe the style of the cloth fashion. If there is no people detected, please answer with None."
 
 
 if __name__ == "__main__":
-    processor = GPT4VProcessor("close_fashion_processor")  # type: ignore[no-untyped-call] # FIX ME
+    # Instantiate the specific subclass for the cloth fashion task
+    processor = GPT4VClothFashionProcessor()
     image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    # Providing a valid query and ensuring `ask_info` is correctly implemented in the base class
+    summary: str = processor.ask_info(
+        query="Describe the fashion style", image=image_path
+    )
     print(summary)

ctm/processors/processor_gpt4v_face_emotion.py

@@ -1,15 +1,26 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4v import GPT4VProcessor
 
 
-@GPT4VProcessor.register_processor("gpt4v_face_emotion_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assume register_processor method has been properly typed
+@GPT4VProcessor.register_processor("gpt4v_face_emotion_processor")
 class GPT4VFaceEmotionProcessor(GPT4VProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(
+            *args, **kwargs
+        )  # Properly initialize the parent class
+
+    def init_task_info(self) -> None:
         self.task_instruction = "Besides the main scene in the image, can you describe the face emotion that is on people's faces within this picture?"
 
 
 if __name__ == "__main__":
-    processor = GPT4VProcessor("face_emotion_processor")  # type: ignore[no-untyped-call] # FIX ME
+    # Instantiate the specific subclass for face emotion processing
+    processor = GPT4VFaceEmotionProcessor()
     image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    # Providing a valid query and ensuring that the method ask_info accepts the correct parameters
+    summary: str = processor.ask_info(
+        query="Describe face emotions", image=image_path
+    )
     print(summary)

ctm/processors/processor_gpt4v_ocr.py

@@ -1,15 +1,24 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4v import GPT4VProcessor
 
 
-@GPT4VProcessor.register_processor("gpt4v_ocr_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Correct the registration method to include type checking if possible in the GPT4VProcessor class
+@GPT4VProcessor.register_processor("gpt4v_ocr_processor")
 class GPT4VOCRProcessor(GPT4VProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(
+            *args, **kwargs
+        )  # Ensure the parent constructor is called properly
+
+    def init_task_info(self) -> None:
         self.task_instruction = "You should act like an OCR model. Please extract the text from the image. If there is no text detected, please answer with None."
 
 
 if __name__ == "__main__":
-    processor = GPT4VProcessor("ocr_processor")  # type: ignore[no-untyped-call] # FIX ME
+    # Ensure that we're instantiating the correct processor for the job
+    processor = GPT4VOCRProcessor()
     image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    # Provide a valid query string; ensure `ask_info` can handle all provided parameters
+    summary: str = processor.ask_info(query="Extract text", image=image_path)
     print(summary)

ctm/processors/processor_gpt4v_posture.py

@@ -1,15 +1,25 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4v import GPT4VProcessor
 
 
-@GPT4VProcessor.register_processor("gpt4v_posture_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming the GPT4VProcessor has a properly typed `register_processor` method:
+@GPT4VProcessor.register_processor("gpt4v_posture_processor")
 class GPT4VPostureProcessor(GPT4VProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(
+            *args, **kwargs
+        )  # Properly call the parent's constructor
+
+    def init_task_info(self) -> None:
         self.task_instruction = "Besides the main scene in the image, can you describe the posture that is going on within this picture?"
 
 
 if __name__ == "__main__":
-    processor = GPT4VProcessor("posture_processor")  # type: ignore[no-untyped-call] # FIX ME
+    # Instantiate the specific subclass for the posture analysis task
+    processor = GPT4VPostureProcessor()
     image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    summary: str = processor.ask_info(
+        query="Analyze the posture.", image=image_path
+    )
     print(summary)

ctm/processors/processor_gpt4v_scene_location.py

@@ -1,15 +1,24 @@
+from typing import Any
+
 from ctm.processors.processor_gpt4v import GPT4VProcessor
 
 
-@GPT4VProcessor.register_processor("gpt4v_scene_location_processor")  # type: ignore[no-untyped-call] # FIX ME
+# Assuming GPT4VProcessor has a properly typed `register_processor` method:
+@GPT4VProcessor.register_processor("gpt4v_scene_location_processor")
 class GPT4VSceneLocationProcessor(GPT4VProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)  # Initialize the parent processor
+
+    def init_task_info(self) -> None:
         self.task_instruction = "Besides the main activity in the image, can you describe the potential location or the event that is going on within this picture?"
 
 
 if __name__ == "__main__":
-    processor = GPT4VProcessor("scene_location_processor")  # type: ignore[no-untyped-call] # FIX ME
+    # Instantiate the specific subclass for the scene location task
+    processor = GPT4VSceneLocationProcessor()
     image_path = "../ctmai-test1.png"
-    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    # The `ask_info` method should also be corrected to include all necessary parameters properly typed.
+    summary: str = processor.ask_info(
+        query="Describe the scene and location.", image=image_path
+    )
     print(summary)

ctm/processors/processor_roberta_text_sentiment.py

@@ -1,46 +1,46 @@
+import json
 import os
+from typing import Any, Dict, Optional
 
-from huggingface_hub.inference_api import (
-    InferenceApi,  # type: ignore[import] # FIX ME
-)
+from huggingface_hub import InferenceClient
 
-from ctm.messengers.messenger_base import BaseMessenger
-from ctm.processors.processor_base import BaseProcessor
+from ..messengers.messenger_base import BaseMessenger
+from .processor_base import BaseProcessor
 
 
-@BaseProcessor.register_processor("roberta_text_sentiment_processor")  # type: ignore[no-untyped-call] # FIX ME
+@BaseProcessor.register_processor("roberta_text_sentiment_processor")
 class RobertaTextSentimentProcessor(BaseProcessor):
-    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
-        self.init_processor()  # type: ignore[no-untyped-call] # FIX ME
+    def init_executor(self) -> None:
+        self.executor = InferenceClient(token=os.environ["HF_TOKEN"])
 
-    def init_processor(self):  # type: ignore[no-untyped-def] # FIX ME
-        self.model = InferenceApi(
-            token=os.environ["HF_TOKEN"],
-            repo_id="cardiffnlp/twitter-roberta-base-sentiment-latest",
-        )
-        self.messenger = BaseMessenger("roberta_text_sentiment_messenger")  # type: ignore[no-untyped-call] # FIX ME
-        return
+    def init_task_info(self) -> None:
+        pass
 
-    def update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+    def init_messenger(self) -> None:
+        self.messenger = BaseMessenger("roberta_text_sentiment_messenger")
+
+    def update_info(self, feedback: str) -> None:
         self.messenger.add_assistant_message(feedback)
 
-    def ask_info(  # type: ignore[override] # FIX ME
+    def ask_info(
         self,
-        query: str,
-        context: str = None,  # type: ignore[assignment] # FIX ME
-        image_path: str = None,  # type: ignore[assignment] # FIX ME
-        audio_path: str = None,  # type: ignore[assignment] # FIX ME
-        video_path: str = None,  # type: ignore[assignment] # FIX ME
+        query: Optional[str],
+        text: Optional[str] = None,
+        *args: Any,
+        **kwargs: Any,
     ) -> str:
-        if self.messenger.check_iter_round_num() == 0:  # type: ignore[no-untyped-call] # FIX ME
-            self.messenger.add_user_message(context)
-
-        response = self.model(self.messenger.get_messages())  # type: ignore[no-untyped-call] # FIX ME
-        results = response[0]
-        # choose the label with the highest score
-        pos_score = 0
-        neg_score = 0
-        neutral_score = 0
+        if text and self.messenger.check_iter_round_num() == 0:
+            self.messenger.add_user_message(text)
+
+        results: Dict[str, Any] = json.loads(
+            self.executor.post(
+                json={"inputs": self.messenger.get_messages()},
+                model="cardiffnlp/twitter-roberta-base-sentiment-latest",
+            )
+        )[0]
+        pos_score = (
+            neg_score
+        ) = neutral_score = 0.0  # Initialize scores as floats
         for result in results:
             if result["label"] == "POSITIVE":
                 pos_score = result["score"]
@@ -48,23 +48,11 @@ class RobertaTextSentimentProcessor(BaseProcessor):
                 neg_score = result["score"]
             else:
                 neutral_score = result["score"]
-        if max(pos_score, neg_score, neutral_score) == pos_score:
+
+        # Simplified decision structure
+        max_score = max(pos_score, neg_score, neutral_score)
+        if max_score == pos_score:
             return "This text is positive."
-        elif max(pos_score, neg_score, neutral_score) == neg_score:
+        elif max_score == neg_score:
             return "This text is negative."
-        else:
-            return "This text is neutral."
-
-
-if __name__ == "__main__":
-    processor = BaseProcessor("roberta_text_sentiment_processor")  # type: ignore[no-untyped-call] # FIX ME
-    image_path = "../ctmai-test1.png"
-    text: str = (
-        "In a shocking turn of events, Hugging Face has released a new version of Transformers "
-        "that brings several enhancements and bug fixes. Users are thrilled with the improvements "
-        "and are finding the new version to be significantly better than the previous one. "
-        "The Hugging Face team is thankful for the community's support and continues to work "
-        "towards making the library the best it can be."
-    )
-    label = processor.ask_info(query=None, context=text, image_path=image_path)  # type: ignore[no-untyped-call] # FIX ME
-    print(label)
+        return "This text is neutral."

ctm/supervisors/__init__.py

@@ -1,5 +1,7 @@
-from .supervisor_gpt4 import GPT4Supervisior
+from ctm.supervisors.supervisor_base import BaseSupervisor
+from ctm.supervisors.supervisor_gpt4 import GPT4Supervisor
 
 __all__ = [
-    "GPT4Supervisior",
+    "GPT4Supervisor",
+    "BaseSupervisor",
 ]

ctm/supervisors/supervisor_base.py

@@ -1,18 +1,21 @@
 import base64
+from typing import Any, Dict, Optional, Tuple, Type
 
 
 class BaseSupervisor(object):
-    _supervisor_registry = {}
+    _supervisor_registry: Dict[str, Type["BaseSupervisor"]] = {}
 
     @classmethod
-    def register_supervisor(cls, supervisor_name):
-        def decorator(subclass):
+    def register_supervisor(cls, supervisor_name: str) -> Any:
+        def decorator(
+            subclass: Type["BaseSupervisor"],
+        ) -> Type["BaseSupervisor"]:
             cls._supervisor_registry[supervisor_name] = subclass
             return subclass
 
         return decorator
 
-    def __new__(cls, supervisor_name, *args, **kwargs):
+    def __new__(cls, supervisor_name: str, *args: Any, **kwargs: Any) -> Any:
         if supervisor_name not in cls._supervisor_registry:
             raise ValueError(
                 f"No supervisor registered with name '{supervisor_name}'"
@@ -21,31 +24,24 @@ class BaseSupervisor(object):
             cls._supervisor_registry[supervisor_name]
         )
 
-    def set_model(self):
+    def set_model(
+        self,
+    ) -> None:
         raise NotImplementedError(
             "The 'set_model' method must be implemented in derived classes."
         )
 
-    @staticmethod
-    def process_image(image_path):
-        with open(image_path, "rb") as image_file:
-            return base64.b64encode(image_file.read()).decode("utf-8")
-
-    @staticmethod
-    def process_audio(audio_path):
-        return None
-
-    @staticmethod
-    def process_video(video_path):
-        return None
-
-    def ask(self, query, image_path):
+    def ask(self, query: str, image_path: str) -> Tuple[str, float]:
         gist = self.ask_info(query, image_path)
         score = self.ask_score(query, gist, verbose=True)
         return gist, score
 
-    def ask_info(self, query: str, context: str = None) -> str:
-        return None
+    def ask_info(self, query: str, context: Optional[str] = None) -> str:
+        raise NotImplementedError(
+            "The 'ask_info' method must be implemented in derived classes."
+        )
 
     def ask_score(self, query: str, gist: str, verbose: bool = False) -> float:
-        return None
+        raise NotImplementedError(
+            "The 'ask_score' method must be implemented in derived classes."
+        )

ctm/supervisors/supervisor_gpt4.py

@@ -1,32 +1,49 @@
+from typing import Any, Optional
+
 from openai import OpenAI
 
-from ctm.supervisors.supervisor_base import BaseSupervisor
-from ctm.utils.exponential_backoff import exponential_backoff
+from ..utils import info_exponential_backoff, score_exponential_backoff
+from .supervisor_base import BaseSupervisor
 
 
 @BaseSupervisor.register_supervisor("gpt4_supervisor")
-class GPT4Supervisior(BaseSupervisor):
-    def __init__(self, *args, **kwargs):
+class GPT4Supervisor(BaseSupervisor):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         self.init_supervisor()
 
-    def init_supervisor(self):
+    def init_supervisor(self) -> None:
         self.model = OpenAI()
 
-    @exponential_backoff(retries=5, base_wait_time=1)
-    def ask_info(self, query: str, context: str = None) -> str:
-        prompt = [
-            {
-                "role": "user",
-                "content": f"The following is detailed information on the topic: {context}. Based on this information, answer the question: {query}. Answer with a few words:",
-            }
-        ]
+    @info_exponential_backoff(retries=5, base_wait_time=1)
+    def ask_info(self, query: str, context: Optional[str] = None) -> str | Any:
+
         responses = self.model.chat.completions.create(
-            model="gpt-4-turbo-preview", messages=prompt, max_tokens=300, n=1
+            model="gpt-4-turbo-preview",
+            messages=[
+                {
+                    "role": "user",
+                    "content": f"The following is detailed information on the topic: {context}. Based on this information, answer the question: {query}. Answer with a few words:",
+                }
+            ],
+            max_tokens=300,
+            n=1,
+        )
+        answer = (
+            responses.choices[0].message.content
+            if responses.choices[0].message.content
+            else "FAILED"
         )
-        answer = responses.choices[0].message.content
         return answer
 
-    def ask_score(self, query, gist, verbose=False, *args, **kwargs):
+    @score_exponential_backoff(retries=5, base_wait_time=1)
+    def ask_score(
+        self,
+        query: str,
+        gist: str,
+        verbose: bool = False,
+        *args: Any,
+        **kwargs: Any,
+    ) -> float:
         max_attempts = 5
         for attempt in range(max_attempts):
             try:
@@ -35,14 +52,16 @@ class GPT4Supervisior(BaseSupervisor):
                     messages=[
                         {
                             "role": "user",
-                            "content": "How related is the information ({}) with the query ({})? We want to make sure that the information includes a person's name as the answer. Answer with a number from 0 to 5 and do not add any other thing.".format(
-                                gist, query
-                            ),
+                            "content": f"How related is the information ({gist}) with the query ({query})? We want to make sure that the information includes a person's name as the answer. Answer with a number from 0 to 5 and do not add any other thing.",
                         },
                     ],
                     max_tokens=50,
                 )
-                score = int(response.choices[0].message.content.strip()) / 5
+                score = (
+                    float(response.choices[0].message.content.strip()) / 5
+                    if response.choices[0].message.content
+                    else 0.0
+                )
                 return score
             except Exception as e:
                 print(f"Attempt {attempt + 1} failed: {e}")
@@ -50,11 +69,4 @@ class GPT4Supervisior(BaseSupervisor):
                     print("Retrying...")
                 else:
                     print("Max attempts reached. Returning default score.")
-        return 0
-
-
-if __name__ == "__main__":
-    supervisor = BaseSupervisor("cloth_fashion_supervisor")
-    image_path = "../ctmai-test1.png"
-    summary: str = supervisor.ask_info(query=None, image_path=image_path)
-    print(summary)
+        return 0.0

ctm/utils/__init__.py

@@ -0,0 +1,13 @@
+from .decorator import (
+    info_exponential_backoff,
+    score_exponential_backoff,
+)
+from .loader import load_audio, load_image, load_video
+
+__all__ = [
+    "score_exponential_backoff",
+    "info_exponential_backoff",
+    "load_audio",
+    "load_image",
+    "load_video",
+]

ctm/utils/{exponential_backoff.py → decorator.py}

@@ -1,18 +1,55 @@
 import math
 import time
 from functools import wraps
+from typing import Any, Callable, Optional
 
+INF = float(math.inf)
 
-def exponential_backoff(retries=5, base_wait_time=1):  # type: ignore[no-untyped-def] # FIX ME
+
+def info_exponential_backoff(
+    retries: int = 5, base_wait_time: int = 1
+) -> Callable[[Callable[..., str]], Callable[..., str]]:
+    """
+    Decorator for applying exponential backoff to a function.
+    :param retries: Maximum number of retries.
+    :param base_wait_time: Base wait time in seconds for the exponential backoff.
+    """
+
+    def decorator(func: Callable[..., str]) -> Callable[..., str]:
+        @wraps(func)
+        def wrapper(*args: Any, **kwargs: Any) -> str:
+            attempts = 0
+            while attempts < retries:
+                try:
+                    return func(*args, **kwargs)
+                except Exception as e:
+                    wait_time = base_wait_time * (2**attempts)
+                    print(f"Attempt {attempts + 1} failed: {e}")
+                    print(f"Waiting {wait_time} seconds before retrying...")
+                    time.sleep(wait_time)
+                    attempts += 1
+            print(
+                f"Failed to execute '{func.__name__}' after {retries} retries."
+            )
+            return "FAILED"
+
+        return wrapper
+
+    return decorator
+
+
+def score_exponential_backoff(
+    retries: int = 5, base_wait_time: int = 1
+) -> Callable[[Callable[..., float]], Callable[..., float]]:
     """
     Decorator for applying exponential backoff to a function.
     :param retries: Maximum number of retries.
     :param base_wait_time: Base wait time in seconds for the exponential backoff.
     """
 
-    def decorator(func):  # type: ignore[no-untyped-def] # FIX ME
+    def decorator(func: Callable[..., float]) -> Callable[..., float]:
         @wraps(func)
-        def wrapper(*args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        def wrapper(*args: Any, **kwargs: Any) -> float:
             attempts = 0
             while attempts < retries:
                 try:
@@ -26,7 +63,7 @@ def exponential_backoff(retries=5, base_wait_time=1):  # type: ignore[no-untyped
             print(
                 f"Failed to execute '{func.__name__}' after {retries} retries."
             )
-            return None
+            return -INF
 
         return wrapper
 

ctm/utils/loader.py

@@ -0,0 +1,41 @@
+import base64
+from typing import Any, List, Tuple
+
+import numpy as np
+from numpy.typing import NDArray
+
+
+def load_audio(audio_path: str) -> Tuple[NDArray[np.float32], int]:
+    import librosa
+
+    audio, sr = librosa.load(audio_path, sr=None)
+    import pdb
+
+    pdb.set_trace()  # Debugging breakpoint
+    return (audio.astype(np.float32), int(sr))
+
+
+def load_image(image_path: str) -> str:
+    with open(image_path, "rb") as image_file:
+        encoded_image = base64.b64encode(image_file.read()).decode("utf-8")
+        return encoded_image
+
+
+def load_video(video_path: str, frame_num: int = 5) -> List[NDArray[np.uint8]]:
+    import cv2
+
+    cap = cv2.VideoCapture(video_path)
+    frames: List[np.ndarray[np.uint8, Any]] = []
+    try:
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            frames.append(frame.astype(np.uint8))
+    finally:
+        cap.release()
+
+    if len(frames) >= frame_num:
+        step = len(frames) // frame_num
+        frames = [frames[i] for i in range(0, len(frames), step)]
+    return frames