Introduce MemoryEvents to monitor their usage (#3098)

* feat: emit events about memory usage

* test: add tests about memory events usage

* fixed linter issues

* test: use scoped_handlers while listener Memory events

src/crewai/memory/entity/entity_memory.py

@@ -1,10 +1,20 @@
 from typing import Optional
+import time
 
 from pydantic import PrivateAttr
 
 from crewai.memory.entity.entity_memory_item import EntityMemoryItem
 from crewai.memory.memory import Memory
 from crewai.memory.storage.rag_storage import RAGStorage
+from crewai.utilities.events.crewai_event_bus import crewai_event_bus
+from crewai.utilities.events.memory_events import (
+    MemoryQueryStartedEvent,
+    MemoryQueryCompletedEvent,
+    MemoryQueryFailedEvent,
+    MemorySaveStartedEvent,
+    MemorySaveCompletedEvent,
+    MemorySaveFailedEvent,
+)
 
 
 class EntityMemory(Memory):
@@ -48,16 +58,96 @@ class EntityMemory(Memory):
 
     def save(self, item: EntityMemoryItem) -> None:  # type: ignore # BUG?: Signature of "save" incompatible with supertype "Memory"
         """Saves an entity item into the SQLite storage."""
-        if self._memory_provider == "mem0":
-            data = f"""
-            Remember details about the following entity:
-            Name: {item.name}
-            Type: {item.type}
-            Entity Description: {item.description}
-            """
-        else:
-            data = f"{item.name}({item.type}): {item.description}"
-        super().save(data, item.metadata)
+        crewai_event_bus.emit(
+            self,
+            event=MemorySaveStartedEvent(
+                metadata=item.metadata,
+                source_type="entity_memory",
+            ),
+        )
+
+        start_time = time.time()
+        try:
+            if self._memory_provider == "mem0":
+                data = f"""
+                Remember details about the following entity:
+                Name: {item.name}
+                Type: {item.type}
+                Entity Description: {item.description}
+                """
+            else:
+                data = f"{item.name}({item.type}): {item.description}"
+
+            super().save(data, item.metadata)
+
+            # Emit memory save completed event
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveCompletedEvent(
+                    value=data,
+                    metadata=item.metadata,
+                    save_time_ms=(time.time() - start_time) * 1000,
+                    source_type="entity_memory",
+                ),
+            )
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveFailedEvent(
+                    metadata=item.metadata,
+                    error=str(e),
+                    source_type="entity_memory",
+                ),
+            )
+            raise
+
+    def search(
+        self,
+        query: str,
+        limit: int = 3,
+        score_threshold: float = 0.35,
+    ):
+        crewai_event_bus.emit(
+            self,
+            event=MemoryQueryStartedEvent(
+                query=query,
+                limit=limit,
+                score_threshold=score_threshold,
+                source_type="entity_memory",
+            ),
+        )
+
+        start_time = time.time()
+        try:
+            results = super().search(
+                query=query, limit=limit, score_threshold=score_threshold
+            )
+
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryCompletedEvent(
+                    query=query,
+                    results=results,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    query_time_ms=(time.time() - start_time) * 1000,
+                    source_type="entity_memory",
+                ),
+            )
+
+            return results
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryFailedEvent(
+                    query=query,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    error=str(e),
+                    source_type="entity_memory",
+                ),
+            )
+            raise
 
     def reset(self) -> None:
         try:

src/crewai/memory/external/external_memory.py

@@ -1,8 +1,18 @@
 from typing import TYPE_CHECKING, Any, Dict, Optional
+import time
 
 from crewai.memory.external.external_memory_item import ExternalMemoryItem
 from crewai.memory.memory import Memory
 from crewai.memory.storage.interface import Storage
+from crewai.utilities.events.crewai_event_bus import crewai_event_bus
+from crewai.utilities.events.memory_events import (
+    MemoryQueryStartedEvent,
+    MemoryQueryCompletedEvent,
+    MemoryQueryFailedEvent,
+    MemorySaveStartedEvent,
+    MemorySaveCompletedEvent,
+    MemorySaveFailedEvent,
+)
 
 if TYPE_CHECKING:
     from crewai.memory.storage.mem0_storage import Mem0Storage
@@ -46,8 +56,91 @@ class ExternalMemory(Memory):
         agent: Optional[str] = None,
     ) -> None:
         """Saves a value into the external storage."""
