JDSS/mem0
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(memory): expose expiration controls in client docs (#5874)

Browse Source 
Co-authored-by: Soumil Rathi <soumilrathi@gmail.com>
This commit is contained in:
soumil-rathi
2026-06-25 17:14:33 -07:00
committed by GitHub
co-authored by Soumil Rathi
parent 818c2981b7
commit 0fbbb2f525
19 changed files with 519 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mem0/memory/main.py
+144 -24
View File
@@ -9,7 +9,7 @@ import time
import uuid
import warnings
from copy import deepcopy
from datetime import datetime, timezone
from datetime import date, datetime, timezone
from typing import Any, Dict, Optional
from pydantic import ValidationError
@@ -378,9 +378,37 @@ def _entity_collection_name(provider: str, collection_name: str) -> str:
return f"{collection_name}{separator}entities"
def _normalize_expiration_date(value: Any) -> Optional[str]:
if value is None:
return None
if isinstance(value, datetime):
return value.date().isoformat()
if isinstance(value, date):
return value.isoformat()
if isinstance(value, str):
try:
return date.fromisoformat(value).isoformat()
except ValueError as exc:
raise ValueError("expiration_date must be a valid date in YYYY-MM-DD format.") from exc
raise ValueError("expiration_date must be a date string in YYYY-MM-DD format.")
def _payload_is_expired(payload: Optional[Dict[str, Any]]) -> bool:
if not payload:
return False
expiration_date = payload.get("expiration_date")
if not expiration_date:
return False
try:
return date.fromisoformat(str(expiration_date)) < datetime.now(timezone.utc).date()
except ValueError:
return False
setup_config()
logger = logging.getLogger(__name__)
_UNSET = object()
_PROJECT_UPDATE_UNSUPPORTED_ERROR = "Project updates are not supported by the OSS Memory SDK."
@@ -694,6 +722,7 @@ class Memory(MemoryBase):
run_id: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
timestamp: Optional[Any] = None,
expiration_date: Optional[Any] = None,
infer: bool = True,
memory_type: Optional[str] = None,
prompt: Optional[str] = None,
@@ -712,6 +741,8 @@ class Memory(MemoryBase):
run_id (str, optional): ID of the run creating the memory. Defaults to None.
metadata (dict, optional): Metadata to store with the memory. Defaults to None.
timestamp (Any, optional): Platform-only temporal parameter. Not supported in OSS.
expiration_date (Any, optional): Date in YYYY-MM-DD format. Expired memories are hidden
from search and get_all unless show_expired is True.
infer (bool, optional): If True (default), an LLM is used to extract key facts from
'messages' and decide whether to add, update, or delete related memories.
If False, 'messages' are added as raw memories directly.
@@ -737,6 +768,7 @@ class Memory(MemoryBase):
if timestamp is not None:
raise ValueError(get_temporal_feature_error_message("sync", "add", "timestamp"))
normalized_expiration_date = _normalize_expiration_date(expiration_date)
temporal_usage_notice = detect_temporal_usage_from_metadata(metadata)
processed_metadata, effective_filters = _build_filters_and_metadata(
user_id=user_id,
@@ -744,6 +776,8 @@ class Memory(MemoryBase):
run_id=run_id,
input_metadata=metadata,
)
if normalized_expiration_date is not None:
processed_metadata["expiration_date"] = normalized_expiration_date
if memory_type is not None and memory_type != MemoryType.PROCEDURAL.value:
raise Mem0ValidationError(
@@ -1141,6 +1175,7 @@ class Memory(MemoryBase):
"actor_id",
"role",
"attributed_to",
"expiration_date",
]
core_and_promoted_keys = {"data", "hash", "created_at", "updated_at", "id", "text_lemmatized", "attributed_to", *promoted_payload_keys}
@@ -1169,6 +1204,7 @@ class Memory(MemoryBase):
*,
filters: Optional[Dict[str, Any]] = None,
top_k: int = 20,
show_expired: bool = False,
**kwargs,
):
"""
@@ -1179,6 +1215,7 @@ class Memory(MemoryBase):
Must contain at least one of: user_id, agent_id, run_id.
Example: filters={"user_id": "u1", "agent_id": "a1"}
top_k (int, optional): The maximum number of memories to return. Defaults to 20.
show_expired (bool, optional): Include expired memories. Defaults to False.
Returns:
dict: A dictionary containing a list of memories under the "results" key.
@@ -1217,6 +1254,7 @@ class Memory(MemoryBase):
)
limit = top_k
fetch_limit = limit if show_expired else max(limit * 4, 60)
scale_threshold_notice = detect_scale_threshold_from_top_k(top_k)
keys, encoded_ids = process_telemetry_filters(effective_filters)
@@ -1224,7 +1262,7 @@ class Memory(MemoryBase):
"mem0.get_all", self, {"limit": limit, "keys": keys, "encoded_ids": encoded_ids, "sync_type": "sync"}
)
all_memories_result = self._get_all_from_vector_store(effective_filters, limit)
all_memories_result = self._get_all_from_vector_store(effective_filters, fetch_limit, show_expired, limit)
if scale_threshold_notice:
display_scale_threshold_notice(self, "sync", "get_all", *scale_threshold_notice)
@@ -1232,7 +1270,7 @@ class Memory(MemoryBase):
display_first_run_notice(self, "sync", "get_all")
return {"results": all_memories_result}
def _get_all_from_vector_store(self, filters, limit):
def _get_all_from_vector_store(self, filters, limit, show_expired=False, output_limit=None):
memories_result = self.vector_store.list(filters=filters, top_k=limit)
# Handle different vector store return formats by inspecting first element
@@ -1255,11 +1293,14 @@ class Memory(MemoryBase):
"actor_id",
"role",
"attributed_to",
"expiration_date",
]
core_and_promoted_keys = {"data", "hash", "created_at", "updated_at", "id", "text_lemmatized", "attributed_to", *promoted_payload_keys}
formatted_memories = []
for mem in actual_memories:
if not show_expired and _payload_is_expired(mem.payload):
continue
memory_item_dict = MemoryItem(
id=mem.id,
memory=mem.payload.get("data", ""),
@@ -1277,6 +1318,8 @@ class Memory(MemoryBase):
memory_item_dict["metadata"] = additional_metadata
formatted_memories.append(memory_item_dict)
if output_limit is not None and len(formatted_memories) >= output_limit:
break
return formatted_memories
@@ -1290,6 +1333,7 @@ class Memory(MemoryBase):
rerank: bool = False,
explain: bool = False,
reference_date: Optional[Any] = None,
show_expired: bool = False,
**kwargs,
):
"""
@@ -1322,6 +1366,7 @@ class Memory(MemoryBase):
rerank (bool, optional): Whether to rerank results. Defaults to False.
explain (bool, optional): Whether to include score_details for each result. Defaults to False.
reference_date (Any, optional): Platform-only temporal parameter. Not supported in OSS.
show_expired (bool, optional): Include expired memories. Defaults to False.
Returns:
dict: A dictionary containing the search results under a "results" key.
@@ -1393,7 +1438,9 @@ class Memory(MemoryBase):
)
search_start = time.perf_counter()
original_memories = self._search_vector_store(query, effective_filters, limit, threshold, explain=explain)
original_memories = self._search_vector_store(
query, effective_filters, limit, threshold, explain=explain, show_expired=show_expired
)
search_elapsed_seconds = time.perf_counter() - search_start
# Apply reranking if enabled and reranker is available
@@ -1525,7 +1572,7 @@ class Memory(MemoryBase):
return True
return False
def _search_vector_store(self, query, filters, limit, threshold=0.1, explain=False):
def _search_vector_store(self, query, filters, limit, threshold=0.1, explain=False, show_expired=False):
# Guard against None threshold (backward compat)
if threshold is None:
threshold = 0.1
@@ -1566,11 +1613,14 @@ class Memory(MemoryBase):
# Step 7: Build candidate set from semantic results
candidates = []
for mem in semantic_results:
payload = mem.payload if hasattr(mem, 'payload') else {}
if not show_expired and _payload_is_expired(payload):
continue
mem_id = str(mem.id)
candidates.append({
"id": mem_id,
"score": mem.score,
"payload": mem.payload if hasattr(mem, 'payload') else {},
"payload": payload,
})
# Step 8: Score and rank
@@ -1591,6 +1641,7 @@ class Memory(MemoryBase):
"actor_id",
"role",
"attributed_to",
"expiration_date",
]
core_and_promoted_keys = {"data", "hash", "created_at", "updated_at", "id", "text_lemmatized", "attributed_to", *promoted_payload_keys}
@@ -1708,14 +1759,21 @@ class Memory(MemoryBase):
return memory_boosts
def update(self, memory_id, data, metadata: Optional[Dict[str, Any]] = None):
def update(
self,
memory_id,
data: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
expiration_date: Any = _UNSET,
):
"""
Update a memory by ID.
Args:
memory_id (str): ID of the memory to update.
data (str): New content to update the memory with.
data (str, optional): New content to update the memory with.
metadata (dict, optional): Metadata to update with the memory. Defaults to None.
expiration_date (Any, optional): Date in YYYY-MM-DD format, or None to clear it.
Returns:
dict: Success message indicating the memory was updated.
@@ -1726,9 +1784,19 @@ class Memory(MemoryBase):
"""
capture_event("mem0.update", self, {"memory_id": memory_id, "sync_type": "sync"})
existing_embeddings = {data: self.embedding_model.embed(data, "update")}
if data is None and metadata is None and expiration_date is _UNSET:
raise ValueError("At least one of data, metadata, or expiration_date must be provided.")
self._update_memory(memory_id, data, existing_embeddings, metadata)
update_metadata = deepcopy(metadata) if metadata is not None else None
if expiration_date is not _UNSET:
update_metadata = update_metadata or {}
update_metadata["expiration_date"] = _normalize_expiration_date(expiration_date)
existing_embeddings = {}
if data is not None:
existing_embeddings[data] = self.embedding_model.embed(data, "update")
self._update_memory(memory_id, data, existing_embeddings, update_metadata)
display_first_run_notice(self, "sync", "update")
return {"message": "Memory updated successfully!"}
@@ -1895,6 +1963,11 @@ class Memory(MemoryBase):
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:
@@ -1936,8 +2009,9 @@ class Memory(MemoryBase):
# 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)}
self._remove_memory_from_entity_store(memory_id, session_filters)
self._link_entities_for_memory(memory_id, data, session_filters)
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
@@ -2284,6 +2358,7 @@ class AsyncMemory(MemoryBase):
run_id: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
timestamp: Optional[Any] = None,
expiration_date: Optional[Any] = None,
infer: bool = True,
memory_type: Optional[str] = None,
prompt: Optional[str] = None,
@@ -2299,6 +2374,8 @@ class AsyncMemory(MemoryBase):
run_id (str, optional): ID of the run creating the memory. Defaults to None.
metadata (dict, optional): Metadata to store with the memory. Defaults to None.
timestamp (Any, optional): Platform-only temporal parameter. Not supported in OSS.
expiration_date (Any, optional): Date in YYYY-MM-DD format. Expired memories are hidden
from search and get_all unless show_expired is True.
infer (bool, optional): Whether to infer the memories. Defaults to True.
memory_type (str, optional): Type of memory to create. Defaults to None.
Pass "procedural_memory" to create procedural memories.
@@ -2310,10 +2387,13 @@ class AsyncMemory(MemoryBase):
if timestamp is not None:
raise ValueError(await get_temporal_feature_error_message_async("async", "add", "timestamp"))
normalized_expiration_date = _normalize_expiration_date(expiration_date)
temporal_usage_notice = detect_temporal_usage_from_metadata(metadata)
processed_metadata, effective_filters = _build_filters_and_metadata(
user_id=user_id, agent_id=agent_id, run_id=run_id, input_metadata=metadata
)
if normalized_expiration_date is not None:
processed_metadata["expiration_date"] = normalized_expiration_date
if memory_type is not None and memory_type != MemoryType.PROCEDURAL.value:
raise ValueError(
@@ -2716,6 +2796,7 @@ class AsyncMemory(MemoryBase):
"actor_id",
"role",
"attributed_to",
"expiration_date",
]
core_and_promoted_keys = {"data", "hash", "created_at", "updated_at", "id", "text_lemmatized", "attributed_to", *promoted_payload_keys}
@@ -2744,6 +2825,7 @@ class AsyncMemory(MemoryBase):
*,
filters: Optional[Dict[str, Any]] = None,
top_k: int = 20,
show_expired: bool = False,
**kwargs,
):
"""
@@ -2754,6 +2836,7 @@ class AsyncMemory(MemoryBase):
Must contain at least one of: user_id, agent_id, run_id.
Example: filters={"user_id": "u1", "agent_id": "a1"}
top_k (int, optional): The maximum number of memories to return. Defaults to 20.
show_expired (bool, optional): Include expired memories. Defaults to False.
Returns:
dict: A dictionary containing a list of memories under the "results" key.
@@ -2792,6 +2875,7 @@ class AsyncMemory(MemoryBase):
)
limit = top_k
fetch_limit = limit if show_expired else max(limit * 4, 60)
scale_threshold_notice = detect_scale_threshold_from_top_k(top_k)
keys, encoded_ids = process_telemetry_filters(effective_filters)
@@ -2799,7 +2883,7 @@ class AsyncMemory(MemoryBase):
"mem0.get_all", self, {"limit": limit, "keys": keys, "encoded_ids": encoded_ids, "sync_type": "async"}
)
all_memories_result = await self._get_all_from_vector_store(effective_filters, limit)
all_memories_result = await self._get_all_from_vector_store(effective_filters, fetch_limit, show_expired, limit)
if scale_threshold_notice:
await display_scale_threshold_notice_async(self, "async", "get_all", *scale_threshold_notice)
@@ -2807,7 +2891,7 @@ class AsyncMemory(MemoryBase):
await display_first_run_notice_async(self, "async", "get_all")
return {"results": all_memories_result}
async def _get_all_from_vector_store(self, filters, limit):
async def _get_all_from_vector_store(self, filters, limit, show_expired=False, output_limit=None):
memories_result = await asyncio.to_thread(self.vector_store.list, filters=filters, top_k=limit)
# Handle different vector store return formats by inspecting first element
@@ -2830,11 +2914,14 @@ class AsyncMemory(MemoryBase):
"actor_id",
"role",
"attributed_to",
"expiration_date",
]
core_and_promoted_keys = {"data", "hash", "created_at", "updated_at", "id", "text_lemmatized", "attributed_to", *promoted_payload_keys}
formatted_memories = []
for mem in actual_memories:
if not show_expired and _payload_is_expired(mem.payload):
continue
memory_item_dict = MemoryItem(
id=mem.id,
memory=mem.payload.get("data", ""),
@@ -2852,6 +2939,8 @@ class AsyncMemory(MemoryBase):
memory_item_dict["metadata"] = additional_metadata
formatted_memories.append(memory_item_dict)
if output_limit is not None and len(formatted_memories) >= output_limit:
break
return formatted_memories
@@ -2865,6 +2954,7 @@ class AsyncMemory(MemoryBase):
rerank: bool = False,
explain: bool = False,
reference_date: Optional[Any] = None,
show_expired: bool = False,
**kwargs,
):
"""
@@ -2897,6 +2987,7 @@ class AsyncMemory(MemoryBase):
rerank (bool, optional): Whether to rerank results. Defaults to False.
explain (bool, optional): Whether to include score_details for each result. Defaults to False.
reference_date (Any, optional): Platform-only temporal parameter. Not supported in OSS.
show_expired (bool, optional): Include expired memories. Defaults to False.
Returns:
dict: A dictionary containing the search results under a "results" key.
@@ -2972,7 +3063,9 @@ class AsyncMemory(MemoryBase):
)
search_start = time.perf_counter()
original_memories = await self._search_vector_store(query, effective_filters, limit, threshold, explain=explain)
original_memories = await self._search_vector_store(
query, effective_filters, limit, threshold, explain=explain, show_expired=show_expired
)
search_elapsed_seconds = time.perf_counter() - search_start
# Apply reranking if enabled and reranker is available
@@ -3107,7 +3200,7 @@ class AsyncMemory(MemoryBase):
return True
return False
async def _search_vector_store(self, query, filters, limit, threshold=0.1, explain=False):
async def _search_vector_store(self, query, filters, limit, threshold=0.1, explain=False, show_expired=False):
if threshold is None:
threshold = 0.1
@@ -3147,11 +3240,14 @@ class AsyncMemory(MemoryBase):
# Step 7: Build candidate set from semantic results
candidates = []
for mem in semantic_results:
payload = mem.payload if hasattr(mem, 'payload') else {}
if not show_expired and _payload_is_expired(payload):
continue
mem_id = str(mem.id)
candidates.append({
"id": mem_id,
"score": mem.score,
"payload": mem.payload if hasattr(mem, 'payload') else {},
"payload": payload,
})
# Step 8: Score and rank
@@ -3172,6 +3268,7 @@ class AsyncMemory(MemoryBase):
"actor_id",
"role",
"attributed_to",
"expiration_date",
]
core_and_promoted_keys = {"data", "hash", "created_at", "updated_at", "id", "text_lemmatized", "attributed_to", *promoted_payload_keys}
@@ -3280,14 +3377,21 @@ class AsyncMemory(MemoryBase):
return memory_boosts
async def update(self, memory_id, data, metadata: Optional[Dict[str, Any]] = None):
async def update(
self,
memory_id,
data: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
expiration_date: Any = _UNSET,
):
"""
Update a memory by ID asynchronously.
Args:
memory_id (str): ID of the memory to update.
data (str): New content to update the memory with.
data (str, optional): New content to update the memory with.
metadata (dict, optional): Metadata to update with the memory. Defaults to None.
expiration_date (Any, optional): Date in YYYY-MM-DD format, or None to clear it.
Returns:
dict: Success message indicating the memory was updated.
@@ -3298,10 +3402,20 @@ class AsyncMemory(MemoryBase):
"""
capture_event("mem0.update", self, {"memory_id": memory_id, "sync_type": "async"})
embeddings = await asyncio.to_thread(self.embedding_model.embed, data, "update")
existing_embeddings = {data: embeddings}
if data is None and metadata is None and expiration_date is _UNSET:
raise ValueError("At least one of data, metadata, or expiration_date must be provided.")
await self._update_memory(memory_id, data, existing_embeddings, metadata)
update_metadata = deepcopy(metadata) if metadata is not None else None
if expiration_date is not _UNSET:
update_metadata = update_metadata or {}
update_metadata["expiration_date"] = _normalize_expiration_date(expiration_date)
existing_embeddings = {}
if data is not None:
embeddings = await asyncio.to_thread(self.embedding_model.embed, data, "update")
existing_embeddings[data] = embeddings
await self._update_memory(memory_id, data, existing_embeddings, update_metadata)
await display_first_run_notice_async(self, "async", "update")
return {"message": "Memory updated successfully!"}
@@ -3499,6 +3613,11 @@ class AsyncMemory(MemoryBase):
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:
@@ -3542,8 +3661,9 @@ class AsyncMemory(MemoryBase):
# 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)}
await self._remove_memory_from_entity_store(memory_id, session_filters)
await self._link_entities_for_memory(memory_id, data, session_filters)
if text_changed:
await self._remove_memory_from_entity_store(memory_id, session_filters)
await self._link_entities_for_memory(memory_id, data, session_filters)
return memory_id