File size: 2,695 Bytes
acb3380 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
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
|