mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-24 16:54:43 +00:00
617 lines
24 KiB
Python
617 lines
24 KiB
Python
"""Mem0 memory plugin — MemoryProvider interface.
|
|
|
|
Server-side LLM fact extraction, semantic search with reranking, and
|
|
automatic deduplication via the Mem0 Platform API.
|
|
|
|
Original PR #2933 by kartik-mem0, adapted to MemoryProvider ABC.
|
|
|
|
Config via environment variables:
|
|
MEM0_API_KEY — Mem0 Platform API key (required)
|
|
MEM0_USER_ID — User identifier (default: hermes-user)
|
|
MEM0_AGENT_ID — Agent identifier (default: hermes)
|
|
|
|
Or via $HERMES_HOME/mem0.json.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import threading
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from agent.memory_provider import MemoryProvider
|
|
from tools.registry import tool_error
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Circuit breaker: after this many consecutive failures, pause API calls
|
|
# for _BREAKER_COOLDOWN_SECS to avoid hammering a down server.
|
|
_BREAKER_THRESHOLD = 5
|
|
_BREAKER_COOLDOWN_SECS = 120
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _load_config() -> dict:
|
|
"""Load config from env vars, with $HERMES_HOME/mem0.json overrides.
|
|
|
|
Environment variables provide defaults; mem0.json (if present) overrides
|
|
individual keys. This avoids a silent failure when the JSON file exists
|
|
but is missing fields like ``api_key`` that the user set in ``.env``.
|
|
"""
|
|
from hermes_constants import get_hermes_home
|
|
|
|
config = {
|
|
"api_key": os.environ.get("MEM0_API_KEY", ""),
|
|
"user_id": os.environ.get("MEM0_USER_ID", "hermes-user"),
|
|
"agent_id": os.environ.get("MEM0_AGENT_ID", "hermes"),
|
|
"rerank": True,
|
|
"keyword_search": False,
|
|
}
|
|
|
|
config_path = get_hermes_home() / "mem0.json"
|
|
if config_path.exists():
|
|
try:
|
|
file_cfg = json.loads(config_path.read_text(encoding="utf-8"))
|
|
config.update({k: v for k, v in file_cfg.items()
|
|
if v is not None and v != ""})
|
|
except Exception:
|
|
pass
|
|
|
|
return config
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
PROFILE_SCHEMA = {
|
|
"name": "mem0_profile",
|
|
"description": (
|
|
"Retrieve all stored memories about the user — preferences, facts, "
|
|
"project context. Fast, no reranking. Use at conversation start."
|
|
),
|
|
"parameters": {"type": "object", "properties": {}, "required": []},
|
|
}
|
|
|
|
SEARCH_SCHEMA = {
|
|
"name": "mem0_search",
|
|
"description": (
|
|
"Search memories by meaning. Returns relevant facts ranked by similarity. "
|
|
"Set rerank=true for higher accuracy on important queries."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string", "description": "What to search for."},
|
|
"rerank": {"type": "boolean", "description": "Enable reranking for precision (default: false)."},
|
|
"top_k": {"type": "integer", "description": "Max results (default: 10, max: 50)."},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
}
|
|
|
|
CONCLUDE_SCHEMA = {
|
|
"name": "mem0_conclude",
|
|
"description": (
|
|
"Store a durable fact about the user. Stored verbatim (no LLM extraction). "
|
|
"Use for explicit preferences, corrections, decisions, or operational facts. "
|
|
"Optional metadata lets callers separate preferences, doctrine, facts, and history."
|
|
),
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"conclusion": {"type": "string", "description": "The fact to store."},
|
|
"memory_class": {
|
|
"type": "string",
|
|
"enum": ["preference", "doctrine", "factual", "temporal"],
|
|
"description": "Optional class override for the stored memory.",
|
|
},
|
|
"time_scope": {
|
|
"type": "string",
|
|
"enum": ["current", "past", "future", "timeless"],
|
|
"description": "Optional time-scope override for the stored memory.",
|
|
},
|
|
"metadata": {
|
|
"type": "object",
|
|
"description": "Optional extra metadata stored alongside the memory.",
|
|
},
|
|
},
|
|
"required": ["conclusion"],
|
|
},
|
|
}
|
|
|
|
_EXPLICIT_PREFIX_RE = re.compile(r"^\[(?:PREF|RULE|FACT|HIST)(?:/(?:CURRENT|PAST|FUTURE|TIMELESS))?\]\s*")
|
|
_MEMORY_CLASS_PREFIX = {
|
|
"preference": "PREF",
|
|
"doctrine": "RULE",
|
|
"factual": "FACT",
|
|
"temporal": "HIST",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MemoryProvider implementation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Mem0MemoryProvider(MemoryProvider):
|
|
"""Mem0 Platform memory with server-side extraction and semantic search."""
|
|
|
|
def __init__(self):
|
|
self._config = None
|
|
self._client = None
|
|
self._client_lock = threading.Lock()
|
|
self._api_key = ""
|
|
self._user_id = "hermes-user"
|
|
self._agent_id = "hermes"
|
|
self._rerank = True
|
|
self._sync_turn_mode = "full"
|
|
self._prefetch_result = ""
|
|
self._prefetch_lock = threading.Lock()
|
|
self._prefetch_thread = None
|
|
self._sync_thread = None
|
|
self._write_thread = None
|
|
# Circuit breaker state
|
|
self._consecutive_failures = 0
|
|
self._breaker_open_until = 0.0
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return "mem0"
|
|
|
|
def is_available(self) -> bool:
|
|
cfg = _load_config()
|
|
return bool(cfg.get("api_key"))
|
|
|
|
def save_config(self, values, hermes_home):
|
|
"""Write config to $HERMES_HOME/mem0.json."""
|
|
import json
|
|
from pathlib import Path
|
|
config_path = Path(hermes_home) / "mem0.json"
|
|
existing = {}
|
|
if config_path.exists():
|
|
try:
|
|
existing = json.loads(config_path.read_text())
|
|
except Exception:
|
|
pass
|
|
existing.update(values)
|
|
config_path.write_text(json.dumps(existing, indent=2))
|
|
|
|
def get_config_schema(self):
|
|
return [
|
|
{"key": "api_key", "description": "Mem0 Platform API key", "secret": True, "required": True, "env_var": "MEM0_API_KEY", "url": "https://app.mem0.ai"},
|
|
{"key": "user_id", "description": "User identifier", "default": "hermes-user"},
|
|
{"key": "agent_id", "description": "Agent identifier", "default": "hermes"},
|
|
{"key": "rerank", "description": "Enable reranking for recall", "default": "true", "choices": ["true", "false"]},
|
|
{"key": "sync_turn_mode", "description": "Automatic turn-to-memory extraction mode", "default": "full", "choices": ["full", "off"]},
|
|
]
|
|
|
|
def _get_client(self):
|
|
"""Thread-safe client accessor with lazy initialization."""
|
|
with self._client_lock:
|
|
if self._client is not None:
|
|
return self._client
|
|
try:
|
|
from mem0 import MemoryClient
|
|
self._client = MemoryClient(api_key=self._api_key)
|
|
return self._client
|
|
except ImportError:
|
|
raise RuntimeError("mem0 package not installed. Run: pip install mem0ai")
|
|
|
|
def _is_breaker_open(self) -> bool:
|
|
"""Return True if the circuit breaker is tripped (too many failures)."""
|
|
if self._consecutive_failures < _BREAKER_THRESHOLD:
|
|
return False
|
|
if time.monotonic() >= self._breaker_open_until:
|
|
# Cooldown expired — reset and allow a retry
|
|
self._consecutive_failures = 0
|
|
return False
|
|
return True
|
|
|
|
def _record_success(self):
|
|
self._consecutive_failures = 0
|
|
|
|
def _record_failure(self):
|
|
self._consecutive_failures += 1
|
|
if self._consecutive_failures >= _BREAKER_THRESHOLD:
|
|
self._breaker_open_until = time.monotonic() + _BREAKER_COOLDOWN_SECS
|
|
logger.warning(
|
|
"Mem0 circuit breaker tripped after %d consecutive failures. "
|
|
"Pausing API calls for %ds.",
|
|
self._consecutive_failures, _BREAKER_COOLDOWN_SECS,
|
|
)
|
|
|
|
def initialize(self, session_id: str, **kwargs) -> None:
|
|
self._config = _load_config()
|
|
self._api_key = self._config.get("api_key", "")
|
|
# Prefer gateway-provided user_id for per-user memory scoping;
|
|
# fall back to config/env default for CLI (single-user) sessions.
|
|
self._user_id = kwargs.get("user_id") or self._config.get("user_id", "hermes-user")
|
|
self._agent_id = self._config.get("agent_id", "hermes")
|
|
self._rerank = self._config.get("rerank", True)
|
|
if "sync_turn_mode" in self._config:
|
|
self._sync_turn_mode = str(self._config.get("sync_turn_mode", "full") or "full").strip().lower()
|
|
elif self._config.get("sync_turns") is False:
|
|
self._sync_turn_mode = "off"
|
|
else:
|
|
self._sync_turn_mode = "full"
|
|
if self._sync_turn_mode not in {"full", "off"}:
|
|
self._sync_turn_mode = "full"
|
|
|
|
def _read_filters(self) -> Dict[str, Any]:
|
|
"""Filters for search/get_all — scoped to user only for cross-session recall."""
|
|
return {"user_id": self._user_id}
|
|
|
|
def _scoped_read_filters(self) -> Dict[str, Any]:
|
|
"""Fallback filters for stores that only return agent-scoped memories."""
|
|
filters = self._read_filters().copy()
|
|
if self._agent_id:
|
|
filters["agent_id"] = self._agent_id
|
|
return filters
|
|
|
|
def _write_filters(self) -> Dict[str, Any]:
|
|
"""Filters for add — scoped to user + agent for attribution."""
|
|
return {"user_id": self._user_id, "agent_id": self._agent_id}
|
|
|
|
@staticmethod
|
|
def _unwrap_results(response: Any) -> list:
|
|
"""Normalize Mem0 API response — v2 wraps results in {"results": [...]}."""
|
|
if isinstance(response, dict):
|
|
return response.get("results", [])
|
|
if isinstance(response, list):
|
|
return response
|
|
return []
|
|
|
|
@staticmethod
|
|
def _strip_explicit_prefix(text: str) -> str:
|
|
return _EXPLICIT_PREFIX_RE.sub("", (text or "").strip())
|
|
|
|
@classmethod
|
|
def _normalize_memory_class(cls, value: Optional[str], *, target: str = "memory", content: str = "") -> str:
|
|
raw = str(value or "").strip().lower()
|
|
aliases = {
|
|
"pref": "preference",
|
|
"preference": "preference",
|
|
"user": "preference",
|
|
"rule": "doctrine",
|
|
"doctrine": "doctrine",
|
|
"policy": "doctrine",
|
|
"fact": "factual",
|
|
"factual": "factual",
|
|
"current_fact": "factual",
|
|
"history": "temporal",
|
|
"historical": "temporal",
|
|
"temporal": "temporal",
|
|
"past": "temporal",
|
|
}
|
|
if raw in aliases:
|
|
return aliases[raw]
|
|
if target == "user":
|
|
return "preference"
|
|
base = cls._strip_explicit_prefix(content).lower()
|
|
if re.search(r"\b(do not|don't|never|must|should|unless|only if|expected to|keep .* on)\b", base):
|
|
return "doctrine"
|
|
if re.search(r"\b(previously|earlier|before|used to|was changed|retired|on 20\d{2}-\d{2}-\d{2}|as of)\b", base):
|
|
return "temporal"
|
|
return "factual"
|
|
|
|
@staticmethod
|
|
def _normalize_time_scope(value: Optional[str], content: str = "") -> str:
|
|
raw = str(value or "").strip().lower()
|
|
aliases = {
|
|
"current": "current",
|
|
"now": "current",
|
|
"present": "current",
|
|
"past": "past",
|
|
"previous": "past",
|
|
"historical": "past",
|
|
"future": "future",
|
|
"planned": "future",
|
|
"timeless": "timeless",
|
|
"rule": "timeless",
|
|
}
|
|
if raw in aliases:
|
|
return aliases[raw]
|
|
base = Mem0MemoryProvider._strip_explicit_prefix(content).lower()
|
|
if re.search(r"\b(plan|planned|later|upcoming|will|future)\b", base):
|
|
return "future"
|
|
if re.search(r"\b(always|never|must|should|unless|only if|expected to)\b", base):
|
|
return "timeless"
|
|
if re.search(r"\b(previously|earlier|before|used to|was|were|retired|on 20\d{2}-\d{2}-\d{2}|as of)\b", base):
|
|
return "past"
|
|
return "current"
|
|
|
|
@classmethod
|
|
def _decorate_explicit_memory(
|
|
cls,
|
|
content: str,
|
|
*,
|
|
target: str = "memory",
|
|
memory_class: Optional[str] = None,
|
|
time_scope: Optional[str] = None,
|
|
) -> str:
|
|
base = cls._strip_explicit_prefix(content)
|
|
memory_class_norm = cls._normalize_memory_class(memory_class, target=target, content=base)
|
|
time_scope_norm = cls._normalize_time_scope(time_scope, base)
|
|
prefix = _MEMORY_CLASS_PREFIX.get(memory_class_norm, "FACT")
|
|
if memory_class_norm == "preference":
|
|
return f"[{prefix}] {base}"
|
|
return f"[{prefix}/{time_scope_norm.upper()}] {base}"
|
|
|
|
@classmethod
|
|
def _build_explicit_metadata(
|
|
cls,
|
|
*,
|
|
target: str,
|
|
content: str,
|
|
memory_class: Optional[str] = None,
|
|
time_scope: Optional[str] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
source: str = "mem0_conclude",
|
|
action: str = "add",
|
|
) -> Dict[str, Any]:
|
|
merged = dict(metadata or {})
|
|
merged.setdefault("memory_target", target)
|
|
merged.setdefault("source", source)
|
|
merged.setdefault("write_action", action)
|
|
merged["memory_class"] = cls._normalize_memory_class(memory_class, target=target, content=content)
|
|
merged["time_scope"] = cls._normalize_time_scope(time_scope, content)
|
|
merged["explicit_memory"] = True
|
|
return merged
|
|
|
|
@classmethod
|
|
def _format_memory_for_display(cls, text: str) -> str:
|
|
return cls._strip_explicit_prefix(text)
|
|
|
|
def system_prompt_block(self) -> str:
|
|
return (
|
|
"# Mem0 Memory\n"
|
|
f"Active. User: {self._user_id}.\n"
|
|
"Use mem0_search to find memories, mem0_conclude to store facts, "
|
|
"mem0_profile for a full overview."
|
|
)
|
|
|
|
def _search_with_fallback(self, client: Any, *, query: str, rerank: bool, top_k: int) -> list:
|
|
"""Search user-wide first, then fall back to agent-scoped reads if empty."""
|
|
results = self._unwrap_results(client.search(
|
|
query=query,
|
|
filters=self._read_filters(),
|
|
rerank=rerank,
|
|
top_k=top_k,
|
|
))
|
|
if results or not self._agent_id:
|
|
return results
|
|
return self._unwrap_results(client.search(
|
|
query=query,
|
|
filters=self._scoped_read_filters(),
|
|
rerank=rerank,
|
|
top_k=top_k,
|
|
))
|
|
|
|
def _get_all_with_fallback(self, client: Any) -> list:
|
|
"""Load user-wide memories first, then fall back to agent-scoped reads if empty."""
|
|
memories = self._unwrap_results(client.get_all(filters=self._read_filters()))
|
|
if memories or not self._agent_id:
|
|
return memories
|
|
return self._unwrap_results(client.get_all(filters=self._scoped_read_filters()))
|
|
|
|
def prefetch(self, query: str, *, session_id: str = "") -> str:
|
|
if self._prefetch_thread and self._prefetch_thread.is_alive():
|
|
self._prefetch_thread.join(timeout=3.0)
|
|
with self._prefetch_lock:
|
|
result = self._prefetch_result
|
|
self._prefetch_result = ""
|
|
if not result:
|
|
return ""
|
|
return f"## Mem0 Memory\n{result}"
|
|
|
|
def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
|
|
if self._is_breaker_open():
|
|
return
|
|
|
|
def _run():
|
|
try:
|
|
client = self._get_client()
|
|
results = self._search_with_fallback(
|
|
client,
|
|
query=query,
|
|
rerank=self._rerank,
|
|
top_k=5,
|
|
)
|
|
if results:
|
|
lines = [self._format_memory_for_display(r.get("memory", "")) for r in results if r.get("memory")]
|
|
with self._prefetch_lock:
|
|
self._prefetch_result = "\n".join(f"- {l}" for l in lines if l)
|
|
self._record_success()
|
|
except Exception as e:
|
|
self._record_failure()
|
|
logger.debug("Mem0 prefetch failed: %s", e)
|
|
|
|
self._prefetch_thread = threading.Thread(target=_run, daemon=True, name="mem0-prefetch")
|
|
self._prefetch_thread.start()
|
|
|
|
def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
|
|
"""Send the turn to Mem0 for server-side fact extraction (non-blocking)."""
|
|
if self._is_breaker_open():
|
|
return
|
|
if self._sync_turn_mode == "off":
|
|
return
|
|
|
|
def _sync():
|
|
try:
|
|
client = self._get_client()
|
|
messages = [
|
|
{"role": "user", "content": user_content},
|
|
{"role": "assistant", "content": assistant_content},
|
|
]
|
|
client.add(messages, **self._write_filters())
|
|
self._record_success()
|
|
except Exception as e:
|
|
self._record_failure()
|
|
logger.warning("Mem0 sync failed: %s", e)
|
|
|
|
# Wait for any previous sync before starting a new one
|
|
if self._sync_thread and self._sync_thread.is_alive():
|
|
self._sync_thread.join(timeout=5.0)
|
|
|
|
self._sync_thread = threading.Thread(target=_sync, daemon=True, name="mem0-sync")
|
|
self._sync_thread.start()
|
|
|
|
def get_tool_schemas(self) -> List[Dict[str, Any]]:
|
|
return [PROFILE_SCHEMA, SEARCH_SCHEMA, CONCLUDE_SCHEMA]
|
|
|
|
def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
|
|
if self._is_breaker_open():
|
|
return json.dumps({
|
|
"error": "Mem0 API temporarily unavailable (multiple consecutive failures). Will retry automatically."
|
|
})
|
|
|
|
try:
|
|
client = self._get_client()
|
|
except Exception as e:
|
|
return tool_error(str(e))
|
|
|
|
if tool_name == "mem0_profile":
|
|
try:
|
|
memories = self._get_all_with_fallback(client)
|
|
self._record_success()
|
|
if not memories:
|
|
return json.dumps({"result": "No memories stored yet."})
|
|
lines = [self._format_memory_for_display(m.get("memory", "")) for m in memories if m.get("memory")]
|
|
return json.dumps({"result": "\n".join(l for l in lines if l), "count": len(lines)})
|
|
except Exception as e:
|
|
self._record_failure()
|
|
return tool_error(f"Failed to fetch profile: {e}")
|
|
|
|
elif tool_name == "mem0_search":
|
|
query = args.get("query", "")
|
|
if not query:
|
|
return tool_error("Missing required parameter: query")
|
|
rerank = args.get("rerank", False)
|
|
top_k = min(int(args.get("top_k", 10)), 50)
|
|
try:
|
|
results = self._search_with_fallback(
|
|
client,
|
|
query=query,
|
|
rerank=rerank,
|
|
top_k=top_k,
|
|
)
|
|
self._record_success()
|
|
if not results:
|
|
return json.dumps({"result": "No relevant memories found."})
|
|
items = [
|
|
{
|
|
"memory": self._format_memory_for_display(r.get("memory", "")),
|
|
"score": r.get("score", 0),
|
|
"metadata": r.get("metadata", {}),
|
|
}
|
|
for r in results
|
|
]
|
|
return json.dumps({"results": items, "count": len(items)})
|
|
except Exception as e:
|
|
self._record_failure()
|
|
return tool_error(f"Search failed: {e}")
|
|
|
|
elif tool_name == "mem0_conclude":
|
|
conclusion = args.get("conclusion", "")
|
|
if not conclusion:
|
|
return tool_error("Missing required parameter: conclusion")
|
|
try:
|
|
explicit_text = self._decorate_explicit_memory(
|
|
conclusion,
|
|
target="memory",
|
|
memory_class=args.get("memory_class"),
|
|
time_scope=args.get("time_scope"),
|
|
)
|
|
explicit_metadata = self._build_explicit_metadata(
|
|
target="memory",
|
|
content=conclusion,
|
|
memory_class=args.get("memory_class"),
|
|
time_scope=args.get("time_scope"),
|
|
metadata=args.get("metadata") if isinstance(args.get("metadata"), dict) else None,
|
|
source="mem0_conclude",
|
|
action="add",
|
|
)
|
|
client.add(
|
|
[{"role": "user", "content": explicit_text}],
|
|
**self._write_filters(),
|
|
infer=False,
|
|
metadata=explicit_metadata,
|
|
)
|
|
self._record_success()
|
|
return json.dumps({
|
|
"result": "Fact stored.",
|
|
"stored": self._format_memory_for_display(explicit_text),
|
|
"memory_class": explicit_metadata.get("memory_class"),
|
|
"time_scope": explicit_metadata.get("time_scope"),
|
|
})
|
|
except Exception as e:
|
|
self._record_failure()
|
|
return tool_error(f"Failed to store: {e}")
|
|
|
|
return tool_error(f"Unknown tool: {tool_name}")
|
|
|
|
def on_memory_write(
|
|
self,
|
|
action: str,
|
|
target: str,
|
|
content: str,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
) -> None:
|
|
"""Mirror built-in memory writes to Mem0 using explicit classification metadata."""
|
|
if self._is_breaker_open():
|
|
return
|
|
if action not in {"add", "replace"} or not (content or "").strip():
|
|
return
|
|
|
|
def _write():
|
|
try:
|
|
client = self._get_client()
|
|
explicit_text = self._decorate_explicit_memory(
|
|
content,
|
|
target=target,
|
|
memory_class=(metadata or {}).get("memory_class") if isinstance(metadata, dict) else None,
|
|
time_scope=(metadata or {}).get("time_scope") if isinstance(metadata, dict) else None,
|
|
)
|
|
explicit_metadata = self._build_explicit_metadata(
|
|
target=target,
|
|
content=content,
|
|
memory_class=(metadata or {}).get("memory_class") if isinstance(metadata, dict) else None,
|
|
time_scope=(metadata or {}).get("time_scope") if isinstance(metadata, dict) else None,
|
|
metadata=metadata,
|
|
source="builtin_memory_tool",
|
|
action=action,
|
|
)
|
|
client.add(
|
|
[{"role": "user", "content": explicit_text}],
|
|
**self._write_filters(),
|
|
infer=False,
|
|
metadata=explicit_metadata,
|
|
)
|
|
self._record_success()
|
|
except Exception as e:
|
|
self._record_failure()
|
|
logger.debug("Mem0 on_memory_write failed: %s", e)
|
|
|
|
if self._write_thread and self._write_thread.is_alive():
|
|
self._write_thread.join(timeout=2.0)
|
|
self._write_thread = threading.Thread(target=_write, daemon=True, name="mem0-memwrite")
|
|
self._write_thread.start()
|
|
|
|
def shutdown(self) -> None:
|
|
for t in (self._prefetch_thread, self._sync_thread, self._write_thread):
|
|
if t and t.is_alive():
|
|
t.join(timeout=5.0)
|
|
with self._client_lock:
|
|
self._client = None
|
|
|
|
|
|
def register(ctx) -> None:
|
|
"""Register Mem0 as a memory provider plugin."""
|
|
ctx.register_memory_provider(Mem0MemoryProvider())
|