Feature/support ctm (#16)

* support inner loop ctm

* support pre-commit

* support running

* support ctm ai

.gitignore

@@ -157,5 +157,4 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
-**/ctm/*
+#.idea/

app.py

@@ -2,11 +2,8 @@ import os
 import sys
 
 import gradio as gr
-import base64
-import io
-from PIL import Image
-
-sys.path.append("../CTM/")
+import sys
+sys.path.append('./ctm')
 from ctm.ctms.ctm_base import BaseConsciousnessTuringMachine
 
 ctm = BaseConsciousnessTuringMachine()
@@ -35,14 +32,12 @@ def introduction():
             """
         )
 
-
 def add_processor(processor_name, display_name, state):
-    print("add processor ", processor_name)
+    print('add processor ', processor_name)
     ctm.add_processor(processor_name)
     print(ctm.processor_group_map)
     print(len(ctm.processor_list))
-    return display_name + " (added)"
-
+    return display_name + ' (added)'
 
 def processor_tab():
     # Categorized model names
@@ -50,14 +45,14 @@ def processor_tab():
         "gpt4_text_emotion_processor",
         "gpt4_text_summary_processor",
         "gpt4_speaker_intent_processor",
-        "roberta_text_sentiment_processor",
+        "roberta_text_sentiment_processor"
     ]
     vision_processors = [
         "gpt4v_cloth_fashion_processor",
         "gpt4v_face_emotion_processor",
         "gpt4v_ocr_processor",
-        "gpt4v_posture_processor",
-        "gpt4v_scene_location_processor",
+        "gpt4v_posture",
+        "gpt4v_scene_location_processor"
     ]
 
     with gr.Blocks():
@@ -65,54 +60,37 @@ def processor_tab():
             with gr.Column(scale=1):
                 gr.Markdown("### Text Processors")
                 for model_name in text_processors:
-                    display_name = (
-                        model_name.replace("processor", "")
-                        .replace("_", " ")
-                        .title()
-                    )
+                    display_name = model_name.replace("processor", "").replace("_", " ").title()
 
                     button = gr.Button(display_name)
-                    processor_name = gr.Textbox(
-                        value=model_name, visible=False
-                    )
-                    display_name = gr.Textbox(
-                        value=display_name, visible=False
-                    )
+                    processor_name = gr.Textbox(value=model_name, visible=False)
+                    display_name = gr.Textbox(value=display_name, visible=False)
                     button.click(
                         fn=add_processor,
                         inputs=[processor_name, display_name, gr.State()],
-                        outputs=[button],
+                        outputs=[button]
                     )
 
             with gr.Column(scale=1):
                 gr.Markdown("### Vision Processors")
                 for model_name in vision_processors:
-                    display_name = (
-                        model_name.replace("processor", "")
-                        .replace("_", " ")
-                        .title()
-                    )
+                    display_name = model_name.replace("processor", "").replace("_", " ").title()
 
                     button = gr.Button(display_name)
-                    processor_name = gr.Textbox(
-                        value=model_name, visible=False
-                    )
-                    display_name = gr.Textbox(
-                        value=display_name, visible=False
-                    )
+                    processor_name = gr.Textbox(value=model_name, visible=False)
+                    display_name = gr.Textbox(value=display_name, visible=False)
                     button.click(
                         fn=add_processor,
                         inputs=[processor_name, display_name, gr.State()],
-                        outputs=[button],
+                        outputs=[button]
                     )
 
 
-def forward(query, text, image, state):
-    state["question"] = query
-    image = convert_base64(image)
-    ask_processors_output_info, state = ask_processors(
-        query, text, image, state
-    )
+
+
+def forward(query, content, image, state):
+    state['question'] = query
+    ask_processors_output_info, state = ask_processors(query, content, image, state)
     uptree_competition_output_info, state = uptree_competition(state)
     ask_supervisor_output_info, state = ask_supervisor(state)
 

ctm

@@ -1 +0,0 @@
-/Users/yuhaofei/github_repo/CTM-AI/ctm/

ctm/configs/ctm_config_base.py

@@ -0,0 +1,44 @@
+import json
+
+
+class BaseConsciousnessTuringMachineConfig(object):
+    # Initialize with default values or those passed to the constructor
+    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
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+    def to_json_string(self):
+        """Serializes this instance to a JSON string."""
+        return json.dumps(self.__dict__, indent=2) + "\n"
+
+    @classmethod
+    def from_json_file(cls, json_file):
+        """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):
+        """
+        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"
+        return cls.from_json_file(config_file)

ctm/configs/sarcasm_ctm_config.json

@@ -0,0 +1,18 @@
+{
+    "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/__init__.py

@@ -0,0 +1,5 @@
+from .ctm_base import BaseConsciousnessTuringMachine
+
+__all__ = [
+    "BaseConsciousnessTuringMachine",
+]

ctm/ctms/ctm_base.py

@@ -0,0 +1,227 @@
+import concurrent.futures
+from collections import defaultdict
+
+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
+
+
+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)
+        if ctm_name:
+            self.config = BaseConsciousnessTuringMachineConfig.from_ctm(
+                ctm_name
+            )
+        else:
+            self.config = BaseConsciousnessTuringMachineConfig()
+        self.processor_list = []
+        self.processor_group_map = defaultdict(list)
+        self.load_ctm()
+
+    def add_processor(self, processor_name, group_name=None):
+        processor_instance = BaseProcessor(processor_name)
+        self.processor_list.append(
+            {
+                "processor_name": processor_name,
+                "processor_instance": processor_instance,
+            }
+        )
+        if group_name:
+            self.processor_group_map[processor_name] = group_name
+
+    def add_supervisor(self, supervisor_name):
+        supervisor_instance = BaseSupervisor(supervisor_name)
+        self.supervisor = {
+            "supervisor_name": supervisor_name,
+            "supervisor_instance": supervisor_instance,
+        }
+
+    @staticmethod
+    def ask_processor(
+        processor, question, context, image_path, audio_path, video_path
+    ):
+        processor_instance = processor["processor_instance"]
+        processor_name = processor["processor_name"]
+        gist, score = processor_instance.ask(
+            question, context, image_path, audio_path, video_path
+        )
+        return {"name": processor_name, "gist": gist, "score": score}
+
+    def ask_processors(
+        self, question, context, image_path, audio_path, video_path
+    ):
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            futures = [
+                executor.submit(
+                    self.ask_processor,
+                    processor,
+                    question,
+                    context,
+                    image_path,
+                    audio_path,
+                    video_path,
+                )
+                for processor in self.processor_list
+            ]
+            results = [
+                future.result()
+                for future in concurrent.futures.as_completed(futures)
+            ]
+
+        output = {}
+        for result in results:
+            output[result["name"]] = {
+                "gist": result["gist"],
+                "score": result["score"],
+            }
+
+        assert len(output) == len(self.processor_list)
+        return output
+
+    def uptree_competition(self, processor_output):
+        # Unpack processor outputs into lists for easier processing
+        gists, scores, names = [], [], []
+        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())
+
+        # Prepare to track the best processor by group
+        best_processor_by_group = {
+            group: (None, -1) for group in unique_groups
+        }  # (processor_name, score)
+
+        # Iterate through processors to find the best in each group
+        for name, score in zip(names, scores):
+            group = self.processor_group_map[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_processor_by_group.values(), key=lambda x: x[1]
+        )
+        best_name = best_overall[0]
+        index = names.index(best_name)
+
+        winning_info = {
+            "name": best_name,
+            "gist": gists[index],
+            "score": scores[index],
+        }
+        return winning_info
+
+    def ask_supervisor(self, question, processor_info):
+        final_answer, score = self.supervisor["supervisor_instance"].ask(
+            question, processor_info["gist"]
+        )
+        return final_answer, score
+
+    def downtree_broadcast(self, winning_output):
+        winning_processor_name = winning_output["name"]
+        winning_processor_gist = winning_output["gist"]
+        for processor in self.processor_list:
+            if processor["processor_name"] != winning_processor_name:
+                processor["processor_instance"].update_info(
+                    winning_processor_gist
+                )
+        return
+
+    def calc_processor_sim(self, processor_output):
+        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):
+        sim = self.calc_processor_sim(processor_output)
+        print(sim)
+        # iterate on each sim pair
+        # if sim > threshold, then link the two processors by combining them into the same group
+        link_threshold = 0.5
+        for i in range(len(sim)):
+            for j in range(i + 1, len(sim)):
+                if sim[i][j] > 0.5:
+                    processor1_name = list(processor_output.keys())[i]
+                    processor2_name = list(processor_output.keys())[j]
+                    # choose the group that includes more processors
+                    # processor_group_map is a dict with processor_name as key and group_name as value
+                    group1 = self.processor_group_map[processor1_name]
+                    group2 = self.processor_group_map[processor2_name]
+                    # calculate the number of processors in each group
+                    group1_count = sum(
+                        [
+                            1
+                            for group in self.processor_group_map.values()
+                            if group == group1
+                        ]
+                    )
+                    group2_count = sum(
+                        [
+                            1
+                            for group in self.processor_group_map.values()
+                            if group == group2
+                        ]
+                    )
+                    # choose the group with more processors
+                    group_name = (
+                        group1 if group1_count > group2_count else group2
+                    )
+                    self.processor_group_map[processor1_name] = group_name
+                    self.processor_group_map[processor2_name] = group_name
+        return
+
+    def processor_fuse(self, infos, scores):
+        return infos, scores
+
+    def forward(
+        self,
+        question=None,
+        context=None,
+        image_path=None,
+        audio_path=None,
+        video_path=None,
+    ):
+        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,
+            )
+            import pdb; pdb.set_trace()
+            winning_output = self.uptree_competition(processor_output)
+            answer, score = self.ask_supervisor(question, winning_output)
+            if score > answer_threshold:
+                break
+            else:
+                self.downtree_broadcast(winning_output)
+                self.link_form(processor_output)
+        return answer, score
+
+    def load_ctm(self):
+        for (
+            group_name,
+            processor_list,
+        ) in self.config.groups_of_processors.items():
+            for processor_name in processor_list:
+                self.add_processor(processor_name, group_name=group_name)
+        self.add_supervisor(self.config.supervisor)

ctm/messengers/__init__.py

@@ -0,0 +1,15 @@
+from .messenger_bart_text_summ import BartTextSummarizationMessenger
+from .messenger_base import BaseMessenger
+from .messenger_gpt4 import GPT4Messenger
+from .messenger_gpt4v import GPT4VMessenger
+from .messenger_roberta_text_sentiment import (
+    RobertaTextSentimentMessenger,
+)
+
+__all__ = [
+    "BaseMessenger",
+    "GPT4VMessenger",
+    "GPT4Messenger",
+    "BartTextSummarizationMessenger",
+    "RobertaTextSentimentMessenger",
+]

ctm/messengers/messenger_bart_text_summ.py

@@ -0,0 +1,22 @@
+from typing import Dict, List, Union
+
+from ctm.messengers.messenger_base import BaseMessenger
+
+
+@BaseMessenger.register_messenger("bart_text_summ_messenger")  # type: ignore[no-untyped-call] # FIX ME
+class BartTextSummarizationMessenger(BaseMessenger):
+    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        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 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 check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
+        return 1 if len(self.messages) > 0 else 0

ctm/messengers/messenger_base.py

@@ -0,0 +1,63 @@
+from typing import Dict, List, Union
+
+
+class BaseMessenger(object):
+    _messenger_registry = {}  # type: ignore[var-annotated] # FIX ME
+
+    @classmethod
+    def register_messenger(cls, messenger_name):  # type: ignore[no-untyped-def] # FIX ME
+        def decorator(subclass):  # type: ignore[no-untyped-def] # FIX ME
+            cls._messenger_registry[messenger_name] = subclass
+            return subclass
+
+        return decorator
+
+    def __new__(cls, messenger_name, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        if messenger_name not in cls._messenger_registry:
+            raise ValueError(
+                f"No messenger registered with name '{messenger_name}'"
+            )
+        return super(BaseMessenger, cls).__new__(
+            cls._messenger_registry[messenger_name]
+        )
+
+    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        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 update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+        pass
+
+    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
+        pass
+
+    def add_system_message(self, message: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+        self.update_message("system", message)
+
+    def add_assistant_message(self, message: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+        self.update_message("assistant", message)
+
+    def add_user_message(self, message: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+        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
+            "user",
+            {
+                "type": "image_url",
+                "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 clear(self):  # type: ignore[no-untyped-def] # FIX ME
+        self.messages.clear()  # type: ignore[attr-defined] # FIX ME
+
+    def get_messages(self):  # type: ignore[no-untyped-def] # FIX ME
+        return self.messages  # type: ignore[attr-defined] # FIX ME

ctm/messengers/messenger_gpt4.py

@@ -0,0 +1,22 @@
+from typing import Dict, List, Union
+
+from ctm.messengers.messenger_base import BaseMessenger
+
+
+@BaseMessenger.register_messenger("gpt4_messenger")  # type: ignore[no-untyped-call] # FIX ME
+class GPT4Messenger(BaseMessenger):
+    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        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 update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+        self.messages.append({"role": role, "content": content})
+
+    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
+        return len(self.messages)

ctm/messengers/messenger_gpt4v.py

@@ -0,0 +1,22 @@
+from typing import Dict, List, Union
+
+from ctm.messengers.messenger_base import BaseMessenger
+
+
+@BaseMessenger.register_messenger("gpt4v_messenger")  # type: ignore[no-untyped-call] # FIX ME
+class GPT4VMessenger(BaseMessenger):
+    def __init__(self, role=None, content=None, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        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 update_message(self, role: str, content: Union[str, Dict, List]):  # type: ignore[no-untyped-def, type-arg] # FIX ME
+        self.messages.append({"role": role, "content": content})
+
+    def check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
+        return len(self.messages)

ctm/messengers/messenger_roberta_text_sentiment.py

@@ -0,0 +1,23 @@
+from typing import Dict, List, Union
+
+from ctm.messengers.messenger_base import BaseMessenger
+
+
+@BaseMessenger.register_messenger("roberta_text_sentiment_messenger")  # type: ignore[no-untyped-call] # FIX ME
+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_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 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 check_iter_round_num(self):  # type: ignore[no-untyped-def] # FIX ME
+        return 1 if len(self.messages) > 0 else 0

ctm/processors/__init__.py

@@ -0,0 +1,31 @@
+from .processor_bart_text_summary import BartTextSummaryProcessor
+from .processor_base import BaseProcessor
+from .processor_gpt4 import GPT4Processor
+from .processor_gpt4_speaker_intent import GPT4SpeakerIntentProcessor
+from .processor_gpt4_text_emotion import GPT4TextEmotionProcessor
+from .processor_gpt4_text_summary import GPT4TextSummaryProcessor
+from .processor_gpt4v import GPT4VProcessor
+from .processor_gpt4v_cloth_fashion import GPT4VClothFashionProcessor
+from .processor_gpt4v_face_emotion import GPT4VFaceEmotionProcessor
+from .processor_gpt4v_ocr import GPT4VOCRProcessor
+from .processor_gpt4v_posture import GPT4VPostureProcessor
+from .processor_gpt4v_scene_location import GPT4VSceneLocationProcessor
+from .processor_roberta_text_sentiment import (
+    RobertaTextSentimentProcessor,
+)
+
+__all__ = [
+    "BaseProcessor",
+    "GPT4VProcessor",
+    "GPT4VSceneLocationProcessor",
+    "GPT4VOCRProcessor",
+    "GPT4VClothFashionProcessor",
+    "GPT4VFaceEmotionProcessor",
+    "GPT4VPostureProcessor",
+    "RobertaTextSentimentProcessor",
+    "BartTextSummaryProcessor",
+    "GPT4SpeakerIntentProcessor",
+    "GPT4TextEmotionProcessor",
+    "GPT4TextSummaryProcessor",
+    "GPT4Processor",
+]

ctm/processors/processor_bart_text_summary.py

@@ -0,0 +1,55 @@
+import os
+
+from huggingface_hub.inference_api import (
+    InferenceApi,  # type: ignore[import] # FIX ME
+)
+
+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
+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_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 update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+        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)
+
+        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
+
+
+if __name__ == "__main__":
+    processor = BaseProcessor("bart_text_summ_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."
+    )
+    summary: str = processor.ask_info(  # type: ignore[no-untyped-call] # FIX ME
+        query=None, context=text, image_path=image_path
+    )
+    print(summary)

ctm/processors/processor_base.py

@@ -0,0 +1,117 @@
+import base64
+
+from ctm.utils.exponential_backoff import exponential_backoff
+
+
+class BaseProcessor(object):
+    _processor_registry = {}  # type: ignore[var-annotated] # FIX ME
+
+    @classmethod
+    def register_processor(cls, processor_name):  # type: ignore[no-untyped-def] # FIX ME
+        def decorator(subclass):  # type: ignore[no-untyped-def] # FIX ME
+            cls._processor_registry[processor_name] = subclass
+            return subclass
+
+        return decorator
+
+    def __new__(cls, processor_name, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        if processor_name not in cls._processor_registry:
+            raise ValueError(
+                f"No processor registered with name '{processor_name}'"
+            )
+        return super(BaseProcessor, cls).__new__(
+            cls._processor_registry[processor_name]
+        )
+
+    def set_model(self):  # type: ignore[no-untyped-def] # FIX ME
+        raise NotImplementedError(
+            "The 'set_model' 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
+
+    @staticmethod
+    def process_video(video_path):  # type: ignore[no-untyped-def] # FIX ME
+        return None
+
+    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
+        )
+        score = self.ask_score(query, gist, verbose=True)  # type: ignore[no-untyped-call] # FIX ME
+        return gist, score
+
+    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[misc, no-untyped-call] # FIX ME
+    def ask_relevance(self, query: str, gist: str) -> float:
+        response = self.model.chat.completions.create(  # type: ignore[attr-defined] # FIX ME
+            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
+                    ),
+                },
+            ],
+            max_tokens=50,
+        )
+        score = int(response.choices[0].message.content.strip()) / 5
+        return score
+
+    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[misc, no-untyped-call] # FIX ME
+    def ask_confidence(self, query: str, gist: str) -> float:
+        response = self.model.chat.completions.create(  # type: ignore[attr-defined] # FIX ME
+            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
+                    ),
+                },
+            ],
+            max_tokens=50,
+        )
+        score = int(response.choices[0].message.content.strip()) / 5
+        return score
+
+    @exponential_backoff(retries=5, base_wait_time=1)  # type: ignore[misc, no-untyped-call] # FIX ME
+    def ask_surprise(
+        self, query: str, gist: str, history_gists: str = None  # type: ignore[assignment] # FIX ME
+    ) -> float:
+        response = self.model.chat.completions.create(  # type: ignore[attr-defined] # FIX ME
+            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
+                    ),
+                },
+            ],
+            max_tokens=50,
+        )
+        score = int(response.choices[0].message.content.strip()) / 5
+        return score
+
+    def ask_score(self, query, gist, verbose=False, *args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+        relevance = self.ask_relevance(query, gist, *args, **kwargs)
+        confidence = self.ask_confidence(query, gist, *args, **kwargs)
+        surprise = self.ask_surprise(query, gist, *args, **kwargs)
+        if verbose:
+            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
+        raise NotImplementedError(
+            "The 'ask_information' method must be implemented in derived classes."
+        )

ctm/processors/processor_gpt4.py

@@ -0,0 +1,59 @@
+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
+
+
+@BaseProcessor.register_processor("gpt4_processor")  # type: ignore[no-untyped-call] # FIX ME
+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_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 process(self, payload: dict) -> dict:  # type: ignore[type-arg] # FIX ME
+        return  # type: ignore[return-value] # FIX ME
+
+    def update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+        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(
+            model="gpt-4-turbo-preview",
+            messages=self.messenger.get_messages(),  # type: ignore[no-untyped-call] # FIX ME
+            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
+    ) -> 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
+            )
+
+        response = self.gpt4_requst()
+        description = response.choices[0].message.content
+        return description  # type: ignore[no-any-return] # FIX ME
+
+
+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
+    print(summary)

ctm/processors/processor_gpt4_speaker_intent.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4 import GPT4Processor
+
+
+@GPT4Processor.register_processor("gpt4_speaker_intent_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    print(summary)

ctm/processors/processor_gpt4_text_emotion.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4 import GPT4Processor
+
+
+@GPT4Processor.register_processor("gpt4_text_emotion_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    print(summary)

ctm/processors/processor_gpt4_text_summary.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4 import GPT4Processor
+
+
+@GPT4Processor.register_processor("gpt4_text_summary_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    print(summary)

ctm/processors/processor_gpt4v.py

@@ -0,0 +1,64 @@
+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
+
+
+@BaseProcessor.register_processor("gpt4v_processor")  # type: ignore[no-untyped-call] # FIX ME
+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_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 process(self, payload: dict) -> dict:  # type: ignore[type-arg] # FIX ME
+        return  # type: ignore[return-value] # FIX ME
+
+    def update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+        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(
+            model="gpt-4-vision-preview",
+            messages=self.messenger.get_messages(),  # type: ignore[no-untyped-call] # FIX ME
+            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
+    ) -> 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},
+                    {
+                        "type": "image_url",
+                        "image_url": f"data:image/jpeg;base64,{image}",
+                    },
+                ]
+            )
+
+        response = self.gpt4v_requst()
+        description = response.choices[0].message.content
+        return description  # type: ignore[no-any-return] # FIX ME
+
+
+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
+    print(summary)

ctm/processors/processor_gpt4v_cloth_fashion.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4v import GPT4VProcessor
+
+
+@GPT4VProcessor.register_processor("gpt4v_cloth_fashion_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    image_path = "../ctmai-test1.png"
+    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    print(summary)

ctm/processors/processor_gpt4v_face_emotion.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4v import GPT4VProcessor
+
+
+@GPT4VProcessor.register_processor("gpt4v_face_emotion_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    image_path = "../ctmai-test1.png"
+    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    print(summary)

ctm/processors/processor_gpt4v_ocr.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4v import GPT4VProcessor
+
+
+@GPT4VProcessor.register_processor("gpt4v_ocr_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    image_path = "../ctmai-test1.png"
+    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    print(summary)

ctm/processors/processor_gpt4v_posture.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4v import GPT4VProcessor
+
+
+@GPT4VProcessor.register_processor("gpt4v_posture_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    image_path = "../ctmai-test1.png"
+    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    print(summary)

ctm/processors/processor_gpt4v_scene_location.py

@@ -0,0 +1,15 @@
+from ctm.processors.processor_gpt4v import GPT4VProcessor
+
+
+@GPT4VProcessor.register_processor("gpt4v_scene_location_processor")  # type: ignore[no-untyped-call] # FIX ME
+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
+        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
+    image_path = "../ctmai-test1.png"
+    summary: str = processor.ask_info(query=None, image_path=image_path)  # type: ignore[arg-type] # FIX ME
+    print(summary)

ctm/processors/processor_roberta_text_sentiment.py

@@ -0,0 +1,70 @@
+import os
+
+from huggingface_hub.inference_api import (
+    InferenceApi,  # type: ignore[import] # FIX ME
+)
+
+from ctm.messengers.messenger_base import BaseMessenger
+from ctm.processors.processor_base import BaseProcessor
+
+
+@BaseProcessor.register_processor("roberta_text_sentiment_processor")  # type: ignore[no-untyped-call] # FIX ME
+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_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 update_info(self, feedback: str):  # type: ignore[no-untyped-def] # FIX ME
+        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)
+
+        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
+        for result in results:
+            if result["label"] == "POSITIVE":
+                pos_score = result["score"]
+            elif result["label"] == "NEGATIVE":
+                neg_score = result["score"]
+            else:
+                neutral_score = result["score"]
+        if max(pos_score, neg_score, neutral_score) == pos_score:
+            return "This text is positive."
+        elif max(pos_score, neg_score, neutral_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)

ctm/supervisors/__init__.py

@@ -0,0 +1,5 @@
+from .supervisor_gpt4 import GPT4Supervisior
+
+__all__ = [
+    "GPT4Supervisior",
+]

ctm/supervisors/supervisor_base.py

@@ -0,0 +1,51 @@
+import base64
+
+
+class BaseSupervisor(object):
+    _supervisor_registry = {}
+
+    @classmethod
+    def register_supervisor(cls, supervisor_name):
+        def decorator(subclass):
+            cls._supervisor_registry[supervisor_name] = subclass
+            return subclass
+
+        return decorator
+
+    def __new__(cls, supervisor_name, *args, **kwargs):
+        if supervisor_name not in cls._supervisor_registry:
+            raise ValueError(
+                f"No supervisor registered with name '{supervisor_name}'"
+            )
+        return super(BaseSupervisor, cls).__new__(
+            cls._supervisor_registry[supervisor_name]
+        )
+
+    def set_model(self):
+        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):
+        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_score(self, query: str, gist: str, verbose: bool = False) -> float:
+        return None

ctm/supervisors/supervisor_gpt4.py

@@ -0,0 +1,60 @@
+from openai import OpenAI
+
+from ctm.supervisors.supervisor_base import BaseSupervisor
+from ctm.utils.exponential_backoff import exponential_backoff
+
+
+@BaseSupervisor.register_supervisor("gpt4_supervisor")
+class GPT4Supervisior(BaseSupervisor):
+    def __init__(self, *args, **kwargs):
+        self.init_supervisor()
+
+    def init_supervisor(self):
+        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:",
+            }
+        ]
+        responses = self.model.chat.completions.create(
+            model="gpt-4-turbo-preview", messages=prompt, max_tokens=300, n=1
+        )
+        answer = responses.choices[0].message.content
+        return answer
+
+    def ask_score(self, query, gist, verbose=False, *args, **kwargs):
+        max_attempts = 5
+        for attempt in range(max_attempts):
+            try:
+                response = self.model.chat.completions.create(
+                    model="gpt-4-0125-preview",
+                    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
+                            ),
+                        },
+                    ],
+                    max_tokens=50,
+                )
+                score = int(response.choices[0].message.content.strip()) / 5
+                return score
+            except Exception as e:
+                print(f"Attempt {attempt + 1} failed: {e}")
+                if attempt < max_attempts - 1:
+                    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)

ctm/utils/exponential_backoff.py

@@ -0,0 +1,33 @@
+import math
+import time
+from functools import wraps
+
+
+def exponential_backoff(retries=5, base_wait_time=1):  # type: ignore[no-untyped-def] # FIX ME
+    """
+    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
+        @wraps(func)
+        def wrapper(*args, **kwargs):  # type: ignore[no-untyped-def] # FIX ME
+            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 None
+
+        return wrapper
+
+    return decorator