Spaces:
Running
Running
File size: 1,802 Bytes
bdafe83 |
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 65 66 67 |
from abc import abstractmethod
from typing import List
from ..config import BackendConfig, Configurable
from ..message import Message
class IntelligenceBackend(Configurable):
"""An abstraction of the intelligence source of the agents."""
stateful = None
type_name = None
@abstractmethod
def __init__(self, **kwargs):
super().__init__(**kwargs) # registers the arguments with Configurable
def __init_subclass__(cls, **kwargs):
# check if the subclass has the required attributes
for required in (
"stateful",
"type_name",
):
if getattr(cls, required) is None:
raise TypeError(
f"Can't instantiate abstract class {cls.__name__} without {required} attribute defined"
)
return super().__init_subclass__(**kwargs)
def to_config(self) -> BackendConfig:
self._config_dict["backend_type"] = self.type_name
return BackendConfig(**self._config_dict)
@abstractmethod
def query(
self,
agent_name: str,
role_desc: str,
history_messages: List[Message],
global_prompt: str = None,
request_msg: Message = None,
*args,
**kwargs,
) -> str:
raise NotImplementedError
@abstractmethod
async def async_query(
self,
agent_name: str,
role_desc: str,
history_messages: List[Message],
global_prompt: str = None,
request_msg: Message = None,
*args,
**kwargs,
) -> str:
"""Async querying."""
raise NotImplementedError
# reset the state of the backend
def reset(self):
if self.stateful:
raise NotImplementedError
else:
pass
|