Skip to content

neurotrace.core.hippocampus.stm

neurotrace.core.hippocampus.stm

BaseShortTermMemory

Bases: ABC

Abstract base class for short-term memory management.

This class defines the interface for managing a temporary message store with token limit constraints.

Source code in neurotrace/core/hippocampus/stm.py 
10
11
12
13
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
class BaseShortTermMemory(ABC):
    """Abstract base class for short-term memory management.

    This class defines the interface for managing a temporary message store
    with token limit constraints.
    """

    @abstractmethod
    def append(self, message: Message) -> None:
        """Add a message to short-term memory.

        Args:
            message (Message): The message to be added to memory.
        """
        ...

    @abstractmethod
    def get_messages(self) -> List[Message]:
        """Retrieve all messages from short-term memory.

        Returns:
            List[Message]: List of all messages currently in memory.
        """
        ...

    @abstractmethod
    def clear(self) -> None:
        """Remove all messages from short-term memory."""
        ...

    @abstractmethod
    def set_messages(self, messages: List[Message]) -> None:
        """Replace all messages in memory with a new list.

        Args:
            messages (List[Message]): New list of messages to store.
        """
        ...

    @abstractmethod
    def total_tokens(self) -> int:
        """Calculate total tokens used by all messages.

        Returns:
            int: Sum of estimated token lengths of all messages.
        """
        ...

    def __len__(self) -> int:
        """Get the number of messages in memory.

        Returns:
            int: Count of messages currently stored.
        """
        return len(self.get_messages())

__len__()

Get the number of messages in memory.

Returns:

Name Type Description
int int

Count of messages currently stored.
Source code in neurotrace/core/hippocampus/stm.py 
58
59
60
61
62
63
64
def __len__(self) -> int:
    """Get the number of messages in memory.

    Returns:
        int: Count of messages currently stored.
    """
    return len(self.get_messages())

append(message) abstractmethod

Add a message to short-term memory.

Parameters:

Name Type Description Default
message Message

The message to be added to memory.

 required
Source code in neurotrace/core/hippocampus/stm.py 
17
18
19
20
21
22
23
24
@abstractmethod
def append(self, message: Message) -> None:
    """Add a message to short-term memory.

    Args:
        message (Message): The message to be added to memory.
    """
    ...

clear() abstractmethod

Remove all messages from short-term memory.

Source code in neurotrace/core/hippocampus/stm.py 
35
36
37
38
@abstractmethod
def clear(self) -> None:
    """Remove all messages from short-term memory."""
    ...

get_messages() abstractmethod

Retrieve all messages from short-term memory.

Returns:

Type Description
List[Message]

List[Message]: List of all messages currently in memory.
Source code in neurotrace/core/hippocampus/stm.py 
26
27
28
29
30
31
32
33
@abstractmethod
def get_messages(self) -> List[Message]:
    """Retrieve all messages from short-term memory.

    Returns:
        List[Message]: List of all messages currently in memory.
    """
    ...

set_messages(messages) abstractmethod

Replace all messages in memory with a new list.

Parameters:

Name Type Description Default
messages List[Message]

New list of messages to store.

 required
Source code in neurotrace/core/hippocampus/stm.py 
40
41
42
43
44
45
46
47
@abstractmethod
def set_messages(self, messages: List[Message]) -> None:
    """Replace all messages in memory with a new list.

    Args:
        messages (List[Message]): New list of messages to store.
    """
    ...

total_tokens() abstractmethod

Calculate total tokens used by all messages.

Returns:

Name Type Description
int int

Sum of estimated token lengths of all messages.
Source code in neurotrace/core/hippocampus/stm.py 
49
50
51
52
53
54
55
56
@abstractmethod
def total_tokens(self) -> int:
    """Calculate total tokens used by all messages.

    Returns:
        int: Sum of estimated token lengths of all messages.
    """
    ...

ShortTermMemory

Bases: BaseShortTermMemory

Implementation of token-limited short-term memory.

This class maintains a list of messages while ensuring the total token count stays within a specified limit. When the limit is exceeded, older messages are automatically evicted.

Parameters:

Name Type Description Default
max_tokens int

Maximum number of tokens to store. Set to 0 to disable memory (all messages will be evicted). Defaults to 2048.

 2048
