Add text to Mem0 memory.
(self, text: str, **kwargs)
| 289 | return f"context_{self._context_id}" if self._context_id else "default_user" |
| 290 | |
| 291 | def memorize(self, text: str, **kwargs) -> MemoryBuildResult: |
| 292 | """Add text to Mem0 memory.""" |
| 293 | user_id = self._get_user_id() |
| 294 | |
| 295 | chunks = self._split_text_into_chunks(text, self.max_input_tokens) |
| 296 | memory_entries = [] |
| 297 | |
| 298 | for chunk in chunks: |
| 299 | memory_messages = [ |
| 300 | {"role": "user", "content": chunk}, |
| 301 | {"role": "assistant", "content": "I'll make sure to remember this information."} |
| 302 | ] |
| 303 | result = self._memory.add(memory_messages, user_id=user_id) |
| 304 | if result and "results" in result: |
| 305 | for entry in result.get("results", []): |
| 306 | memory_entries.append({ |
| 307 | "event": entry.get("event", "ADD"), |
| 308 | "memory": entry.get("memory", ""), |
| 309 | "id": entry.get("id", ""), |
| 310 | }) |
| 311 | |
| 312 | self._memory_chunks.append(text) |
| 313 | self._is_initialized = True |
| 314 | |
| 315 | return MemoryBuildResult( |
| 316 | success=True, |
| 317 | method="mem0", |
| 318 | action="add_to_memory", |
| 319 | input_content=text, |
| 320 | stored_content=text, |
| 321 | memory_entries=memory_entries, |
| 322 | chunk_count=len(self._memory_chunks), |
| 323 | extra={"input_chunks": len(chunks)}, |
| 324 | ) |
| 325 | |
| 326 | def _retrieve(self, query: str) -> List[Dict[str, Any]]: |
| 327 | """Retrieve relevant memories from Mem0.""" |
nothing calls this directly
no test coverage detected