Memory Lifecycle

Copyclass MemoryItem(BaseModel):
    id: str = Field(..., description="The unique identifier for the text data")
    memory: str = Field(
        ..., description="The memory deduced from the text data"
    )  # TODO After prompt changes from platform, update this
    hash: Optional[str] = Field(None, description="The hash of the memory")
    # The metadata value can be anything and not just string. Fix it
    metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata for the text data")
    score: Optional[float] = Field(None, description="The score associated with the text data")
    created_at: Optional[str] = Field(None, description="The timestamp when the memory was created")
    updated_at: Optional[str] = Field(None, description="The timestamp when the memory was updated")
 
 
class MemoryConfig(BaseModel):
    vector_store: VectorStoreConfig = Field(
        description="Configuration for the vector store",
        default_factory=VectorStoreConfig,
    )
    llm: LlmConfig = Field(
        description="Configuration for the language model",
        default_factory=LlmConfig,
    )
    embedder: EmbedderConfig = Field(
        description="Configuration for the embedding model",
        default_factory=EmbedderConfig,
    )
    history_db_path: str = Field(
        description="Path to the history database",
        default=os.path.join(mem0_dir, "history.db"),
    )
    reranker: Optional[RerankerConfig] = Field(
        description="Configuration for the reranker",
        default=None,
    )
    version: str = Field(
        description="The version of the API",
        default="v1.1",
    )
    custom_instructions: Optional[str] = Field(
        description="Custom instructions for fact extraction",
        default=None,
    )
 

Copydef _create_procedural_memory(self, messages, metadata=None, prompt=None):
    """
    Create a procedural memory
 
    Args:
        messages (list): List of messages to create a procedural memory from.
        metadata (dict): Metadata to create a procedural memory from.
        prompt (str, optional): Prompt to use for the procedural memory creation. Defaults to None.
    """
    logger.info("Creating procedural memory")
 
    parsed_messages = [
        {"role": "system", "content": prompt or PROCEDURAL_MEMORY_SYSTEM_PROMPT},
        *messages,
        {
            "role": "user",
            "content": "Create procedural memory of the above conversation.",
        },
    ]
 
    try:
        procedural_memory = self.llm.generate_response(messages=parsed_messages)
        procedural_memory = remove_code_blocks(procedural_memory)
    except Exception as e:
        logger.error(f"Error generating procedural memory summary: {e}")
        raise
 
    if not procedural_memory:
        raise ValueError(
            "The LLM returned no content for the procedural memory summary. "
            "The model may have declined the request or returned an empty response."
        )
 
    if metadata is None:
        raise ValueError("Metadata cannot be done for procedural memory.")
 
    metadata = {**metadata, "memory_type": MemoryType.PROCEDURAL.value}
    embeddings = self.embedding_model.embed(procedural_memory, memory_action="add")
    memory_id = self._create_memory(procedural_memory, {procedural_memory: embeddings}, metadata=metadata)
    capture_event("mem0._create_procedural_memory", self, {"memory_id": memory_id, "sync_type": "sync"})
 
    result = {"results": [{"id": memory_id, "memory": procedural_memory, "event": "ADD"}]}
 
    return result

Copydef _create_memory(self, data, existing_embeddings, metadata=None):
    logger.debug(f"Creating memory with {data=}")
    if data in existing_embeddings:
        embeddings = existing_embeddings[data]
    else:
        embeddings = self.embedding_model.embed(data, memory_action="add")
    memory_id = str(uuid.uuid4())
    new_metadata = deepcopy(metadata) if metadata is not None else {}
    new_metadata["data"] = data
    new_metadata["hash"] = hashlib.md5(data.encode()).hexdigest()
    if "created_at" not in new_metadata:
        new_metadata["created_at"] = datetime.now(timezone.utc).isoformat()
    new_metadata["updated_at"] = new_metadata["created_at"]
    new_metadata["text_lemmatized"] = lemmatize_for_bm25(data)
 
    self.vector_store.insert(
        vectors=[embeddings],
        ids=[memory_id],
        payloads=[new_metadata],
    )

Copyself.db.add_history(
    memory_id,
    None,
    data,
    "ADD",
    created_at=new_metadata.get("created_at"),
    updated_at=new_metadata.get("updated_at"),
    actor_id=new_metadata.get("actor_id"),
    role=new_metadata.get("role"),
)
return memory_id

Copydef _update_memory(self, memory_id, data, existing_embeddings, metadata=None):
    logger.info(f"Updating memory with {data=}")
 
    try:
        existing_memory = self.vector_store.get(vector_id=memory_id)
    except Exception:
        # Backing-store failure, not a bad memory_id: re-raise the original so the REST layer maps it to 5xx, not 4xx.
        logger.error(f"Error getting memory with ID {memory_id} during update.")
        raise
 
    if existing_memory is None:
        raise ValueError(f"Memory with id {memory_id} not found. Please provide a valid 'memory_id'")
 
    prev_value = existing_memory.payload.get("data")
    if data is None:
        data = prev_value
    if not isinstance(data, str):
        raise ValueError(f"Memory with id {memory_id} does not have text content to update")
    text_changed = data != prev_value
 
    new_metadata = deepcopy(existing_memory.payload)
    if metadata is not None:
        new_metadata.update(_strip_identity_keys(metadata, existing_memory.payload))
 
    new_metadata["data"] = data
    new_metadata["hash"] = hashlib.md5(data.encode()).hexdigest()
    new_metadata["text_lemmatized"] = lemmatize_for_bm25(data)
    new_metadata["created_at"] = existing_memory.payload.get("created_at")
    new_metadata["updated_at"] = datetime.now(timezone.utc).isoformat()
 
    if data in existing_embeddings:
        embeddings = existing_embeddings[data]
    else:
        embeddings = self.embedding_model.embed(data, "update")
 
    self.vector_store.update(
        vector_id=memory_id,
        vector=embeddings,
        payload=new_metadata,
    )
    logger.info(f"Updating memory with ID {memory_id=} with {data=}")
 
    self.db.add_history(
        memory_id,
        prev_value,
        data,
        "UPDATE",
        created_at=new_metadata["created_at"],
        updated_at=new_metadata["updated_at"],
        actor_id=new_metadata.get("actor_id"),
        role=new_metadata.get("role"),
    )
 
    # Entity-store cleanup: strip this memory's id from old-text entities,
    # then re-extract entities from the new text and link them back.
    session_filters = {k: new_metadata[k] for k in ("user_id", "agent_id", "run_id") if new_metadata.get(k)}
    if text_changed:
        self._remove_memory_from_entity_store(memory_id, session_filters)
        self._link_entities_for_memory(memory_id, data, session_filters)
 
    return memory_id

