memoryscope.core.worker.backend

class memoryscope.core.worker.backend.ContraRepeatWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

The ContraRepeatWorker class specializes in processing memory nodes to identify and handle contradictory and repetitive information. It extends the base functionality of MemoryBaseWorker.

Responsibilities: - Collects observation nodes from various memory categories. - Constructs a prompt with these observations for language model analysis. - Parses the model’s response to detect contradictions or redundancies. - Adjusts the status of memory nodes based on the analysis. - Persists the updated node statuses back into memory.

FILE_PATH: str = '/home/runner/work/MemoryScope/MemoryScope/memoryscope/core/worker/backend/contra_repeat_worker.py'

class memoryscope.core.worker.backend.GetObservationWithTimeWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: GetObservationWorker

A specialized worker class that extends GetObservationWorker functionality to handle retrieval of observations which include associated timestamp information from chat messages.

FILE_PATH: str = '/home/runner/work/MemoryScope/MemoryScope/memoryscope/core/worker/backend/get_observation_with_time_worker.py'

OBS_STORE_KEY: str = 'new_obs_with_time_nodes'

filter_messages() → List[Message][source]

Filters the chat messages to only include those which contain time-related keywords.

Returns:: A list of filtered messages that mention time.
Return type:: List[Message]

build_message(filter_messages: List[Message]) → List[Message][source]

Constructs a prompt message for obtaining observations with timestamp information: based on filtered chat messages.

This method processes each filtered message with the timestamp information. It then organizes these timestamped messages into a structured prompt that includes a system prompt, few-shot examples, and the concatenated user queries.

Parameters:: filter_messages (List[Message]) – A list of Message objects that have been filtered for processing.
Returns:: A list containing the newly constructed Message object for further interaction.
Return type:: List[Message]

class memoryscope.core.worker.backend.GetObservationWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

A specialized worker class to generate the observations from the original chat histories.

FILE_PATH: str = '/home/runner/work/MemoryScope/MemoryScope/memoryscope/core/worker/backend/get_observation_worker.py'

OBS_STORE_KEY: str = 'new_obs_nodes'

add_observation(message: Message, time_infer: str, obs_content: str, keywords: str)[source]

Builds a MemoryNode containing the observation details.

Parameters:

message (Message) – The source message from which the observation is derived.
time_infer (str) – The inferred time if available.
obs_content (str) – The content of the observation.
keywords (str) – Keywords associated with the observation.

Returns:

The constructed MemoryNode containing the observation.

Return type:

MemoryNode

filter_messages() → List[Message][source]

Filters the chat messages to only include those which not contain time-related keywords.

Returns:: A list of filtered messages that mention time.
Return type:: List[Message]

build_message(filter_messages: List[Message]) → List[Message][source]

Constructs a formatted message for observation based on input messages, incorporating system prompts, few-shot examples, and user queries.

Parameters:: filter_messages (List[Message]) – A list of messages filtered for observation processing.
Returns:: A list containing the constructed message ready for observation.
Return type:: List[Message]

class memoryscope.core.worker.backend.GetReflectionSubjectWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

A specialized worker class responsible for retrieving unreflected memory nodes, generating reflection prompts with current insights, invoking an LLM for fresh insights, parsing the LLM responses, forming new insight nodes, and updating memory statuses accordingly.

FILE_PATH: str = '/home/runner/work/MemoryScope/MemoryScope/memoryscope/core/worker/backend/get_reflection_subject_worker.py'

new_insight_node(insight_key: str) → MemoryNode[source]

Creates a new MemoryNode for an insight with the given key, enriched with current datetime metadata.

Parameters:: insight_key (str) – The unique identifier for the insight.
Returns:: A new MemoryNode instance representing the insight, marked as new and of type INSIGHT.
Return type:: MemoryNode

class memoryscope.core.worker.backend.InfoFilterWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

This worker filters and modifies the chat message history (self.chat_messages) by retaining only the messages that include significant information. It then constructs a prompt from these filtered messages, utilizes an AI model to process this prompt, parses the AI’s generated response to allocate scores, and ultimately retains messages in self.chat_messages based on these assigned scores.

FILE_PATH: str = '/home/runner/work/MemoryScope/MemoryScope/memoryscope/core/worker/backend/info_filter_worker.py'

class memoryscope.core.worker.backend.LoadMemoryWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

retrieve_not_reflected_memory(**kwargs)

The wrapper function that manages the timing of the original function.

Parameters:

*args – Variable length argument list for the decorated function.
**kwargs – Arbitrary keyword arguments for the decorated function.

Returns:

The result of the decorated function.

Return type:

Any

retrieve_not_updated_memory(**kwargs)

The wrapper function that manages the timing of the original function.

Parameters:

*args – Variable length argument list for the decorated function.
**kwargs – Arbitrary keyword arguments for the decorated function.

