Description
The current implementation of Mem0ContextProvider fails to retrieve stored memories. There are two distinct bugs in how the provider interacts with the Mem0 API, causing before_run to consistently return an empty list [].
🚨 Bug 1: The app_id vs metadata Mismatch
In after_run, the application ID is incorrectly stuffed into a custom metadata dictionary instead of using Mem0's native top-level parameter. However, during retrieval (_build_filters), the code searches for the native top-level app_id.
Where it breaks:
# In after_run (Saving):
metadata={"application_id": self.application_id} # Stored in metadata
# In _build_filters (Retrieval):
filters["app_id"] = self.application_id # Searches top-level
Because the top-level app_id was never set during insertion, Mem0 filters out all records and returns nothing.
🚨 Bug 2: The Entity Isolation "AND" Trap
When initialized with both user_id and agent_id, the _build_filters method bundles both into a single filters dictionary.
Mem0's LLM extraction engine parses conversations and assigns facts to either the user bucket or the agent bucket—it does not tag single records with both. By passing a bundled dictionary, the provider is forcing a strict logical AND intersection (user_id == "user_1" AND agent_id == "agent_1"). Since Mem0 isolates these entities, this intersection is always empty.
Proposed Solution
Fix Retrieval (before_run):
user_id and agent_id buckets must be queried independently and their results merged.
Code Sample
# ==========================================
# 1. Proposed Fix for ingestion (after_run)
# ==========================================
# ... inside after_run ...
if messages:
await self.mem0_client.add(
messages=messages,
user_id=self.user_id,
agent_id=self.agent_id,
app_id=self.application_id, # Fixed: Map application_id to native app_id parameter
)
# ==========================================
# 2. Proposed Fix for retrieval (before_run)
# ==========================================
async def before_run(
self,
*,
agent: SupportsAgentRun,
session: AgentSession,
context: SessionContext,
state: dict[str, Any],
) -> None:
"""Search Mem0 for relevant memories and add to the session context."""
self._validate_filters()
input_text = "\n".join(msg.text for msg in context.input_messages if msg and msg.text and msg.text.strip())
if not input_text.strip():
return
# Query entity partitions independently to bypass strict logical AND limitations
search_tasks = []
if not self.application_id:
return
# Query User partition independently
if self.user_id:
search_tasks.append(self.mem0_client.search(query=input_text, filters={"user_id" : self.user_id, "app_id" : self.application_id}, top_k=5, threshold=0.25))
# Query Agent partition independently
if self.agent_id:
search_tasks.append(self.mem0_client.search(query=input_text, filters={"agent_id" : self.agent_id, "app_id" : self.application_id}, top_k=5, threshold=0.25))
if not search_tasks:
return
results = await asyncio.gather(*search_tasks, return_exceptions=True)
# Merge and deduplicate results
memories = []
seen_memory_ids = set()
for search_response in results:
if isinstance(search_response, Exception):
continue
current_memories = []
if isinstance(search_response, list):
current_memories = search_response
elif isinstance(search_response, dict) and "results" in search_response:
current_memories = search_response["results"]
else:
current_memories = [search_response]
for mem in current_memories:
mem_id = mem.get("id")
if mem_id and mem_id not in seen_memory_ids:
seen_memory_ids.add(mem_id)
memories.append(mem)
line_separated_memories = "\n".join(memory.get("memory", "") for memory in memories)
if line_separated_memories:
context.extend_messages(
self.source_id,
[Message(role="user", contents=[f"{self.context_prompt}\n{line_separated_memories}"])],
)Error Messages / Stack Traces
Package Versions
agent-framework: 1.7.0, agent-framework-mem0: 1.0.0b260521
Python Version
Python 3.13.12
Additional Context
How mem0 stores memories :
Description
The current implementation of
Mem0ContextProviderfails to retrieve stored memories. There are two distinct bugs in how the provider interacts with the Mem0 API, causingbefore_runto consistently return an empty list[].🚨 Bug 1: The
app_idvsmetadataMismatchIn
after_run, the application ID is incorrectly stuffed into a custom metadata dictionary instead of using Mem0's native top-level parameter. However, during retrieval (_build_filters), the code searches for the native top-levelapp_id.Where it breaks:
Because the top-level
app_idwas never set during insertion, Mem0 filters out all records and returns nothing.🚨 Bug 2: The Entity Isolation "AND" Trap
When initialized with both
user_idandagent_id, the_build_filtersmethod bundles both into a single filters dictionary.Mem0's LLM extraction engine parses conversations and assigns facts to either the user bucket or the agent bucket—it does not tag single records with both. By passing a bundled dictionary, the provider is forcing a strict logical AND intersection (
user_id == "user_1" AND agent_id == "agent_1"). Since Mem0 isolates these entities, this intersection is always empty.Proposed Solution
Fix Retrieval (
before_run):user_idandagent_idbuckets must be queried independently and their results merged.Code Sample
Error Messages / Stack Traces
Package Versions
agent-framework: 1.7.0, agent-framework-mem0: 1.0.0b260521
Python Version
Python 3.13.12
Additional Context
How mem0 stores memories :