memoryscope.core.operation

class memoryscope.core.operation.BackendOperation(interval_time: int, **kwargs)

基类：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__(interval_time: int, **kwargs)

Initializes a new instance of the BaseOperation.

init_workflow(**kwargs)

Initializes the workflow by setting up workers with provided keyword arguments.

参数:

**kwargs -- Arbitrary keyword arguments to be passed during worker initialization.

start_operation_backend(**kwargs)

Initiates the background operation loop if it's not already running. Sets the _loop_switch to True and submits the _loop_operation to a thread from the global thread pool.

stop_operation_backend(wait_operation: bool = False)

Stops the background operation loop by setting the _loop_switch to False.

class memoryscope.core.operation.BaseOperation(name: str, user_name: str, target_names: List[str], chat_messages: List[List[Message]], description: str)

基类：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)

Initializes a new instance of the BaseOperation.

init_workflow(**kwargs)

Initialize the workflow with additional keyword arguments if needed.

参数:

**kwargs -- Additional parameters for initializing the workflow.

abstract run_operation(target_name: str, **kwargs)

Abstract method to define the operation to be run. Subclasses must implement this method.

参数:

• target_name (str) -- target_name(human name).

• **kwargs -- Keyword arguments for running the operation.

抛出:

NotImplementedError -- If the subclass does not implement this method.

start_operation_backend(**kwargs)

Placeholder method for running an operation specific to the backend. Intended to be overridden by subclasses if backend operations are required.

stop_operation_backend(wait_operation: bool = False)

Placeholder method to stop any ongoing backend operations. Should be implemented in subclasses where backend operations are managed.

class memoryscope.core.operation.BaseWorkflow(name: str, memoryscope_context: ~memoryscope.core.utils.singleton.singleton.<locals>._singleton, workflow: str = '', **kwargs)

基类：object

__init__(name: str, memoryscope_context: ~memoryscope.core.utils.singleton.singleton.<locals>._singleton, workflow: str = '', **kwargs)

workflow_print_console(*args, **kwargs)

init_workers(is_backend: bool = False, **kwargs)

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.

参数:

• 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.

抛出:

RuntimeError -- If a worker mentioned in self.worker_dict does not exist in G_CONTEXT.worker_config.

备注

This method modifies self.worker_dict in-place, replacing the keys with actual worker instances.

run_workflow(**kwargs)

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.

参数:

**kwargs -- Additional keyword arguments to be passed to context.

class memoryscope.core.operation.ConsolidateMemoryOp(message_lock, contextual_msg_min_count: int = 0, **kwargs)

基类：BackendOperation

__init__(message_lock, contextual_msg_min_count: int = 0, **kwargs)

Initializes a new instance of the BaseOperation.

run_operation(target_name: str, **kwargs)

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.

参数:

• target_name (str) -- target_name(human name).

• **kwargs -- Keyword arguments for chat operation configuration.

返回:

The result obtained from running the workflow.

返回类型:

Any

class memoryscope.core.operation.FrontendOperation(name: str, user_name: str, target_names: List[str], chat_messages: List[List[Message]], description: str, **kwargs)

基类：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)

Initializes a new instance of the BaseOperation.

init_workflow(**kwargs)

Initializes the workflow by setting up workers with provided keyword arguments.

参数:

**kwargs -- Arbitrary keyword arguments to be passed during worker initialization.

run_operation(target_name: str, **kwargs)

Executes the main operation of reading recent chat messages, initializing workflow, and returning the result of the workflow execution.

参数:

• target_name (str) -- target_name(human name).

• **kwargs -- Additional keyword arguments used in the operation context.

返回:

The result obtained from executing the workflow.

返回类型:

Any