Returns:

The result of the decorated function.

Return type:

Any

retrieve_insight_memory(**kwargs)

The wrapper function that manages the timing of the original function.

Parameters:

*args – Variable length argument list for the decorated function.
**kwargs – Arbitrary keyword arguments for the decorated function.

Returns:

The result of the decorated function.

Return type:

Any

retrieve_today_memory(**kwargs)

The wrapper function that manages the timing of the original function.

Parameters:

*args – Variable length argument list for the decorated function.
**kwargs – Arbitrary keyword arguments for the decorated function.

Returns:

The result of the decorated function.

Return type:

Any

class memoryscope.core.worker.backend.LongContraRepeatWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

Manages and updates memory entries within a conversation scope by identifying and handling contradictions or redundancies. It extends the base MemoryBaseWorker to provide specialized functionality for long conversations with potential contradictory or repetitive statements.

FILE_PATH: str = '/home/runner/work/MemoryScope/MemoryScope/memoryscope/core/worker/backend/long_contra_repeat_worker.py'

retrieve_similar_content(node: ~memoryscope.scheme.memory_node.MemoryNode) -> (<class 'memoryscope.scheme.memory_node.MemoryNode'>, typing.List[memoryscope.scheme.memory_node.MemoryNode])[source]

Retrieves memory nodes with content similar to the given node, filtering by user/target/status/memory_type. Only returns nodes whose similarity score meets or exceeds the predefined threshold.

Parameters:: node (MemoryNode) – The reference node used to find similar content in memory.
Returns:: A tuple containing the original node and a list of similar nodes that passed the similarity threshold.
Return type:: Tuple[MemoryNode, List[MemoryNode]]

class memoryscope.core.worker.backend.UpdateInsightWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

This class is responsible for updating insight value in a memory system. It filters insight nodes based on their association with observed nodes, utilizes a ranking model to prioritize them, generates refreshed insights via an LLM, and manages node statuses and content updates, incorporating features for concurrent execution and logging.

FILE_PATH: str = '/home/runner/work/MemoryScope/MemoryScope/memoryscope/core/worker/backend/update_insight_worker.py'

filter_obs_nodes(insight_node: ~memoryscope.scheme.memory_node.MemoryNode, obs_nodes: ~typing.List[~memoryscope.scheme.memory_node.MemoryNode]) -> (<class 'memoryscope.scheme.memory_node.MemoryNode'>, typing.List[memoryscope.scheme.memory_node.MemoryNode], <class 'float'>)[source]

Filters observed nodes based on their relevance to a given insight node using a ranking model.

Parameters:

insight_node (MemoryNode) – The insight node used as the basis for filtering.
obs_nodes (List[MemoryNode]) – A list of observed nodes to be filtered.

Returns:

A tuple containing:

The original insight node.
A list of filtered observed nodes that are relevant to the insight node.
The maximum relevance score among the filtered nodes.

Return type:

tuple

update_insight_node(insight_node: MemoryNode, insight_value: str)[source]

Updates the MemoryNode with the new insight value.

Parameters:

insight_node (MemoryNode) – The MemoryNode whose insight value needs to be updated.
insight_value (str) – The new insight value.

Returns:

The updated MemoryNode with potentially revised insight value.

Return type:

MemoryNode

update_insight(insight_node: MemoryNode, filtered_nodes: List[MemoryNode]) → MemoryNode[source]

Updates the insight value of a given MemoryNode based on the context from a list of filtered MemoryNodes.

Parameters:

insight_node (MemoryNode) – The MemoryNode whose insight value needs to be updated.
filtered_nodes (List[MemoryNode]) – A list of MemoryNodes used as context for updating the insight.

Returns:

The updated MemoryNode with potentially revised insight value.

Return type:

MemoryNode

class memoryscope.core.worker.backend.UpdateMemoryWorker(embedding_model: str = '', generation_model: str = '', rank_model: str = '', **kwargs)[source]

Bases: MemoryBaseWorker

from_query()[source]

Creates a MemoryNode from the provided query if present in chat_kwargs.

Returns:: A list containing a single MemoryNode created from the query.
Return type:: List[MemoryNode]

from_memory_key()[source]

Retrieves memories based on the memory key if it exists.

Returns:: A list of MemoryNode objects retrieved using the memory key.
Return type:: List[MemoryNode]

delete_all()[source]

Marks all memories for deletion by setting their action_status to ‘DELETE’.

Returns:: A list of all MemoryNode objects marked for deletion.
Return type:: List[MemoryNode]

delete_memory()[source]

Marks specific memories for deletion based on query or memory_id present in chat_kwargs.

Returns:: A list of MemoryNode objects marked for deletion based on the query or memory_id.
Return type:: List[MemoryNode]