-        item = ExternalMemoryItem(value=value, metadata=metadata, agent=agent)
-        super().save(value=item.value, metadata=item.metadata, agent=item.agent)
+        crewai_event_bus.emit(
+            self,
+            event=MemorySaveStartedEvent(
+                value=value,
+                metadata=metadata,
+                agent_role=agent,
+                source_type="external_memory",
+            ),
+        )
+
+        start_time = time.time()
+        try:
+            item = ExternalMemoryItem(value=value, metadata=metadata, agent=agent)
+            super().save(value=item.value, metadata=item.metadata, agent=item.agent)
+
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveCompletedEvent(
+                    value=value,
+                    metadata=metadata,
+                    agent_role=agent,
+                    save_time_ms=(time.time() - start_time) * 1000,
+                    source_type="external_memory",
+                ),
+            )
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveFailedEvent(
+                    value=value,
+                    metadata=metadata,
+                    agent_role=agent,
+                    error=str(e),
+                    source_type="external_memory",
+                ),
+            )
+            raise
+
+    def search(
+        self,
+        query: str,
+        limit: int = 3,
+        score_threshold: float = 0.35,
+    ):
+        crewai_event_bus.emit(
+            self,
+            event=MemoryQueryStartedEvent(
+                query=query,
+                limit=limit,
+                score_threshold=score_threshold,
+                source_type="external_memory",
+            ),
+        )
+
+        start_time = time.time()
+        try:
+            results = super().search(
+                query=query, limit=limit, score_threshold=score_threshold
+            )
+
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryCompletedEvent(
+                    query=query,
+                    results=results,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    query_time_ms=(time.time() - start_time) * 1000,
+                    source_type="external_memory",
+                ),
+            )
+
+            return results
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryFailedEvent(
+                    query=query,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    error=str(e),
+                    source_type="external_memory",
+                ),
+            )
+            raise
 
     def reset(self) -> None:
         self.storage.reset()

src/crewai/memory/long_term/long_term_memory.py

@@ -1,7 +1,17 @@
 from typing import Any, Dict, List
+import time
 
 from crewai.memory.long_term.long_term_memory_item import LongTermMemoryItem
 from crewai.memory.memory import Memory
+from crewai.utilities.events.crewai_event_bus import crewai_event_bus
+from crewai.utilities.events.memory_events import (
+    MemoryQueryStartedEvent,
+    MemoryQueryCompletedEvent,
+    MemoryQueryFailedEvent,
+    MemorySaveStartedEvent,
+    MemorySaveCompletedEvent,
+    MemorySaveFailedEvent,
+)
 from crewai.memory.storage.ltm_sqlite_storage import LTMSQLiteStorage
 
 
@@ -20,17 +30,87 @@ class LongTermMemory(Memory):
         super().__init__(storage=storage)
 
     def save(self, item: LongTermMemoryItem) -> None:  # type: ignore # BUG?: Signature of "save" incompatible with supertype "Memory"
-        metadata = item.metadata
-        metadata.update({"agent": item.agent, "expected_output": item.expected_output})
-        self.storage.save(  # type: ignore # BUG?: Unexpected keyword argument "task_description","score","datetime" for "save" of "Storage"
-            task_description=item.task,
-            score=metadata["quality"],
-            metadata=metadata,
-            datetime=item.datetime,
+        crewai_event_bus.emit(
+            self,
+            event=MemorySaveStartedEvent(
+                value=item.task,
+                metadata=item.metadata,
+                agent_role=item.agent,
+                source_type="long_term_memory",
+            ),
         )
 
+        start_time = time.time()
+        try:
+            metadata = item.metadata
+            metadata.update({"agent": item.agent, "expected_output": item.expected_output})
+            self.storage.save(  # type: ignore # BUG?: Unexpected keyword argument "task_description","score","datetime" for "save" of "Storage"
+                task_description=item.task,
+                score=metadata["quality"],
+                metadata=metadata,
+                datetime=item.datetime,
+            )
+
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveCompletedEvent(
+                    value=item.task,
+                    metadata=item.metadata,
+                    agent_role=item.agent,
+                    save_time_ms=(time.time() - start_time) * 1000,
+                    source_type="long_term_memory",
+                ),
+            )
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveFailedEvent(
+                    value=item.task,
+                    metadata=item.metadata,
+                    agent_role=item.agent,
+                    error=str(e),
+                    source_type="long_term_memory",
+                ),
+            )
+            raise
+
     def search(self, task: str, latest_n: int = 3) -> List[Dict[str, Any]]:  # type: ignore # signature of "search" incompatible with supertype "Memory"
