fix: properly test OpenMemory import error scenario

Co-Authored-By: João <joao@crewai.com>

lib/crewai/pyproject.toml

@@ -61,6 +61,7 @@ openpyxl = [
     "openpyxl~=3.1.5",
 ]
 mem0 = ["mem0ai~=0.1.94"]
+openmemory = ["openmemory-py>=1.0.0"]
 docling = [
     "docling~=2.63.0",
 ]

lib/crewai/src/crewai/memory/external/external_memory.py

@@ -20,6 +20,7 @@ from crewai.rag.embeddings.types import ProviderSpec
 
 if TYPE_CHECKING:
     from crewai.memory.storage.mem0_storage import Mem0Storage
+    from crewai.memory.storage.openmemory_storage import OpenMemoryStorage
 
 
 class ExternalMemory(Memory):
@@ -32,10 +33,19 @@ class ExternalMemory(Memory):
 
         return Mem0Storage(type="external", crew=crew, config=config)  # type: ignore[no-untyped-call]
 
+    @staticmethod
+    def _configure_openmemory(
+        crew: Any, config: dict[str, Any]
+    ) -> OpenMemoryStorage:
+        from crewai.memory.storage.openmemory_storage import OpenMemoryStorage
+
+        return OpenMemoryStorage(type="external", crew=crew, config=config)
+
     @staticmethod
     def external_supported_storages() -> dict[str, Any]:
         return {
             "mem0": ExternalMemory._configure_mem0,
+            "openmemory": ExternalMemory._configure_openmemory,
         }
 
     @staticmethod

