Skip to content

neurotrace.core.hippocampus.ltm

neurotrace.core.hippocampus.ltm

BaseLongTermMemory

Bases: ABC

Abstract base class for long-term memory storage.

This class defines the interface for persistent storage of conversation messages. Implementations should handle the storage and retrieval of messages across different chat sessions.

Source code in neurotrace/core/hippocampus/ltm.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class BaseLongTermMemory(ABC):
    """Abstract base class for long-term memory storage.

    This class defines the interface for persistent storage of conversation
    messages. Implementations should handle the storage and retrieval of
    messages across different chat sessions.
    """

    @abstractmethod
    def add_message(self, message: Message) -> None:
        """Store a message in long-term memory.

        Args:
            message (Message): The message object to be stored.
        """
        pass

    @abstractmethod
    def add_user_message(self, content: str) -> None:
        """Add a user message to long-term memory.

        Args:
            content (str): The content of the user's message.
        """
        pass

    @abstractmethod
    def add_ai_message(self, content: str) -> None:
        """Add an AI message to long-term memory.

        Args:
            content (str): The content of the AI's message.
        """
        pass

    @abstractmethod
    def get_messages(self, session_id: str) -> List[Message]:
        """Retrieve all messages for a given session.

        Args:
            session_id (str): The identifier for the chat session.

        Returns:
            List[Message]: List of messages associated with the session.
        """
        pass

    @abstractmethod
    def clear(self, session_id: str) -> None:
        """Clear messages for a given session.

        Args:
            session_id (str): The identifier for the chat session to clear.
        """
        pass

add_ai_message(content) abstractmethod

Add an AI message to long-term memory.

Parameters:

Name Type Description Default
content str

The content of the AI's message.

 required
Source code in neurotrace/core/hippocampus/ltm.py 
40
41
42
43
44
45
46
47
@abstractmethod
def add_ai_message(self, content: str) -> None:
    """Add an AI message to long-term memory.

    Args:
        content (str): The content of the AI's message.
    """
    pass

add_message(message) abstractmethod

Store a message in long-term memory.

Parameters:

Name Type Description Default
message Message

The message object to be stored.

 required
Source code in neurotrace/core/hippocampus/ltm.py 
22
23
24
25
26
27
28
29
@abstractmethod
def add_message(self, message: Message) -> None:
    """Store a message in long-term memory.

    Args:
        message (Message): The message object to be stored.
    """
    pass

add_user_message(content) abstractmethod

Add a user message to long-term memory.

Parameters:

Name Type Description Default
content str

The content of the user's message.

 required
Source code in neurotrace/core/hippocampus/ltm.py 
31
32
33
34
35
36
37
38
@abstractmethod
def add_user_message(self, content: str) -> None:
    """Add a user message to long-term memory.

    Args:
        content (str): The content of the user's message.
    """
    pass

clear(session_id) abstractmethod

Clear messages for a given session.

Parameters:

Name Type Description Default
session_id str

The identifier for the chat session to clear.

 required
Source code in neurotrace/core/hippocampus/ltm.py 
61
62
63
64
65
66
67
68
@abstractmethod
def clear(self, session_id: str) -> None:
    """Clear messages for a given session.

    Args:
        session_id (str): The identifier for the chat session to clear.
    """
    pass

get_messages(session_id) abstractmethod

Retrieve all messages for a given session.

Parameters:

Name Type Description Default
session_id str

The identifier for the chat session.

 required

Returns:

Type Description
List[Message]

List[Message]: List of messages associated with the session.
Source code in neurotrace/core/hippocampus/ltm.py 
49
50
51
52
53
54
55
56
57
58
59
@abstractmethod
def get_messages(self, session_id: str) -> List[Message]:
    """Retrieve all messages for a given session.

    Args:
        session_id (str): The identifier for the chat session.

    Returns:
        List[Message]: List of messages associated with the session.
    """
    pass

LongTermMemory

Bases: BaseLongTermMemory

LangChain chat history adapter for long-term memory storage.

This adapter implements the BaseLongTermMemory interface using LangChain's chat history components for persistent storage.

Parameters:

Name Type Description Default
history BaseChatMessageHistory

LangChain chat history implementation to use for storage.

 required
session_id str

Default session identifier. Defaults to "default".

 'default'
