Skip to content

Short Term Memory Module

Service

unique_toolkit.short_term_memory.service

ShortTermMemoryService

Provides methods to manage short term memory.

Source code in unique_toolkit/unique_toolkit/short_term_memory/service.py 
 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
 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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
class ShortTermMemoryService:
    """
    Provides methods to manage short term memory.
    """

    @deprecated(
        "Use __init__ with company_id and user_id instead or use the classmethod `from_event`"
    )
    @overload
    def __init__(self, event: Event | ChatEvent | BaseEvent): ...

    """
        Initialize the ShortTermMemoryService with an event (deprecated)
    """

    @overload
    def __init__(
        self,
        *,
        company_id: str,
        user_id: str,
        chat_id: str | None,
        message_id: str | None,
    ): ...

    """
        Initialize the ShortTermMemoryService with a company_id, user_id, chat_id and message_id.
    """

    def __init__(
        self,
        event: Event | ChatEvent | BaseEvent | None = None,
        user_id: str | None = None,
        company_id: str | None = None,
        chat_id: str | None = None,
        message_id: str | None = None,
    ):
        self._event = event
        if event:
            self._company_id: str = event.company_id
            self._user_id: str = event.user_id
            if isinstance(event, (ChatEvent, Event)):
                self._chat_id = event.payload.chat_id
                self._message_id = event.payload.user_message.id
        else:
            [company_id, user_id] = validate_required_values([company_id, user_id])

            if not (chat_id or message_id):
                raise ValueError("Chat_id or message_id must be provided")

            self._company_id: str = company_id
            self._user_id: str = user_id
            self._chat_id: str | None = chat_id
            self._message_id: str | None = message_id

    @classmethod
    def from_event(cls, event: ChatEvent):
        """
        Initialize the ShortTermMemoryService with a chat event.
        """
        return cls(
            company_id=event.company_id,
            user_id=event.user_id,
            chat_id=event.payload.chat_id,
            message_id=event.payload.user_message.id,
        )

    @property
    @deprecated(
        "The event property is deprecated and will be removed in a future version."
    )
    def event(self) -> Event | BaseEvent | None:
        """
        Get the event object (deprecated).

        Returns:
            Event | BaseEvent | None: The event object.
        """
        return self._event

    @property
    @deprecated(
        "The company_id property is deprecated and will be removed in a future version."
    )
    def company_id(self) -> str | None:
        """
        Get the company identifier (deprecated).

        Returns:
            str | None: The company identifier.
        """
        return self._company_id

    @company_id.setter
    @deprecated(
        "The company_id setter is deprecated and will be removed in a future version."
    )
    def company_id(self, value: str) -> None:
        """
        Set the company identifier (deprecated).

        Args:
            value (str | None): The company identifier.
        """
        self._company_id = value

    @property
    @deprecated(
        "The user_id property is deprecated and will be removed in a future version."
    )
    def user_id(self) -> str | None:
        """
        Get the user identifier (deprecated).

        Returns:
            str | None: The user identifier.
        """
        return self._user_id

    @user_id.setter
    @deprecated(
        "The user_id setter is deprecated and will be removed in a future version."
    )
    def user_id(self, value: str) -> None:
        """
        Set the user identifier (deprecated).

        Args:
            value (str | None): The user identifier.
        """
        self._user_id = value

    @property
    @deprecated(
        "The chat_id property is deprecated and will be removed in a future version."
    )
    def chat_id(self) -> str | None:
        """
        Get the chat identifier (deprecated).

        Returns:
            str | None: The chat identifier.
        """
        return self._chat_id

    @chat_id.setter
    @deprecated(
        "The chat_id setter is deprecated and will be removed in a future version."
    )
    def chat_id(self, value: str | None) -> None:
        """
        Set the chat identifier (deprecated).

        Args:
            value (str | None): The chat identifier.
        """
        self._chat_id = value

    @property
    @deprecated(
        "The message_id property is deprecated and will be removed in a future version."
    )
    def message_id(self) -> str | None:
        """
        Get the message identifier (deprecated).

        Returns:
            str | None: The message identifier.
        """
        return self._message_id

    @message_id.setter
    @deprecated(
        "The message_id setter is deprecated and will be removed in a future version."
    )
    def message_id(self, value: str | None) -> None:
        """
        Set the message identifier (deprecated).

        Args:
            value (str | None): The message identifier.
        """
        self._message_id = value

    @classmethod
    @deprecated("Instantiate class directly from event")
    def from_chat_event(cls, chat_event: Event) -> "ShortTermMemoryService":
        return cls(
            user_id=chat_event.user_id,
            company_id=chat_event.company_id,
            chat_id=chat_event.payload.chat_id,
            message_id=chat_event.payload.user_message.id,
        )

    async def find_latest_memory_async(self, key: str) -> ShortTermMemory:
        """
        Find the latest short term memory.

        Args:
            key (str): The key.

        Returns:
            ShortTermMemory: The latest short term memory.

        Raises:
            Exception: If an error occurs.
        """

        return await find_latest_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            chat_id=self._chat_id,
            message_id=self._message_id,
        )

    def find_latest_memory(self, key: str) -> ShortTermMemory:
        """
        Find the latest short term memory.

        Args:
            key (str): The key.

        Returns:
            ShortTermMemory: The latest short term memory.

        Raises:
            Exception: If an error occurs.
        """

        return find_latest_memory(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            chat_id=self._chat_id,
            message_id=self._message_id,
        )

    async def create_memory_async(self, key: str, value: str | dict):
        """
        Create a short term memory.

        Args:
            key (str): The key.
            value (str | dict): The value.

        Returns:
            ShortTermMemory: The created short term memory.

        Raises:
            Exception: If an error occurs.
        """

        return await create_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            value=value,
            chat_id=self._chat_id,
            message_id=self._message_id,
        )

    def create_memory(self, key: str, value: str | dict):
        """
        Create a short term memory.

        Args:
            key (str): The key.
            value (str | dict): The value.
        Returns:
            ShortTermMemory: The created short term memory.

        Raises:
            Exception: If an error occurs.
        """

        return create_memory(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            value=value,
            chat_id=self._chat_id,
            message_id=self._message_id,
        )

    @deprecated("Use create_memory_async instead")
    async def set(self, key: str, value: str | dict):
        return await create_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            value=value,
            chat_id=self._chat_id,
            message_id=self._message_id,
        )

    @deprecated("Use find_latest_memory_async instead")
    async def get(self, key: str) -> ShortTermMemory:
        return await find_latest_memory_async(
            user_id=self._user_id,
            company_id=self._company_id,
            key=key,
            chat_id=self._chat_id,
            message_id=self._message_id,
        )

chat_id property writable

Get the chat identifier (deprecated).

Returns:

Type Description
str | None

str | None: The chat identifier.

company_id property writable

Get the company identifier (deprecated).

Returns:

Type Description
str | None

str | None: The company identifier.

event property

Get the event object (deprecated).

Returns:

Type Description
Event | BaseEvent | None

Event | BaseEvent | None: The event object.

message_id property writable

Get the message identifier (deprecated).

Returns:

Type Description
str | None

str | None: The message identifier.

user_id property writable

Get the user identifier (deprecated).

Returns:

Type Description
str | None

str | None: The user identifier.

create_memory(key, value)

Create a short term memory.

Parameters:

Name Type Description Default
key str

The key.

 required
value str | dict

The value.

 required

Raises:

Type Description
Exception

If an error occurs.
Source code in unique_toolkit/unique_toolkit/short_term_memory/service.py 
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def create_memory(self, key: str, value: str | dict):
    """
    Create a short term memory.

    Args:
        key (str): The key.
        value (str | dict): The value.
    Returns:
        ShortTermMemory: The created short term memory.

    Raises:
        Exception: If an error occurs.
    """

    return create_memory(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        value=value,
        chat_id=self._chat_id,
        message_id=self._message_id,
    )

create_memory_async(key, value) async

Create a short term memory.

Parameters:

Name Type Description Default
key str

The key.

 required
value str | dict

The value.

 required

Returns:

Name Type Description
ShortTermMemory

The created short term memory.

Raises:

Type Description
Exception

If an error occurs.
Source code in unique_toolkit/unique_toolkit/short_term_memory/service.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
async def create_memory_async(self, key: str, value: str | dict):
    """
    Create a short term memory.

    Args:
        key (str): The key.
        value (str | dict): The value.

    Returns:
        ShortTermMemory: The created short term memory.

    Raises:
        Exception: If an error occurs.
    """

    return await create_memory_async(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        value=value,
        chat_id=self._chat_id,
        message_id=self._message_id,
    )

find_latest_memory(key)

Find the latest short term memory.

Parameters:

Name Type Description Default
key str

The key.

 required

Returns:

Name Type Description
ShortTermMemory ShortTermMemory

The latest short term memory.

Raises:

Type Description
Exception

If an error occurs.
Source code in unique_toolkit/unique_toolkit/short_term_memory/service.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def find_latest_memory(self, key: str) -> ShortTermMemory:
    """
    Find the latest short term memory.

    Args:
        key (str): The key.

    Returns:
        ShortTermMemory: The latest short term memory.

    Raises:
        Exception: If an error occurs.
    """

    return find_latest_memory(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        chat_id=self._chat_id,
        message_id=self._message_id,
    )

find_latest_memory_async(key) async

Find the latest short term memory.

Parameters:

Name Type Description Default
key str

The key.

 required

Returns:

Name Type Description
ShortTermMemory ShortTermMemory

The latest short term memory.

Raises:

Type Description
Exception

If an error occurs.
Source code in unique_toolkit/unique_toolkit/short_term_memory/service.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
async def find_latest_memory_async(self, key: str) -> ShortTermMemory:
    """
    Find the latest short term memory.

    Args:
        key (str): The key.

    Returns:
        ShortTermMemory: The latest short term memory.

    Raises:
        Exception: If an error occurs.
    """

    return await find_latest_memory_async(
        user_id=self._user_id,
        company_id=self._company_id,
        key=key,
        chat_id=self._chat_id,
        message_id=self._message_id,
    )

from_event(event) classmethod

Initialize the ShortTermMemoryService with a chat event.

Source code in unique_toolkit/unique_toolkit/short_term_memory/service.py 
73
74
75
76
77
78
79
80
81
82
83
@classmethod
def from_event(cls, event: ChatEvent):
    """
    Initialize the ShortTermMemoryService with a chat event.
    """
    return cls(
        company_id=event.company_id,
        user_id=event.user_id,
        chat_id=event.payload.chat_id,
        message_id=event.payload.user_message.id,
    )