lib/crewai/src/crewai/memory/storage/openmemory_storage.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from crewai.memory.storage.interface import Storage
+from crewai.utilities.paths import get_crewai_storage_dir
+
+
+class OpenMemoryStorage(Storage):
+    """
+    Extends Storage to handle embedding and searching using OpenMemory.
+
+    OpenMemory is a local-first persistent memory engine for AI applications.
+    It provides cognitive memory capabilities without requiring external vector databases.
+    """
+
+    def __init__(
+        self,
+        type: str,
+        crew: Any | None = None,
+        config: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__()
+
+        self._validate_type(type)
+        self.memory_type = type
+        self.crew = crew
+        self.config = config or {}
+
+        self._extract_config_values()
+        self._initialize_memory()
+
+    def _validate_type(self, type: str) -> None:
+        supported_types = {"short_term", "long_term", "entities", "external"}
+        if type not in supported_types:
+            raise ValueError(
+                f"Invalid type '{type}' for OpenMemoryStorage. "
+                f"Must be one of: {', '.join(supported_types)}"
+            )
+
+    def _extract_config_values(self) -> None:
+        self.user_id = self.config.get("user_id")
+        self.path = self.config.get("path")
+        self.tier = self.config.get("tier", "fast")
+        self.embeddings = self.config.get(
+            "embeddings", {"provider": "synthetic"}
+        )
+
+        if not self.path:
+            storage_dir = get_crewai_storage_dir()
+            self.path = os.path.join(
+                storage_dir, f"openmemory_{self.memory_type}.sqlite"
+            )
+
+    def _initialize_memory(self) -> None:
+        try:
+            from openmemory import (
+                OpenMemory,  # type: ignore[import-untyped,import-not-found]
+            )
+        except ImportError as e:
+            raise ImportError(
+                "OpenMemory is not installed. "
+                "Please install it with `pip install openmemory-py`."
+            ) from e
+
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+
+        self.memory = OpenMemory(
+            mode="local",
+            path=self.path,
+            tier=self.tier,
+            embeddings=self.embeddings,
+        )
+
+    def save(self, value: Any, metadata: dict[str, Any]) -> None:
+        """
+        Save a memory item to OpenMemory.
+
+        Args:
+            value: The content to save.
+            metadata: Additional metadata to associate with the memory.
+        """
+        params: dict[str, Any] = {}
+
+        if self.user_id:
+            params["userId"] = self.user_id
+
+        tags = metadata.pop("tags", None)
+        if tags:
+            params["tags"] = tags
+
+        params["metadata"] = {
+            "type": self.memory_type,
+            **metadata,
+        }
+
+        self.memory.add(str(value), **params)
+
+    def search(
+        self, query: str, limit: int = 5, score_threshold: float = 0.6
+    ) -> list[Any]:
+        """
+        Search for relevant memories in OpenMemory.
+
+        Args:
+            query: The search query.
+            limit: Maximum number of results to return.
+            score_threshold: Minimum similarity score for results.
+
+        Returns:
+            List of matching memory entries with 'content' field.
+        """
+        params: dict[str, Any] = {
+            "k": limit,
+        }
+
+        filters: dict[str, Any] = {}
+        if self.user_id:
+            filters["user_id"] = self.user_id
+
+        if filters:
+            params["filters"] = filters
+
+        results = self.memory.query(query, **params)
+
+        normalized_results = []
+        if isinstance(results, list):
+            for result in results:
+                if isinstance(result, dict):
+                    normalized = dict(result)
+                    if "content" not in normalized and "memory" in normalized:
+                        normalized["content"] = normalized["memory"]
+                    elif "content" not in normalized:
+                        normalized["content"] = str(result)
+
+                    score = normalized.get("score", 1.0)
+                    if score >= score_threshold:
+                        normalized_results.append(normalized)
+                else:
+                    normalized_results.append({"content": str(result)})
+
+        return normalized_results
+
+    def reset(self) -> None:
+        """
+        Reset the OpenMemory storage by closing and deleting the database file.
+        """
+        if hasattr(self, "memory") and self.memory:
+            try:
+                self.memory.close()
+            except Exception:
+                self.memory = None
+
+        if self.path and os.path.exists(self.path):
+            os.remove(self.path)
+
+        self._initialize_memory()

lib/crewai/tests/storage/test_openmemory_storage.py

@@ -0,0 +1,257 @@
+import os
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+class MockOpenMemory:
+    def __init__(self, mode=None, path=None, tier=None, embeddings=None):
+        self.mode = mode
+        self.path = path
+        self.tier = tier
+        self.embeddings = embeddings
+        self._memories: list[dict] = []
+
+    def add(self, content, userId=None, tags=None, metadata=None, **kwargs):
+        memory_id = f"mem_{len(self._memories)}"
+        self._memories.append(
+            {
+                "id": memory_id,
+                "content": content,
+                "userId": userId,
+                "tags": tags,
+                "metadata": metadata,
+            }
+        )
+        return {"id": memory_id, "primarySector": "semantic", "sectors": ["semantic"]}
+
+    def query(self, query, k=10, filters=None, **kwargs):
+        results = []
+        for mem in self._memories:
+            if query.lower() in mem["content"].lower():
+                results.append(
+                    {
+                        "content": mem["content"],
+                        "score": 0.9,
+                        "metadata": mem["metadata"],
+                    }
+                )
+        return results[:k]
+
+    def close(self):
+        pass
+
+
+class MockCrew:
+    def __init__(self):
+        self.agents = [MagicMock(role="Test Agent")]
+
+
+@pytest.fixture
+def mock_openmemory():
+    return MockOpenMemory
+
+
+@pytest.fixture
+def temp_storage_path():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        yield os.path.join(tmpdir, "test_memory.sqlite")
+
+
+@pytest.fixture
+def openmemory_storage_with_mock(mock_openmemory, temp_storage_path):
+    with patch(
+        "crewai.memory.storage.openmemory_storage.OpenMemory", mock_openmemory
+    ):
+        from crewai.memory.storage.openmemory_storage import OpenMemoryStorage
+
+        config = {
+            "path": temp_storage_path,
+            "tier": "fast",
+            "embeddings": {"provider": "synthetic"},
+            "user_id": "test_user",
+        }
+        crew = MockCrew()
+        storage = OpenMemoryStorage(type="external", crew=crew, config=config)
+        return storage
+
+
+def test_openmemory_storage_initialization(openmemory_storage_with_mock):
+    storage = openmemory_storage_with_mock
+    assert storage.memory_type == "external"
+    assert storage.tier == "fast"
+    assert storage.user_id == "test_user"
+    assert storage.embeddings == {"provider": "synthetic"}
+
+
+def test_openmemory_storage_invalid_type():
+    with patch(
+        "crewai.memory.storage.openmemory_storage.OpenMemory", MockOpenMemory
+    ):
+        from crewai.memory.storage.openmemory_storage import OpenMemoryStorage
+
+        with pytest.raises(ValueError, match="Invalid type"):
+            OpenMemoryStorage(type="invalid_type", config={"path": "/tmp/test.sqlite"})
+
+
+def test_openmemory_storage_save(openmemory_storage_with_mock):
+    storage = openmemory_storage_with_mock
+
+    test_value = "This is a test memory about AI agents"
+    test_metadata = {"description": "Test description", "agent": "Test Agent"}
+
+    storage.save(test_value, test_metadata)
+
+    assert len(storage.memory._memories) == 1
+    saved_memory = storage.memory._memories[0]
+    assert saved_memory["content"] == test_value
+    assert saved_memory["userId"] == "test_user"
+    assert saved_memory["metadata"]["type"] == "external"
+    assert saved_memory["metadata"]["description"] == "Test description"
+
+
+def test_openmemory_storage_save_with_tags(openmemory_storage_with_mock):
+    storage = openmemory_storage_with_mock
+
+    test_value = "Memory with tags"
+    test_metadata = {"tags": ["tag1", "tag2"], "description": "Tagged memory"}
+
+    storage.save(test_value, test_metadata)
+
+    saved_memory = storage.memory._memories[0]
+    assert saved_memory["tags"] == ["tag1", "tag2"]
+
+
+def test_openmemory_storage_search(openmemory_storage_with_mock):
+    storage = openmemory_storage_with_mock
+
+    storage.save("Memory about Python programming", {"description": "Python"})
+    storage.save("Memory about JavaScript", {"description": "JavaScript"})
+    storage.save("Memory about Python frameworks", {"description": "Frameworks"})
+
+    results = storage.search("Python", limit=5, score_threshold=0.5)
+
+    assert len(results) == 2
+    assert all("content" in r for r in results)
+    assert any("Python programming" in r["content"] for r in results)
+    assert any("Python frameworks" in r["content"] for r in results)
+
+
+def test_openmemory_storage_search_with_score_threshold(openmemory_storage_with_mock):
+    storage = openmemory_storage_with_mock
+
+    storage.save("Test memory", {"description": "Test"})
+
+    results = storage.search("Test", limit=5, score_threshold=0.95)
+    assert len(results) == 0
+
+    results = storage.search("Test", limit=5, score_threshold=0.5)
+    assert len(results) == 1
+
+
+def test_openmemory_storage_reset(temp_storage_path):
+    with patch(
+        "crewai.memory.storage.openmemory_storage.OpenMemory", MockOpenMemory
+    ):
+        from crewai.memory.storage.openmemory_storage import OpenMemoryStorage
+
+        config = {
+            "path": temp_storage_path,
+            "tier": "fast",
+            "embeddings": {"provider": "synthetic"},
+        }
+        storage = OpenMemoryStorage(type="external", config=config)
+
+        storage.save("Test memory", {"description": "Test"})
+        assert len(storage.memory._memories) == 1
+
+        storage.reset()
+
+        assert len(storage.memory._memories) == 0
+
+
+def test_openmemory_storage_default_path():
+    with patch(
+        "crewai.memory.storage.openmemory_storage.OpenMemory", MockOpenMemory
+    ):
+        from crewai.memory.storage.openmemory_storage import OpenMemoryStorage
+
+        config = {
+            "tier": "fast",
+            "embeddings": {"provider": "synthetic"},
+        }
+        storage = OpenMemoryStorage(type="external", config=config)
+
+        assert storage.path is not None
+        assert "openmemory_external.sqlite" in storage.path
+
+
+def test_openmemory_storage_different_memory_types():
+    with patch(
+        "crewai.memory.storage.openmemory_storage.OpenMemory", MockOpenMemory
+    ):
+        from crewai.memory.storage.openmemory_storage import OpenMemoryStorage
+
+        for memory_type in ["short_term", "long_term", "entities", "external"]:
+            config = {
+                "tier": "fast",
+                "embeddings": {"provider": "synthetic"},
+            }
+            storage = OpenMemoryStorage(type=memory_type, config=config)
+            assert storage.memory_type == memory_type
+            assert f"openmemory_{memory_type}.sqlite" in storage.path
+
+
+def test_openmemory_import_error():
+    import sys
+    import importlib
+
+    original_modules = sys.modules.copy()
+    sys.modules["openmemory"] = None  # type: ignore[assignment]
+
+    try:
+        if "crewai.memory.storage.openmemory_storage" in sys.modules:
+            del sys.modules["crewai.memory.storage.openmemory_storage"]
+
+        with pytest.raises(ImportError, match="OpenMemory is not installed"):
+            from crewai.memory.storage import openmemory_storage
+
+            importlib.reload(openmemory_storage)
+            openmemory_storage.OpenMemoryStorage(
+                type="external", config={"path": "/tmp/test.sqlite"}
+            )
+    finally:
+        sys.modules.update(original_modules)
+
+
+def test_external_memory_openmemory_provider():
+    with patch(
+        "crewai.memory.storage.openmemory_storage.OpenMemory", MockOpenMemory
+    ):
+        from crewai.memory.external.external_memory import ExternalMemory
+
+        supported = ExternalMemory.external_supported_storages()
+        assert "openmemory" in supported
+
+
+def test_external_memory_create_openmemory_storage(temp_storage_path):
+    with patch(
+        "crewai.memory.storage.openmemory_storage.OpenMemory", MockOpenMemory
+    ):
+        from crewai.memory.external.external_memory import ExternalMemory
+
+        crew = MockCrew()
+        embedder_config = {
+            "provider": "openmemory",
+            "config": {
+                "path": temp_storage_path,
+                "tier": "fast",
+                "embeddings": {"provider": "synthetic"},
+            },
+        }
+
+        storage = ExternalMemory.create_storage(crew, embedder_config)
+
+        assert storage is not None
+        assert storage.memory_type == "external"