Memory Module

pipelines.pipelines.agents.memory.conversation_memory

ConversationMemory

A memory class that stores conversation history.

Source code in pipelines/pipelines/agents/memory/conversation_memory.py

class ConversationMemory(Memory):
    """
    A memory class that stores conversation history.
    """

    def __init__(self, input_key: str = "input", output_key: str = "output"):
        """
        Initialize ConversationMemory with input and output keys.

        :param input_key: The key to use for storing user input.
        :param output_key: The key to use for storing model output.
        """
        self.list: List[OrderedDict] = []
        self.input_key = input_key
        self.output_key = output_key

    def load(self, keys: Optional[List[str]] = None, **kwargs) -> str:
        """
        Load conversation history as a formatted string.

        :param keys: Optional list of keys (ignored in this implementation).
        :param kwargs: Optional keyword arguments
            - window_size: integer specifying the number of most recent conversation snippets to load.
        :return: A formatted string containing the conversation history.
        """
        chat_transcript = ""
        window_size = kwargs.get("window_size", None)

        if window_size is not None:
            chat_list = self.list[-window_size:]  # pylint: disable=invalid-unary-operand-type
        else:
            chat_list = self.list

        for chat_snippet in chat_list:
            chat_transcript += f"Human: {chat_snippet['Human']}\n"
            chat_transcript += f"AI: {chat_snippet['AI']}\n"
        return chat_transcript

    def save(self, data: Dict[str, Any]) -> None:
        """
        Save a conversation snippet to memory.

        :param data: A dictionary containing the conversation snippet to save.
        """
        chat_snippet = collections.OrderedDict()
        chat_snippet["Human"] = data[self.input_key]
        chat_snippet["AI"] = data[self.output_key]
        self.list.append(chat_snippet)

    def clear(self) -> None:
        """
        Clear the conversation history.
        """
        self.list = []

__init__

__init__(input_key: str = 'input', output_key: str = 'output')

Initialize ConversationMemory with input and output keys.

Parameters:

Name	Type	Description	Default
input_key	str	The key to use for storing user input.	'input'
output_key	str	The key to use for storing model output.	'output'

Source code in pipelines/pipelines/agents/memory/conversation_memory.py

def __init__(self, input_key: str = "input", output_key: str = "output"):
    """
    Initialize ConversationMemory with input and output keys.

    :param input_key: The key to use for storing user input.
    :param output_key: The key to use for storing model output.
    """
    self.list: List[OrderedDict] = []
    self.input_key = input_key
    self.output_key = output_key

clear

clear() -> None

Clear the conversation history.

Source code in pipelines/pipelines/agents/memory/conversation_memory.py

def clear(self) -> None:
    """
    Clear the conversation history.
    """
    self.list = []

load

load(keys: Optional[List[str]] = None, **kwargs) -> str

Load conversation history as a formatted string.

Parameters:

Name	Type	Description	Default
keys	Optional[List[str]]	Optional list of keys (ignored in this implementation).	None
kwargs		Optional keyword arguments - window_size: integer specifying the number of most recent conversation snippets to load.	{}

Returns:

Type	Description
str	A formatted string containing the conversation history.

Source code in pipelines/pipelines/agents/memory/conversation_memory.py

def load(self, keys: Optional[List[str]] = None, **kwargs) -> str:
    """
    Load conversation history as a formatted string.

    :param keys: Optional list of keys (ignored in this implementation).
    :param kwargs: Optional keyword arguments
        - window_size: integer specifying the number of most recent conversation snippets to load.
    :return: A formatted string containing the conversation history.
    """
    chat_transcript = ""
    window_size = kwargs.get("window_size", None)

    if window_size is not None:
        chat_list = self.list[-window_size:]  # pylint: disable=invalid-unary-operand-type
    else:
        chat_list = self.list

    for chat_snippet in chat_list:
        chat_transcript += f"Human: {chat_snippet['Human']}\n"
        chat_transcript += f"AI: {chat_snippet['AI']}\n"
    return chat_transcript

save

save(data: Dict[str, Any]) -> None

Save a conversation snippet to memory.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	A dictionary containing the conversation snippet to save.	required

Source code in pipelines/pipelines/agents/memory/conversation_memory.py

def save(self, data: Dict[str, Any]) -> None:
    """
    Save a conversation snippet to memory.

    :param data: A dictionary containing the conversation snippet to save.
    """
    chat_snippet = collections.OrderedDict()
    chat_snippet["Human"] = data[self.input_key]
    chat_snippet["AI"] = data[self.output_key]
    self.list.append(chat_snippet)

pipelines.pipelines.agents.memory.no_memory

NoMemory

A memory class that doesn't store any data.

Source code in pipelines/pipelines/agents/memory/no_memory.py

class NoMemory(Memory):
    """
    A memory class that doesn't store any data.
    """

    def load(self, keys: Optional[List[str]] = None, **kwargs) -> str:
        """
        Load an empty dictionary.

        :param keys: Optional list of keys (ignored in this implementation).
        :return: An empty str.
        """
        return ""

    def save(self, data: Dict[str, Any]) -> None:
        """
        Save method that does nothing.

        :param data: A dictionary containing the data to save (ignored in this implementation).
        """
        pass

    def clear(self) -> None:
        """
        Clear method that does nothing.
        """
        pass

clear

clear() -> None

Clear method that does nothing.

Source code in pipelines/pipelines/agents/memory/no_memory.py

def clear(self) -> None:
    """
    Clear method that does nothing.
    """
    pass

load

load(keys: Optional[List[str]] = None, **kwargs) -> str

Load an empty dictionary.

Parameters:

Name	Type	Description	Default
keys	Optional[List[str]]	Optional list of keys (ignored in this implementation).	None

Returns:

Type	Description
str	An empty str.

Source code in pipelines/pipelines/agents/memory/no_memory.py

def load(self, keys: Optional[List[str]] = None, **kwargs) -> str:
    """
    Load an empty dictionary.

    :param keys: Optional list of keys (ignored in this implementation).
    :return: An empty str.
    """
    return ""

save

save(data: Dict[str, Any]) -> None

Save method that does nothing.

Parameters:

Name	Type	Description	Default
data	Dict[str, Any]	A dictionary containing the data to save (ignored in this implementation).	required

Source code in pipelines/pipelines/agents/memory/no_memory.py

def save(self, data: Dict[str, Any]) -> None:
    """
    Save method that does nothing.

    :param data: A dictionary containing the data to save (ignored in this implementation).
    """
    pass