Source code in neurotrace/core/hippocampus/ltm.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class LongTermMemory(BaseLongTermMemory):
    """LangChain chat history adapter for long-term memory storage.

    This adapter implements the BaseLongTermMemory interface using LangChain's
    chat history components for persistent storage.

    Args:
        history (BaseChatMessageHistory): LangChain chat history implementation
            to use for storage.
        session_id (str, optional): Default session identifier. Defaults to "default".
    """

    def __init__(self, history: BaseChatMessageHistory, session_id: str = "default"):
        """Initialize the LangChain history adapter.

        Args:
            history (BaseChatMessageHistory): LangChain chat history implementation.
            session_id (str, optional): Default session identifier. Defaults to "default".
        """
        self.history = history
        self.session_id = session_id

    def add_message(self, message: Message) -> None:
        """Add a message to the LangChain chat history.

        Converts the Message object to LangChain's message format before storing.

        Args:
            message (Message): The message to store.
        """
        lc_msg: BaseMessage = message.to_langchain_message()
        self.history.add_message(lc_msg)

    def add_user_message(self, content: str) -> None:
        """Add a user message to the chat history.

        Creates a Message object with HUMAN role and adds it to storage.

        Args:
            content (str): The content of the user's message.
        """
        self.add_message(Message(role=Role.HUMAN.value, content=content))

    def add_ai_message(self, content: str) -> None:
        """Add an AI message to the chat history.

        Creates a Message object with AI role and adds it to storage.

        Args:
            content (str): The content of the AI's message.
        """
        self.add_message(Message(role=Role.AI.value, content=content))

    def get_messages(self, session_id: str = None) -> List[Message]:
        """Retrieve all messages from the chat history.

        Args:
            session_id (str, optional): Session identifier. Currently unused as
                LangChain history doesn't support session filtering. Defaults to None.

        Returns:
            List[Message]: All messages in the chat history.
        """
        lc_msgs = self.history.messages
        return [from_langchain_message(m) for m in lc_msgs]

    def clear(self, session_id: str = None) -> None:
        """Clear all messages from the chat history.

        Args:
            session_id (str, optional): Session identifier. Currently unused as
                LangChain history doesn't support session filtering. Defaults to None.
        """
        self.history.clear()

__init__(history, session_id='default')

Initialize the LangChain history adapter.

Parameters:

Name Type Description Default
history BaseChatMessageHistory

LangChain chat history implementation.

 required
session_id str

Default session identifier. Defaults to "default".

 'default'
Source code in neurotrace/core/hippocampus/ltm.py 
83
84
85
86
87
88
89
90
91
def __init__(self, history: BaseChatMessageHistory, session_id: str = "default"):
    """Initialize the LangChain history adapter.

    Args:
        history (BaseChatMessageHistory): LangChain chat history implementation.
        session_id (str, optional): Default session identifier. Defaults to "default".
    """
    self.history = history
    self.session_id = session_id

add_ai_message(content)

Add an AI message to the chat history.

Creates a Message object with AI role and adds it to storage.

Parameters:

Name Type Description Default
content str

The content of the AI's message.

 required
Source code in neurotrace/core/hippocampus/ltm.py 
114
115
116
117
118
119
120
121
122
def add_ai_message(self, content: str) -> None:
    """Add an AI message to the chat history.

    Creates a Message object with AI role and adds it to storage.

    Args:
        content (str): The content of the AI's message.
    """
    self.add_message(Message(role=Role.AI.value, content=content))

add_message(message)

Add a message to the LangChain chat history.

Converts the Message object to LangChain's message format before storing.

Parameters:

Name Type Description Default
message Message

The message to store.

 required
Source code in neurotrace/core/hippocampus/ltm.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
def add_message(self, message: Message) -> None:
    """Add a message to the LangChain chat history.

    Converts the Message object to LangChain's message format before storing.

    Args:
        message (Message): The message to store.
    """
    lc_msg: BaseMessage = message.to_langchain_message()
    self.history.add_message(lc_msg)

add_user_message(content)

Add a user message to the chat history.

Creates a Message object with HUMAN role and adds it to storage.

Parameters:

Name Type Description Default
content str

The content of the user's message.

 required
Source code in neurotrace/core/hippocampus/ltm.py 
104
105
106
107
108
109
110
111
112
def add_user_message(self, content: str) -> None:
    """Add a user message to the chat history.

    Creates a Message object with HUMAN role and adds it to storage.

    Args:
        content (str): The content of the user's message.
    """
    self.add_message(Message(role=Role.HUMAN.value, content=content))

clear(session_id=None)

Clear all messages from the chat history.

Parameters:

Name Type Description Default
session_id str

Session identifier. Currently unused as LangChain history doesn't support session filtering. Defaults to None.

 None
Source code in neurotrace/core/hippocampus/ltm.py 
137
138
139
140
141
142
143
144
def clear(self, session_id: str = None) -> None:
    """Clear all messages from the chat history.

    Args:
        session_id (str, optional): Session identifier. Currently unused as
            LangChain history doesn't support session filtering. Defaults to None.
    """
    self.history.clear()

get_messages(session_id=None)

Retrieve all messages from the chat history.

Parameters:

Name Type Description Default
session_id str

Session identifier. Currently unused as LangChain history doesn't support session filtering. Defaults to None.

 None

Returns:

Type Description
List[Message]

List[Message]: All messages in the chat history.
Source code in neurotrace/core/hippocampus/ltm.py 
124
125
126
127
128
129
130
131
132
133
134
135
def get_messages(self, session_id: str = None) -> List[Message]:
    """Retrieve all messages from the chat history.

    Args:
        session_id (str, optional): Session identifier. Currently unused as
            LangChain history doesn't support session filtering. Defaults to None.

    Returns:
        List[Message]: All messages in the chat history.
    """
    lc_msgs = self.history.messages
    return [from_langchain_message(m) for m in lc_msgs]