agora.sender.memory
SenderMemory Objects
class SenderMemory(ProtocolMemory)
Manages the memory for the Sender, including protocol suitability and task conversations.
__init__
def __init__(storage: Storage)
Initializes SenderMemory with a storage backend.
Arguments:
storage
Storage - The storage backend for memory.
get_suitability
def get_suitability(protocol_id: str, task_id: str,
target: Optional[str]) -> Suitability
Retrieves the suitability status for a given protocol ID and task ID.
Arguments:
protocol_id
str - The protocol identifier.task_id
str - The task identifier.target
Optional[str] - The target system or service.
Returns:
Suitability
- The stored suitability status.
get_known_suitable_protocol_ids
def get_known_suitable_protocol_ids(task_id, target)
Returns known suitable protocol IDs for the given task and target.
Arguments:
task_id
- The task identifier.target
- The target system or service.
Returns:
list
- A list of known suitable protocol IDs.
get_suitable_protocol
def get_suitable_protocol(task_id, target) -> Optional[Protocol]
Retrieves a suitable protocol object for the given task and target if available.
Arguments:
task_id
- The task identifier.target
- The target system or service.
Returns:
Optional[Protocol]
- The first suitable protocol found or None if none available.
increment_task_conversations
def increment_task_conversations(task_id, target)
Increments the conversation counter for a given task and target.
Arguments:
task_id
- The task identifier.target
- The target system or service.
get_task_conversations
def get_task_conversations(task_id, target)
Retrieves the number of stored conversations for a task and target.
Arguments:
task_id
- The task identifier.target
- The target system or service.
Returns:
int
- The number of conversations.
increment_protocol_conversations
def increment_protocol_conversations(protocol_id)
Increments the conversation counter for a given protocol.
Arguments:
protocol_id
- The protocol identifier.
get_protocol_conversations
def get_protocol_conversations(protocol_id)
Retrieves the number of stored conversations for a protocol.
Arguments:
protocol_id
- The protocol identifier.
Returns:
int
- The number of conversations.
has_suitable_protocol
def has_suitable_protocol(task_id, target)
Checks whether a suitable protocol exists for a given task and target.
Arguments:
task_id
- The task identifier.target
- The target system or service.
Returns:
bool
- True if a suitable protocol exists, otherwise False.
get_unclassified_protocols
def get_unclassified_protocols(task_id)
Get protocols that have not been classified for a specific task.
Arguments:
task_id
- The identifier of the task.
Returns:
List[str]
- A list of unclassified protocol IDs.
set_default_suitability
def set_default_suitability(protocol_id: str, task_id: str,
suitability: Suitability)
Set the default suitability for a protocol and task.
Arguments:
protocol_id
str - The identifier of the protocol.task_id
str - The identifier of the task.suitability
Suitability - The default suitability status to set.
set_suitability_override
def set_suitability_override(protocol_id: str, task_id: str, target: str,
suitability: Suitability)
Override the suitability of a protocol for a specific task and target.
Arguments:
protocol_id
str - The identifier of the protocol.task_id
str - The identifier of the task.target
str - The target for which the suitability is overridden.suitability
Suitability - The overridden suitability status.
register_new_protocol
def register_new_protocol(protocol_id: str, protocol_document: str,
sources: list, metadata: dict)
Register a new protocol with the given sources, document, and metadata.
Arguments:
protocol_id
str - The identifier of the new protocol.protocol_document
str - The document describing the protocol.sources
list - A list of sources where the protocol is referenced.metadata
dict - Additional metadata related to the protocol.