Copydef _delete_memory(self, memory_id, existing_memory=None):
    logger.info(f"Deleting memory with {memory_id=}")
    if existing_memory is None:
        existing_memory = self.vector_store.get(vector_id=memory_id)
        if existing_memory is None:
            raise ValueError(f"Memory with id {memory_id} not found. Please provide a valid 'memory_id'")
    prev_value = existing_memory.payload.get("data", "")
    created_at = _normalize_iso_timestamp_to_utc(existing_memory.payload.get("created_at"))
    updated_at = datetime.now(timezone.utc).isoformat()
    payload = existing_memory.payload or {}
    session_filters = {k: payload[k] for k in ("user_id", "agent_id", "run_id") if payload.get(k)}
    self.vector_store.delete(vector_id=memory_id)
    self.db.add_history(
        memory_id,
        prev_value,
        None,
        "DELETE",
        created_at=created_at,
        updated_at=updated_at,
        actor_id=existing_memory.payload.get("actor_id"),
        role=existing_memory.payload.get("role"),
        is_deleted=1,
    )
 
    # Entity-store cleanup: strip this memory's id from any entity records
    # that linked to it. Non-fatal — the helper swallows errors.
    self._remove_memory_from_entity_store(memory_id, session_filters)
 
    return memory_id
 
def reset(self):
    """
    Reset the memory store by:
        Deletes the vector store collection
        Resets the database
        Recreates the vector store with a new client
    """
    logger.warning("Resetting all memories")
 
    self.db.reset()
    self.db.close()
    self.db = SQLiteManager(self.config.history_db_path)
 
    if hasattr(self.vector_store, "reset"):
        self.vector_store = VectorStoreFactory.reset(self.vector_store)
    else:
        logger.warning("Vector store does not support reset. Skipping.")
        self.vector_store.delete_col()
        self.vector_store = VectorStoreFactory.create(

CopyNOTICE_ID = "first_run"
TEMPORAL_FEATURE_NOTICE_ID = "temporal_stub"
TEMPORAL_USAGE_NOTICE_ID = "temporal_usage"
DECAY_FEATURE_NOTICE_ID = "decay_stub"
DECAY_USAGE_NOTICE_ID = "decay_usage"
SCALE_THRESHOLD_NOTICE_ID = "scale_threshold"
PERFORMANCE_SLOW_QUERY_NOTICE_ID = "performance_slow_query"
NOTICE_EVENT = "mem0.notice_displayed"
DISPLAYED_VARIANT = "displayed"
HOLDOUT_VARIANT = "holdout"
STATE_SECTION = "notice_state"
STATE_KEY = "first_run"
TEMPORAL_USAGE_STATE_KEY = "temporal_usage"
TEMPORAL_USAGE_CAP = 10
TEMPORAL_USAGE_WINDOW = timedelta(days=7)
DECAY_USAGE_STATE_KEY = "decay_usage"
DECAY_USAGE_CAP = 10
DECAY_USAGE_WINDOW = timedelta(days=7)
DECAY_USAGE_DELETE_THRESHOLD = 5
SCALE_THRESHOLD_STATE_KEY = "scale_threshold"
SCALE_THRESHOLD_CAP = 10
SCALE_THRESHOLD_WINDOW = timedelta(days=7)
SCALE_MEMORY_COUNT_THRESHOLD = 2000
SCALE_MEMORY_COUNT_CHECK_INTERVAL = 100
SCALE_TOP_K_THRESHOLD = 50
PERFORMANCE_SLOW_QUERY_STATE_KEY = "performance_slow_query"
PERFORMANCE_SLOW_QUERY_CAP = 10
PERFORMANCE_SLOW_QUERY_WINDOW = timedelta(days=7)
PERFORMANCE_SLOW_QUERY_THRESHOLD_SECONDS = 2.0

Copy_ISO_DATE_RE = re.compile(
    r"\b\d{4}-\d{2}-\d{2}(?:[T\s]\d{2}:\d{2}(?::\d{2})?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)?\b"
)
_RELATIVE_TIME_RE = re.compile(
    r"\b("
    r"today|yesterday|tomorrow|"
    r"last\s+(?:night|week|month|year)|"
    r"this\s+(?:week|month|year)|"
    r"next\s+(?:week|month|year)|"
    r"(?:past|last)\s+\d+\s+(?:day|days|week|weeks|month|months|year|years)|"
    r"(?:since|before|after|until)\s+(?:today|yesterday|tomorrow|\d{4}-\d{2}-\d{2}|last\s+(?:week|month|year))"
    r")\b",
    re.IGNORECASE,
)
_RANGE_OPERATORS = {"gt", "gte", "lt", "lte"}