Source code in neurotrace/core/hippocampus/stm.py 
 67
 68
 69
 70
 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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class ShortTermMemory(BaseShortTermMemory):
    """Implementation of token-limited short-term memory.

    This class maintains a list of messages while ensuring the total token
    count stays within a specified limit. When the limit is exceeded, older
    messages are automatically evicted.

    Args:
        max_tokens (int, optional): Maximum number of tokens to store. Set to 0
            to disable memory (all messages will be evicted). Defaults to 2048.
    """

    def __init__(self, max_tokens: int = 2048):
        """Initialize short-term memory with token limit.

        Args:
            max_tokens (int, optional): Maximum number of tokens to store.
                Defaults to 2048.
        """
        self.messages: List[Message] = []
        self.max_tokens = max_tokens

    def append(self, message: Message) -> None:
        """Add a message to memory and evict old messages if needed.

        If the message doesn't have an ID, generates a UUID for it. After
        adding the message, ensures token limit compliance by evicting old
        messages if necessary.

        Args:
            message (Message): The message to add to memory.
        """
        if not message.id:
            message.id = str(uuid.uuid4())

        self.messages.append(message)
        MemoryLogger.log_add(message, destination="stm")
        self._evict_if_needed()

    def get_messages(self) -> List[Message]:
        """Get all messages currently in memory.

        Returns:
            List[Message]: List of all stored messages.
        """
        return self.messages

    def clear(self) -> None:
        """Remove all messages from memory."""
        self.messages = []

    def _evict_if_needed(self) -> None:
        """Maintain token limit by removing oldest messages.

        If max_tokens is 0, clears all messages. Otherwise, removes oldest
        messages until total token count is within limit, always keeping at
        least one message.
        """
        # If max_tokens is 0, clear everything (user wants no memory)
        if self.max_tokens == 0:
            self.messages.clear()
            return

        total = sum(msg.estimated_token_length() for msg in self.messages)

        # Keep at least 1 message even if over limit (unless max_tokens is zero)
        while total > self.max_tokens and len(self.messages) > 1:
            evicted = self.messages.pop(0)
            total -= evicted.estimated_token_length()
            MemoryLogger.log_evict(evicted)

    def set_messages(self, messages: List[Message]) -> None:
        """Replace current messages with new list and maintain token limit.

        Args:
            messages (List[Message]): New messages to store in memory.
        """
        self.messages = messages
        self._evict_if_needed()

    def total_tokens(self) -> int:
        """Calculate total tokens used by all messages.

        Returns:
            int: Sum of estimated token lengths across all messages.
        """
        return sum(m.estimated_token_length() for m in self.messages)

    def __len__(self):
        """Get number of messages in memory.

        Returns:
            int: Count of stored messages.
        """
        return len(self.messages)

    def __repr__(self):
        """Get string representation of memory state.

        Returns:
            str: String showing message count and token usage/limit.
        """
        return f"<STM messages={len(self.messages)} tokens={self.total_tokens()}/{self.max_tokens}>"

__init__(max_tokens=2048)

Initialize short-term memory with token limit.

Parameters:

Name Type Description Default
max_tokens int

Maximum number of tokens to store. Defaults to 2048.

 2048
Source code in neurotrace/core/hippocampus/stm.py 
79
80
81
82
83
84
85
86
87
def __init__(self, max_tokens: int = 2048):
    """Initialize short-term memory with token limit.

    Args:
        max_tokens (int, optional): Maximum number of tokens to store.
            Defaults to 2048.
    """
    self.messages: List[Message] = []
    self.max_tokens = max_tokens

__len__()

Get number of messages in memory.

Returns:

Name Type Description
int

Count of stored messages.
Source code in neurotrace/core/hippocampus/stm.py 
155
156
157
158
159
160
161
def __len__(self):
    """Get number of messages in memory.

    Returns:
        int: Count of stored messages.
    """
    return len(self.messages)

__repr__()

Get string representation of memory state.

Returns:

Name Type Description
str

String showing message count and token usage/limit.
Source code in neurotrace/core/hippocampus/stm.py 
163
164
165
166
167
168
169
def __repr__(self):
    """Get string representation of memory state.

    Returns:
        str: String showing message count and token usage/limit.
    """
    return f"<STM messages={len(self.messages)} tokens={self.total_tokens()}/{self.max_tokens}>"

append(message)

Add a message to memory and evict old messages if needed.

If the message doesn't have an ID, generates a UUID for it. After adding the message, ensures token limit compliance by evicting old messages if necessary.

Parameters:

Name Type Description Default
message Message

The message to add to memory.

 required
Source code in neurotrace/core/hippocampus/stm.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def append(self, message: Message) -> None:
    """Add a message to memory and evict old messages if needed.

    If the message doesn't have an ID, generates a UUID for it. After
    adding the message, ensures token limit compliance by evicting old
    messages if necessary.

    Args:
        message (Message): The message to add to memory.
    """
    if not message.id:
        message.id = str(uuid.uuid4())

    self.messages.append(message)
    MemoryLogger.log_add(message, destination="stm")
    self._evict_if_needed()

clear()

Remove all messages from memory.

Source code in neurotrace/core/hippocampus/stm.py 
114
115
116
def clear(self) -> None:
    """Remove all messages from memory."""
    self.messages = []

get_messages()

Get all messages currently in memory.

Returns:

Type Description
List[Message]

List[Message]: List of all stored messages.
Source code in neurotrace/core/hippocampus/stm.py 
106
107
108
109
110
111
112
def get_messages(self) -> List[Message]:
    """Get all messages currently in memory.

    Returns:
        List[Message]: List of all stored messages.
    """
    return self.messages

set_messages(messages)

Replace current messages with new list and maintain token limit.

Parameters:

Name Type Description Default
messages List[Message]

New messages to store in memory.

 required
Source code in neurotrace/core/hippocampus/stm.py 
138
139
140
141
142
143
144
145
def set_messages(self, messages: List[Message]) -> None:
    """Replace current messages with new list and maintain token limit.

    Args:
        messages (List[Message]): New messages to store in memory.
    """
    self.messages = messages
    self._evict_if_needed()

total_tokens()

Calculate total tokens used by all messages.

Returns:

Name Type Description
int int

Sum of estimated token lengths across all messages.
Source code in neurotrace/core/hippocampus/stm.py 
147
148
149
150
151
152
153
def total_tokens(self) -> int:
    """Calculate total tokens used by all messages.

    Returns:
        int: Sum of estimated token lengths across all messages.
    """
    return sum(m.estimated_token_length() for m in self.messages)