-        return self.storage.load(task, latest_n)  # type: ignore # BUG?: "Storage" has no attribute "load"
+        crewai_event_bus.emit(
+            self,
+            event=MemoryQueryStartedEvent(
+                query=task,
+                limit=latest_n,
+                source_type="long_term_memory",
+            ),
+        )
+
+        start_time = time.time()
+        try:
+            results = self.storage.load(task, latest_n)  # type: ignore # BUG?: "Storage" has no attribute "load"
+
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryCompletedEvent(
+                    query=task,
+                    results=results,
+                    limit=latest_n,
+                    query_time_ms=(time.time() - start_time) * 1000,
+                    source_type="long_term_memory",
+                ),
+            )
+
+            return results
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryFailedEvent(
+                    query=task,
+                    limit=latest_n,
+                    error=str(e),
+                    source_type="long_term_memory",
+                ),
+            )
+            raise
 
     def reset(self) -> None:
         self.storage.reset()

src/crewai/memory/short_term/short_term_memory.py

@@ -1,10 +1,20 @@
 from typing import Any, Dict, Optional
+import time
 
 from pydantic import PrivateAttr
 
 from crewai.memory.memory import Memory
 from crewai.memory.short_term.short_term_memory_item import ShortTermMemoryItem
 from crewai.memory.storage.rag_storage import RAGStorage
+from crewai.utilities.events.crewai_event_bus import crewai_event_bus
+from crewai.utilities.events.memory_events import (
+    MemoryQueryStartedEvent,
+    MemoryQueryCompletedEvent,
+    MemoryQueryFailedEvent,
+    MemorySaveStartedEvent,
+    MemorySaveCompletedEvent,
+    MemorySaveFailedEvent,
+)
 
 
 class ShortTermMemory(Memory):
@@ -52,11 +62,46 @@ class ShortTermMemory(Memory):
         metadata: Optional[Dict[str, Any]] = None,
         agent: Optional[str] = None,
     ) -> None:
-        item = ShortTermMemoryItem(data=value, metadata=metadata, agent=agent)
-        if self._memory_provider == "mem0":
-            item.data = f"Remember the following insights from Agent run: {item.data}"
+        crewai_event_bus.emit(
+            self,
+            event=MemorySaveStartedEvent(
+                value=value,
+                metadata=metadata,
+                agent_role=agent,
+                source_type="short_term_memory",
+            ),
+        )
 
-        super().save(value=item.data, metadata=item.metadata, agent=item.agent)
+        start_time = time.time()
+        try:
+            item = ShortTermMemoryItem(data=value, metadata=metadata, agent=agent)
+            if self._memory_provider == "mem0":
+                item.data = f"Remember the following insights from Agent run: {item.data}"
+
+            super().save(value=item.data, metadata=item.metadata, agent=item.agent)
+
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveCompletedEvent(
+                    value=value,
+                    metadata=metadata,
+                    agent_role=agent,
+                    save_time_ms=(time.time() - start_time) * 1000,
+                    source_type="short_term_memory",
+                ),
+            )
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemorySaveFailedEvent(
+                    value=value,
+                    metadata=metadata,
+                    agent_role=agent,
+                    error=str(e),
+                    source_type="short_term_memory",
+                ),
+            )
+            raise
 
     def search(
         self,
@@ -64,9 +109,47 @@ class ShortTermMemory(Memory):
         limit: int = 3,
         score_threshold: float = 0.35,
     ):
-        return self.storage.search(
-            query=query, limit=limit, score_threshold=score_threshold
-        )  # type: ignore # BUG? The reference is to the parent class, but the parent class does not have this parameters
+        crewai_event_bus.emit(
+            self,
+            event=MemoryQueryStartedEvent(
+                query=query,
+                limit=limit,
+                score_threshold=score_threshold,
+                source_type="short_term_memory",
+            ),
+        )
+
+        start_time = time.time()
+        try:
+            results = self.storage.search(
+                query=query, limit=limit, score_threshold=score_threshold
+            )  # type: ignore # BUG? The reference is to the parent class, but the parent class does not have this parameters
+
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryCompletedEvent(
+                    query=query,
+                    results=results,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    query_time_ms=(time.time() - start_time) * 1000,
+                    source_type="short_term_memory",
+                ),
+            )
+
+            return results
+        except Exception as e:
+            crewai_event_bus.emit(
+                self,
+                event=MemoryQueryFailedEvent(
+                    query=query,
+                    limit=limit,
+                    score_threshold=score_threshold,
+                    error=str(e),
+                    source_type="short_term_memory",
+                ),
+            )
+            raise
 
     def reset(self) -> None:
         try: