feat: support defining any memory in an isolated way

This change makes it easier to use a specific memory type without unintentionally enabling all others.

Previously, setting memory=True would implicitly configure all available memories (like LTM and STM), which might not be ideal in all cases. For example, when building a chatbot that only needs an external memory, users were forced to also configure LTM and STM — which rely on default OpenAPI embeddings — even if they weren’t needed.

With this update, users can now define a single memory in isolation, making the configuration process simpler and more flexible.
This commit is contained in:
Lucas Gomide
2025-04-12 15:18:48 -03:00
parent 6a1eb10830
commit 9eeed380a8
3 changed files with 71 additions and 38 deletions

View File

@@ -156,6 +156,23 @@ class Agent(BaseAgent):
except (TypeError, ValueError) as e:
raise ValueError(f"Invalid Knowledge Configuration: {str(e)}")
def _is_any_available_memory(self) -> bool:
"""Check if any memory is available."""
if not self.crew:
return False
memory_attributes = [
"memory",
"memory_config",
"_short_term_memory",
"_long_term_memory",
"_entity_memory",
"_user_memory",
"_external_memory",
]
return any(getattr(self.crew, attr) for attr in memory_attributes)
def execute_task(
self,
task: Task,
@@ -200,7 +217,7 @@ class Agent(BaseAgent):
task=task_prompt, context=context
)
if self.crew and self.crew.memory:
if self._is_any_available_memory():
contextual_memory = ContextualMemory(
self.crew.memory_config,
self.crew._short_term_memory,

View File

@@ -275,46 +275,51 @@ class Crew(BaseModel):
return self
def _initialize_user_memory(self):
if (
self.memory_config
and "user_memory" in self.memory_config
and self.memory_config.get("provider") == "mem0"
): # Check for user_memory in config
user_memory_config = self.memory_config["user_memory"]
if isinstance(
user_memory_config, dict
): # Check if it's a configuration dict
self._user_memory = UserMemory(crew=self)
else:
raise TypeError("user_memory must be a configuration dictionary")
def _initialize_default_memories(self):
self._long_term_memory = self._long_term_memory or LongTermMemory()
self._short_term_memory = self._short_term_memory or ShortTermMemory(
crew=self,
embedder_config=self.embedder,
)
self._entity_memory = self.entity_memory or EntityMemory(
crew=self, embedder_config=self.embedder
)
@model_validator(mode="after")
def create_crew_memory(self) -> "Crew":
"""Set private attributes."""
"""Initialize private memory attributes."""
self._external_memory = (
# External memory doesnt support a default value since it was designed to be managed entirely externally
self.external_memory.set_crew(self)
if self.external_memory
else None
)
self._long_term_memory = self.long_term_memory
self._short_term_memory = self.short_term_memory
self._entity_memory = self.entity_memory
# UserMemory is gonna to be deprecated in the future, but we have to initialize a default value for now
self._user_memory = None
if self.memory:
self._long_term_memory = (
self.long_term_memory if self.long_term_memory else LongTermMemory()
)
self._short_term_memory = (
self.short_term_memory
if self.short_term_memory
else ShortTermMemory(
crew=self,
embedder_config=self.embedder,
)
)
self._entity_memory = (
self.entity_memory
if self.entity_memory
else EntityMemory(crew=self, embedder_config=self.embedder)
)
self._external_memory = (
# External memory doesnt support a default value since it was designed to be managed entirely externally
self.external_memory.set_crew(self)
if self.external_memory
else None
)
if (
self.memory_config
and "user_memory" in self.memory_config
and self.memory_config.get("provider") == "mem0"
): # Check for user_memory in config
user_memory_config = self.memory_config["user_memory"]
if isinstance(
user_memory_config, dict
): # Check if it's a configuration dict
self._user_memory = UserMemory(crew=self)
else:
raise TypeError("user_memory must be a configuration dictionary")
else:
self._user_memory = None # No user memory if not in config
self._initialize_default_memories()
self._initialize_user_memory()
return self
@model_validator(mode="after")

View File

@@ -53,6 +53,10 @@ class ContextualMemory:
Fetches recent relevant insights from STM related to the task's description and expected_output,
formatted as bullet points.
"""
if self.stm is None:
return ""
stm_results = self.stm.search(query)
formatted_results = "\n".join(
[
@@ -67,6 +71,10 @@ class ContextualMemory:
Fetches historical data or insights from LTM that are relevant to the task's description and expected_output,
formatted as bullet points.
"""
if self.ltm is None:
return ""
ltm_results = self.ltm.search(task, latest_n=2)
if not ltm_results:
return None
@@ -86,6 +94,9 @@ class ContextualMemory:
Fetches relevant entity information from Entity Memory related to the task's description and expected_output,
formatted as bullet points.
"""
if self.em is None:
return ""
em_results = self.em.search(query)
formatted_results = "\n".join(
[