From 968cbef157882d63b2f85b8876f0fdbe18741d80 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:05:29 -0700 Subject: [PATCH] refactor(memory/mem0): table-driven tool dispatch, shared _oss_hint/_truthy/_add helpers, packed provider tables --- plugins/memory/mem0/__init__.py | 245 +++++++++----------------- plugins/memory/mem0/_backend.py | 29 +-- plugins/memory/mem0/_openai_llm.py | 22 +-- plugins/memory/mem0/_oss_providers.py | 31 +--- 4 files changed, 106 insertions(+), 221 deletions(-) diff --git a/plugins/memory/mem0/__init__.py b/plugins/memory/mem0/__init__.py index 9bd5a9ca49..b986e0bb8d 100644 --- a/plugins/memory/mem0/__init__.py +++ b/plugins/memory/mem0/__init__.py @@ -5,8 +5,7 @@ Platform API (cloud), a self-hosted Mem0 server (MEM0_HOST, HTTP), or OSS Memory Secrets live in $HERMES_HOME/.env (MEM0_API_KEY, MEM0_HOST); behavioral settings in $HERMES_HOME/mem0.json via `hermes memory setup`: mode ("platform"|"oss"), host, user_id (canonical id across every gateway so one human gets one merged store; -unset → gateway-native id), agent_id. MEM0_MODE/MEM0_USER_ID/MEM0_AGENT_ID env -vars remain a backward-compatible fallback. +unset → gateway-native id), agent_id. MEM0_* env vars remain a fallback. """ from __future__ import annotations @@ -41,6 +40,13 @@ def _is_client_error(exc: Exception) -> bool: return type(exc).__name__ in _CLIENT_ERROR_TYPES or any(s in err_str for s in ("404", "not found", "valid uuid")) +def _truthy(value: Any, falsy_strings: tuple[str, ...] | None = None) -> bool: + """Coerce a flag; strings match ``falsy_strings`` (deny-list) when given, else a truthy allow-list.""" + if not isinstance(value, str): + return bool(value) + return value.lower() not in falsy_strings if falsy_strings else value.lower() in ("true", "1", "yes") + + def _read_mem0_json(config_path: Path) -> dict: """Best-effort read of mem0.json; missing/corrupt file -> {}.""" if config_path.exists(): @@ -56,15 +62,8 @@ def _load_config() -> dict: Layering avoids a silent failure when the JSON file exists but lacks fields like ``api_key`` that the user set in ``.env``.""" from hermes_constants import get_hermes_home - config = { - "mode": os.environ.get("MEM0_MODE", "platform"), - "api_key": get_secret("MEM0_API_KEY", ""), - "host": os.environ.get("MEM0_HOST", ""), - "agent_id": os.environ.get("MEM0_AGENT_ID", "hermes"), - "oss": {}, - } - # Only carry user_id when explicitly configured so initialize() can fall - # back to the gateway-native id. + config = {"mode": os.environ.get("MEM0_MODE", "platform"), "api_key": get_secret("MEM0_API_KEY", ""), "host": os.environ.get("MEM0_HOST", ""), "agent_id": os.environ.get("MEM0_AGENT_ID", "hermes"), "oss": {}} + # Only carry user_id when explicitly configured so initialize() can fall back to the gateway-native id. if os.environ.get("MEM0_USER_ID"): config["user_id"] = os.environ["MEM0_USER_ID"] file_cfg = _read_mem0_json(get_hermes_home() / "mem0.json") @@ -76,59 +75,42 @@ def _load_config() -> dict: # Tool schemas # --------------------------------------------------------------------------- -def _schema(name: str, description: str, properties: dict, required: list[str]) -> dict: - return {"name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": required}} - - -def _param(type_: str, description: str) -> dict: - return {"type": type_, "description": description} +def _schema(name: str, description: str, properties: dict[str, tuple[str, str]], required: list[str]) -> dict: + props = {k: {"type": t, "description": d} for k, (t, d) in properties.items()} + return {"name": name, "description": description, "parameters": {"type": "object", "properties": props, "required": required}} SEARCH_SCHEMA = _schema( "mem0_search", - "Search the user's memories by meaning; returns facts ranked by " - "relevance. Use this before answering any question that may depend on " - "what you know about the user (preferences, facts, history, people, " - "projects, past decisions). For multi-part or multi-hop questions, " - "call it several times — vary the wording and run follow-up searches " - "on what earlier results reveal; one search is rarely enough.", - { - "query": _param("string", "What to search for."), - "top_k": _param("integer", "Max results (default: 10, max: 50)."), - "rerank": _param("boolean", "Rerank results for relevance (default: false, platform mode only)."), - }, + "Search the user's memories by meaning; returns facts ranked by relevance. Use this before answering any question that may depend on what you know about the user (preferences, facts, history, people, projects, past decisions). For multi-part or multi-hop questions, call it several times — vary the wording and run follow-up searches on what earlier results reveal; one search is rarely enough.", + {"query": ("string", "What to search for."), "top_k": ("integer", "Max results (default: 10, max: 50)."), "rerank": ("boolean", "Rerank results for relevance (default: false, platform mode only).")}, ["query"], ) - ADD_SCHEMA = _schema( "mem0_add", - "Store a durable fact about the user, verbatim (no LLM extraction). " - "Call this the moment the user states a lasting preference, correction, " - "decision, or personal detail worth recalling on future turns — don't " - "wait to be asked to remember. Skip transient chit-chat and facts you've " - "already stored.", - {"content": _param("string", "The fact to store.")}, + "Store a durable fact about the user, verbatim (no LLM extraction). Call this the moment the user states a lasting preference, correction, decision, or personal detail worth recalling on future turns — don't wait to be asked to remember. Skip transient chit-chat and facts you've already stored.", + {"content": ("string", "The fact to store.")}, ["content"], ) - UPDATE_SCHEMA = _schema( "mem0_update", - "Replace the text of an existing memory by its ID (take the ID from a " - "mem0_search result). Use when a stored fact has changed " - "or was wrong — correct it in place instead of adding a duplicate.", - {"memory_id": _param("string", "Memory UUID to update."), "text": _param("string", "New text content.")}, + "Replace the text of an existing memory by its ID (take the ID from a mem0_search result). Use when a stored fact has changed or was wrong — correct it in place instead of adding a duplicate.", + {"memory_id": ("string", "Memory UUID to update."), "text": ("string", "New text content.")}, ["memory_id", "text"], ) - DELETE_SCHEMA = _schema( "mem0_delete", - "Delete a memory by its ID (take the ID from a mem0_search " - "result). Use when a stored fact is obsolete or the user asks you to " - "forget it; prefer mem0_update if the fact merely changed.", - {"memory_id": _param("string", "Memory UUID to delete.")}, + "Delete a memory by its ID (take the ID from a mem0_search result). Use when a stored fact is obsolete or the user asks you to forget it; prefer mem0_update if the fact merely changed.", + {"memory_id": ("string", "Memory UUID to delete.")}, ["memory_id"], ) +_PROMPT_BODY = ( + "You have persistent memory of this user from past conversations. You should call mem0_search before answering anything that could depend on prior context (the user's preferences, facts, history, people, projects, or earlier decisions) — do not rely on the chat window alone, and do not assume you have no memory.\n" + "For multi-part or multi-hop questions, run several searches with different wording/angles and follow-up searches on what the first results surface; one search is rarely enough. Keep searching until you have every fact the question needs before you answer.\n" + "Tools: mem0_search to find memories, mem0_add to store facts, mem0_update and mem0_delete to manage by ID." +) + # --------------------------------------------------------------------------- # MemoryProvider implementation @@ -157,8 +139,7 @@ class Mem0MemoryProvider(MemoryProvider): cfg = _load_config() if cfg.get("mode", "platform") == "oss": return bool(cfg.get("oss", {}).get("vector_store")) - # Platform needs an api_key; self-hosted needs a host (api_key optional - # when the server runs with AUTH_DISABLED). + # Platform needs an api_key; self-hosted needs a host (api_key optional when the server runs with AUTH_DISABLED). return bool(cfg.get("api_key") or cfg.get("host")) def save_config(self, values, hermes_home): @@ -183,9 +164,11 @@ class Mem0MemoryProvider(MemoryProvider): from ._setup import post_setup post_setup(hermes_home, config) - def _vs_provider(self, default: str) -> str: - """Configured OSS vector-store provider name (for error hints).""" - return self._config.get("oss", {}).get("vector_store", {}).get("provider", default) + def _oss_hint(self, template: str, default: str = "vector store") -> str: + """OSS-only hint; ``{vs}`` is the configured vector-store provider. "" in other modes.""" + if self._mode != "oss": + return "" + return template.format(vs=self._config.get("oss", {}).get("vector_store", {}).get("provider", default)) def _create_backend(self): # Lazy-install the mem0 SDK before either backend imports it. ensure() honors @@ -222,10 +205,8 @@ class Mem0MemoryProvider(MemoryProvider): def _format_error(self, prefix: str, exc: Exception) -> str: msg = f"{prefix}: {exc}" - if self._mode == "oss": - err_str = str(exc).lower() - if "connection" in err_str or "refused" in err_str or "timeout" in err_str: - msg += f" (check that {self._vs_provider('vector store')} is running)" + if any(s in str(exc).lower() for s in ("connection", "refused", "timeout")): + msg += self._oss_hint(" (check that {vs} is running)") return msg def _record_success(self): @@ -239,12 +220,8 @@ class Mem0MemoryProvider(MemoryProvider): if tripped: self._breaker_open_until = time.monotonic() + _BREAKER_COOLDOWN_SECS if tripped: - hint = f" Check that your {self._vs_provider('unknown')} vector store is running and reachable." if self._mode == "oss" else "" - logger.warning( - "Mem0 circuit breaker tripped after %d consecutive failures. " - "Pausing API calls for %ds.%s", - count, _BREAKER_COOLDOWN_SECS, hint, - ) + hint = self._oss_hint(" Check that your {vs} vector store is running and reachable.", "unknown") + logger.warning("Mem0 circuit breaker tripped after %d consecutive failures. Pausing API calls for %ds.%s", count, _BREAKER_COOLDOWN_SECS, hint) def initialize(self, session_id: str, **kwargs) -> None: self._config = _load_config() @@ -259,43 +236,29 @@ class Mem0MemoryProvider(MemoryProvider): self._agent_id = self._config.get("agent_id", "hermes") # Persisted rerank preference: DEFAULT for mem0_search when the model doesn't # pass ``rerank``; per-call args win. Platform-only; other backends ignore it. - _rr = self._config.get("rerank", False) - self._rerank_default = _rr.lower() in ("true", "1", "yes") if isinstance(_rr, str) else bool(_rr) + self._rerank_default = _truthy(self._config.get("rerank", False)) self._channel = kwargs.get("platform") or "cli" self._backend = self._create_backend() if self._backend and not self._atexit_registered: atexit.register(self._shutdown_backend) self._atexit_registered = True - def _read_filters(self) -> Dict[str, Any]: + def _search(self, query: str, top_k: int = 10, rerank: bool = False, backend=None) -> list: # Scoped to user_id only — by design — so recall surfaces memories from any # gateway/agent under this principal; writes attach agent_id and metadata.channel # (dashboard per-channel filtering) so narrower views remain possible at query time. - return {"user_id": self._user_id} + return (backend or self._backend).search(query, filters={"user_id": self._user_id}, top_k=top_k, rerank=rerank) - def _write_metadata(self) -> Dict[str, Any]: - return {"channel": self._channel} if self._channel else {} + def _add(self, messages: list, infer: bool): + metadata = {"channel": self._channel} if self._channel else {} + return self._backend.add(messages, user_id=self._user_id, agent_id=self._agent_id, infer=infer, metadata=metadata) def system_prompt_block(self) -> str: # Mirror _create_backend precedence (oss > host > platform) so the label names # the backend that actually runs. Rerank is a Mem0 Platform feature only. mode_label = "OSS (self-hosted)" if self._mode == "oss" else "self-hosted (HTTP API)" if self._host else "platform (cloud API)" rerank_note = " Rerank is available on search." if (self._mode == "platform" and not self._host) else "" - return ( - "# Mem0 Memory\n" - f"Active. Mode: {mode_label}. User: {self._user_id}.\n" - "You have persistent memory of this user from past conversations. " - "You should call mem0_search before answering anything that could depend " - "on prior context (the user's preferences, facts, history, people, " - "projects, or earlier decisions) — do not rely on the chat window " - "alone, and do not assume you have no memory.\n" - "For multi-part or multi-hop questions, run several searches with " - "different wording/angles and follow-up searches on what the first " - "results surface; one search is rarely enough. Keep searching until " - "you have every fact the question needs before you answer.\n" - "Tools: mem0_search to find memories, mem0_add to store facts, " - f"mem0_update and mem0_delete to manage by ID.{rerank_note}" - ) + return f"# Mem0 Memory\nActive. Mode: {mode_label}. User: {self._user_id}.\n{_PROMPT_BODY}{rerank_note}" def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None: self._start_prefetch(message) @@ -313,17 +276,14 @@ class Mem0MemoryProvider(MemoryProvider): return with self._prefetch_lock: # Same query already answered or still in flight: don't restart it. - if self._prefetch_query == query and ( - self._prefetch_done or (self._prefetch_thread and self._prefetch_thread.is_alive()) - ): + if self._prefetch_query == query and (self._prefetch_done or (self._prefetch_thread and self._prefetch_thread.is_alive())): return self._prefetch_query, self._prefetch_result, self._prefetch_done = query, "", False def _run(): body = "" try: - results = backend.search(query, filters=self._read_filters(), top_k=10, rerank=False) - lines = [r.get("memory", "") for r in (results or []) if r.get("memory")] + lines = [r.get("memory", "") for r in (self._search(query, backend=backend) or []) if r.get("memory")] if lines: body = "## Mem0 Memory\n" + "\n".join(f"- {l}" for l in lines) self._record_success() @@ -332,8 +292,7 @@ class Mem0MemoryProvider(MemoryProvider): logger.debug("Mem0 prefetch failed: %s", e) with self._prefetch_lock: if self._prefetch_query == query: - self._prefetch_result = body - self._prefetch_done = True + self._prefetch_result, self._prefetch_done = body, True t = threading.Thread(target=_run, daemon=True, name="mem0-prefetch") with self._prefetch_lock: @@ -359,12 +318,10 @@ class Mem0MemoryProvider(MemoryProvider): return def _sync(): - backend = self._backend - if backend is None: + if self._backend is None: return try: - messages = [{"role": "user", "content": user_content}, {"role": "assistant", "content": assistant_content}] - backend.add(messages, user_id=self._user_id, agent_id=self._agent_id, infer=True, metadata=self._write_metadata()) + self._add([{"role": "user", "content": user_content}, {"role": "assistant", "content": assistant_content}], infer=True) self._record_success() except Exception as e: self._record_failure() @@ -382,86 +339,56 @@ class Mem0MemoryProvider(MemoryProvider): def get_tool_schemas(self) -> List[Dict[str, Any]]: return [SEARCH_SCHEMA, ADD_SCHEMA, UPDATE_SCHEMA, DELETE_SCHEMA] - # -- tool handlers ------------------------------------------------------- + # -- tool handlers: (required params, error label, body, client-error policy) --- + # Client errors (bad ID / not found) never trip the breaker, except for mem0_add + # where they count as failures; update/delete answer them with "Memory not found". def _tool_search(self, args: dict) -> str: - query = args.get("query", "") - if not query: - return tool_error("Missing required parameter: query") - try: - top_k = max(1, min(int(args.get("top_k", 10)), 50)) - rerank_raw = args.get("rerank", self._rerank_default) - rerank = rerank_raw.lower() not in ("false", "0", "no") if isinstance(rerank_raw, str) else bool(rerank_raw) - results = self._backend.search(query, filters=self._read_filters(), top_k=top_k, rerank=rerank) - self._record_success() - if not results: - return json.dumps({"result": "No relevant memories found."}) - items = [{"id": r.get("id"), "memory": r.get("memory", ""), "score": r.get("score", 0)} for r in results] - return json.dumps({"results": items, "count": len(items)}) - except Exception as e: - if not _is_client_error(e): - self._record_failure() - return tool_error(self._format_error("Search failed", e)) + top_k = max(1, min(int(args.get("top_k", 10)), 50)) + rerank = _truthy(args.get("rerank", self._rerank_default), falsy_strings=("false", "0", "no")) + results = self._search(args["query"], top_k, rerank) + if not results: + return json.dumps({"result": "No relevant memories found."}) + items = [{"id": r.get("id"), "memory": r.get("memory", ""), "score": r.get("score", 0)} for r in results] + return json.dumps({"results": items, "count": len(items)}) def _tool_add(self, args: dict) -> str: - content = args.get("content", "") - if not content: - return tool_error("Missing required parameter: content") - try: - result = self._backend.add( - [{"role": "user", "content": content}], - user_id=self._user_id, agent_id=self._agent_id, infer=False, metadata=self._write_metadata(), - ) - self._record_success() - event_id = result.get("event_id") if isinstance(result, dict) else None - # Cloud add is async (server-side extraction); OSS and self-hosted store synchronously. - msg = "Fact stored." if (self._mode == "oss" or self._host) else "Fact queued for storage." - return json.dumps({"result": msg, "event_id": event_id}) - except Exception as e: - self._record_failure() - return tool_error(self._format_error("Failed to store", e)) - - def _tool_by_id(self, args: dict, required: tuple[str, ...], label: str, method: str) -> str: - """Shared update/delete shape: required-param check, then backend.(*values).""" - values = [args.get(k, "") for k in required] - for k, v in zip(required, values): - if not v: - return tool_error(f"Missing required parameter: {k}") - try: - result = getattr(self._backend, method)(*values) - self._record_success() - return json.dumps(result) - except Exception as e: - if _is_client_error(e): - return tool_error(f"Memory not found: {values[0]}") - self._record_failure() - return tool_error(self._format_error(label, e)) - - def _tool_update(self, args: dict) -> str: - return self._tool_by_id(args, ("memory_id", "text"), "Update failed", "update") - - def _tool_delete(self, args: dict) -> str: - return self._tool_by_id(args, ("memory_id",), "Delete failed", "delete") + result = self._add([{"role": "user", "content": args["content"]}], infer=False) + event_id = result.get("event_id") if isinstance(result, dict) else None + # Cloud add is async (server-side extraction); OSS and self-hosted store synchronously. + msg = "Fact stored." if (self._mode == "oss" or self._host) else "Fact queued for storage." + return json.dumps({"result": msg, "event_id": event_id}) _TOOL_HANDLERS = { - "mem0_search": _tool_search, - "mem0_add": _tool_add, - "mem0_update": _tool_update, - "mem0_delete": _tool_delete, + "mem0_search": (("query",), "Search failed", _tool_search, "skip"), + "mem0_add": (("content",), "Failed to store", _tool_add, "count"), + "mem0_update": (("memory_id", "text"), "Update failed", lambda self, a: json.dumps(self._backend.update(a["memory_id"], a["text"])), "not_found"), + "mem0_delete": (("memory_id",), "Delete failed", lambda self, a: json.dumps(self._backend.delete(a["memory_id"])), "not_found"), } def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str: if self._backend is None: err = getattr(self, "_init_error", "unknown error") - hint = f" Check that {self._vs_provider('vector store')} is running and reachable." if self._mode == "oss" else "" - return json.dumps({"error": f"Mem0 backend not initialized: {err}.{hint}"}) + return json.dumps({"error": f"Mem0 backend not initialized: {err}.{self._oss_hint(' Check that {vs} is running and reachable.')}"}) if self._is_breaker_open(): - hint = f" Check that your {self._vs_provider('vector store')} is running." if self._mode == "oss" else "" - return json.dumps({"error": f"Mem0 temporarily unavailable (multiple consecutive failures). Will retry automatically.{hint}"}) - handler = self._TOOL_HANDLERS.get(tool_name) - if handler is None: + return json.dumps({"error": f"Mem0 temporarily unavailable (multiple consecutive failures). Will retry automatically.{self._oss_hint(' Check that your {vs} is running.')}"}) + if tool_name not in self._TOOL_HANDLERS: return tool_error(f"Unknown tool: {tool_name}") - return handler(self, args) + required, label, body, on_client_error = self._TOOL_HANDLERS[tool_name] + for k in required: + if not args.get(k, ""): + return tool_error(f"Missing required parameter: {k}") + try: + result = body(self, args) + self._record_success() + return result + except Exception as e: + client = _is_client_error(e) + if client and on_client_error == "not_found": + return tool_error(f"Memory not found: {args['memory_id']}") + if not client or on_client_error == "count": + self._record_failure() + return tool_error(self._format_error(label, e)) def _shutdown_backend(self): try: diff --git a/plugins/memory/mem0/_backend.py b/plugins/memory/mem0/_backend.py index 930214958a..e935cef5cf 100644 --- a/plugins/memory/mem0/_backend.py +++ b/plugins/memory/mem0/_backend.py @@ -81,16 +81,12 @@ class SelfHostedBackend(Mem0Backend): def __init__(self, api_key: str, host: str, transport=None): import httpx - headers = {"Content-Type": "application/json"} if api_key: headers["X-API-Key"] = api_key # omitted only for AUTH_DISABLED servers # Connect-level retries keep a single dropped SYN from counting toward the # provider failure breaker. ``transport`` is injectable for tests. - self._client = httpx.Client( - base_url=host.rstrip("/"), headers=headers, timeout=30.0, - transport=transport or httpx.HTTPTransport(retries=2), - ) + self._client = httpx.Client(base_url=host.rstrip("/"), headers=headers, timeout=30.0, transport=transport or httpx.HTTPTransport(retries=2)) def _json(self, method: str, path: str, **kwargs) -> Any: resp = self._client.request(method, path, **kwargs) @@ -105,8 +101,7 @@ class SelfHostedBackend(Mem0Backend): return _unwrap_results(self._json("POST", "/search", json=body)) def add(self, messages: list, *, user_id: str, agent_id: str, infer: bool = False, metadata: dict | None = None) -> dict: - body: dict[str, Any] = {"messages": messages, **_add_kwargs(user_id, agent_id, infer, metadata)} - return self._json("POST", "/memories", json=body) + return self._json("POST", "/memories", json={"messages": messages, **_add_kwargs(user_id, agent_id, infer, metadata)}) def _update(self, memory_id: str, text: str) -> None: self._json("PUT", f"/memories/{memory_id}", json={"text": text}) @@ -133,10 +128,7 @@ def _register_direct_openai_provider() -> None: provider_map = getattr(LlmFactory, "provider_to_class", None) register_provider = getattr(LlmFactory, "register_provider", None) if not isinstance(provider_map, dict) or not callable(register_provider): - raise RuntimeError( - "mem0 LlmFactory does not support the provider registration required " - "for the Hermes OpenAI OSS backend" - ) + raise RuntimeError("mem0 LlmFactory does not support the provider registration required for the Hermes OpenAI OSS backend") if provider_map.get(_DIRECT_OPENAI_PROVIDER) != (_DIRECT_OPENAI_CLASS_PATH, OpenAIConfig): register_provider(_DIRECT_OPENAI_PROVIDER, _DIRECT_OPENAI_CLASS_PATH, OpenAIConfig) @@ -166,7 +158,6 @@ class OSSBackend(Mem0Backend): vs_config = dict(vector_store.get("config", {})) if "path" in vs_config: vs_config["path"] = os.path.expanduser(vs_config["path"]) - embedder_config = oss_config.get("embedder", {}).get("config", {}) dims = embedder_config.get("embedding_dims") or KNOWN_DIMS.get(embedder_config.get("model", "")) if dims: @@ -174,12 +165,7 @@ class OSSBackend(Mem0Backend): self._recreate_collection_if_dims_changed(vector_store.get("provider", "qdrant"), vs_config, dims) vector_store["config"] = vs_config - config = { - "vector_store": vector_store, - "llm": _provider_block("llm"), - "embedder": _provider_block("embedder"), - "version": "v1.1", - } + config = {"vector_store": vector_store, "llm": _provider_block("llm"), "embedder": _provider_block("embedder"), "version": "v1.1"} if str(config["llm"].get("provider") or "").strip().lower() == "openai": # mem0 validates LlmConfig.provider before its factory lookup: build the # supported OpenAI config first, then swap the provider on the validated object. @@ -189,10 +175,7 @@ class OSSBackend(Mem0Backend): try: memory_config.llm.provider = _DIRECT_OPENAI_PROVIDER except (AttributeError, TypeError) as exc: - raise RuntimeError( - "mem0 MemoryConfig does not expose a mutable llm.provider " - "for the Hermes OpenAI OSS backend" - ) from exc + raise RuntimeError("mem0 MemoryConfig does not expose a mutable llm.provider for the Hermes OpenAI OSS backend") from exc self._memory = Memory(memory_config) else: self._memory = Memory.from_config(config) @@ -215,7 +198,7 @@ class OSSBackend(Mem0Backend): if not client.collection_exists(collection_name): return vectors = client.get_collection(collection_name).config.params.vectors - # Named-vector collections expose a dict; unnamed expose an object with .size. + # Named-vector collections expose a dict; unnamed expose an object with .size. if isinstance(vectors, dict): vectors = next(iter(vectors.values()), None) current_dims = getattr(vectors, "size", None) diff --git a/plugins/memory/mem0/_openai_llm.py b/plugins/memory/mem0/_openai_llm.py index 64d1fe6312..4a54adc394 100644 --- a/plugins/memory/mem0/_openai_llm.py +++ b/plugins/memory/mem0/_openai_llm.py @@ -11,6 +11,10 @@ from mem0.configs.llms.openai import OpenAIConfig from mem0.llms.base import LLMBase from mem0.llms.openai import OpenAILLM +# BaseLlmConfig fields copied into OpenAIConfig; the last two may be absent on older mem0. +_COPIED_FIELDS = ("model", "temperature", "api_key", "max_tokens", "top_p", "top_k", "enable_vision", "vision_details", "http_client_proxies") +_OPTIONAL_FIELDS = ("reasoning_effort", "is_reasoning_model") + class DirectOpenAILLM(OpenAILLM): """Use OpenAI credentials and requests regardless of router environment.""" @@ -21,15 +25,9 @@ class DirectOpenAILLM(OpenAILLM): elif isinstance(config, dict): config = OpenAIConfig(**config) elif isinstance(config, BaseLlmConfig) and not isinstance(config, OpenAIConfig): - config = OpenAIConfig( - model=config.model, temperature=config.temperature, api_key=config.api_key, - max_tokens=config.max_tokens, top_p=config.top_p, top_k=config.top_k, - enable_vision=config.enable_vision, vision_details=config.vision_details, - reasoning_effort=getattr(config, "reasoning_effort", None), - http_client_proxies=config.http_client_proxies, - is_reasoning_model=getattr(config, "is_reasoning_model", None), - ) - + fields = {k: getattr(config, k) for k in _COPIED_FIELDS} + fields.update({k: getattr(config, k, None) for k in _OPTIONAL_FIELDS}) + config = OpenAIConfig(**fields) if not config.model: config.model = "gpt-5-mini" # Configs predating the setup marker: keep the default model reasoning-safe @@ -39,17 +37,13 @@ class DirectOpenAILLM(OpenAILLM): # Bypass OpenAILLM.__init__ (it picks OpenRouter when OPENROUTER_API_KEY is # set); LLMBase still owns validation and supported-parameter filtering. LLMBase.__init__(self, config) - api_key = self.config.api_key or os.getenv("OPENAI_API_KEY") if not api_key: raise ValueError("OpenAI API key is required for the Hermes Mem0 OSS provider") from openai import OpenAI self.client = OpenAI(api_key=api_key, base_url=self.config.openai_base_url or os.getenv("OPENAI_BASE_URL") or "https://api.openai.com/v1") - def generate_response( - self, messages: List[Dict[str, str]], response_format=None, - tools: Optional[List[Dict]] = None, tool_choice: str = "auto", **kwargs, - ): + def generate_response(self, messages: List[Dict[str, str]], response_format=None, tools: Optional[List[Dict]] = None, tool_choice: str = "auto", **kwargs): params = self._get_supported_params(messages=messages, **kwargs) params.update({"model": self.config.model, "messages": messages}) # No OpenRouter-only fields; ``store`` is opt-in so OpenAI-compatible diff --git a/plugins/memory/mem0/_oss_providers.py b/plugins/memory/mem0/_oss_providers.py index 69ead2009b..7fb87d998a 100644 --- a/plugins/memory/mem0/_oss_providers.py +++ b/plugins/memory/mem0/_oss_providers.py @@ -6,33 +6,17 @@ import os from typing import Any LLM_PROVIDERS: dict[str, dict[str, Any]] = { - "openai": { - "label": "OpenAI", "needs_key": True, "env_var": "OPENAI_API_KEY", - "default_model": "gpt-5-mini", "base_url_key": "openai_base_url", - }, - "ollama": { - "label": "Ollama (local)", "needs_key": False, "default_model": "llama3.1:8b", - "default_url": "http://localhost:11434", "base_url_key": "ollama_base_url", "pip_dep": "ollama", - }, + "openai": {"label": "OpenAI", "needs_key": True, "env_var": "OPENAI_API_KEY", "default_model": "gpt-5-mini", "base_url_key": "openai_base_url"}, + "ollama": {"label": "Ollama (local)", "needs_key": False, "default_model": "llama3.1:8b", "default_url": "http://localhost:11434", "base_url_key": "ollama_base_url", "pip_dep": "ollama"}, } EMBEDDER_PROVIDERS: dict[str, dict[str, Any]] = { - "openai": { - "label": "OpenAI", "needs_key": True, "env_var": "OPENAI_API_KEY", - "default_model": "text-embedding-3-small", "base_url_key": "openai_base_url", "dims": 1536, - }, - "ollama": { - "label": "Ollama (local)", "needs_key": False, "default_model": "nomic-embed-text", - "default_url": "http://localhost:11434", "base_url_key": "ollama_base_url", "dims": 768, "pip_dep": "ollama", - }, + "openai": {"label": "OpenAI", "needs_key": True, "env_var": "OPENAI_API_KEY", "default_model": "text-embedding-3-small", "base_url_key": "openai_base_url", "dims": 1536}, + "ollama": {"label": "Ollama (local)", "needs_key": False, "default_model": "nomic-embed-text", "default_url": "http://localhost:11434", "base_url_key": "ollama_base_url", "dims": 768, "pip_dep": "ollama"}, } VECTOR_PROVIDERS: dict[str, dict[str, Any]] = { - "qdrant": { - "label": "Qdrant", - "default_config": {"path": os.path.expanduser("~/.hermes/mem0_qdrant")}, - "pip_dep": "qdrant-client", - }, + "qdrant": {"label": "Qdrant", "default_config": {"path": os.path.expanduser("~/.hermes/mem0_qdrant")}, "pip_dep": "qdrant-client"}, "pgvector": { "label": "PGVector", "default_config": {"host": "localhost", "port": 5432, "user": os.getenv("USER", "postgres"), "dbname": "postgres"}, @@ -40,9 +24,7 @@ VECTOR_PROVIDERS: dict[str, dict[str, Any]] = { }, } -KNOWN_DIMS: dict[str, int] = { - "text-embedding-3-small": 1536, "text-embedding-3-large": 3072, "text-embedding-ada-002": 1536, "nomic-embed-text": 768, -} +KNOWN_DIMS: dict[str, int] = {"text-embedding-3-small": 1536, "text-embedding-3-large": 3072, "text-embedding-ada-002": 1536, "nomic-embed-text": 768} SECTION_REGISTRIES = (("llm", LLM_PROVIDERS), ("embedder", EMBEDDER_PROVIDERS), ("vector_store", VECTOR_PROVIDERS)) @@ -56,7 +38,6 @@ def validate_oss_config(oss_config: dict) -> list[str]: errors.append(f"Missing required section: {section}") elif block.get("provider", "") not in registry: errors.append(f"Unknown {section} provider '{block.get('provider', '')}'. Valid: {', '.join(registry.keys())}") - vs = oss_config.get("vector_store", {}) if vs.get("provider") == "pgvector" and not vs.get("config", {}).get("user"): errors.append("PGVector requires 'user' in vector_store.config")