memoryscope.core.operation
- class memoryscope.core.operation.BackendOperation(interval_time: int, **kwargs)[source]
Bases:
FrontendOperation
BaseBackendOperation serves as an abstract base class for defining backend operations. It manages operation status, loop control, and integrates with a global context for thread management.
- operation_type: Literal['frontend', 'backend'] = 'backend'
- init_workflow(**kwargs)[source]
Initializes the workflow by setting up workers with provided keyword arguments.
- Parameters:
**kwargs – Arbitrary keyword arguments to be passed during worker initialization.
- class memoryscope.core.operation.BaseOperation(name: str, user_name: str, target_names: List[str], chat_messages: List[List[Message]], description: str)[source]
Bases:
object
An abstract base class representing an operation that can be categorized as either frontend or backend.
- operation_type
Specifies the type of operation, defaulting to “frontend”.
- Type:
OPERATION_TYPE
- name
The name of the operation.
- Type:
str
- description
A description of the operation.
- Type:
str
- operation_type: Literal['frontend', 'backend'] = 'frontend'
- __init__(name: str, user_name: str, target_names: List[str], chat_messages: List[List[Message]], description: str)[source]
Initializes a new instance of the BaseOperation.
- init_workflow(**kwargs)[source]
Initialize the workflow with additional keyword arguments if needed.
- Parameters:
**kwargs – Additional parameters for initializing the workflow.
- abstract run_operation(target_name: str, **kwargs)[source]
Abstract method to define the operation to be run. Subclasses must implement this method.
- Parameters:
target_name (str) – target_name(human name).
**kwargs – Keyword arguments for running the operation.
- Raises:
NotImplementedError – If the subclass does not implement this method.
- class memoryscope.core.operation.BaseWorkflow(name: str, memoryscope_context: ~memoryscope.core.utils.singleton.singleton.<locals>._singleton, workflow: str = '', **kwargs)[source]
Bases:
object
- __init__(name: str, memoryscope_context: ~memoryscope.core.utils.singleton.singleton.<locals>._singleton, workflow: str = '', **kwargs)[source]
- init_workers(is_backend: bool = False, **kwargs)[source]
Initializes worker instances based on the configuration for each worker defined in G_CONTEXT.worker_config. Each worker can be set to run in a multithreaded mode depending on the is_backend flag or the worker’s individual configuration.
- Parameters:
is_backend (bool, optional) – A flag indicating whether the workers should be initialized in a
False. (backend context. Defaults to)
**kwargs – Additional keyword arguments to be passed during worker initialization.
- Raises:
RuntimeError – If a worker mentioned in self.worker_dict does not exist in G_CONTEXT.worker_config.
Note
This method modifies self.worker_dict in-place, replacing the keys with actual worker instances.
- run_workflow(**kwargs)[source]
Executes the workflow by orchestrating the steps defined in self.workflow_worker_list. This method supports both sequential and parallel execution of sub-workflows based on the structure of self.workflow_worker_list.
If a workflow part consists of a single item, it is executed sequentially. For parts with multiple items, they are submitted for parallel execution using a thread pool. The workflow will stop if any sub-workflow returns False.
- Parameters:
**kwargs – Additional keyword arguments to be passed to context.
- class memoryscope.core.operation.ConsolidateMemoryOp(message_lock, contextual_msg_min_count: int = 0, **kwargs)[source]
Bases:
BackendOperation
- __init__(message_lock, contextual_msg_min_count: int = 0, **kwargs)[source]
Initializes a new instance of the BaseOperation.
- run_operation(target_name: str, **kwargs)[source]
Executes an operation after preparing the chat context, checking message memory status, and updating workflow status accordingly.
If the number of not-memorized messages is less than the contextual message count, the operation is skipped. Otherwise, it sets up the chat context, runs the workflow, captures the result, and updates the memory status.
- Parameters:
target_name (str) – target_name(human name).
**kwargs – Keyword arguments for chat operation configuration.
- Returns:
The result obtained from running the workflow.
- Return type:
Any
- class memoryscope.core.operation.FrontendOperation(name: str, user_name: str, target_names: List[str], chat_messages: List[List[Message]], description: str, **kwargs)[source]
Bases:
BaseWorkflow
,BaseOperation
- operation_type: Literal['frontend', 'backend'] = 'frontend'
- __init__(name: str, user_name: str, target_names: List[str], chat_messages: List[List[Message]], description: str, **kwargs)[source]
Initializes a new instance of the BaseOperation.
- init_workflow(**kwargs)[source]
Initializes the workflow by setting up workers with provided keyword arguments.
- Parameters:
**kwargs – Arbitrary keyword arguments to be passed during worker initialization.
- run_operation(target_name: str, **kwargs)[source]
Executes the main operation of reading recent chat messages, initializing workflow, and returning the result of the workflow execution.
- Parameters:
target_name (str) – target_name(human name).
**kwargs – Additional keyword arguments used in the operation context.
- Returns:
The result obtained from executing the workflow.
- Return type:
Any