wehub-resource-sync
555e282cc4
ci / changelog_check (push) Waiting to run
ci / check_changes (push) Waiting to run
ci / build_mem0 (3.10) (push) Blocked by required conditions
ci / build_mem0 (3.11) (push) Blocked by required conditions
ci / build_mem0 (3.12) (push) Blocked by required conditions
CLI Node CI / lint (push) Waiting to run
CLI Node CI / test (20) (push) Waiting to run
CLI Node CI / test (22) (push) Waiting to run
CLI Node CI / build (push) Waiting to run
CLI Python CI / lint (push) Waiting to run
CLI Python CI / test (3.10) (push) Waiting to run
CLI Python CI / test (3.11) (push) Waiting to run
CLI Python CI / test (3.12) (push) Waiting to run
CLI Python CI / build (push) Waiting to run
openclaw checks / lint (push) Waiting to run
openclaw checks / test (20) (push) Waiting to run
openclaw checks / test (22) (push) Waiting to run
openclaw checks / build (push) Waiting to run
opencode-plugin checks / build (push) Waiting to run
pi-agent-plugin checks / lint (push) Waiting to run
pi-agent-plugin checks / test (20) (push) Waiting to run
pi-agent-plugin checks / test (22) (push) Waiting to run
pi-agent-plugin checks / build (push) Waiting to run
TypeScript SDK CI / check_changes (push) Waiting to run
TypeScript SDK CI / changelog_check (push) Blocked by required conditions
TypeScript SDK CI / build_ts_sdk (20) (push) Blocked by required conditions
TypeScript SDK CI / build_ts_sdk (22) (push) Blocked by required conditions
TypeScript SDK CI / integration_ts_sdk (20) (push) Blocked by required conditions
TypeScript SDK CI / integration_ts_sdk (22) (push) Blocked by required conditions
3780 lines
160 KiB
Python
3780 lines
160 KiB
Python
import asyncio
|
|
import concurrent.futures
|
|
import gc
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
import uuid
|
|
import warnings
|
|
from copy import deepcopy
|
|
from datetime import date, datetime, timezone
|
|
from typing import Any, Dict, Optional
|
|
|
|
from pydantic import ValidationError
|
|
|
|
from mem0.configs.base import MemoryConfig, MemoryItem
|
|
from mem0.configs.enums import MemoryType
|
|
from mem0.configs.prompts import (
|
|
ADDITIVE_EXTRACTION_PROMPT,
|
|
AGENT_CONTEXT_SUFFIX,
|
|
PROCEDURAL_MEMORY_SYSTEM_PROMPT,
|
|
generate_additive_extraction_prompt,
|
|
)
|
|
from mem0.exceptions import LLMError
|
|
from mem0.exceptions import ValidationError as Mem0ValidationError
|
|
from mem0.memory.base import MemoryBase
|
|
from mem0.memory.notices import (
|
|
PERFORMANCE_SLOW_QUERY_THRESHOLD_SECONDS,
|
|
detect_decay_usage_from_delete,
|
|
detect_decay_usage_from_delete_all,
|
|
detect_scale_threshold_from_add_result,
|
|
detect_scale_threshold_from_top_k,
|
|
detect_temporal_usage_from_metadata,
|
|
detect_temporal_usage_from_search,
|
|
display_decay_usage_notice,
|
|
display_decay_usage_notice_async,
|
|
display_first_run_notice,
|
|
display_first_run_notice_async,
|
|
display_performance_slow_query_notice,
|
|
display_performance_slow_query_notice_async,
|
|
display_scale_threshold_notice,
|
|
display_scale_threshold_notice_async,
|
|
display_temporal_usage_notice,
|
|
display_temporal_usage_notice_async,
|
|
get_decay_feature_error_message,
|
|
get_decay_feature_error_message_async,
|
|
get_temporal_feature_error_message,
|
|
get_temporal_feature_error_message_async,
|
|
)
|
|
from mem0.memory.setup import mem0_dir, setup_config
|
|
from mem0.memory.storage import SQLiteManager
|
|
from mem0.memory.telemetry import MEM0_TELEMETRY, capture_event
|
|
from mem0.memory.utils import (
|
|
extract_json,
|
|
parse_messages,
|
|
parse_vision_messages,
|
|
process_telemetry_filters,
|
|
remove_code_blocks,
|
|
)
|
|
from mem0.utils.entity_extraction import extract_entities, extract_entities_batch
|
|
from mem0.utils.factory import (
|
|
EmbedderFactory,
|
|
LlmFactory,
|
|
RerankerFactory,
|
|
VectorStoreFactory,
|
|
)
|
|
from mem0.utils.lemmatization import lemmatize_for_bm25
|
|
from mem0.utils.scoring import (
|
|
ENTITY_BOOST_WEIGHT,
|
|
get_bm25_params,
|
|
normalize_bm25,
|
|
score_and_rank,
|
|
)
|
|
from mem0.vector_stores.base import VectorStoreBase
|
|
|
|
# Suppress SWIG deprecation warnings globally
|
|
warnings.filterwarnings("ignore", category=DeprecationWarning, message=".*SwigPy.*")
|
|
warnings.filterwarnings("ignore", category=DeprecationWarning, message=".*swigvarlink.*")
|
|
|
|
# Initialize logger early for util functions
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _vector_store_list_rows(listed):
|
|
if isinstance(listed, (list, tuple)) and listed and isinstance(listed[0], list):
|
|
return listed[0]
|
|
if isinstance(listed, (list, tuple)):
|
|
return listed
|
|
return []
|
|
|
|
|
|
# Fields that hold runtime auth/connection objects and must be preserved.
|
|
# These are non-serializable objects (e.g. AWSV4SignerAuth, RequestsHttpConnection)
|
|
# needed by clients like OpenSearch — not sensitive strings to redact.
|
|
_RUNTIME_FIELDS = frozenset({
|
|
"http_auth",
|
|
"auth",
|
|
"connection_class",
|
|
"ssl_context",
|
|
})
|
|
|
|
# Fields that are known to contain sensitive secrets and must be redacted.
|
|
_SENSITIVE_FIELDS_EXACT = frozenset({
|
|
"api_key",
|
|
"secret_key",
|
|
"private_key",
|
|
"access_key",
|
|
"password",
|
|
"credentials",
|
|
"credential",
|
|
"secret",
|
|
"token",
|
|
"access_token",
|
|
"refresh_token",
|
|
"auth_token",
|
|
"session_token",
|
|
"client_secret",
|
|
"auth_client_secret",
|
|
"azure_client_secret",
|
|
"service_account_json",
|
|
"aws_session_token",
|
|
})
|
|
|
|
# Suffixes that indicate a field likely holds a secret value.
|
|
_SENSITIVE_SUFFIXES = (
|
|
"_password",
|
|
"_secret",
|
|
"_token",
|
|
"_credential",
|
|
"_credentials",
|
|
)
|
|
|
|
# Entity parameters that must be passed via filters, not top-level kwargs
|
|
ENTITY_PARAMS = frozenset({"user_id", "agent_id", "run_id"})
|
|
|
|
|
|
def _reject_top_level_entity_params(kwargs: Dict[str, Any], method_name: str) -> None:
|
|
"""Reject top-level entity parameters - must use filters instead."""
|
|
invalid_keys = ENTITY_PARAMS & set(kwargs.keys())
|
|
if invalid_keys:
|
|
raise ValueError(
|
|
f"Top-level entity parameters {invalid_keys} are not supported in {method_name}(). "
|
|
f"Use filters={{'user_id': '...'}} instead."
|
|
)
|
|
|
|
|
|
def _validate_and_trim_entity_id(value: Optional[Any], name: str) -> Optional[str]:
|
|
"""
|
|
Validates and normalizes an entity ID.
|
|
- Coerces non-string values (e.g. integer ids) to str
|
|
- Trims leading/trailing whitespace
|
|
- Rejects empty or whitespace-only strings
|
|
- Rejects strings containing internal whitespace
|
|
|
|
Args:
|
|
value: The entity ID value to validate
|
|
name: The parameter name (for error messages)
|
|
|
|
Returns:
|
|
The trimmed entity ID, or None if input is None
|
|
|
|
Raises:
|
|
ValueError: If entity ID is invalid
|
|
"""
|
|
if value is None:
|
|
return None
|
|
# Callers commonly pass integer ids (e.g. a database primary key). Coerce
|
|
# to str at this single validation point so scoping stays consistent across
|
|
# add/search/get_all/delete_all instead of crashing on `.strip()`.
|
|
if not isinstance(value, str):
|
|
value = str(value)
|
|
trimmed = value.strip()
|
|
if trimmed == "":
|
|
raise ValueError(
|
|
f"Invalid {name}: cannot be empty or whitespace-only. Provide a valid identifier."
|
|
)
|
|
if any(c.isspace() for c in trimmed):
|
|
raise ValueError(
|
|
f"Invalid {name}: cannot contain whitespace. Provide a valid identifier without spaces."
|
|
)
|
|
return trimmed
|
|
|
|
|
|
def _validate_search_params(threshold: Optional[float] = None, top_k: Optional[int] = None) -> None:
|
|
"""
|
|
Validates search parameters.
|
|
|
|
Args:
|
|
threshold: Similarity threshold (must be between 0 and 1)
|
|
top_k: Number of results to return (must be non-negative integer)
|
|
|
|
Raises:
|
|
ValueError: If threshold or top_k are invalid
|
|
"""
|
|
if threshold is not None:
|
|
if not isinstance(threshold, (int, float)):
|
|
raise ValueError("threshold must be a valid number")
|
|
if threshold < 0 or threshold > 1:
|
|
raise ValueError(
|
|
f"Invalid threshold: {threshold}. Must be between 0 and 1 (inclusive)."
|
|
)
|
|
if top_k is not None:
|
|
if not isinstance(top_k, int) or isinstance(top_k, bool):
|
|
raise ValueError("top_k must be a valid integer")
|
|
if top_k < 0:
|
|
raise ValueError(
|
|
f"Invalid top_k: {top_k}. Must be a non-negative integer."
|
|
)
|
|
|
|
|
|
def _validate_and_trim_search_query(query: str) -> str:
|
|
"""
|
|
Validates and normalizes a search query before embedding/vector search.
|
|
|
|
Raises:
|
|
ValueError: If query is not a string or is empty/whitespace-only.
|
|
"""
|
|
if not isinstance(query, str):
|
|
raise ValueError("Invalid query: must be a non-empty string.")
|
|
trimmed = query.strip()
|
|
if not trimmed:
|
|
raise ValueError("Invalid query: cannot be empty or whitespace-only.")
|
|
return trimmed
|
|
|
|
|
|
def _is_sensitive_field(field_name: str) -> bool:
|
|
"""Check if a field should be redacted for telemetry safety.
|
|
|
|
Uses a layered approach:
|
|
1. Runtime fields (allowlist) — always preserved, highest priority.
|
|
2. Exact deny list — known secret field names.
|
|
3. Suffix deny list — catches patterns like db_password, auth_secret, etc.
|
|
"""
|
|
name = field_name.lower().strip()
|
|
if name in _RUNTIME_FIELDS:
|
|
return False
|
|
if name in _SENSITIVE_FIELDS_EXACT:
|
|
return True
|
|
return any(name.endswith(suffix) for suffix in _SENSITIVE_SUFFIXES)
|
|
|
|
|
|
def _safe_deepcopy_config(config):
|
|
"""Safely deepcopy config, falling back to dict-based cloning for non-serializable objects."""
|
|
try:
|
|
return deepcopy(config)
|
|
except Exception as e:
|
|
logger.debug(f"Deepcopy failed, using dict-based cloning: {e}")
|
|
|
|
config_class = type(config)
|
|
|
|
if hasattr(config, "model_dump"):
|
|
try:
|
|
clone_dict = config.model_dump()
|
|
except Exception:
|
|
clone_dict = dict(config.__dict__)
|
|
else:
|
|
clone_dict = dict(config.__dict__)
|
|
|
|
# Restore runtime fields, redact sensitive ones
|
|
for field_name in list(clone_dict.keys()):
|
|
if field_name in _RUNTIME_FIELDS and hasattr(config, field_name):
|
|
clone_dict[field_name] = getattr(config, field_name)
|
|
elif _is_sensitive_field(field_name):
|
|
clone_dict[field_name] = None
|
|
|
|
try:
|
|
return config_class(**clone_dict)
|
|
except Exception:
|
|
logger.debug("Config reconstruction failed, returning shallow dict clone")
|
|
return type("Config", (), clone_dict)()
|
|
|
|
|
|
def _normalize_iso_timestamp_to_utc(timestamp: Optional[str]) -> Optional[str]:
|
|
"""Normalize timezone-aware ISO timestamps to UTC without rewriting naive values."""
|
|
if not timestamp:
|
|
return timestamp
|
|
try:
|
|
parsed = datetime.fromisoformat(timestamp)
|
|
except ValueError:
|
|
return timestamp
|
|
if parsed.tzinfo is None:
|
|
return timestamp
|
|
return parsed.astimezone(timezone.utc).isoformat()
|
|
|
|
|
|
def _build_filters_and_metadata(
|
|
*, # Enforce keyword-only arguments
|
|
user_id: Optional[str] = None,
|
|
agent_id: Optional[str] = None,
|
|
run_id: Optional[str] = None,
|
|
actor_id: Optional[str] = None, # For query-time filtering
|
|
input_metadata: Optional[Dict[str, Any]] = None,
|
|
input_filters: Optional[Dict[str, Any]] = None,
|
|
) -> tuple[Dict[str, Any], Dict[str, Any]]:
|
|
"""
|
|
Constructs metadata for storage and filters for querying based on session and actor identifiers.
|
|
|
|
This helper supports multiple session identifiers (`user_id`, `agent_id`, and/or `run_id`)
|
|
for flexible session scoping and optionally narrows queries to a specific `actor_id`. It returns two dicts:
|
|
|
|
1. `base_metadata_template`: Used as a template for metadata when storing new memories.
|
|
It includes all provided session identifier(s) and any `input_metadata`.
|
|
2. `effective_query_filters`: Used for querying existing memories. It includes all
|
|
provided session identifier(s), any `input_filters`, and a resolved actor
|
|
identifier for targeted filtering if specified by any actor-related inputs.
|
|
|
|
Actor filtering precedence: explicit `actor_id` arg → `filters["actor_id"]`
|
|
This resolved actor ID is used for querying but is not added to `base_metadata_template`,
|
|
as the actor for storage is typically derived from message content at a later stage.
|
|
|
|
Args:
|
|
user_id (Optional[str]): User identifier, for session scoping.
|
|
agent_id (Optional[str]): Agent identifier, for session scoping.
|
|
run_id (Optional[str]): Run identifier, for session scoping.
|
|
actor_id (Optional[str]): Explicit actor identifier, used as a potential source for
|
|
actor-specific filtering. See actor resolution precedence in the main description.
|
|
input_metadata (Optional[Dict[str, Any]]): Base dictionary to be augmented with
|
|
session identifiers for the storage metadata template. Defaults to an empty dict.
|
|
input_filters (Optional[Dict[str, Any]]): Base dictionary to be augmented with
|
|
session and actor identifiers for query filters. Defaults to an empty dict.
|
|
|
|
Returns:
|
|
tuple[Dict[str, Any], Dict[str, Any]]: A tuple containing:
|
|
- base_metadata_template (Dict[str, Any]): Metadata template for storing memories,
|
|
scoped to the provided session(s).
|
|
- effective_query_filters (Dict[str, Any]): Filters for querying memories,
|
|
scoped to the provided session(s) and potentially a resolved actor.
|
|
"""
|
|
|
|
base_metadata_template = deepcopy(input_metadata) if input_metadata else {}
|
|
effective_query_filters = deepcopy(input_filters) if input_filters else {}
|
|
|
|
# ---------- validate and add all provided session ids ----------
|
|
session_ids_provided = []
|
|
|
|
# Validate and trim entity IDs
|
|
user_id = _validate_and_trim_entity_id(user_id, "user_id")
|
|
agent_id = _validate_and_trim_entity_id(agent_id, "agent_id")
|
|
run_id = _validate_and_trim_entity_id(run_id, "run_id")
|
|
|
|
if user_id:
|
|
base_metadata_template["user_id"] = user_id
|
|
effective_query_filters["user_id"] = user_id
|
|
session_ids_provided.append("user_id")
|
|
|
|
if agent_id:
|
|
base_metadata_template["agent_id"] = agent_id
|
|
effective_query_filters["agent_id"] = agent_id
|
|
session_ids_provided.append("agent_id")
|
|
|
|
if run_id:
|
|
base_metadata_template["run_id"] = run_id
|
|
effective_query_filters["run_id"] = run_id
|
|
session_ids_provided.append("run_id")
|
|
|
|
if not session_ids_provided:
|
|
raise Mem0ValidationError(
|
|
message="At least one of 'user_id', 'agent_id', or 'run_id' must be provided.",
|
|
error_code="VALIDATION_001",
|
|
details={"provided_ids": {"user_id": user_id, "agent_id": agent_id, "run_id": run_id}},
|
|
suggestion="Please provide at least one identifier to scope the memory operation."
|
|
)
|
|
|
|
# ---------- optional actor filter ----------
|
|
resolved_actor_id = actor_id or effective_query_filters.get("actor_id")
|
|
if resolved_actor_id:
|
|
effective_query_filters["actor_id"] = resolved_actor_id
|
|
|
|
return base_metadata_template, effective_query_filters
|
|
|
|
|
|
def _build_session_scope(filters):
|
|
"""Build deterministic session scope string from entity IDs."""
|
|
parts = []
|
|
for key in sorted(["user_id", "agent_id", "run_id"]):
|
|
val = filters.get(key)
|
|
if val:
|
|
parts.append(f"{key}={val}")
|
|
return "&".join(parts)
|
|
|
|
|
|
def _entity_collection_name(provider: str, collection_name: str) -> str:
|
|
separator = "-" if provider == "s3_vectors" else "_"
|
|
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."
|
|
|
|
|
|
class _OSSProject:
|
|
def update(
|
|
self,
|
|
custom_instructions: Optional[str] = None,
|
|
custom_categories: Optional[list] = None,
|
|
retrieval_criteria: Optional[list] = None,
|
|
multilingual: Optional[bool] = None,
|
|
decay: Optional[bool] = None,
|
|
):
|
|
if decay is True:
|
|
raise ValueError(get_decay_feature_error_message("sync", "project.update", "decay"))
|
|
raise ValueError(_PROJECT_UPDATE_UNSUPPORTED_ERROR)
|
|
|
|
|
|
class _AsyncOSSProject:
|
|
async def update(
|
|
self,
|
|
custom_instructions: Optional[str] = None,
|
|
custom_categories: Optional[list] = None,
|
|
retrieval_criteria: Optional[list] = None,
|
|
multilingual: Optional[bool] = None,
|
|
decay: Optional[bool] = None,
|
|
):
|
|
if decay is True:
|
|
raise ValueError(await get_decay_feature_error_message_async("async", "project.update", "decay"))
|
|
raise ValueError(_PROJECT_UPDATE_UNSUPPORTED_ERROR)
|
|
|
|
|
|
class Memory(MemoryBase):
|
|
def __init__(self, config: MemoryConfig = MemoryConfig()):
|
|
self.config = config
|
|
|
|
self.embedding_model = EmbedderFactory.create(
|
|
self.config.embedder.provider,
|
|
self.config.embedder.config,
|
|
self.config.vector_store.config,
|
|
)
|
|
self.vector_store = VectorStoreFactory.create(
|
|
self.config.vector_store.provider, self.config.vector_store.config
|
|
)
|
|
self.llm = LlmFactory.create(self.config.llm.provider, self.config.llm.config)
|
|
self.db = SQLiteManager(self.config.history_db_path)
|
|
self.collection_name = self.config.vector_store.config.collection_name
|
|
self.api_version = self.config.version
|
|
self.custom_instructions = self.config.custom_instructions
|
|
|
|
# Initialize reranker if configured
|
|
self.reranker = None
|
|
if config.reranker:
|
|
self.reranker = RerankerFactory.create(
|
|
config.reranker.provider,
|
|
config.reranker.config
|
|
)
|
|
|
|
# Entity store is initialized lazily on first use
|
|
self._entity_store = None
|
|
|
|
if MEM0_TELEMETRY:
|
|
# Create telemetry config manually to avoid deepcopy issues with thread locks
|
|
telemetry_config_dict = {}
|
|
if hasattr(self.config.vector_store.config, 'model_dump'):
|
|
# For pydantic models
|
|
telemetry_config_dict = self.config.vector_store.config.model_dump()
|
|
else:
|
|
# For other objects, manually copy common attributes
|
|
for attr in ['host', 'port', 'path', 'api_key', 'index_name', 'dimension', 'metric']:
|
|
if hasattr(self.config.vector_store.config, attr):
|
|
telemetry_config_dict[attr] = getattr(self.config.vector_store.config, attr)
|
|
|
|
# Override collection name for telemetry
|
|
telemetry_config_dict['collection_name'] = "mem0migrations"
|
|
|
|
# Set path for file-based vector stores
|
|
telemetry_config = _safe_deepcopy_config(self.config.vector_store.config)
|
|
if self.config.vector_store.provider in ["faiss", "qdrant"]:
|
|
provider_path = f"migrations_{self.config.vector_store.provider}"
|
|
telemetry_config_dict['path'] = os.path.join(mem0_dir, provider_path)
|
|
os.makedirs(telemetry_config_dict['path'], exist_ok=True)
|
|
|
|
# Create the config object using the same class as the original
|
|
telemetry_config = self.config.vector_store.config.__class__(**telemetry_config_dict)
|
|
self._telemetry_vector_store = VectorStoreFactory.create(
|
|
self.config.vector_store.provider, telemetry_config
|
|
)
|
|
if getattr(type(self.vector_store), "keyword_search", None) is VectorStoreBase.keyword_search:
|
|
logger.warning(
|
|
"The '%s' vector store does not support keyword search. "
|
|
"Hybrid (BM25) scoring will be disabled and search will use "
|
|
"semantic similarity only. To enable hybrid search, switch to a "
|
|
"store with keyword_search support (e.g. qdrant, elasticsearch, pgvector).",
|
|
self.config.vector_store.provider,
|
|
)
|
|
|
|
capture_event("mem0.init", self, {"sync_type": "sync"})
|
|
|
|
@property
|
|
def project(self):
|
|
return _OSSProject()
|
|
|
|
@property
|
|
def entity_store(self):
|
|
"""Lazily initialize entity store on first use."""
|
|
if self._entity_store is None:
|
|
entity_config = _safe_deepcopy_config(self.config.vector_store.config)
|
|
entity_collection = _entity_collection_name(self.config.vector_store.provider, self.collection_name)
|
|
# Set collection name on the cloned config
|
|
if hasattr(entity_config, 'collection_name'):
|
|
entity_config.collection_name = entity_collection
|
|
elif isinstance(entity_config, dict):
|
|
entity_config['collection_name'] = entity_collection
|
|
# For Qdrant, share the existing client to avoid RocksDB lock contention
|
|
# when using embedded mode (path=...). QdrantConfig.client takes precedence
|
|
# over host/port/path.
|
|
if self.config.vector_store.provider == "qdrant" and hasattr(self.vector_store, "client"):
|
|
if hasattr(entity_config, "client"):
|
|
entity_config.client = self.vector_store.client
|
|
elif isinstance(entity_config, dict):
|
|
entity_config["client"] = self.vector_store.client
|
|
self._entity_store = VectorStoreFactory.create(
|
|
self.config.vector_store.provider, entity_config
|
|
)
|
|
return self._entity_store
|
|
|
|
@staticmethod
|
|
def _normalize_entity_text(value: str) -> str:
|
|
return " ".join(value.strip().lower().split())
|
|
|
|
def _existing_entities_by_text(self, filters):
|
|
"""Return existing entity rows keyed by normalized payload data."""
|
|
try:
|
|
listed = self.entity_store.list(filters=filters, top_k=10000)
|
|
except Exception as e:
|
|
logger.debug(f"Exact entity lookup failed, falling back to semantic dedup: {e}")
|
|
return {}
|
|
|
|
rows_by_text = {}
|
|
for row in _vector_store_list_rows(listed):
|
|
payload = getattr(row, "payload", None) or {}
|
|
text = payload.get("data")
|
|
if not isinstance(text, str):
|
|
continue
|
|
normalized = self._normalize_entity_text(text)
|
|
if normalized and normalized not in rows_by_text:
|
|
rows_by_text[normalized] = row
|
|
return rows_by_text
|
|
|
|
def _upsert_entity(self, entity_text, entity_type, memory_id, filters):
|
|
"""Upsert an entity into the entity store, linking it to a memory."""
|
|
try:
|
|
entity_embedding = self.embedding_model.embed(entity_text, "add")
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
exact_match = self._existing_entities_by_text(search_filters).get(self._normalize_entity_text(entity_text))
|
|
|
|
existing = []
|
|
if exact_match is None:
|
|
existing = self.entity_store.search(
|
|
query=entity_text,
|
|
vectors=entity_embedding,
|
|
top_k=1,
|
|
filters=search_filters,
|
|
)
|
|
|
|
semantic_match = existing[0] if existing and existing[0].score >= 0.95 else None
|
|
match = exact_match or semantic_match
|
|
if match:
|
|
# Update existing entity's linked_memory_ids
|
|
payload = match.payload or {}
|
|
linked_ids = payload.get("linked_memory_ids", [])
|
|
if memory_id not in linked_ids:
|
|
linked_ids.append(memory_id)
|
|
payload["linked_memory_ids"] = linked_ids
|
|
self.entity_store.update(
|
|
vector_id=match.id,
|
|
vector=None,
|
|
payload=payload,
|
|
)
|
|
else:
|
|
# Create new entity
|
|
entity_id = str(uuid.uuid4())
|
|
entity_payload = {
|
|
"data": entity_text,
|
|
"entity_type": entity_type,
|
|
"linked_memory_ids": [memory_id],
|
|
**{k: v for k, v in search_filters.items()},
|
|
}
|
|
self.entity_store.insert(
|
|
vectors=[entity_embedding],
|
|
ids=[entity_id],
|
|
payloads=[entity_payload],
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Entity upsert failed for '{entity_text}': {e}")
|
|
|
|
def _remove_memory_from_entity_store(self, memory_id, filters):
|
|
"""Strip `memory_id` from every entity record scoped to `filters`.
|
|
|
|
For each entity whose `linked_memory_ids` contains `memory_id`:
|
|
- remove the id; if the list becomes empty, delete the entity record.
|
|
- otherwise re-embed the entity text and update the payload
|
|
(the vector store's update() requires a vector).
|
|
|
|
No-op if the entity store has never been initialized in this process.
|
|
Errors on individual entities are swallowed at debug level; outer
|
|
failures are swallowed at warning level so the primary delete/update
|
|
path is never broken by entity cleanup.
|
|
"""
|
|
if self._entity_store is None:
|
|
return
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
try:
|
|
listed = self.entity_store.list(filters=search_filters, top_k=10000)
|
|
rows = listed[0] if isinstance(listed, (list, tuple)) and listed and isinstance(listed[0], list) else listed
|
|
for row in rows or []:
|
|
try:
|
|
payload = getattr(row, "payload", None) or {}
|
|
linked = payload.get("linked_memory_ids", [])
|
|
if not isinstance(linked, list) or memory_id not in linked:
|
|
continue
|
|
remaining = [mid for mid in linked if mid != memory_id]
|
|
if not remaining:
|
|
try:
|
|
self.entity_store.delete(vector_id=row.id)
|
|
except Exception as e:
|
|
logger.debug(f"Entity delete failed for id={row.id}: {e}")
|
|
else:
|
|
entity_text = payload.get("data")
|
|
if not isinstance(entity_text, str) or not entity_text:
|
|
logger.debug(f"Entity id={row.id} missing 'data'; skipping update during cleanup")
|
|
continue
|
|
try:
|
|
vec = self.embedding_model.embed(entity_text, "update")
|
|
except Exception as e:
|
|
logger.debug(f"Entity re-embed failed for '{entity_text}': {e}")
|
|
continue
|
|
new_payload = {**payload, "linked_memory_ids": remaining}
|
|
try:
|
|
self.entity_store.update(
|
|
vector_id=row.id,
|
|
vector=vec,
|
|
payload=new_payload,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Entity update failed for id={row.id}: {e}")
|
|
except Exception as e:
|
|
logger.debug(f"Entity cleanup error: {e}")
|
|
except Exception as e:
|
|
logger.warning(f"Entity store cleanup failed for memory_id={memory_id}: {e}")
|
|
|
|
def _link_entities_for_memory(self, memory_id, text, filters):
|
|
"""Extract entities from `text` and link them to `memory_id` in the
|
|
entity store, scoped to `filters`. Simpler single-memory variant of
|
|
Phase 7 in add(): per-entity search-then-update-or-insert via the
|
|
existing `_upsert_entity` helper. Non-fatal on any failure.
|
|
"""
|
|
try:
|
|
entities = extract_entities(text)
|
|
if not entities:
|
|
return
|
|
seen = set()
|
|
for entity_type, entity_text in entities:
|
|
key = self._normalize_entity_text(entity_text)
|
|
if not key or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
try:
|
|
self._upsert_entity(entity_text, entity_type, memory_id, filters)
|
|
except Exception as e:
|
|
logger.debug(f"Entity link failed for '{entity_text}': {e}")
|
|
except Exception as e:
|
|
logger.warning(f"Entity linking failed for memory_id={memory_id}: {e}")
|
|
|
|
@classmethod
|
|
def from_config(cls, config_dict: Dict[str, Any]):
|
|
try:
|
|
config = MemoryConfig(**config_dict)
|
|
except ValidationError as e:
|
|
logger.error(f"Configuration validation error: {e}")
|
|
raise
|
|
return cls(config)
|
|
|
|
def _should_use_agent_memory_extraction(self, messages, metadata):
|
|
"""Determine whether to use agent memory extraction based on the logic:
|
|
- If agent_id is present and messages contain assistant role -> True
|
|
- Otherwise -> False
|
|
|
|
Args:
|
|
messages: List of message dictionaries
|
|
metadata: Metadata containing user_id, agent_id, etc.
|
|
|
|
Returns:
|
|
bool: True if should use agent memory extraction, False for user memory extraction
|
|
"""
|
|
# Check if agent_id is present in metadata
|
|
has_agent_id = metadata.get("agent_id") is not None
|
|
|
|
# Check if there are assistant role messages
|
|
has_assistant_messages = any(msg.get("role") == "assistant" for msg in messages)
|
|
|
|
# Use agent memory extraction if agent_id is present and there are assistant messages
|
|
return has_agent_id and has_assistant_messages
|
|
|
|
def add(
|
|
self,
|
|
messages,
|
|
*,
|
|
user_id: Optional[str] = None,
|
|
agent_id: Optional[str] = None,
|
|
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,
|
|
):
|
|
"""
|
|
Create a new memory.
|
|
|
|
Adds new memories scoped to a single session id (e.g. `user_id`, `agent_id`, or `run_id`). One of those ids is required.
|
|
|
|
Args:
|
|
messages (str or List[Dict[str, str]]): The message content or list of messages
|
|
(e.g., `[{"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi"}]`)
|
|
to be processed and stored.
|
|
user_id (str, optional): ID of the user creating the memory. Defaults to None.
|
|
agent_id (str, optional): ID of the agent creating the memory. Defaults to None.
|
|
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.
|
|
memory_type (str, optional): Specifies the type of memory. Currently, only
|
|
`MemoryType.PROCEDURAL.value` ("procedural_memory") is explicitly handled for
|
|
creating procedural memories (typically requires 'agent_id'). Otherwise, memories
|
|
are treated as general conversational/factual memories.
|
|
prompt (str, optional): Prompt to use for the memory creation. Defaults to None.
|
|
|
|
|
|
Returns:
|
|
dict: A dictionary containing the result of the memory addition operation, typically
|
|
including a list of memory items affected (added, updated) under a "results" key.
|
|
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", "event": "ADD"}]}`
|
|
|
|
Raises:
|
|
Mem0ValidationError: If input validation fails (invalid memory_type, messages format, etc.).
|
|
VectorStoreError: If vector store operations fail.
|
|
EmbeddingError: If embedding generation fails.
|
|
LLMError: If LLM operations fail.
|
|
DatabaseError: If database operations fail.
|
|
"""
|
|
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,
|
|
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 Mem0ValidationError(
|
|
message=f"Invalid 'memory_type'. Please pass {MemoryType.PROCEDURAL.value} to create procedural memories.",
|
|
error_code="VALIDATION_002",
|
|
details={"provided_type": memory_type, "valid_type": MemoryType.PROCEDURAL.value},
|
|
suggestion=f"Use '{MemoryType.PROCEDURAL.value}' to create procedural memories."
|
|
)
|
|
|
|
if isinstance(messages, str):
|
|
messages = [{"role": "user", "content": messages}]
|
|
|
|
elif isinstance(messages, dict):
|
|
messages = [messages]
|
|
|
|
elif not isinstance(messages, list):
|
|
raise Mem0ValidationError(
|
|
message="messages must be str, dict, or list[dict]",
|
|
error_code="VALIDATION_003",
|
|
details={"provided_type": type(messages).__name__, "valid_types": ["str", "dict", "list[dict]"]},
|
|
suggestion="Convert your input to a string, dictionary, or list of dictionaries."
|
|
)
|
|
|
|
if agent_id is not None and memory_type == MemoryType.PROCEDURAL.value:
|
|
results = self._create_procedural_memory(messages, metadata=processed_metadata, prompt=prompt)
|
|
scale_threshold_notice = detect_scale_threshold_from_add_result(self, results)
|
|
if temporal_usage_notice:
|
|
display_temporal_usage_notice(self, "sync", "add", *temporal_usage_notice)
|
|
elif scale_threshold_notice:
|
|
display_scale_threshold_notice(self, "sync", "add", *scale_threshold_notice)
|
|
else:
|
|
display_first_run_notice(self, "sync", "add")
|
|
return results
|
|
|
|
if self.config.llm.config.get("enable_vision"):
|
|
messages = parse_vision_messages(messages, self.llm, self.config.llm.config.get("vision_details"))
|
|
else:
|
|
messages = parse_vision_messages(messages)
|
|
|
|
vector_store_result = self._add_to_vector_store(messages, processed_metadata, effective_filters, infer, prompt=prompt)
|
|
scale_threshold_notice = detect_scale_threshold_from_add_result(self, vector_store_result)
|
|
if temporal_usage_notice:
|
|
display_temporal_usage_notice(self, "sync", "add", *temporal_usage_notice)
|
|
elif scale_threshold_notice:
|
|
display_scale_threshold_notice(self, "sync", "add", *scale_threshold_notice)
|
|
else:
|
|
display_first_run_notice(self, "sync", "add")
|
|
return {"results": vector_store_result}
|
|
|
|
def _add_to_vector_store(self, messages, metadata, filters, infer, prompt=None):
|
|
if not infer:
|
|
returned_memories = []
|
|
for message_dict in messages:
|
|
if (
|
|
not isinstance(message_dict, dict)
|
|
or message_dict.get("role") is None
|
|
or message_dict.get("content") is None
|
|
):
|
|
logger.warning(f"Skipping invalid message format: {message_dict}")
|
|
continue
|
|
|
|
if message_dict["role"] == "system":
|
|
continue
|
|
|
|
per_msg_meta = deepcopy(metadata)
|
|
per_msg_meta["role"] = message_dict["role"]
|
|
|
|
actor_name = message_dict.get("name")
|
|
if actor_name:
|
|
per_msg_meta["actor_id"] = actor_name
|
|
|
|
msg_content = message_dict["content"]
|
|
msg_embeddings = self.embedding_model.embed(msg_content, "add")
|
|
mem_id = self._create_memory(msg_content, {msg_content: msg_embeddings}, per_msg_meta)
|
|
|
|
returned_memories.append(
|
|
{
|
|
"id": mem_id,
|
|
"memory": msg_content,
|
|
"event": "ADD",
|
|
"actor_id": actor_name if actor_name else None,
|
|
"role": message_dict["role"],
|
|
}
|
|
)
|
|
return returned_memories
|
|
|
|
# === V3 PHASED BATCH PIPELINE ===
|
|
|
|
# Phase 0: Context gathering
|
|
session_scope = _build_session_scope(filters)
|
|
last_messages = self.db.get_last_messages(session_scope, limit=10)
|
|
parsed_messages = parse_messages(messages)
|
|
|
|
# Phase 1: Existing memory retrieval
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
query_embedding = self.embedding_model.embed(parsed_messages, "search")
|
|
existing_results = self.vector_store.search(
|
|
query=parsed_messages,
|
|
vectors=query_embedding,
|
|
top_k=10,
|
|
filters=search_filters,
|
|
)
|
|
|
|
# Map UUIDs to integers (anti-hallucination)
|
|
existing_memories = []
|
|
uuid_mapping = {}
|
|
for idx, mem in enumerate(existing_results):
|
|
uuid_mapping[str(idx)] = mem.id
|
|
existing_memories.append({"id": str(idx), "text": mem.payload.get("data", "")})
|
|
|
|
# Phase 2: LLM extraction (single call)
|
|
is_agent_scoped = bool(filters.get("agent_id")) and not filters.get("user_id")
|
|
system_prompt = ADDITIVE_EXTRACTION_PROMPT
|
|
if is_agent_scoped:
|
|
system_prompt += AGENT_CONTEXT_SUFFIX
|
|
|
|
custom_instr = prompt or self.custom_instructions
|
|
|
|
user_prompt = generate_additive_extraction_prompt(
|
|
existing_memories=existing_memories,
|
|
new_messages=parsed_messages,
|
|
last_k_messages=last_messages,
|
|
custom_instructions=custom_instr,
|
|
)
|
|
|
|
try:
|
|
response = self.llm.generate_response(
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt},
|
|
],
|
|
response_format={"type": "json_object"},
|
|
)
|
|
except Exception as e:
|
|
# Re-raise so callers can implement provider fallback / retry.
|
|
# The original silent ``return []`` made upstream callers unable to
|
|
# distinguish "LLM unavailable" (429/5xx/timeout) from "LLM
|
|
# extracted no facts" -- both surfaced as an empty list.
|
|
logger.error(f"LLM extraction failed: {e}")
|
|
raise LLMError(f"LLM extraction failed: {e}") from e
|
|
|
|
# Parse response
|
|
try:
|
|
response = remove_code_blocks(response)
|
|
if not response or not response.strip():
|
|
extracted_memories = []
|
|
else:
|
|
try:
|
|
extracted_memories = json.loads(response, strict=False).get("memory", [])
|
|
except json.JSONDecodeError:
|
|
extracted_json = extract_json(response)
|
|
extracted_memories = json.loads(extracted_json, strict=False).get("memory", [])
|
|
except Exception as e:
|
|
logger.error(f"Error parsing extraction response: {e}")
|
|
extracted_memories = []
|
|
|
|
if not extracted_memories:
|
|
# Save messages even if nothing extracted
|
|
self.db.save_messages(messages, session_scope)
|
|
return []
|
|
|
|
# Phase 3: Batch embed all extracted memory texts
|
|
mem_texts = [m.get("text", "") for m in extracted_memories if m.get("text")]
|
|
try:
|
|
mem_embeddings_list = self.embedding_model.embed_batch(mem_texts, "add")
|
|
embed_map = dict(zip(mem_texts, mem_embeddings_list))
|
|
except Exception:
|
|
# Fallback: embed individually
|
|
embed_map = {}
|
|
for text in mem_texts:
|
|
try:
|
|
embed_map[text] = self.embedding_model.embed(text, "add")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to embed memory text: {e}")
|
|
|
|
# Phase 4: Per-memory CPU processing + Phase 5: Hash dedup
|
|
# Build set of existing hashes for dedup
|
|
existing_hashes = set()
|
|
for mem in existing_results:
|
|
h = mem.payload.get("hash") if hasattr(mem, "payload") and mem.payload else None
|
|
if h:
|
|
existing_hashes.add(h)
|
|
|
|
records = [] # (memory_id, text, embedding, payload)
|
|
seen_hashes = set() # dedup within the current batch
|
|
for mem in extracted_memories:
|
|
text = mem.get("text")
|
|
if not text or text not in embed_map:
|
|
continue
|
|
|
|
mem_hash = hashlib.md5(text.encode()).hexdigest()
|
|
if mem_hash in existing_hashes or mem_hash in seen_hashes:
|
|
logger.debug(f"Skipping duplicate memory (hash match): {text[:50]}")
|
|
continue
|
|
seen_hashes.add(mem_hash)
|
|
|
|
text_lemmatized = lemmatize_for_bm25(text)
|
|
|
|
memory_id = str(uuid.uuid4())
|
|
mem_metadata = deepcopy(metadata)
|
|
mem_metadata["data"] = text
|
|
mem_metadata["text_lemmatized"] = text_lemmatized
|
|
mem_metadata["hash"] = mem_hash
|
|
if "created_at" not in mem_metadata:
|
|
mem_metadata["created_at"] = datetime.now(timezone.utc).isoformat()
|
|
mem_metadata["updated_at"] = mem_metadata["created_at"]
|
|
if mem.get("attributed_to"):
|
|
mem_metadata["attributed_to"] = mem["attributed_to"]
|
|
|
|
records.append((memory_id, text, embed_map[text], mem_metadata))
|
|
|
|
if not records:
|
|
self.db.save_messages(messages, session_scope)
|
|
return []
|
|
|
|
# Phase 6: Batch persist
|
|
all_vectors = [r[2] for r in records]
|
|
all_ids = [r[0] for r in records]
|
|
all_payloads = [r[3] for r in records]
|
|
|
|
try:
|
|
self.vector_store.insert(
|
|
vectors=all_vectors,
|
|
ids=all_ids,
|
|
payloads=all_payloads,
|
|
)
|
|
except Exception:
|
|
# Fallback: insert one by one
|
|
for mid, vec, pay in zip(all_ids, all_vectors, all_payloads):
|
|
try:
|
|
self.vector_store.insert(vectors=[vec], ids=[mid], payloads=[pay])
|
|
except Exception as e:
|
|
logger.error(f"Failed to insert memory {mid}: {e}")
|
|
|
|
# Batch history
|
|
history_records = [
|
|
{
|
|
"memory_id": r[0],
|
|
"old_memory": None,
|
|
"new_memory": r[1],
|
|
"event": "ADD",
|
|
"created_at": r[3].get("created_at"),
|
|
"is_deleted": 0,
|
|
}
|
|
for r in records
|
|
]
|
|
try:
|
|
self.db.batch_add_history(history_records)
|
|
except Exception:
|
|
# Fallback: add one by one
|
|
for hr in history_records:
|
|
try:
|
|
self.db.add_history(hr["memory_id"], None, hr["new_memory"], "ADD", created_at=hr.get("created_at"))
|
|
except Exception as e:
|
|
logger.error(f"Failed to add history for {hr['memory_id']}: {e}")
|
|
|
|
# Phase 7: Batch entity linking
|
|
try:
|
|
all_texts = [r[1] for r in records]
|
|
all_entities = extract_entities_batch(all_texts)
|
|
|
|
# 7a: Global dedup — collect unique entities across all memories
|
|
global_entities = {} # normalized_key -> (entity_type, entity_text, set of memory_ids)
|
|
for idx, (memory_id, text, embedding, payload) in enumerate(records):
|
|
entities = all_entities[idx] if idx < len(all_entities) else []
|
|
for entity_type, entity_text in entities:
|
|
key = self._normalize_entity_text(entity_text)
|
|
if key in global_entities:
|
|
global_entities[key][2].add(memory_id)
|
|
else:
|
|
global_entities[key] = [entity_type, entity_text, {memory_id}]
|
|
|
|
if global_entities:
|
|
ordered_keys = list(global_entities.keys())
|
|
entity_texts = [global_entities[k][1] for k in ordered_keys]
|
|
|
|
# 7b: Single batch embed for all unique entities
|
|
try:
|
|
entity_embeddings = self.embedding_model.embed_batch(entity_texts, "add")
|
|
except Exception:
|
|
# Fallback: embed individually, use None for failures
|
|
entity_embeddings = []
|
|
for t in entity_texts:
|
|
try:
|
|
entity_embeddings.append(self.embedding_model.embed(t, "add"))
|
|
except Exception:
|
|
entity_embeddings.append(None)
|
|
|
|
|
|
if len(entity_embeddings) != len(ordered_keys):
|
|
logger.warning(
|
|
"embed_batch returned %d vectors for %d entity texts — "
|
|
"padding/truncating to avoid dropping entity links",
|
|
len(entity_embeddings),
|
|
len(ordered_keys),
|
|
)
|
|
entity_embeddings = list(entity_embeddings[: len(ordered_keys)])
|
|
entity_embeddings += [None] * (len(ordered_keys) - len(entity_embeddings))
|
|
|
|
# Filter out entities with failed embeddings
|
|
valid = [(i, k) for i, k in enumerate(ordered_keys) if entity_embeddings[i] is not None]
|
|
if valid:
|
|
valid_indices, valid_keys = zip(*valid)
|
|
valid_vectors = [entity_embeddings[i] for i in valid_indices]
|
|
exact_matches = self._existing_entities_by_text(search_filters)
|
|
|
|
# 7c: Batch search for existing entities
|
|
valid_texts = [global_entities[k][1] for k in valid_keys]
|
|
existing_matches = self.entity_store.search_batch(
|
|
queries=valid_texts,
|
|
vectors_list=valid_vectors,
|
|
top_k=1,
|
|
filters=search_filters,
|
|
)
|
|
|
|
# 7d: Separate into inserts vs updates
|
|
to_insert_vectors, to_insert_ids, to_insert_payloads = [], [], []
|
|
for j, key in enumerate(valid_keys):
|
|
entity_type, entity_text, memory_ids = global_entities[key]
|
|
matches = existing_matches[j] if j < len(existing_matches) else []
|
|
exact_match = exact_matches.get(key)
|
|
|
|
semantic_match = matches[0] if matches and matches[0].score >= 0.95 else None
|
|
match = exact_match or semantic_match
|
|
if match:
|
|
# Update existing entity
|
|
payload = match.payload or {}
|
|
linked = set(payload.get("linked_memory_ids", []))
|
|
linked |= memory_ids
|
|
payload["linked_memory_ids"] = sorted(linked)
|
|
try:
|
|
self.entity_store.update(
|
|
vector_id=match.id,
|
|
vector=None,
|
|
payload=payload,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Entity update failed for '{entity_text}': {e}")
|
|
else:
|
|
# New entity — collect for batch insert
|
|
to_insert_vectors.append(valid_vectors[j])
|
|
to_insert_ids.append(str(uuid.uuid4()))
|
|
to_insert_payloads.append({
|
|
"data": entity_text,
|
|
"entity_type": entity_type,
|
|
"linked_memory_ids": sorted(memory_ids),
|
|
**search_filters,
|
|
})
|
|
|
|
# 7e: Single batch insert for all new entities
|
|
if to_insert_vectors:
|
|
try:
|
|
self.entity_store.insert(
|
|
vectors=to_insert_vectors,
|
|
ids=to_insert_ids,
|
|
payloads=to_insert_payloads,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Batch entity insert failed: {e}")
|
|
except Exception as e:
|
|
logger.warning(f"Batch entity linking failed: {e}")
|
|
|
|
# Phase 8: Save messages + return
|
|
self.db.save_messages(messages, session_scope)
|
|
|
|
returned_memories = [
|
|
{"id": r[0], "memory": r[1], "event": "ADD"}
|
|
for r in records
|
|
]
|
|
|
|
keys, encoded_ids = process_telemetry_filters(filters)
|
|
capture_event(
|
|
"mem0.add",
|
|
self,
|
|
{"version": self.api_version, "keys": keys, "encoded_ids": encoded_ids, "sync_type": "sync"},
|
|
)
|
|
return returned_memories
|
|
|
|
def get(self, memory_id):
|
|
"""
|
|
Retrieve a memory by ID.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to retrieve.
|
|
|
|
Returns:
|
|
dict: Retrieved memory.
|
|
"""
|
|
capture_event("mem0.get", self, {"memory_id": memory_id, "sync_type": "sync"})
|
|
memory = self.vector_store.get(vector_id=memory_id)
|
|
if not memory:
|
|
display_first_run_notice(self, "sync", "get")
|
|
return None
|
|
|
|
promoted_payload_keys = [
|
|
"user_id",
|
|
"agent_id",
|
|
"run_id",
|
|
"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}
|
|
|
|
result_item = MemoryItem(
|
|
id=memory.id,
|
|
memory=memory.payload.get("data", ""),
|
|
hash=memory.payload.get("hash"),
|
|
created_at=memory.payload.get("created_at"),
|
|
updated_at=memory.payload.get("updated_at"),
|
|
).model_dump()
|
|
|
|
for key in promoted_payload_keys:
|
|
if key in memory.payload:
|
|
result_item[key] = memory.payload[key]
|
|
|
|
additional_metadata = {k: v for k, v in memory.payload.items() if k not in core_and_promoted_keys}
|
|
if additional_metadata:
|
|
result_item["metadata"] = additional_metadata
|
|
|
|
display_first_run_notice(self, "sync", "get")
|
|
return result_item
|
|
|
|
def get_all(
|
|
self,
|
|
*,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
top_k: int = 20,
|
|
show_expired: bool = False,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
List all memories.
|
|
|
|
Args:
|
|
filters (dict): Filter dict containing entity IDs and optional metadata filters.
|
|
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.
|
|
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", ...}]}`
|
|
|
|
Raises:
|
|
ValueError: If filters doesn't contain at least one of user_id, agent_id, run_id,
|
|
or if top_k is invalid.
|
|
"""
|
|
# Reject top-level entity params - must use filters instead
|
|
_reject_top_level_entity_params(kwargs, "get_all")
|
|
|
|
# Validate top_k
|
|
_validate_search_params(top_k=top_k)
|
|
|
|
# Validate and trim entity IDs in filters
|
|
effective_filters = dict(filters) if filters else {}
|
|
if "user_id" in effective_filters:
|
|
effective_filters["user_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["user_id"], "user_id"
|
|
)
|
|
if "agent_id" in effective_filters:
|
|
effective_filters["agent_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["agent_id"], "agent_id"
|
|
)
|
|
if "run_id" in effective_filters:
|
|
effective_filters["run_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["run_id"], "run_id"
|
|
)
|
|
|
|
# Validate filters contains at least one entity ID
|
|
if not any(key in effective_filters for key in ("user_id", "agent_id", "run_id")):
|
|
raise ValueError(
|
|
"filters must contain at least one of: user_id, agent_id, run_id. "
|
|
"Example: filters={'user_id': 'u1'}"
|
|
)
|
|
|
|
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)
|
|
capture_event(
|
|
"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, fetch_limit, show_expired, limit)
|
|
|
|
if scale_threshold_notice:
|
|
display_scale_threshold_notice(self, "sync", "get_all", *scale_threshold_notice)
|
|
else:
|
|
display_first_run_notice(self, "sync", "get_all")
|
|
return {"results": all_memories_result}
|
|
|
|
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
|
|
if isinstance(memories_result, (tuple, list)) and len(memories_result) > 0:
|
|
first_element = memories_result[0]
|
|
|
|
# If first element is a container, unwrap one level
|
|
if isinstance(first_element, (list, tuple)):
|
|
actual_memories = first_element
|
|
else:
|
|
# First element is a memory object, structure is already flat
|
|
actual_memories = memories_result
|
|
else:
|
|
actual_memories = memories_result
|
|
|
|
promoted_payload_keys = [
|
|
"user_id",
|
|
"agent_id",
|
|
"run_id",
|
|
"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", ""),
|
|
hash=mem.payload.get("hash"),
|
|
created_at=mem.payload.get("created_at"),
|
|
updated_at=mem.payload.get("updated_at"),
|
|
).model_dump(exclude={"score"})
|
|
|
|
for key in promoted_payload_keys:
|
|
if key in mem.payload:
|
|
memory_item_dict[key] = mem.payload[key]
|
|
|
|
additional_metadata = {k: v for k, v in mem.payload.items() if k not in core_and_promoted_keys}
|
|
if additional_metadata:
|
|
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
|
|
|
|
def search(
|
|
self,
|
|
query: str,
|
|
*,
|
|
top_k: int = 20,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
threshold: float = 0.1,
|
|
rerank: bool = False,
|
|
explain: bool = False,
|
|
reference_date: Optional[Any] = None,
|
|
show_expired: bool = False,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
Searches for memories based on a query.
|
|
|
|
Args:
|
|
query (str): Query to search for.
|
|
top_k (int, optional): Maximum number of results to return. Defaults to 20.
|
|
filters (dict): Filter dict containing entity IDs and optional metadata filters.
|
|
Must contain at least one of: user_id, agent_id, run_id.
|
|
Example: filters={"user_id": "u1", "agent_id": "a1"}
|
|
|
|
Enhanced metadata filtering with operators:
|
|
- {"key": "value"} - exact match
|
|
- {"key": {"eq": "value"}} - equals
|
|
- {"key": {"ne": "value"}} - not equals
|
|
- {"key": {"in": ["val1", "val2"]}} - in list
|
|
- {"key": {"nin": ["val1", "val2"]}} - not in list
|
|
- {"key": {"gt": 10}} - greater than
|
|
- {"key": {"gte": 10}} - greater than or equal
|
|
- {"key": {"lt": 10}} - less than
|
|
- {"key": {"lte": 10}} - less than or equal
|
|
- {"key": {"contains": "text"}} - contains text
|
|
- {"key": {"icontains": "text"}} - case-insensitive contains
|
|
- {"key": "*"} - wildcard match (any value)
|
|
- {"AND": [filter1, filter2]} - logical AND
|
|
- {"OR": [filter1, filter2]} - logical OR
|
|
- {"NOT": [filter1]} - logical NOT
|
|
threshold (float, optional): Minimum score for a memory to be included. Defaults to 0.1.
|
|
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.
|
|
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", "score": 0.8, ...}]}`
|
|
|
|
Raises:
|
|
ValueError: If filters doesn't contain at least one of user_id, agent_id, run_id,
|
|
or if threshold/top_k values are invalid.
|
|
"""
|
|
if reference_date is not None:
|
|
raise ValueError(get_temporal_feature_error_message("sync", "search", "reference_date"))
|
|
|
|
# Reject top-level entity params - must use filters instead
|
|
_reject_top_level_entity_params(kwargs, "search")
|
|
|
|
# Validate search parameters (before applying defaults)
|
|
_validate_search_params(threshold=threshold, top_k=top_k)
|
|
query = _validate_and_trim_search_query(query)
|
|
temporal_usage_notice = detect_temporal_usage_from_search(query, filters)
|
|
|
|
# Validate and trim entity IDs in filters
|
|
effective_filters = filters.copy() if filters else {}
|
|
if "user_id" in effective_filters:
|
|
effective_filters["user_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["user_id"], "user_id"
|
|
)
|
|
if "agent_id" in effective_filters:
|
|
effective_filters["agent_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["agent_id"], "agent_id"
|
|
)
|
|
if "run_id" in effective_filters:
|
|
effective_filters["run_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["run_id"], "run_id"
|
|
)
|
|
if not any(key in effective_filters for key in ("user_id", "agent_id", "run_id")):
|
|
raise ValueError(
|
|
"filters must contain at least one of: user_id, agent_id, run_id. "
|
|
"Example: filters={'user_id': 'u1'}"
|
|
)
|
|
|
|
limit = top_k
|
|
scale_threshold_notice = detect_scale_threshold_from_top_k(top_k)
|
|
|
|
# Apply enhanced metadata filtering if advanced operators are detected
|
|
if self._has_advanced_operators(effective_filters):
|
|
processed_filters = self._process_metadata_filters(effective_filters)
|
|
# Remove logical/operator keys that have been reprocessed
|
|
for logical_key in ("AND", "OR", "NOT"):
|
|
effective_filters.pop(logical_key, None)
|
|
for fk in list(effective_filters.keys()):
|
|
if fk not in ("AND", "OR", "NOT", "user_id", "agent_id", "run_id") and isinstance(effective_filters.get(fk), dict):
|
|
effective_filters.pop(fk, None)
|
|
effective_filters.update(processed_filters)
|
|
|
|
keys, encoded_ids = process_telemetry_filters(effective_filters)
|
|
capture_event(
|
|
"mem0.search",
|
|
self,
|
|
{
|
|
"limit": limit,
|
|
"version": self.api_version,
|
|
"keys": keys,
|
|
"encoded_ids": encoded_ids,
|
|
"sync_type": "sync",
|
|
"threshold": threshold,
|
|
"explain": explain,
|
|
"advanced_filters": bool(filters and self._has_advanced_operators(filters)),
|
|
},
|
|
)
|
|
|
|
search_start = time.perf_counter()
|
|
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
|
|
if rerank and self.reranker and original_memories:
|
|
try:
|
|
reranked_memories = self.reranker.rerank(query, original_memories, limit)
|
|
original_memories = reranked_memories
|
|
except Exception as e:
|
|
logger.warning(f"Reranking failed, using original results: {e}")
|
|
|
|
if temporal_usage_notice:
|
|
display_temporal_usage_notice(self, "sync", "search", *temporal_usage_notice)
|
|
elif scale_threshold_notice:
|
|
display_scale_threshold_notice(self, "sync", "search", *scale_threshold_notice)
|
|
elif search_elapsed_seconds > PERFORMANCE_SLOW_QUERY_THRESHOLD_SECONDS:
|
|
display_performance_slow_query_notice(
|
|
self,
|
|
"sync",
|
|
"search",
|
|
search_elapsed_seconds,
|
|
top_k,
|
|
len(original_memories),
|
|
)
|
|
else:
|
|
display_first_run_notice(self, "sync", "search")
|
|
return {"results": original_memories}
|
|
|
|
def _process_metadata_filters(self, metadata_filters: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Process enhanced metadata filters and convert them to vector store compatible format.
|
|
|
|
Args:
|
|
metadata_filters: Enhanced metadata filters with operators
|
|
|
|
Returns:
|
|
Dict of processed filters compatible with vector store
|
|
"""
|
|
processed_filters = {}
|
|
|
|
def process_condition(key: str, condition: Any) -> Dict[str, Any]:
|
|
if not isinstance(condition, dict):
|
|
# Simple equality: {"key": "value"}
|
|
if condition == "*":
|
|
# Wildcard: match everything for this field (implementation depends on vector store)
|
|
return {key: "*"}
|
|
return {key: condition}
|
|
|
|
result = {}
|
|
for operator, value in condition.items():
|
|
# Map platform operators to universal format that can be translated by each vector store
|
|
operator_map = {
|
|
"eq": "eq", "ne": "ne", "gt": "gt", "gte": "gte",
|
|
"lt": "lt", "lte": "lte", "in": "in", "nin": "nin",
|
|
"contains": "contains", "icontains": "icontains"
|
|
}
|
|
|
|
if operator in operator_map:
|
|
result.setdefault(key, {})[operator_map[operator]] = value
|
|
else:
|
|
raise ValueError(f"Unsupported metadata filter operator: {operator}")
|
|
return result
|
|
|
|
def merge_filters(target: Dict[str, Any], source: Dict[str, Any]) -> None:
|
|
"""Merge source into target, deep-merging nested operator dicts for the same key."""
|
|
for key, value in source.items():
|
|
if key in target and isinstance(target[key], dict) and isinstance(value, dict):
|
|
target[key].update(value)
|
|
else:
|
|
target[key] = value
|
|
|
|
for key, value in metadata_filters.items():
|
|
if key == "AND":
|
|
# Logical AND: combine multiple conditions
|
|
if not isinstance(value, list):
|
|
raise ValueError("AND operator requires a list of conditions")
|
|
for condition in value:
|
|
for sub_key, sub_value in condition.items():
|
|
merge_filters(processed_filters, process_condition(sub_key, sub_value))
|
|
elif key == "OR":
|
|
# Logical OR: Pass through to vector store for implementation-specific handling
|
|
if not isinstance(value, list) or not value:
|
|
raise ValueError("OR operator requires a non-empty list of conditions")
|
|
# Store OR conditions in a way that vector stores can interpret
|
|
processed_filters["$or"] = []
|
|
for condition in value:
|
|
or_condition = {}
|
|
for sub_key, sub_value in condition.items():
|
|
merge_filters(or_condition, process_condition(sub_key, sub_value))
|
|
processed_filters["$or"].append(or_condition)
|
|
elif key == "NOT":
|
|
# Logical NOT: Pass through to vector store for implementation-specific handling
|
|
if not isinstance(value, list) or not value:
|
|
raise ValueError("NOT operator requires a non-empty list of conditions")
|
|
processed_filters["$not"] = []
|
|
for condition in value:
|
|
not_condition = {}
|
|
for sub_key, sub_value in condition.items():
|
|
merge_filters(not_condition, process_condition(sub_key, sub_value))
|
|
processed_filters["$not"].append(not_condition)
|
|
else:
|
|
merge_filters(processed_filters, process_condition(key, value))
|
|
|
|
return processed_filters
|
|
|
|
def _has_advanced_operators(self, filters: Dict[str, Any]) -> bool:
|
|
"""
|
|
Check if filters contain advanced operators that need special processing.
|
|
|
|
Args:
|
|
filters: Dictionary of filters to check
|
|
|
|
Returns:
|
|
bool: True if advanced operators are detected
|
|
"""
|
|
if not isinstance(filters, dict):
|
|
return False
|
|
|
|
for key, value in filters.items():
|
|
# Check for platform-style logical operators
|
|
if key in ["AND", "OR", "NOT"]:
|
|
return True
|
|
# Check for comparison operators (without $ prefix for universal compatibility)
|
|
if isinstance(value, dict):
|
|
for op in value.keys():
|
|
if op in ["eq", "ne", "gt", "gte", "lt", "lte", "in", "nin", "contains", "icontains"]:
|
|
return True
|
|
# Check for wildcard values
|
|
if value == "*":
|
|
return True
|
|
return 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
|
|
|
|
# Step 1: Preprocess query
|
|
query_lemmatized = lemmatize_for_bm25(query)
|
|
query_entities = extract_entities(query)
|
|
|
|
# Step 2: Embed query
|
|
embeddings = self.embedding_model.embed(query, "search")
|
|
|
|
# Step 3: Semantic search (over-fetch for scoring pool)
|
|
internal_limit = max(limit * 4, 60)
|
|
semantic_results = self.vector_store.search(
|
|
query=query, vectors=embeddings, top_k=internal_limit, filters=filters
|
|
)
|
|
|
|
# Step 4: Keyword search (if store supports it)
|
|
keyword_results = self.vector_store.keyword_search(
|
|
query=query_lemmatized, top_k=internal_limit, filters=filters
|
|
)
|
|
|
|
# Step 5: Compute BM25 scores from keyword results
|
|
bm25_scores = {}
|
|
if keyword_results is not None:
|
|
midpoint, steepness = get_bm25_params(query, lemmatized=query_lemmatized)
|
|
for mem in keyword_results:
|
|
mem_id = str(mem.id) if hasattr(mem, 'id') else str(mem.get('id', ''))
|
|
raw_score = mem.score if hasattr(mem, 'score') else mem.get('score', 0)
|
|
if raw_score and raw_score > 0:
|
|
bm25_scores[mem_id] = normalize_bm25(raw_score, midpoint, steepness)
|
|
|
|
# Step 6: Compute entity boosts
|
|
entity_boosts = {}
|
|
if query_entities:
|
|
entity_boosts = self._compute_entity_boosts(query_entities, filters)
|
|
|
|
# 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": payload,
|
|
})
|
|
|
|
# Step 8: Score and rank
|
|
scored_results = score_and_rank(
|
|
semantic_results=candidates,
|
|
bm25_scores=bm25_scores,
|
|
entity_boosts=entity_boosts,
|
|
threshold=threshold,
|
|
top_k=limit,
|
|
explain=explain,
|
|
)
|
|
|
|
# Step 9: Format results
|
|
promoted_payload_keys = [
|
|
"user_id",
|
|
"agent_id",
|
|
"run_id",
|
|
"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}
|
|
|
|
original_memories = []
|
|
for scored in scored_results:
|
|
payload = scored.get("payload") or {}
|
|
|
|
if not payload.get("data"):
|
|
continue # Skip candidates with no payload data
|
|
|
|
memory_item_dict = MemoryItem(
|
|
id=scored["id"],
|
|
memory=payload.get("data", ""),
|
|
hash=payload.get("hash"),
|
|
created_at=payload.get("created_at"),
|
|
updated_at=payload.get("updated_at"),
|
|
score=scored["score"],
|
|
).model_dump()
|
|
|
|
for key in promoted_payload_keys:
|
|
if key in payload:
|
|
memory_item_dict[key] = payload[key]
|
|
|
|
additional_metadata = {k: v for k, v in payload.items() if k not in core_and_promoted_keys}
|
|
if additional_metadata:
|
|
if not memory_item_dict.get("metadata"):
|
|
memory_item_dict["metadata"] = {}
|
|
memory_item_dict["metadata"].update(additional_metadata)
|
|
if explain and "score_details" in scored:
|
|
memory_item_dict["score_details"] = scored["score_details"]
|
|
|
|
original_memories.append(memory_item_dict)
|
|
|
|
return original_memories
|
|
|
|
def _compute_entity_boosts(self, query_entities, filters):
|
|
"""Compute per-memory entity boosts from entity store search.
|
|
|
|
For each extracted entity from the query:
|
|
1. Embed the entity text
|
|
2. Search the entity store (threshold >= 0.5)
|
|
3. For each matched entity, boost its linked memories
|
|
|
|
Returns:
|
|
Dict mapping memory_id (str) -> max entity boost [0, 0.5].
|
|
"""
|
|
# Deduplicate entities (max 8)
|
|
seen = set()
|
|
deduped = []
|
|
for entity_type, entity_text in query_entities[:8]:
|
|
key = self._normalize_entity_text(entity_text)
|
|
if key and key not in seen:
|
|
seen.add(key)
|
|
deduped.append((entity_type, entity_text))
|
|
|
|
if not deduped:
|
|
return {}
|
|
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
memory_boosts = {}
|
|
|
|
try:
|
|
entity_texts = [text for _, text in deduped]
|
|
embeddings = self.embedding_model.embed_batch(entity_texts, "search")
|
|
|
|
if len(embeddings) != len(entity_texts):
|
|
logger.warning(
|
|
"embed_batch returned %d vectors for %d texts — skipping entity boost",
|
|
len(embeddings),
|
|
len(entity_texts),
|
|
)
|
|
return memory_boosts
|
|
|
|
entity_store = self.entity_store
|
|
|
|
def _search_entity(entity_text, embedding):
|
|
return entity_store.search(
|
|
query=entity_text, vectors=embedding, top_k=500, filters=search_filters
|
|
)
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
|
|
futures = {
|
|
pool.submit(_search_entity, text, emb): text
|
|
for text, emb in zip(entity_texts, embeddings)
|
|
}
|
|
|
|
for future in concurrent.futures.as_completed(futures):
|
|
try:
|
|
matches = future.result()
|
|
except Exception as e:
|
|
logger.warning("Entity boost search failed for one entity: %s", e)
|
|
continue
|
|
|
|
for match in matches:
|
|
similarity = match.score if hasattr(match, 'score') else 0.0
|
|
if similarity < 0.5:
|
|
continue
|
|
|
|
payload = match.payload if hasattr(match, 'payload') else {}
|
|
linked_memory_ids = payload.get("linked_memory_ids", [])
|
|
if not isinstance(linked_memory_ids, list):
|
|
continue
|
|
|
|
num_linked = max(len(linked_memory_ids), 1)
|
|
memory_count_weight = 1.0 / (1.0 + 0.001 * ((num_linked - 1) ** 2))
|
|
boost = similarity * ENTITY_BOOST_WEIGHT * memory_count_weight
|
|
|
|
for memory_id in linked_memory_ids:
|
|
if memory_id:
|
|
memory_key = str(memory_id)
|
|
memory_boosts[memory_key] = max(memory_boosts.get(memory_key, 0.0), boost)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Entity boost computation failed: {e}")
|
|
|
|
return memory_boosts
|
|
|
|
def update(
|
|
self,
|
|
memory_id,
|
|
text: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
expiration_date: Any = _UNSET,
|
|
data: Optional[str] = None,
|
|
):
|
|
"""
|
|
Update a memory by ID.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to update.
|
|
text (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.
|
|
data (str, optional): Deprecated alias for ``text``. Will be removed in the next
|
|
major release; use ``text`` instead.
|
|
|
|
Returns:
|
|
dict: Success message indicating the memory was updated.
|
|
|
|
Example:
|
|
>>> m.update(memory_id="mem_123", text="Likes to play tennis on weekends")
|
|
{'message': 'Memory updated successfully!'}
|
|
"""
|
|
capture_event("mem0.update", self, {"memory_id": memory_id, "sync_type": "sync"})
|
|
|
|
if data is not None:
|
|
logger.warning(
|
|
"The `data` argument to update() is deprecated and will be removed in the "
|
|
"next major release. Use `text` instead."
|
|
)
|
|
if text is None:
|
|
text = data
|
|
|
|
if text is None and metadata is None and expiration_date is _UNSET:
|
|
raise ValueError("At least one of text, metadata, or expiration_date must be provided.")
|
|
|
|
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 text is not None:
|
|
existing_embeddings[text] = self.embedding_model.embed(text, "update")
|
|
|
|
self._update_memory(memory_id, text, existing_embeddings, update_metadata)
|
|
display_first_run_notice(self, "sync", "update")
|
|
return {"message": "Memory updated successfully!"}
|
|
|
|
def delete(self, memory_id):
|
|
"""
|
|
Delete a memory by ID.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to delete.
|
|
"""
|
|
capture_event("mem0.delete", self, {"memory_id": memory_id, "sync_type": "sync"})
|
|
|
|
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")
|
|
|
|
self._delete_memory(memory_id, existing_memory)
|
|
decay_usage_notice = detect_decay_usage_from_delete()
|
|
if decay_usage_notice:
|
|
display_decay_usage_notice(self, "sync", "delete", *decay_usage_notice)
|
|
else:
|
|
display_first_run_notice(self, "sync", "delete")
|
|
return {"message": "Memory deleted successfully!"}
|
|
|
|
def delete_all(self, user_id: Optional[str] = None, agent_id: Optional[str] = None, run_id: Optional[str] = None):
|
|
"""
|
|
Delete all memories.
|
|
|
|
Args:
|
|
user_id (str, optional): ID of the user to delete memories for. Defaults to None.
|
|
agent_id (str, optional): ID of the agent to delete memories for. Defaults to None.
|
|
run_id (str, optional): ID of the run to delete memories for. Defaults to None.
|
|
"""
|
|
user_id = _validate_and_trim_entity_id(user_id, "user_id")
|
|
agent_id = _validate_and_trim_entity_id(agent_id, "agent_id")
|
|
run_id = _validate_and_trim_entity_id(run_id, "run_id")
|
|
|
|
filters: Dict[str, Any] = {}
|
|
if user_id:
|
|
filters["user_id"] = user_id
|
|
if agent_id:
|
|
filters["agent_id"] = agent_id
|
|
if run_id:
|
|
filters["run_id"] = run_id
|
|
|
|
if not filters:
|
|
raise ValueError(
|
|
"At least one filter is required to delete all memories. If you want to delete all memories, use the `reset()` method."
|
|
)
|
|
|
|
keys, encoded_ids = process_telemetry_filters(filters)
|
|
capture_event("mem0.delete_all", self, {"keys": keys, "encoded_ids": encoded_ids, "sync_type": "sync"})
|
|
# delete all vector memories and reset the collections
|
|
memories = self.vector_store.list(filters=filters)[0]
|
|
for memory in memories:
|
|
self._delete_memory(memory.id)
|
|
|
|
logger.info(f"Deleted {len(memories)} memories")
|
|
|
|
decay_usage_notice = detect_decay_usage_from_delete_all(len(memories))
|
|
if decay_usage_notice:
|
|
display_decay_usage_notice(self, "sync", "delete_all", *decay_usage_notice)
|
|
else:
|
|
display_first_run_notice(self, "sync", "delete_all")
|
|
return {"message": "Memories deleted successfully!"}
|
|
|
|
def history(self, memory_id):
|
|
"""
|
|
Get the history of changes for a memory by ID.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to get history for.
|
|
|
|
Returns:
|
|
list: List of changes for the memory.
|
|
"""
|
|
capture_event("mem0.history", self, {"memory_id": memory_id, "sync_type": "sync"})
|
|
history = self.db.get_history(memory_id)
|
|
display_first_run_notice(self, "sync", "history")
|
|
return history
|
|
|
|
def _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],
|
|
)
|
|
self.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
|
|
|
|
def _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 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
|
|
|
|
def _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(metadata)
|
|
|
|
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()
|
|
|
|
# actor_id is immutable after creation (issue #4490)
|
|
if "actor_id" in existing_memory.payload:
|
|
new_metadata["actor_id"] = existing_memory.payload["actor_id"]
|
|
|
|
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
|
|
|
|
def _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(
|
|
self.config.vector_store.provider, self.config.vector_store.config
|
|
)
|
|
# Reset entity store if initialized
|
|
if self._entity_store is not None:
|
|
try:
|
|
self._entity_store.reset()
|
|
except Exception as e:
|
|
logger.warning(f"Failed to reset entity store: {e}")
|
|
self._entity_store = None
|
|
|
|
capture_event("mem0.reset", self, {"sync_type": "sync"})
|
|
display_first_run_notice(self, "sync", "reset")
|
|
|
|
def close(self):
|
|
"""Release resources held by this Memory instance (SQLite connections, etc.)."""
|
|
if hasattr(self, "db") and self.db is not None:
|
|
self.db.close()
|
|
self.db = None
|
|
|
|
def chat(self, query):
|
|
raise NotImplementedError("Chat function not implemented yet.")
|
|
|
|
|
|
class AsyncMemory(MemoryBase):
|
|
def __init__(self, config: MemoryConfig = MemoryConfig()):
|
|
self.config = config
|
|
|
|
self.embedding_model = EmbedderFactory.create(
|
|
self.config.embedder.provider,
|
|
self.config.embedder.config,
|
|
self.config.vector_store.config,
|
|
)
|
|
self.vector_store = VectorStoreFactory.create(
|
|
self.config.vector_store.provider, self.config.vector_store.config
|
|
)
|
|
self.llm = LlmFactory.create(self.config.llm.provider, self.config.llm.config)
|
|
self.db = SQLiteManager(self.config.history_db_path)
|
|
self.collection_name = self.config.vector_store.config.collection_name
|
|
self.api_version = self.config.version
|
|
self.custom_instructions = self.config.custom_instructions
|
|
self._entity_store = None
|
|
|
|
# Initialize reranker if configured
|
|
self.reranker = None
|
|
if config.reranker:
|
|
self.reranker = RerankerFactory.create(
|
|
config.reranker.provider,
|
|
config.reranker.config
|
|
)
|
|
|
|
if MEM0_TELEMETRY:
|
|
telemetry_config = _safe_deepcopy_config(self.config.vector_store.config)
|
|
telemetry_config.collection_name = "mem0migrations"
|
|
if self.config.vector_store.provider in ["faiss", "qdrant"]:
|
|
provider_path = f"migrations_{self.config.vector_store.provider}"
|
|
telemetry_config.path = os.path.join(mem0_dir, provider_path)
|
|
os.makedirs(telemetry_config.path, exist_ok=True)
|
|
self._telemetry_vector_store = VectorStoreFactory.create(self.config.vector_store.provider, telemetry_config)
|
|
|
|
if getattr(type(self.vector_store), "keyword_search", None) is VectorStoreBase.keyword_search:
|
|
logger.warning(
|
|
"The '%s' vector store does not support keyword search. "
|
|
"Hybrid (BM25) scoring will be disabled and search will use "
|
|
"semantic similarity only. To enable hybrid search, switch to a "
|
|
"store with keyword_search support (e.g. qdrant, elasticsearch, pgvector).",
|
|
self.config.vector_store.provider,
|
|
)
|
|
|
|
capture_event("mem0.init", self, {"sync_type": "async"})
|
|
|
|
@property
|
|
def project(self):
|
|
return _AsyncOSSProject()
|
|
|
|
@property
|
|
def entity_store(self):
|
|
"""Lazily initialize entity store on first use."""
|
|
if self._entity_store is None:
|
|
entity_config = _safe_deepcopy_config(self.config.vector_store.config)
|
|
entity_collection = _entity_collection_name(self.config.vector_store.provider, self.collection_name)
|
|
if hasattr(entity_config, 'collection_name'):
|
|
entity_config.collection_name = entity_collection
|
|
elif isinstance(entity_config, dict):
|
|
entity_config['collection_name'] = entity_collection
|
|
# For Qdrant, share the existing client to avoid RocksDB lock contention
|
|
# when using embedded mode (path=...). QdrantConfig.client takes precedence
|
|
# over host/port/path.
|
|
if self.config.vector_store.provider == "qdrant" and hasattr(self.vector_store, "client"):
|
|
if hasattr(entity_config, "client"):
|
|
entity_config.client = self.vector_store.client
|
|
elif isinstance(entity_config, dict):
|
|
entity_config["client"] = self.vector_store.client
|
|
self._entity_store = VectorStoreFactory.create(
|
|
self.config.vector_store.provider, entity_config
|
|
)
|
|
return self._entity_store
|
|
|
|
@staticmethod
|
|
def _normalize_entity_text(value: str) -> str:
|
|
return " ".join(value.strip().lower().split())
|
|
|
|
def _existing_entities_by_text(self, filters):
|
|
"""Return existing entity rows keyed by normalized payload data."""
|
|
try:
|
|
listed = self.entity_store.list(filters=filters, top_k=10000)
|
|
except Exception as e:
|
|
logger.debug(f"Exact entity lookup failed, falling back to semantic dedup: {e}")
|
|
return {}
|
|
|
|
rows_by_text = {}
|
|
for row in _vector_store_list_rows(listed):
|
|
payload = getattr(row, "payload", None) or {}
|
|
text = payload.get("data")
|
|
if not isinstance(text, str):
|
|
continue
|
|
normalized = self._normalize_entity_text(text)
|
|
if normalized and normalized not in rows_by_text:
|
|
rows_by_text[normalized] = row
|
|
return rows_by_text
|
|
|
|
async def _upsert_entity_async(self, entity_text, entity_type, memory_id, filters):
|
|
"""Async variant of `_upsert_entity` — per-entity search-then-update-or-insert."""
|
|
try:
|
|
entity_embedding = await asyncio.to_thread(self.embedding_model.embed, entity_text, "add")
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
exact_match = (
|
|
await asyncio.to_thread(self._existing_entities_by_text, search_filters)
|
|
).get(self._normalize_entity_text(entity_text))
|
|
|
|
existing = []
|
|
if exact_match is None:
|
|
existing = await asyncio.to_thread(
|
|
self.entity_store.search,
|
|
query=entity_text,
|
|
vectors=entity_embedding,
|
|
top_k=1,
|
|
filters=search_filters,
|
|
)
|
|
|
|
semantic_match = existing[0] if existing and existing[0].score >= 0.95 else None
|
|
match = exact_match or semantic_match
|
|
if match:
|
|
payload = match.payload or {}
|
|
linked_ids = payload.get("linked_memory_ids", [])
|
|
if memory_id not in linked_ids:
|
|
linked_ids.append(memory_id)
|
|
payload["linked_memory_ids"] = linked_ids
|
|
await asyncio.to_thread(
|
|
self.entity_store.update,
|
|
vector_id=match.id,
|
|
vector=None,
|
|
payload=payload,
|
|
)
|
|
else:
|
|
entity_id = str(uuid.uuid4())
|
|
entity_payload = {
|
|
"data": entity_text,
|
|
"entity_type": entity_type,
|
|
"linked_memory_ids": [memory_id],
|
|
**{k: v for k, v in search_filters.items()},
|
|
}
|
|
await asyncio.to_thread(
|
|
self.entity_store.insert,
|
|
vectors=[entity_embedding],
|
|
ids=[entity_id],
|
|
payloads=[entity_payload],
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Entity upsert failed for '{entity_text}' (async): {e}")
|
|
|
|
async def _bulk_clear_entity_store(self, filters):
|
|
"""Delete all entity records matching the given scope filters.
|
|
|
|
Used by delete_all to avoid the race condition that occurs when
|
|
concurrent _delete_memory coroutines each try to read-modify-write
|
|
the same entity rows' linked_memory_ids lists.
|
|
"""
|
|
if self._entity_store is None:
|
|
return
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
try:
|
|
listed = await asyncio.to_thread(self.entity_store.list, filters=search_filters, top_k=10000)
|
|
rows = listed[0] if isinstance(listed, (list, tuple)) and listed and isinstance(listed[0], list) else listed
|
|
for row in rows or []:
|
|
try:
|
|
await asyncio.to_thread(self.entity_store.delete, vector_id=row.id)
|
|
except Exception as e:
|
|
logger.debug(f"Bulk entity delete failed for id={row.id}: {e}")
|
|
except Exception as e:
|
|
logger.warning(f"Bulk entity store cleanup failed: {e}")
|
|
|
|
async def _remove_memory_from_entity_store(self, memory_id, filters):
|
|
"""Async variant of `Memory._remove_memory_from_entity_store`."""
|
|
if self._entity_store is None:
|
|
return
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
try:
|
|
listed = await asyncio.to_thread(self.entity_store.list, filters=search_filters, top_k=10000)
|
|
rows = listed[0] if isinstance(listed, (list, tuple)) and listed and isinstance(listed[0], list) else listed
|
|
for row in rows or []:
|
|
try:
|
|
payload = getattr(row, "payload", None) or {}
|
|
linked = payload.get("linked_memory_ids", [])
|
|
if not isinstance(linked, list) or memory_id not in linked:
|
|
continue
|
|
remaining = [mid for mid in linked if mid != memory_id]
|
|
if not remaining:
|
|
try:
|
|
await asyncio.to_thread(self.entity_store.delete, vector_id=row.id)
|
|
except Exception as e:
|
|
logger.debug(f"Entity delete failed for id={row.id} (async): {e}")
|
|
else:
|
|
entity_text = payload.get("data")
|
|
if not isinstance(entity_text, str) or not entity_text:
|
|
logger.debug(f"Entity id={row.id} missing 'data'; skipping update during cleanup (async)")
|
|
continue
|
|
try:
|
|
vec = await asyncio.to_thread(self.embedding_model.embed, entity_text, "update")
|
|
except Exception as e:
|
|
logger.debug(f"Entity re-embed failed for '{entity_text}' (async): {e}")
|
|
continue
|
|
new_payload = {**payload, "linked_memory_ids": remaining}
|
|
try:
|
|
await asyncio.to_thread(
|
|
self.entity_store.update,
|
|
vector_id=row.id,
|
|
vector=vec,
|
|
payload=new_payload,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Entity update failed for id={row.id} (async): {e}")
|
|
except Exception as e:
|
|
logger.debug(f"Entity cleanup error (async): {e}")
|
|
except Exception as e:
|
|
logger.warning(f"Entity store cleanup failed for memory_id={memory_id} (async): {e}")
|
|
|
|
async def _link_entities_for_memory(self, memory_id, text, filters):
|
|
"""Async variant of `Memory._link_entities_for_memory`."""
|
|
try:
|
|
entities = await asyncio.to_thread(extract_entities, text)
|
|
if not entities:
|
|
return
|
|
seen = set()
|
|
for entity_type, entity_text in entities:
|
|
key = self._normalize_entity_text(entity_text)
|
|
if not key or key in seen:
|
|
continue
|
|
seen.add(key)
|
|
try:
|
|
await self._upsert_entity_async(entity_text, entity_type, memory_id, filters)
|
|
except Exception as e:
|
|
logger.debug(f"Entity link failed for '{entity_text}' (async): {e}")
|
|
except Exception as e:
|
|
logger.warning(f"Entity linking failed for memory_id={memory_id} (async): {e}")
|
|
|
|
@classmethod
|
|
def from_config(cls, config_dict: Dict[str, Any]):
|
|
try:
|
|
config = MemoryConfig(**config_dict)
|
|
except ValidationError as e:
|
|
logger.error(f"Configuration validation error: {e}")
|
|
raise
|
|
return cls(config)
|
|
|
|
def _should_use_agent_memory_extraction(self, messages, metadata):
|
|
"""Determine whether to use agent memory extraction based on the logic:
|
|
- If agent_id is present and messages contain assistant role -> True
|
|
- Otherwise -> False
|
|
|
|
Args:
|
|
messages: List of message dictionaries
|
|
metadata: Metadata containing user_id, agent_id, etc.
|
|
|
|
Returns:
|
|
bool: True if should use agent memory extraction, False for user memory extraction
|
|
"""
|
|
# Check if agent_id is present in metadata
|
|
has_agent_id = metadata.get("agent_id") is not None
|
|
|
|
# Check if there are assistant role messages
|
|
has_assistant_messages = any(msg.get("role") == "assistant" for msg in messages)
|
|
|
|
# Use agent memory extraction if agent_id is present and there are assistant messages
|
|
return has_agent_id and has_assistant_messages
|
|
|
|
async def add(
|
|
self,
|
|
messages,
|
|
*,
|
|
user_id: Optional[str] = None,
|
|
agent_id: Optional[str] = None,
|
|
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,
|
|
llm=None,
|
|
):
|
|
"""
|
|
Create a new memory asynchronously.
|
|
|
|
Args:
|
|
messages (str or List[Dict[str, str]]): Messages to store in the memory.
|
|
user_id (str, optional): ID of the user creating the memory.
|
|
agent_id (str, optional): ID of the agent creating the memory. Defaults to None.
|
|
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.
|
|
prompt (str, optional): Prompt to use for the memory creation. Defaults to None.
|
|
llm (BaseChatModel, optional): LLM class to use for generating procedural memories. Defaults to None. Useful when user is using LangChain ChatModel.
|
|
Returns:
|
|
dict: A dictionary containing the result of the memory addition operation.
|
|
"""
|
|
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(
|
|
f"Invalid 'memory_type'. Please pass {MemoryType.PROCEDURAL.value} to create procedural memories."
|
|
)
|
|
|
|
if isinstance(messages, str):
|
|
messages = [{"role": "user", "content": messages}]
|
|
|
|
elif isinstance(messages, dict):
|
|
messages = [messages]
|
|
|
|
elif not isinstance(messages, list):
|
|
raise Mem0ValidationError(
|
|
message="messages must be str, dict, or list[dict]",
|
|
error_code="VALIDATION_003",
|
|
details={"provided_type": type(messages).__name__, "valid_types": ["str", "dict", "list[dict]"]},
|
|
suggestion="Convert your input to a string, dictionary, or list of dictionaries."
|
|
)
|
|
|
|
if agent_id is not None and memory_type == MemoryType.PROCEDURAL.value:
|
|
results = await self._create_procedural_memory(
|
|
messages, metadata=processed_metadata, prompt=prompt, llm=llm
|
|
)
|
|
scale_threshold_notice = await asyncio.to_thread(detect_scale_threshold_from_add_result, self, results)
|
|
if temporal_usage_notice:
|
|
await display_temporal_usage_notice_async(self, "async", "add", *temporal_usage_notice)
|
|
elif scale_threshold_notice:
|
|
await display_scale_threshold_notice_async(self, "async", "add", *scale_threshold_notice)
|
|
else:
|
|
await display_first_run_notice_async(self, "async", "add")
|
|
return results
|
|
|
|
if self.config.llm.config.get("enable_vision"):
|
|
messages = parse_vision_messages(messages, self.llm, self.config.llm.config.get("vision_details"))
|
|
else:
|
|
messages = parse_vision_messages(messages)
|
|
|
|
vector_store_result = await self._add_to_vector_store(messages, processed_metadata, effective_filters, infer, prompt=prompt)
|
|
scale_threshold_notice = await asyncio.to_thread(detect_scale_threshold_from_add_result, self, vector_store_result)
|
|
if temporal_usage_notice:
|
|
await display_temporal_usage_notice_async(self, "async", "add", *temporal_usage_notice)
|
|
elif scale_threshold_notice:
|
|
await display_scale_threshold_notice_async(self, "async", "add", *scale_threshold_notice)
|
|
else:
|
|
await display_first_run_notice_async(self, "async", "add")
|
|
return {"results": vector_store_result}
|
|
|
|
async def _add_to_vector_store(
|
|
self,
|
|
messages: list,
|
|
metadata: dict,
|
|
effective_filters: dict,
|
|
infer: bool,
|
|
prompt: Optional[str] = None,
|
|
):
|
|
if not infer:
|
|
returned_memories = []
|
|
for message_dict in messages:
|
|
if (
|
|
not isinstance(message_dict, dict)
|
|
or message_dict.get("role") is None
|
|
or message_dict.get("content") is None
|
|
):
|
|
logger.warning(f"Skipping invalid message format (async): {message_dict}")
|
|
continue
|
|
|
|
if message_dict["role"] == "system":
|
|
continue
|
|
|
|
per_msg_meta = deepcopy(metadata)
|
|
per_msg_meta["role"] = message_dict["role"]
|
|
|
|
actor_name = message_dict.get("name")
|
|
if actor_name:
|
|
per_msg_meta["actor_id"] = actor_name
|
|
|
|
msg_content = message_dict["content"]
|
|
msg_embeddings = await asyncio.to_thread(self.embedding_model.embed, msg_content, "add")
|
|
mem_id = await self._create_memory(msg_content, {msg_content: msg_embeddings}, per_msg_meta)
|
|
|
|
returned_memories.append(
|
|
{
|
|
"id": mem_id,
|
|
"memory": msg_content,
|
|
"event": "ADD",
|
|
"actor_id": actor_name if actor_name else None,
|
|
"role": message_dict["role"],
|
|
}
|
|
)
|
|
return returned_memories
|
|
|
|
# === V3 PHASED BATCH PIPELINE (async) ===
|
|
|
|
# Phase 0: Context gathering
|
|
session_scope = _build_session_scope(effective_filters)
|
|
last_messages = await asyncio.to_thread(self.db.get_last_messages, session_scope, 10)
|
|
parsed_messages = parse_messages(messages)
|
|
|
|
# Phase 1: Existing memory retrieval
|
|
search_filters = {k: v for k, v in effective_filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
query_embedding = await asyncio.to_thread(self.embedding_model.embed, parsed_messages, "search")
|
|
existing_results = await asyncio.to_thread(
|
|
self.vector_store.search,
|
|
query=parsed_messages,
|
|
vectors=query_embedding,
|
|
top_k=10,
|
|
filters=search_filters,
|
|
)
|
|
|
|
# Map UUIDs to integers (anti-hallucination)
|
|
existing_memories = []
|
|
uuid_mapping = {}
|
|
for idx, mem in enumerate(existing_results):
|
|
uuid_mapping[str(idx)] = mem.id
|
|
existing_memories.append({"id": str(idx), "text": mem.payload.get("data", "")})
|
|
|
|
# Phase 2: LLM extraction (single call)
|
|
is_agent_scoped = bool(effective_filters.get("agent_id")) and not effective_filters.get("user_id")
|
|
system_prompt = ADDITIVE_EXTRACTION_PROMPT
|
|
if is_agent_scoped:
|
|
system_prompt += AGENT_CONTEXT_SUFFIX
|
|
|
|
custom_instr = prompt or self.custom_instructions
|
|
|
|
user_prompt = generate_additive_extraction_prompt(
|
|
existing_memories=existing_memories,
|
|
new_messages=parsed_messages,
|
|
last_k_messages=last_messages,
|
|
custom_instructions=custom_instr,
|
|
)
|
|
|
|
try:
|
|
response = await asyncio.to_thread(
|
|
self.llm.generate_response,
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt},
|
|
],
|
|
response_format={"type": "json_object"},
|
|
)
|
|
except Exception as e:
|
|
# Re-raise so callers can implement provider fallback / retry
|
|
# (see sync counterpart for rationale).
|
|
logger.error(f"LLM extraction failed (async): {e}")
|
|
raise LLMError(f"LLM extraction failed: {e}") from e
|
|
|
|
# Parse response
|
|
try:
|
|
response = remove_code_blocks(response)
|
|
if not response or not response.strip():
|
|
extracted_memories = []
|
|
else:
|
|
try:
|
|
extracted_memories = json.loads(response, strict=False).get("memory", [])
|
|
except json.JSONDecodeError:
|
|
extracted_json = extract_json(response)
|
|
extracted_memories = json.loads(extracted_json, strict=False).get("memory", [])
|
|
except Exception as e:
|
|
logger.error(f"Error parsing extraction response (async): {e}")
|
|
extracted_memories = []
|
|
|
|
if not extracted_memories:
|
|
await asyncio.to_thread(self.db.save_messages, messages, session_scope)
|
|
return []
|
|
|
|
# Phase 3: Batch embed all extracted memory texts
|
|
mem_texts = [m.get("text", "") for m in extracted_memories if m.get("text")]
|
|
try:
|
|
mem_embeddings_list = await asyncio.to_thread(self.embedding_model.embed_batch, mem_texts, "add")
|
|
embed_map = dict(zip(mem_texts, mem_embeddings_list))
|
|
except Exception:
|
|
embed_map = {}
|
|
for text in mem_texts:
|
|
try:
|
|
embed_map[text] = await asyncio.to_thread(self.embedding_model.embed, text, "add")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to embed memory text (async): {e}")
|
|
|
|
# Phase 4: Per-memory CPU processing + Phase 5: Hash dedup
|
|
existing_hashes = set()
|
|
for mem in existing_results:
|
|
h = mem.payload.get("hash") if hasattr(mem, "payload") and mem.payload else None
|
|
if h:
|
|
existing_hashes.add(h)
|
|
|
|
records = []
|
|
seen_hashes = set()
|
|
for mem in extracted_memories:
|
|
text = mem.get("text")
|
|
if not text or text not in embed_map:
|
|
continue
|
|
|
|
mem_hash = hashlib.md5(text.encode()).hexdigest()
|
|
if mem_hash in existing_hashes or mem_hash in seen_hashes:
|
|
logger.debug(f"Skipping duplicate memory (hash match, async): {text[:50]}")
|
|
continue
|
|
seen_hashes.add(mem_hash)
|
|
|
|
text_lemmatized = lemmatize_for_bm25(text)
|
|
|
|
memory_id = str(uuid.uuid4())
|
|
mem_metadata = deepcopy(metadata)
|
|
mem_metadata["data"] = text
|
|
mem_metadata["text_lemmatized"] = text_lemmatized
|
|
mem_metadata["hash"] = mem_hash
|
|
if "created_at" not in mem_metadata:
|
|
mem_metadata["created_at"] = datetime.now(timezone.utc).isoformat()
|
|
mem_metadata["updated_at"] = mem_metadata["created_at"]
|
|
if mem.get("attributed_to"):
|
|
mem_metadata["attributed_to"] = mem["attributed_to"]
|
|
|
|
records.append((memory_id, text, embed_map[text], mem_metadata))
|
|
|
|
if not records:
|
|
await asyncio.to_thread(self.db.save_messages, messages, session_scope)
|
|
return []
|
|
|
|
# Phase 6: Batch persist
|
|
all_vectors = [r[2] for r in records]
|
|
all_ids = [r[0] for r in records]
|
|
all_payloads = [r[3] for r in records]
|
|
|
|
try:
|
|
await asyncio.to_thread(
|
|
self.vector_store.insert,
|
|
vectors=all_vectors,
|
|
ids=all_ids,
|
|
payloads=all_payloads,
|
|
)
|
|
except Exception:
|
|
for mid, vec, pay in zip(all_ids, all_vectors, all_payloads):
|
|
try:
|
|
await asyncio.to_thread(self.vector_store.insert, vectors=[vec], ids=[mid], payloads=[pay])
|
|
except Exception as e:
|
|
logger.error(f"Failed to insert memory {mid} (async): {e}")
|
|
|
|
# Batch history
|
|
history_records = [
|
|
{
|
|
"memory_id": r[0],
|
|
"old_memory": None,
|
|
"new_memory": r[1],
|
|
"event": "ADD",
|
|
"created_at": r[3].get("created_at"),
|
|
"is_deleted": 0,
|
|
}
|
|
for r in records
|
|
]
|
|
try:
|
|
await asyncio.to_thread(self.db.batch_add_history, history_records)
|
|
except Exception:
|
|
for hr in history_records:
|
|
try:
|
|
await asyncio.to_thread(
|
|
self.db.add_history, hr["memory_id"], None, hr["new_memory"], "ADD",
|
|
created_at=hr.get("created_at")
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to add history for {hr['memory_id']} (async): {e}")
|
|
|
|
# Phase 7: Batch entity linking
|
|
try:
|
|
all_texts = [r[1] for r in records]
|
|
all_entities = await asyncio.to_thread(extract_entities_batch, all_texts)
|
|
|
|
# 7a: Global dedup
|
|
global_entities = {}
|
|
for idx, (memory_id, text, embedding, payload) in enumerate(records):
|
|
entities = all_entities[idx] if idx < len(all_entities) else []
|
|
for entity_type, entity_text in entities:
|
|
key = self._normalize_entity_text(entity_text)
|
|
if key in global_entities:
|
|
global_entities[key][2].add(memory_id)
|
|
else:
|
|
global_entities[key] = [entity_type, entity_text, {memory_id}]
|
|
|
|
if global_entities:
|
|
ordered_keys = list(global_entities.keys())
|
|
entity_texts = [global_entities[k][1] for k in ordered_keys]
|
|
|
|
# 7b: Batch embed entities
|
|
try:
|
|
entity_embeddings = await asyncio.to_thread(self.embedding_model.embed_batch, entity_texts, "add")
|
|
except Exception:
|
|
entity_embeddings = []
|
|
for t in entity_texts:
|
|
try:
|
|
entity_embeddings.append(await asyncio.to_thread(self.embedding_model.embed, t, "add"))
|
|
except Exception:
|
|
entity_embeddings.append(None)
|
|
|
|
if len(entity_embeddings) != len(ordered_keys):
|
|
logger.warning(
|
|
"embed_batch returned %d vectors for %d entity texts — "
|
|
"padding/truncating to avoid dropping entity links",
|
|
len(entity_embeddings),
|
|
len(ordered_keys),
|
|
)
|
|
entity_embeddings = list(entity_embeddings[: len(ordered_keys)])
|
|
entity_embeddings += [None] * (len(ordered_keys) - len(entity_embeddings))
|
|
|
|
valid = [(i, k) for i, k in enumerate(ordered_keys) if entity_embeddings[i] is not None]
|
|
if valid:
|
|
valid_indices, valid_keys = zip(*valid)
|
|
valid_vectors = [entity_embeddings[i] for i in valid_indices]
|
|
exact_matches = await asyncio.to_thread(self._existing_entities_by_text, search_filters)
|
|
|
|
# 7c: Batch search for existing entities
|
|
valid_texts = [global_entities[k][1] for k in valid_keys]
|
|
existing_matches = await asyncio.to_thread(
|
|
self.entity_store.search_batch,
|
|
queries=valid_texts,
|
|
vectors_list=valid_vectors,
|
|
top_k=1,
|
|
filters=search_filters,
|
|
)
|
|
|
|
# 7d: Separate into inserts vs updates
|
|
to_insert_vectors, to_insert_ids, to_insert_payloads = [], [], []
|
|
for j, key in enumerate(valid_keys):
|
|
entity_type, entity_text, memory_ids = global_entities[key]
|
|
matches = existing_matches[j] if j < len(existing_matches) else []
|
|
exact_match = exact_matches.get(key)
|
|
|
|
semantic_match = matches[0] if matches and matches[0].score >= 0.95 else None
|
|
match = exact_match or semantic_match
|
|
if match:
|
|
payload = match.payload or {}
|
|
linked = set(payload.get("linked_memory_ids", []))
|
|
linked |= memory_ids
|
|
payload["linked_memory_ids"] = sorted(linked)
|
|
try:
|
|
await asyncio.to_thread(
|
|
self.entity_store.update,
|
|
vector_id=match.id,
|
|
vector=None,
|
|
payload=payload,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Entity update failed for '{entity_text}' (async): {e}")
|
|
else:
|
|
to_insert_vectors.append(valid_vectors[j])
|
|
to_insert_ids.append(str(uuid.uuid4()))
|
|
to_insert_payloads.append({
|
|
"data": entity_text,
|
|
"entity_type": entity_type,
|
|
"linked_memory_ids": sorted(memory_ids),
|
|
**search_filters,
|
|
})
|
|
|
|
# 7e: Batch insert new entities
|
|
if to_insert_vectors:
|
|
try:
|
|
await asyncio.to_thread(
|
|
self.entity_store.insert,
|
|
vectors=to_insert_vectors,
|
|
ids=to_insert_ids,
|
|
payloads=to_insert_payloads,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Batch entity insert failed (async): {e}")
|
|
except Exception as e:
|
|
logger.warning(f"Batch entity linking failed (async): {e}")
|
|
|
|
# Phase 8: Save messages + return
|
|
await asyncio.to_thread(self.db.save_messages, messages, session_scope)
|
|
|
|
returned_memories = [
|
|
{"id": r[0], "memory": r[1], "event": "ADD"}
|
|
for r in records
|
|
]
|
|
|
|
keys, encoded_ids = process_telemetry_filters(effective_filters)
|
|
capture_event(
|
|
"mem0.add",
|
|
self,
|
|
{"version": self.api_version, "keys": keys, "encoded_ids": encoded_ids, "sync_type": "async"},
|
|
)
|
|
return returned_memories
|
|
|
|
async def get(self, memory_id):
|
|
"""
|
|
Retrieve a memory by ID asynchronously.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to retrieve.
|
|
|
|
Returns:
|
|
dict: Retrieved memory.
|
|
"""
|
|
capture_event("mem0.get", self, {"memory_id": memory_id, "sync_type": "async"})
|
|
memory = await asyncio.to_thread(self.vector_store.get, vector_id=memory_id)
|
|
if not memory:
|
|
await display_first_run_notice_async(self, "async", "get")
|
|
return None
|
|
|
|
promoted_payload_keys = [
|
|
"user_id",
|
|
"agent_id",
|
|
"run_id",
|
|
"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}
|
|
|
|
result_item = MemoryItem(
|
|
id=memory.id,
|
|
memory=memory.payload.get("data", ""),
|
|
hash=memory.payload.get("hash"),
|
|
created_at=memory.payload.get("created_at"),
|
|
updated_at=memory.payload.get("updated_at"),
|
|
).model_dump()
|
|
|
|
for key in promoted_payload_keys:
|
|
if key in memory.payload:
|
|
result_item[key] = memory.payload[key]
|
|
|
|
additional_metadata = {k: v for k, v in memory.payload.items() if k not in core_and_promoted_keys}
|
|
if additional_metadata:
|
|
result_item["metadata"] = additional_metadata
|
|
|
|
await display_first_run_notice_async(self, "async", "get")
|
|
return result_item
|
|
|
|
async def get_all(
|
|
self,
|
|
*,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
top_k: int = 20,
|
|
show_expired: bool = False,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
List all memories.
|
|
|
|
Args:
|
|
filters (dict): Filter dict containing entity IDs and optional metadata filters.
|
|
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.
|
|
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", ...}]}`
|
|
|
|
Raises:
|
|
ValueError: If filters doesn't contain at least one of user_id, agent_id, run_id,
|
|
or if top_k is invalid.
|
|
"""
|
|
# Reject top-level entity params - must use filters instead
|
|
_reject_top_level_entity_params(kwargs, "get_all")
|
|
|
|
# Validate top_k
|
|
_validate_search_params(top_k=top_k)
|
|
|
|
# Validate and trim entity IDs in filters
|
|
effective_filters = dict(filters) if filters else {}
|
|
if "user_id" in effective_filters:
|
|
effective_filters["user_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["user_id"], "user_id"
|
|
)
|
|
if "agent_id" in effective_filters:
|
|
effective_filters["agent_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["agent_id"], "agent_id"
|
|
)
|
|
if "run_id" in effective_filters:
|
|
effective_filters["run_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["run_id"], "run_id"
|
|
)
|
|
|
|
# Validate filters contains at least one entity ID
|
|
if not any(key in effective_filters for key in ("user_id", "agent_id", "run_id")):
|
|
raise ValueError(
|
|
"filters must contain at least one of: user_id, agent_id, run_id. "
|
|
"Example: filters={'user_id': 'u1'}"
|
|
)
|
|
|
|
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)
|
|
capture_event(
|
|
"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, fetch_limit, show_expired, limit)
|
|
|
|
if scale_threshold_notice:
|
|
await display_scale_threshold_notice_async(self, "async", "get_all", *scale_threshold_notice)
|
|
else:
|
|
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, 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
|
|
if isinstance(memories_result, (tuple, list)) and len(memories_result) > 0:
|
|
first_element = memories_result[0]
|
|
|
|
# If first element is a container, unwrap one level
|
|
if isinstance(first_element, (list, tuple)):
|
|
actual_memories = first_element
|
|
else:
|
|
# First element is a memory object, structure is already flat
|
|
actual_memories = memories_result
|
|
else:
|
|
actual_memories = memories_result
|
|
|
|
promoted_payload_keys = [
|
|
"user_id",
|
|
"agent_id",
|
|
"run_id",
|
|
"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", ""),
|
|
hash=mem.payload.get("hash"),
|
|
created_at=mem.payload.get("created_at"),
|
|
updated_at=mem.payload.get("updated_at"),
|
|
).model_dump(exclude={"score"})
|
|
|
|
for key in promoted_payload_keys:
|
|
if key in mem.payload:
|
|
memory_item_dict[key] = mem.payload[key]
|
|
|
|
additional_metadata = {k: v for k, v in mem.payload.items() if k not in core_and_promoted_keys}
|
|
if additional_metadata:
|
|
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
|
|
|
|
async def search(
|
|
self,
|
|
query: str,
|
|
*,
|
|
top_k: int = 20,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
threshold: float = 0.1,
|
|
rerank: bool = False,
|
|
explain: bool = False,
|
|
reference_date: Optional[Any] = None,
|
|
show_expired: bool = False,
|
|
**kwargs,
|
|
):
|
|
"""
|
|
Searches for memories based on a query.
|
|
|
|
Args:
|
|
query (str): Query to search for.
|
|
top_k (int, optional): Maximum number of results to return. Defaults to 20.
|
|
filters (dict): Filter dict containing entity IDs and optional metadata filters.
|
|
Must contain at least one of: user_id, agent_id, run_id.
|
|
Example: filters={"user_id": "u1", "agent_id": "a1"}
|
|
|
|
Enhanced metadata filtering with operators:
|
|
- {"key": "value"} - exact match
|
|
- {"key": {"eq": "value"}} - equals
|
|
- {"key": {"ne": "value"}} - not equals
|
|
- {"key": {"in": ["val1", "val2"]}} - in list
|
|
- {"key": {"nin": ["val1", "val2"]}} - not in list
|
|
- {"key": {"gt": 10}} - greater than
|
|
- {"key": {"gte": 10}} - greater than or equal
|
|
- {"key": {"lt": 10}} - less than
|
|
- {"key": {"lte": 10}} - less than or equal
|
|
- {"key": {"contains": "text"}} - contains text
|
|
- {"key": {"icontains": "text"}} - case-insensitive contains
|
|
- {"key": "*"} - wildcard match (any value)
|
|
- {"AND": [filter1, filter2]} - logical AND
|
|
- {"OR": [filter1, filter2]} - logical OR
|
|
- {"NOT": [filter1]} - logical NOT
|
|
threshold (float, optional): Minimum score for a memory to be included. Defaults to 0.1.
|
|
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.
|
|
Example for v1.1+: `{"results": [{"id": "...", "memory": "...", "score": 0.8, ...}]}`
|
|
|
|
Raises:
|
|
ValueError: If filters doesn't contain at least one of user_id, agent_id, run_id,
|
|
or if threshold/top_k values are invalid.
|
|
"""
|
|
if reference_date is not None:
|
|
raise ValueError(
|
|
await get_temporal_feature_error_message_async("async", "search", "reference_date")
|
|
)
|
|
|
|
# Reject top-level entity params - must use filters instead
|
|
_reject_top_level_entity_params(kwargs, "search")
|
|
|
|
# Validate search parameters (before applying defaults)
|
|
_validate_search_params(threshold=threshold, top_k=top_k)
|
|
query = _validate_and_trim_search_query(query)
|
|
temporal_usage_notice = detect_temporal_usage_from_search(query, filters)
|
|
|
|
# Validate and trim entity IDs in filters
|
|
effective_filters = filters.copy() if filters else {}
|
|
if "user_id" in effective_filters:
|
|
effective_filters["user_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["user_id"], "user_id"
|
|
)
|
|
if "agent_id" in effective_filters:
|
|
effective_filters["agent_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["agent_id"], "agent_id"
|
|
)
|
|
if "run_id" in effective_filters:
|
|
effective_filters["run_id"] = _validate_and_trim_entity_id(
|
|
effective_filters["run_id"], "run_id"
|
|
)
|
|
|
|
# Validate filters contains at least one entity ID
|
|
if not any(key in effective_filters for key in ("user_id", "agent_id", "run_id")):
|
|
raise ValueError(
|
|
"filters must contain at least one of: user_id, agent_id, run_id. "
|
|
"Example: filters={'user_id': 'u1'}"
|
|
)
|
|
|
|
limit = top_k
|
|
scale_threshold_notice = detect_scale_threshold_from_top_k(top_k)
|
|
|
|
# Apply enhanced metadata filtering if advanced operators are detected
|
|
if self._has_advanced_operators(effective_filters):
|
|
processed_filters = self._process_metadata_filters(effective_filters)
|
|
# Remove logical/operator keys that have been reprocessed
|
|
for logical_key in ("AND", "OR", "NOT"):
|
|
effective_filters.pop(logical_key, None)
|
|
for fk in list(effective_filters.keys()):
|
|
if fk not in ("AND", "OR", "NOT", "user_id", "agent_id", "run_id") and isinstance(effective_filters.get(fk), dict):
|
|
effective_filters.pop(fk, None)
|
|
effective_filters.update(processed_filters)
|
|
|
|
keys, encoded_ids = process_telemetry_filters(effective_filters)
|
|
capture_event(
|
|
"mem0.search",
|
|
self,
|
|
{
|
|
"limit": limit,
|
|
"version": self.api_version,
|
|
"keys": keys,
|
|
"encoded_ids": encoded_ids,
|
|
"sync_type": "async",
|
|
"threshold": threshold,
|
|
"explain": explain,
|
|
"advanced_filters": bool(filters and self._has_advanced_operators(filters)),
|
|
},
|
|
)
|
|
|
|
search_start = time.perf_counter()
|
|
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
|
|
if rerank and self.reranker and original_memories:
|
|
try:
|
|
# Run reranking in thread pool to avoid blocking async loop
|
|
reranked_memories = await asyncio.to_thread(
|
|
self.reranker.rerank, query, original_memories, limit
|
|
)
|
|
original_memories = reranked_memories
|
|
except Exception as e:
|
|
logger.warning(f"Reranking failed, using original results: {e}")
|
|
|
|
if temporal_usage_notice:
|
|
await display_temporal_usage_notice_async(self, "async", "search", *temporal_usage_notice)
|
|
elif scale_threshold_notice:
|
|
await display_scale_threshold_notice_async(self, "async", "search", *scale_threshold_notice)
|
|
elif search_elapsed_seconds > PERFORMANCE_SLOW_QUERY_THRESHOLD_SECONDS:
|
|
await display_performance_slow_query_notice_async(
|
|
self,
|
|
"async",
|
|
"search",
|
|
search_elapsed_seconds,
|
|
top_k,
|
|
len(original_memories),
|
|
)
|
|
else:
|
|
await display_first_run_notice_async(self, "async", "search")
|
|
return {"results": original_memories}
|
|
|
|
def _process_metadata_filters(self, metadata_filters: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Process enhanced metadata filters and convert them to vector store compatible format.
|
|
|
|
Args:
|
|
metadata_filters: Enhanced metadata filters with operators
|
|
|
|
Returns:
|
|
Dict of processed filters compatible with vector store
|
|
"""
|
|
processed_filters = {}
|
|
|
|
def process_condition(key: str, condition: Any) -> Dict[str, Any]:
|
|
if not isinstance(condition, dict):
|
|
# Simple equality: {"key": "value"}
|
|
if condition == "*":
|
|
# Wildcard: match everything for this field (implementation depends on vector store)
|
|
return {key: "*"}
|
|
return {key: condition}
|
|
|
|
result = {}
|
|
for operator, value in condition.items():
|
|
# Map platform operators to universal format that can be translated by each vector store
|
|
operator_map = {
|
|
"eq": "eq", "ne": "ne", "gt": "gt", "gte": "gte",
|
|
"lt": "lt", "lte": "lte", "in": "in", "nin": "nin",
|
|
"contains": "contains", "icontains": "icontains"
|
|
}
|
|
|
|
if operator in operator_map:
|
|
result.setdefault(key, {})[operator_map[operator]] = value
|
|
else:
|
|
raise ValueError(f"Unsupported metadata filter operator: {operator}")
|
|
return result
|
|
|
|
def merge_filters(target: Dict[str, Any], source: Dict[str, Any]) -> None:
|
|
"""Merge source into target, deep-merging nested operator dicts for the same key."""
|
|
for key, value in source.items():
|
|
if key in target and isinstance(target[key], dict) and isinstance(value, dict):
|
|
target[key].update(value)
|
|
else:
|
|
target[key] = value
|
|
|
|
for key, value in metadata_filters.items():
|
|
if key == "AND":
|
|
# Logical AND: combine multiple conditions
|
|
if not isinstance(value, list):
|
|
raise ValueError("AND operator requires a list of conditions")
|
|
for condition in value:
|
|
for sub_key, sub_value in condition.items():
|
|
merge_filters(processed_filters, process_condition(sub_key, sub_value))
|
|
elif key == "OR":
|
|
# Logical OR: Pass through to vector store for implementation-specific handling
|
|
if not isinstance(value, list) or not value:
|
|
raise ValueError("OR operator requires a non-empty list of conditions")
|
|
# Store OR conditions in a way that vector stores can interpret
|
|
processed_filters["$or"] = []
|
|
for condition in value:
|
|
or_condition = {}
|
|
for sub_key, sub_value in condition.items():
|
|
merge_filters(or_condition, process_condition(sub_key, sub_value))
|
|
processed_filters["$or"].append(or_condition)
|
|
elif key == "NOT":
|
|
# Logical NOT: Pass through to vector store for implementation-specific handling
|
|
if not isinstance(value, list) or not value:
|
|
raise ValueError("NOT operator requires a non-empty list of conditions")
|
|
processed_filters["$not"] = []
|
|
for condition in value:
|
|
not_condition = {}
|
|
for sub_key, sub_value in condition.items():
|
|
merge_filters(not_condition, process_condition(sub_key, sub_value))
|
|
processed_filters["$not"].append(not_condition)
|
|
else:
|
|
merge_filters(processed_filters, process_condition(key, value))
|
|
|
|
return processed_filters
|
|
|
|
def _has_advanced_operators(self, filters: Dict[str, Any]) -> bool:
|
|
"""
|
|
Check if filters contain advanced operators that need special processing.
|
|
|
|
Args:
|
|
filters: Dictionary of filters to check
|
|
|
|
Returns:
|
|
bool: True if advanced operators are detected
|
|
"""
|
|
if not isinstance(filters, dict):
|
|
return False
|
|
|
|
for key, value in filters.items():
|
|
# Check for platform-style logical operators
|
|
if key in ["AND", "OR", "NOT"]:
|
|
return True
|
|
# Check for comparison operators (without $ prefix for universal compatibility)
|
|
if isinstance(value, dict):
|
|
for op in value.keys():
|
|
if op in ["eq", "ne", "gt", "gte", "lt", "lte", "in", "nin", "contains", "icontains"]:
|
|
return True
|
|
# Check for wildcard values
|
|
if value == "*":
|
|
return True
|
|
return 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
|
|
|
|
# Step 1: Preprocess query (CPU-bound)
|
|
query_lemmatized = await asyncio.to_thread(lemmatize_for_bm25, query)
|
|
query_entities = await asyncio.to_thread(extract_entities, query)
|
|
|
|
# Step 2: Embed query
|
|
embeddings = await asyncio.to_thread(self.embedding_model.embed, query, "search")
|
|
|
|
# Step 3: Semantic search (over-fetch)
|
|
internal_limit = max(limit * 4, 60)
|
|
semantic_results = await asyncio.to_thread(
|
|
self.vector_store.search, query=query, vectors=embeddings, top_k=internal_limit, filters=filters
|
|
)
|
|
|
|
# Step 4: Keyword search (if store supports it)
|
|
keyword_results = await asyncio.to_thread(
|
|
self.vector_store.keyword_search, query=query_lemmatized, top_k=internal_limit, filters=filters
|
|
)
|
|
|
|
# Step 5: Compute BM25 scores
|
|
bm25_scores = {}
|
|
if keyword_results is not None:
|
|
midpoint, steepness = get_bm25_params(query, lemmatized=query_lemmatized)
|
|
for mem in keyword_results:
|
|
mem_id = str(mem.id) if hasattr(mem, 'id') else str(mem.get('id', ''))
|
|
raw_score = mem.score if hasattr(mem, 'score') else mem.get('score', 0)
|
|
if raw_score and raw_score > 0:
|
|
bm25_scores[mem_id] = normalize_bm25(raw_score, midpoint, steepness)
|
|
|
|
# Step 6: Compute entity boosts
|
|
entity_boosts = {}
|
|
if query_entities:
|
|
entity_boosts = await self._compute_entity_boosts_async(query_entities, filters)
|
|
|
|
# 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": payload,
|
|
})
|
|
|
|
# Step 8: Score and rank
|
|
scored_results = score_and_rank(
|
|
semantic_results=candidates,
|
|
bm25_scores=bm25_scores,
|
|
entity_boosts=entity_boosts,
|
|
threshold=threshold,
|
|
top_k=limit,
|
|
explain=explain,
|
|
)
|
|
|
|
# Step 9: Format results
|
|
promoted_payload_keys = [
|
|
"user_id",
|
|
"agent_id",
|
|
"run_id",
|
|
"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}
|
|
|
|
original_memories = []
|
|
for scored in scored_results:
|
|
payload = scored.get("payload") or {}
|
|
if not payload.get("data"):
|
|
continue
|
|
|
|
memory_item_dict = MemoryItem(
|
|
id=scored["id"],
|
|
memory=payload.get("data", ""),
|
|
hash=payload.get("hash"),
|
|
created_at=payload.get("created_at"),
|
|
updated_at=payload.get("updated_at"),
|
|
score=scored["score"],
|
|
).model_dump()
|
|
|
|
for key in promoted_payload_keys:
|
|
if key in payload:
|
|
memory_item_dict[key] = payload[key]
|
|
|
|
additional_metadata = {k: v for k, v in payload.items() if k not in core_and_promoted_keys}
|
|
if additional_metadata:
|
|
if not memory_item_dict.get("metadata"):
|
|
memory_item_dict["metadata"] = {}
|
|
memory_item_dict["metadata"].update(additional_metadata)
|
|
if explain and "score_details" in scored:
|
|
memory_item_dict["score_details"] = scored["score_details"]
|
|
|
|
original_memories.append(memory_item_dict)
|
|
|
|
return original_memories
|
|
|
|
async def _compute_entity_boosts_async(self, query_entities, filters):
|
|
"""Async version of entity boost computation."""
|
|
seen = set()
|
|
deduped = []
|
|
for entity_type, entity_text in query_entities[:8]:
|
|
key = self._normalize_entity_text(entity_text)
|
|
if key and key not in seen:
|
|
seen.add(key)
|
|
deduped.append((entity_type, entity_text))
|
|
|
|
if not deduped:
|
|
return {}
|
|
|
|
search_filters = {k: v for k, v in filters.items() if k in ("user_id", "agent_id", "run_id") and v}
|
|
memory_boosts = {}
|
|
|
|
try:
|
|
entity_texts = [text for _, text in deduped]
|
|
embeddings = await asyncio.to_thread(self.embedding_model.embed_batch, entity_texts, "search")
|
|
|
|
if len(embeddings) != len(entity_texts):
|
|
logger.warning(
|
|
"embed_batch returned %d vectors for %d texts — skipping entity boost",
|
|
len(embeddings),
|
|
len(entity_texts),
|
|
)
|
|
return memory_boosts
|
|
|
|
sem = asyncio.Semaphore(4)
|
|
|
|
async def _search_entity(entity_text, embedding):
|
|
async with sem:
|
|
return await asyncio.to_thread(
|
|
self.entity_store.search,
|
|
query=entity_text,
|
|
vectors=embedding,
|
|
top_k=500,
|
|
filters=search_filters,
|
|
)
|
|
|
|
results = await asyncio.gather(
|
|
*(_search_entity(text, emb) for text, emb in zip(entity_texts, embeddings)),
|
|
return_exceptions=True,
|
|
)
|
|
|
|
for matches in results:
|
|
if isinstance(matches, BaseException):
|
|
logger.warning("Entity boost search failed for one entity: %s", matches)
|
|
continue
|
|
|
|
for match in matches:
|
|
similarity = match.score if hasattr(match, 'score') else 0.0
|
|
if similarity < 0.5:
|
|
continue
|
|
|
|
payload = match.payload if hasattr(match, 'payload') else {}
|
|
linked_memory_ids = payload.get("linked_memory_ids", [])
|
|
if not isinstance(linked_memory_ids, list):
|
|
continue
|
|
|
|
num_linked = max(len(linked_memory_ids), 1)
|
|
memory_count_weight = 1.0 / (1.0 + 0.001 * ((num_linked - 1) ** 2))
|
|
boost = similarity * ENTITY_BOOST_WEIGHT * memory_count_weight
|
|
|
|
for memory_id in linked_memory_ids:
|
|
if memory_id:
|
|
memory_key = str(memory_id)
|
|
memory_boosts[memory_key] = max(memory_boosts.get(memory_key, 0.0), boost)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Entity boost computation failed: {e}")
|
|
|
|
return memory_boosts
|
|
|
|
async def update(
|
|
self,
|
|
memory_id,
|
|
text: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
expiration_date: Any = _UNSET,
|
|
data: Optional[str] = None,
|
|
):
|
|
"""
|
|
Update a memory by ID asynchronously.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to update.
|
|
text (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.
|
|
data (str, optional): Deprecated alias for ``text``. Will be removed in the next
|
|
major release; use ``text`` instead.
|
|
|
|
Returns:
|
|
dict: Success message indicating the memory was updated.
|
|
|
|
Example:
|
|
>>> await m.update(memory_id="mem_123", text="Likes to play tennis on weekends")
|
|
{'message': 'Memory updated successfully!'}
|
|
"""
|
|
capture_event("mem0.update", self, {"memory_id": memory_id, "sync_type": "async"})
|
|
|
|
if data is not None:
|
|
logger.warning(
|
|
"The `data` argument to update() is deprecated and will be removed in the "
|
|
"next major release. Use `text` instead."
|
|
)
|
|
if text is None:
|
|
text = data
|
|
|
|
if text is None and metadata is None and expiration_date is _UNSET:
|
|
raise ValueError("At least one of text, metadata, or expiration_date must be provided.")
|
|
|
|
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 text is not None:
|
|
embeddings = await asyncio.to_thread(self.embedding_model.embed, text, "update")
|
|
existing_embeddings[text] = embeddings
|
|
|
|
await self._update_memory(memory_id, text, existing_embeddings, update_metadata)
|
|
await display_first_run_notice_async(self, "async", "update")
|
|
return {"message": "Memory updated successfully!"}
|
|
|
|
async def delete(self, memory_id):
|
|
"""
|
|
Delete a memory by ID asynchronously.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to delete.
|
|
"""
|
|
capture_event("mem0.delete", self, {"memory_id": memory_id, "sync_type": "async"})
|
|
|
|
existing_memory = await asyncio.to_thread(self.vector_store.get, vector_id=memory_id)
|
|
if existing_memory is None:
|
|
raise ValueError(f"Memory with id {memory_id} not found")
|
|
|
|
await self._delete_memory(memory_id, existing_memory)
|
|
decay_usage_notice = detect_decay_usage_from_delete()
|
|
if decay_usage_notice:
|
|
await display_decay_usage_notice_async(self, "async", "delete", *decay_usage_notice)
|
|
else:
|
|
await display_first_run_notice_async(self, "async", "delete")
|
|
return {"message": "Memory deleted successfully!"}
|
|
|
|
async def delete_all(self, user_id=None, agent_id=None, run_id=None):
|
|
"""
|
|
Delete all memories asynchronously.
|
|
|
|
Args:
|
|
user_id (str, optional): ID of the user to delete memories for. Defaults to None.
|
|
agent_id (str, optional): ID of the agent to delete memories for. Defaults to None.
|
|
run_id (str, optional): ID of the run to delete memories for. Defaults to None.
|
|
"""
|
|
user_id = _validate_and_trim_entity_id(user_id, "user_id")
|
|
agent_id = _validate_and_trim_entity_id(agent_id, "agent_id")
|
|
run_id = _validate_and_trim_entity_id(run_id, "run_id")
|
|
|
|
filters = {}
|
|
if user_id:
|
|
filters["user_id"] = user_id
|
|
if agent_id:
|
|
filters["agent_id"] = agent_id
|
|
if run_id:
|
|
filters["run_id"] = run_id
|
|
|
|
if not filters:
|
|
raise ValueError(
|
|
"At least one filter is required to delete all memories. If you want to delete all memories, use the `reset()` method."
|
|
)
|
|
|
|
keys, encoded_ids = process_telemetry_filters(filters)
|
|
capture_event("mem0.delete_all", self, {"keys": keys, "encoded_ids": encoded_ids, "sync_type": "async"})
|
|
memories = await asyncio.to_thread(self.vector_store.list, filters=filters)
|
|
|
|
delete_tasks = []
|
|
for memory in memories[0]:
|
|
delete_tasks.append(self._delete_memory(memory.id, skip_entity_cleanup=True))
|
|
|
|
results = await asyncio.gather(*delete_tasks, return_exceptions=True)
|
|
|
|
if self._entity_store is not None:
|
|
await self._bulk_clear_entity_store(filters)
|
|
|
|
errors = [r for r in results if isinstance(r, BaseException)]
|
|
if errors:
|
|
logger.warning("Failed to delete %d out of %d memories", len(errors), len(results))
|
|
for err in errors:
|
|
logger.warning("Delete error: %s", err)
|
|
|
|
logger.info(f"Deleted {len(results) - len(errors)} memories")
|
|
|
|
decay_usage_notice = detect_decay_usage_from_delete_all(len(memories[0]))
|
|
if decay_usage_notice:
|
|
await display_decay_usage_notice_async(self, "async", "delete_all", *decay_usage_notice)
|
|
else:
|
|
await display_first_run_notice_async(self, "async", "delete_all")
|
|
return {"message": "Memories deleted successfully!"}
|
|
|
|
async def history(self, memory_id):
|
|
"""
|
|
Get the history of changes for a memory by ID asynchronously.
|
|
|
|
Args:
|
|
memory_id (str): ID of the memory to get history for.
|
|
|
|
Returns:
|
|
list: List of changes for the memory.
|
|
"""
|
|
capture_event("mem0.history", self, {"memory_id": memory_id, "sync_type": "async"})
|
|
history = await asyncio.to_thread(self.db.get_history, memory_id)
|
|
await display_first_run_notice_async(self, "async", "history")
|
|
return history
|
|
|
|
async def _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 = await asyncio.to_thread(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)
|
|
|
|
await asyncio.to_thread(
|
|
self.vector_store.insert,
|
|
vectors=[embeddings],
|
|
ids=[memory_id],
|
|
payloads=[new_metadata],
|
|
)
|
|
|
|
await asyncio.to_thread(
|
|
self.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
|
|
|
|
async def _create_procedural_memory(self, messages, metadata=None, llm=None, prompt=None):
|
|
"""
|
|
Create a procedural memory asynchronously
|
|
|
|
Args:
|
|
messages (list): List of messages to create a procedural memory from.
|
|
metadata (dict): Metadata to create a procedural memory from.
|
|
llm (llm, optional): LLM to use for the procedural memory creation. Defaults to None.
|
|
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:
|
|
if llm is not None:
|
|
# langchain-core is only needed to adapt messages for a custom
|
|
# LangChain LLM. The default path uses self.llm and must not
|
|
# require the optional dependency, mirroring the sync version.
|
|
try:
|
|
from langchain_core.messages.utils import (
|
|
convert_to_messages, # type: ignore
|
|
)
|
|
except ImportError as e:
|
|
raise ImportError(
|
|
"langchain-core is required to pass a custom LLM to procedural memory. "
|
|
"Install it with 'pip install langchain-core'."
|
|
) from e
|
|
|
|
parsed_messages = convert_to_messages(parsed_messages)
|
|
response = await asyncio.to_thread(llm.invoke, input=parsed_messages)
|
|
procedural_memory = remove_code_blocks(response.content)
|
|
else:
|
|
procedural_memory = await asyncio.to_thread(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 metadata is None:
|
|
raise ValueError("Metadata cannot be done for procedural memory.")
|
|
|
|
metadata = {**metadata, "memory_type": MemoryType.PROCEDURAL.value}
|
|
embeddings = await asyncio.to_thread(self.embedding_model.embed, procedural_memory, memory_action="add")
|
|
memory_id = await self._create_memory(procedural_memory, {procedural_memory: embeddings}, metadata=metadata)
|
|
capture_event("mem0._create_procedural_memory", self, {"memory_id": memory_id, "sync_type": "async"})
|
|
|
|
result = {"results": [{"id": memory_id, "memory": procedural_memory, "event": "ADD"}]}
|
|
|
|
return result
|
|
|
|
async def _update_memory(self, memory_id, data, existing_embeddings, metadata=None):
|
|
logger.info(f"Updating memory with {data=}")
|
|
|
|
try:
|
|
existing_memory = await asyncio.to_thread(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(metadata)
|
|
|
|
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()
|
|
|
|
# actor_id is immutable after creation (issue #4490)
|
|
if "actor_id" in existing_memory.payload:
|
|
new_metadata["actor_id"] = existing_memory.payload["actor_id"]
|
|
|
|
if data in existing_embeddings:
|
|
embeddings = existing_embeddings[data]
|
|
else:
|
|
embeddings = await asyncio.to_thread(self.embedding_model.embed, data, "update")
|
|
|
|
await asyncio.to_thread(
|
|
self.vector_store.update,
|
|
vector_id=memory_id,
|
|
vector=embeddings,
|
|
payload=new_metadata,
|
|
)
|
|
logger.info(f"Updating memory with ID {memory_id=} with {data=}")
|
|
|
|
await asyncio.to_thread(
|
|
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:
|
|
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
|
|
|
|
async def _delete_memory(self, memory_id, existing_memory=None, skip_entity_cleanup=False):
|
|
logger.info(f"Deleting memory with {memory_id=}")
|
|
if existing_memory is None:
|
|
existing_memory = await asyncio.to_thread(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)}
|
|
|
|
await asyncio.to_thread(self.vector_store.delete, vector_id=memory_id)
|
|
await asyncio.to_thread(
|
|
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,
|
|
)
|
|
|
|
if not skip_entity_cleanup:
|
|
await self._remove_memory_from_entity_store(memory_id, session_filters)
|
|
|
|
return memory_id
|
|
|
|
async def reset(self):
|
|
"""
|
|
Reset the memory store asynchronously by:
|
|
Deletes the vector store collection
|
|
Resets the database
|
|
Recreates the vector store with a new client
|
|
"""
|
|
logger.warning("Resetting all memories")
|
|
await asyncio.to_thread(self.vector_store.delete_col)
|
|
|
|
gc.collect()
|
|
|
|
if hasattr(self.vector_store, "client") and hasattr(self.vector_store.client, "close"):
|
|
await asyncio.to_thread(self.vector_store.client.close)
|
|
|
|
await asyncio.to_thread(self.db.reset)
|
|
await asyncio.to_thread(self.db.close)
|
|
self.db = SQLiteManager(self.config.history_db_path)
|
|
|
|
self.vector_store = VectorStoreFactory.create(
|
|
self.config.vector_store.provider, self.config.vector_store.config
|
|
)
|
|
|
|
if self._entity_store is not None:
|
|
try:
|
|
await asyncio.to_thread(self._entity_store.reset)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to reset entity store: {e}")
|
|
self._entity_store = None
|
|
|
|
capture_event("mem0.reset", self, {"sync_type": "async"})
|
|
await display_first_run_notice_async(self, "async", "reset")
|
|
|
|
def close(self):
|
|
"""Release resources held by this AsyncMemory instance."""
|
|
if hasattr(self, "db") and self.db is not None:
|
|
self.db.close()
|
|
self.db = None
|
|
|
|
async def chat(self, query):
|
|
raise NotImplementedError("Chat function not implemented yet.")
|