Repository: mem0ai/mem0
Pull request #2640 (2025-05-06 19:58:33)

Update Client
## Description

Update Client

Fixes # (issue)

## Type of change

Please delete options that are not relevant.

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] Refactor (does not change functionality, e.g. code style improvements, linting)
- [ ] Documentation update

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration

Please delete options that are not relevant.

- [ ] Unit Test
- [ ] Test Script (please provide)

## Checklist:

- [ ] My code follows the style guidelines of this project
- [ ] I have performed a self-review of my own code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream modules
- [ ] I have checked my code and corrected any misspellings

## Maintainer Checklist

- [ ] closes #xxxx (Replace xxxx with the GitHub issue number)
- [ ] Made sure Checks passed



Patch:
diff --git a/docs/platform/quickstart.mdx b/docs/platform/quickstart.mdx
index 89bca87d..b7791774 100644
--- a/docs/platform/quickstart.mdx
+++ b/docs/platform/quickstart.mdx
@@ -269,10 +269,12 @@ curl -X POST "https://api.mem0.ai/v1/memories/" \
 </Note>
 
 #### Long-term memory for both users and agents
-When you provide both `user_id` and `agent_id`, Mem0 will store memories with both identifiers attached:
-- Each memory will be tagged with both the specified `user_id` and `agent_id`
-- During retrieval, you'll need to provide both IDs to access the memories
-- This enables tracking the full context of conversations between specific users and agents
+When you provide both `user_id` and `agent_id`, Mem0 will store memories for both identifiers separately:
+- Memories from messages with `"role": "user"` are automatically tagged with the provided `user_id`
+- Memories from messages with `"role": "assistant"` are automatically tagged with the provided `agent_id`
+- During retrieval, you can provide either `user_id` or `agent_id` to access the respective memories
+- You can continuously enrich existing memory collections by adding new memories to the same `user_id` or `agent_id` in subsequent API calls, either together or separately, allowing for progressive memory building over time
+- This dual-tagging approach enables personalized experiences for both users and AI agents in your application
 
 <CodeGroup>
 
@@ -314,11 +316,13 @@ curl -X POST "https://api.mem0.ai/v1/memories/" \
 {
     "results": [
         {
+            // memory from user1
             "id": "c57abfa2-f0ac-48af-896a-21728dbcecee0",
             "data": {"memory": "Travelling to San Francisco"},
             "event": "ADD"
         },
         {
+            // memory from agent1
             "id": "0e8c003f-7db7-426a-9fdc-a46f9331a0c2",
             "data": {"memory": "Going to Dubai next month"},
             "event": "ADD"
diff --git a/mem0/client/main.py b/mem0/client/main.py
index 35b25d10..891e64a0 100644
--- a/mem0/client/main.py
+++ b/mem0/client/main.py
@@ -6,6 +6,7 @@ from functools import wraps
 from typing import Any, Dict, List, Optional, Union
 
 import httpx
+import requests
 
 from mem0.memory.setup import get_user_id, setup_config
 from mem0.memory.telemetry import capture_client_event
@@ -62,6 +63,7 @@ class MemoryClient:
         host: Optional[str] = None,
         org_id: Optional[str] = None,
         project_id: Optional[str] = None,
+        client: Optional[httpx.Client] = None,
     ):
         """Initialize the MemoryClient.
 
@@ -71,6 +73,8 @@ class MemoryClient:
             host: The base URL for the Mem0 API. Defaults to "https://api.mem0.ai".
             org_id: The ID of the organization.
             project_id: The ID of the project.
+            client: A custom httpx.Client instance. If provided, it will be used instead of creating a new one.
+                   Note that base_url and headers will be set/overridden as needed.
 
         Raises:
             ValueError: If no API key is provided or found in the environment.
@@ -87,11 +91,20 @@ class MemoryClient:
         # Create MD5 hash of API key for user_id
         self.user_id = hashlib.md5(self.api_key.encode()).hexdigest()
 
-        self.client = httpx.Client(
-            base_url=self.host,
-            headers={"Authorization": f"Token {self.api_key}", "Mem0-User-ID": self.user_id},
-            timeout=300,
-        )
+        if client is not None:
+            self.client = client
+            # Ensure the client has the correct base_url and headers
+            self.client.base_url = httpx.URL(self.host)
+            self.client.headers.update({
+                "Authorization": f"Token {self.api_key}",
+                "Mem0-User-ID": self.user_id
+            })
+        else:
+            self.client = httpx.Client(
+                base_url=self.host,
+                headers={"Authorization": f"Token {self.api_key}", "Mem0-User-ID": self.user_id},
+                timeout=300,
+            )
         self.user_email = self._validate_api_key()
         capture_client_event("client.init", self, {"sync_type": "sync"})
 
@@ -696,10 +709,6 @@ class AsyncMemoryClient:
 
     This class provides asynchronous versions of all MemoryClient methods.
     It uses httpx.AsyncClient for making non-blocking API requests.
-
-    Attributes:
-        sync_client (MemoryClient): Underlying synchronous client instance.
-        async_client (httpx.AsyncClient): Async HTTP client for making API requests.
     """
 
     def __init__(
@@ -708,13 +717,121 @@ class AsyncMemoryClient:
         host: Optional[str] = None,
         org_id: Optional[str] = None,
         project_id: Optional[str] = None,
+        client: Optional[httpx.AsyncClient] = None,
     ):
-        self.sync_client = MemoryClient(api_key, host, org_id, project_id)
-        self.async_client = httpx.AsyncClient(
-            base_url=self.sync_client.host,
-            headers=self.sync_client.client.headers,
-            timeout=300,
-        )
+        """Initialize the AsyncMemoryClient.
+
+        Args:
+            api_key: The API key for authenticating with the Mem0 API. If not provided,
+                     it will attempt to use the MEM0_API_KEY environment variable.
+            host: The base URL for the Mem0 API. Defaults to "https://api.mem0.ai".
+            org_id: The ID of the organization.
+            project_id: The ID of the project.
+            client: A custom httpx.AsyncClient instance. If provided, it will be used instead
+                   of creating a new one. Note that base_url and headers will be set/overridden
+                   as needed.
+
+        Raises:
+            ValueError: If no API key is provided or found in the environment.
+        """
+        self.api_key = api_key or os.getenv("MEM0_API_KEY")
+        self.host = host or "https://api.mem0.ai"
+        self.org_id = org_id
+        self.project_id = project_id
+        self.user_id = get_user_id()
+
+        if not self.api_key:
+            raise ValueError("Mem0 API Key not provided. Please provide an API Key.")
+
+        # Create MD5 hash of API key for user_id
+        self.user_id = hashlib.md5(self.api_key.encode()).hexdigest()
+
+        if client is not None:
+            self.async_client = client
+            # Ensure the client has the correct base_url and headers
+            self.async_client.base_url = httpx.URL(self.host)
+            self.async_client.headers.update({
+                "Authorization": f"Token {self.api_key}",
+                "Mem0-User-ID": self.user_id
+            })
+        else:
+            self.async_client = httpx.AsyncClient(
+                base_url=self.host,
+                headers={"Authorization": f"Token {self.api_key}", "Mem0-User-ID": self.user_id},
+                timeout=300,
+            )
+
+        self.user_email = self._validate_api_key()
+        capture_client_event("client.init", self, {"sync_type": "async"})
+
+    def _validate_api_key(self):
+        """Validate the API key by making a test request."""
+        try:
+            params = self._prepare_params()
+            response = requests.get(f"{self.host}/v1/ping/", headers={"Authorization": f"Token {self.api_key}", "Mem0-User-ID": self.user_id}, params=params)
+            data = response.json()
+
+            response.raise_for_status()
+
+            if data.get("org_id") and data.get("project_id"):
+                self.org_id = data.get("org_id")
+                self.project_id = data.get("project_id")
+
+            return data.get("user_email")
+
+        except requests.HTTPStatusError as e:
+            try:
+                error_data = e.response.json()
+                error_message = error_data.get("detail", str(e))
+            except Exception:
+                error_message = str(e)
+            raise ValueError(f"Error: {error_message}")
+
+    def _prepare_payload(
+        self, messages: Union[str, List[Dict[str, str]], None], kwargs: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Prepare the payload for API requests.
+
+        Args:
+            messages: The messages to include in the payload.
+            kwargs: Additional keyword arguments to include in the payload.
+
+        Returns:
+            A dictionary containing the prepared payload.
+        """
+        payload = {}
+        if isinstance(messages, str):
+            payload["messages"] = [{"role": "user", "content": messages}]
+        elif isinstance(messages, list):
+            payload["messages"] = messages
+
+        payload.update({k: v for k, v in kwargs.items() if v is not None})
+        return payload
+
+    def _prepare_params(self, kwargs: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """Prepare query parameters for API requests.
+
+        Args:
+            kwargs: Keyword arguments to include in the parameters.
+
+        Returns:
+            A dictionary containing the prepared parameters.
+
+        Raises:
+            ValueError: If either org_id or project_id is provided but not both.
+        """
+
+        if kwargs is None:
+            kwargs = {}
+
+        # Add org_id and project_id if both are available
+        if self.org_id and self.project_id:
+            kwargs["org_id"] = self.org_id
+            kwargs["project_id"] = self.project_id
+        elif self.org_id or self.project_id:
+            raise ValueError("Please provide both org_id and project_id")
+
+        return {k: v for k, v in kwargs.items() if v is not None}
 
     async def __aenter__(self):
         return self
@@ -724,89 +841,102 @@ class AsyncMemoryClient:
 
     @api_error_handler
     async def add(self, messages: Union[str, List[Dict[str, str]]], **kwargs) -> Dict[str, Any]:
-        kwargs = self.sync_client._prepare_params(kwargs)
-        payload = self.sync_client._prepare_payload(messages, kwargs)
+        kwargs = self._prepare_params(kwargs)
+        if kwargs.get("output_format") != "v1.1":
+            kwargs["output_format"] = "v1.1"
+            warnings.warn(
+                "output_format='v1.0' is deprecated therefore setting it to 'v1.1' by default."
+                "Check out the docs for more information: https://docs.mem0.ai/platform/quickstart#4-1-create-memories",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        kwargs["version"] = "v2"
+        payload = self._prepare_payload(messages, kwargs)
         response = await self.async_client.post("/v1/memories/", json=payload)
         response.raise_for_status()
         if "metadata" in kwargs:
             del kwargs["metadata"]
-        capture_client_event("client.add", self.sync_client, {"keys": list(kwargs.keys()), "sync_type": "async"})
+        capture_client_event("client.add", self, {"keys": list(kwargs.keys()), "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def get(self, memory_id: str) -> Dict[str, Any]:
-        params = self.sync_client._prepare_params()
+        params = self._prepare_params()
         response = await self.async_client.get(f"/v1/memories/{memory_id}/", params=params)
         response.raise_for_status()
-        capture_client_event("client.get", self.sync_client, {"memory_id": memory_id, "sync_type": "async"})
+        capture_client_event("client.get", self, {"memory_id": memory_id, "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def get_all(self, version: str = "v1", **kwargs) -> List[Dict[str, Any]]:
-        params = self.sync_client._prepare_params(kwargs)
+        params = self._prepare_params(kwargs)
         if version == "v1":
             response = await self.async_client.get(f"/{version}/memories/", params=params)
         elif version == "v2":
-            response = await self.async_client.post(f"/{version}/memories/", json=params)
+            if "page" in params and "page_size" in params:
+                query_params = {"page": params.pop("page"), "page_size": params.pop("page_size")}
+                response = await self.async_client.post(f"/{version}/memories/", json=params, params=query_params)
+            else:
+                response = await self.async_client.post(f"/{version}/memories/", json=params)
         response.raise_for_status()
         if "metadata" in kwargs:
             del kwargs["metadata"]
         capture_client_event(
-            "client.get_all", self.sync_client, {"api_version": version, "keys": list(kwargs.keys()), "sync_type": "async"}
+            "client.get_all", self, {"api_version": version, "keys": list(kwargs.keys()), "sync_type": "async"}
         )
         return response.json()
 
     @api_error_handler
     async def search(self, query: str, version: str = "v1", **kwargs) -> List[Dict[str, Any]]:
         payload = {"query": query}
-        payload.update(self.sync_client._prepare_params(kwargs))
+        payload.update(self._prepare_params(kwargs))
         response = await self.async_client.post(f"/{version}/memories/search/", json=payload)
         response.raise_for_status()
         if "metadata" in kwargs:
             del kwargs["metadata"]
         capture_client_event(
-            "client.search", self.sync_client, {"api_version": version, "keys": list(kwargs.keys()), "sync_type": "async"}
+            "client.search", self, {"api_version": version, "keys": list(kwargs.keys()), "sync_type": "async"}
         )
         return response.json()
 
     @api_error_handler
     async def update(self, memory_id: str, data: str) -> Dict[str, Any]:
-        params = self.sync_client._prepare_params()
+        params = self._prepare_params()
         response = await self.async_client.put(f"/v1/memories/{memory_id}/", json={"text": data}, params=params)
         response.raise_for_status()
-        capture_client_event("client.update", self.sync_client, {"memory_id": memory_id, "sync_type": "async"})
+        capture_client_event("client.update", self, {"memory_id": memory_id, "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def delete(self, memory_id: str) -> Dict[str, Any]:
-        params = self.sync_client._prepare_params()
+        params = self._prepare_params()
         response = await self.async_client.delete(f"/v1/memories/{memory_id}/", params=params)
         response.raise_for_status()
-        capture_client_event("client.delete", self.sync_client, {"memory_id": memory_id, "sync_type": "async"})
+        capture_client_event("client.delete", self, {"memory_id": memory_id, "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def delete_all(self, **kwargs) -> Dict[str, str]:
-        params = self.sync_client._prepare_params(kwargs)
+        params = self._prepare_params(kwargs)
         response = await self.async_client.delete("/v1/memories/", params=params)
         response.raise_for_status()
-        capture_client_event("client.delete_all", self.sync_client, {"keys": list(kwargs.keys()), "sync_type": "async"})
+        capture_client_event("client.delete_all", self, {"keys": list(kwargs.keys()), "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def history(self, memory_id: str) -> List[Dict[str, Any]]:
-        params = self.sync_client._prepare_params()
+        params = self._prepare_params()
         response = await self.async_client.get(f"/v1/memories/{memory_id}/history/", params=params)
         response.raise_for_status()
-        capture_client_event("client.history", self.sync_client, {"memory_id": memory_id, "sync_type": "async"})
+        capture_client_event("client.history", self, {"memory_id": memory_id, "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def users(self) -> Dict[str, Any]:
-        params = self.sync_client._prepare_params()
+        params = self._prepare_params()
         response = await self.async_client.get("/v1/entities/", params=params)
         response.raise_for_status()
-        capture_client_event("client.users", self.sync_client, {"sync_type": "async"})
+        capture_client_event("client.users", self, {"sync_type": "async"})
         return response.json()
 
     @api_error_handler
@@ -848,7 +978,7 @@ class AsyncMemoryClient:
                 for entity in entities["results"]
             ]
 
-        params = self.sync_client._prepare_params()
+        params = self._prepare_params()
 
         if not to_delete:
             raise ValueError("No entities to delete")
@@ -858,7 +988,7 @@ class AsyncMemoryClient:
             response = await self.async_client.delete(f"/v2/entities/{entity['type']}/{entity['name']}/", params=params)
             response.raise_for_status()
 
-        capture_client_event("client.delete_users", self.sync_client, {"sync_type": "async"})
+        capture_client_event("client.delete_users", self, {"user_id": user_id, "agent_id": agent_id, "app_id": app_id, "run_id": run_id, "sync_type": "async"})
         return {
             "message": "Entity deleted successfully."
             if (user_id or agent_id or app_id or run_id)
@@ -868,7 +998,7 @@ class AsyncMemoryClient:
     @api_error_handler
     async def reset(self) -> Dict[str, str]:
         await self.delete_users()
-        capture_client_event("client.reset", self.sync_client, {"sync_type": "async"})
+        capture_client_event("client.reset", self, {"sync_type": "async"})
         return {"message": "Client reset successful. All users and memories deleted."}
 
     @api_error_handler
@@ -889,7 +1019,7 @@ class AsyncMemoryClient:
         response = await self.async_client.put("/v1/batch/", json={"memories": memories})
         response.raise_for_status()
 
-        capture_client_event("client.batch_update", self.sync_client, {"sync_type": "async"})
+        capture_client_event("client.batch_update", self, {"sync_type": "async"})
         return response.json()
 
     @api_error_handler
@@ -909,7 +1039,7 @@ class AsyncMemoryClient:
         response = await self.async_client.request("DELETE", "/v1/batch/", json={"memories": memories})
         response.raise_for_status()
 
-        capture_client_event("client.batch_delete", self.sync_client, {"sync_type": "async"})
+        capture_client_event("client.batch_delete", self, {"sync_type": "async"})
         return response.json()
 
     @api_error_handler
@@ -926,7 +1056,7 @@ class AsyncMemoryClient:
         response = await self.async_client.post("/v1/exports/", json={"schema": schema, **self._prepare_params(kwargs)})
         response.raise_for_status()
         capture_client_event(
-            "client.create_memory_export", self.sync_client, {"schema": schema, "keys": list(kwargs.keys()), "sync_type": "async"}
+            "client.create_memory_export", self, {"schema": schema, "keys": list(kwargs.keys()), "sync_type": "async"}
         )
         return response.json()
 
@@ -942,21 +1072,21 @@ class AsyncMemoryClient:
         """
         response = await self.async_client.post("/v1/exports/get/", json=self._prepare_params(kwargs))
         response.raise_for_status()
-        capture_client_event("client.get_memory_export", self.sync_client, {"keys": list(kwargs.keys()), "sync_type": "async"})
+        capture_client_event("client.get_memory_export", self, {"keys": list(kwargs.keys()), "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def get_project(self, fields: Optional[List[str]] = None) -> Dict[str, Any]:
-        if not (self.sync_client.org_id and self.sync_client.project_id):
+        if not (self.org_id and self.project_id):
             raise ValueError("org_id and project_id must be set to access instructions or categories")
 
-        params = self.sync_client._prepare_params({"fields": fields})
+        params = self._prepare_params({"fields": fields})
         response = await self.async_client.get(
-            f"/api/v1/orgs/organizations/{self.sync_client.org_id}/projects/{self.sync_client.project_id}/",
+            f"/api/v1/orgs/organizations/{self.org_id}/projects/{self.project_id}/",
             params=params,
         )
         response.raise_for_status()
-        capture_client_event("client.get_project", self.sync_client, {"fields": fields, "sync_type": "async"})
+        capture_client_event("client.get_project", self, {"fields": fields, "sync_type": "async"})
         return response.json()
 
     @api_error_handler
@@ -964,7 +1094,7 @@ class AsyncMemoryClient:
         self, custom_instructions: Optional[str] = None, custom_categories: Optional[List[str]] = None,
         retrieval_criteria: Optional[List[Dict[str, Any]]] = None
     ) -> Dict[str, Any]:
-        if not (self.sync_client.org_id and self.sync_client.project_id):
+        if not (self.org_id and self.project_id):
             raise ValueError("org_id and project_id must be set to update instructions or categories")
 
         if custom_instructions is None and custom_categories is None and retrieval_criteria is None:
@@ -972,17 +1102,17 @@ class AsyncMemoryClient:
                 "Currently we only support updating custom_instructions or custom_categories or retrieval_criteria, so you must provide at least one of them"
             )
 
-        payload = self.sync_client._prepare_params(
+        payload = self._prepare_params(
             {"custom_instructions": custom_instructions, "custom_categories": custom_categories, "retrieval_criteria": retrieval_criteria}
         )
         response = await self.async_client.patch(
-            f"/api/v1/orgs/organizations/{self.sync_client.org_id}/projects/{self.sync_client.project_id}/",
+            f"/api/v1/orgs/organizations/{self.org_id}/projects/{self.project_id}/",
             json=payload,
         )
         response.raise_for_status()
         capture_client_event(
             "client.update_project",
-            self.sync_client,
+            self,
             {"custom_instructions": custom_instructions, "custom_categories": custom_categories, "retrieval_criteria": retrieval_criteria, "sync_type": "async"},
         )
         return response.json()
@@ -996,7 +1126,7 @@ class AsyncMemoryClient:
             f"api/v1/webhooks/projects/{project_id}/",
         )
         response.raise_for_status()
-        capture_client_event("client.get_webhook", self.sync_client, {"sync_type": "async"})
+        capture_client_event("client.get_webhook", self, {"sync_type": "async"})
         return response.json()
 
     @api_error_handler
@@ -1004,7 +1134,7 @@ class AsyncMemoryClient:
         payload = {"url": url, "name": name, "event_types": event_types}
         response = await self.async_client.post(f"api/v1/webhooks/projects/{project_id}/", json=payload)
         response.raise_for_status()
-        capture_client_event("client.create_webhook", self.sync_client, {"sync_type": "async"})
+        capture_client_event("client.create_webhook", self, {"sync_type": "async"})
         return response.json()
 
     @api_error_handler
@@ -1018,14 +1148,14 @@ class AsyncMemoryClient:
         payload = {k: v for k, v in {"name": name, "url": url, "event_types": event_types}.items() if v is not None}
         response = await self.async_client.put(f"api/v1/webhooks/{webhook_id}/", json=payload)
         response.raise_for_status()
-        capture_client_event("client.update_webhook", self.sync_client, {"webhook_id": webhook_id, "sync_type": "async"})
+        capture_client_event("client.update_webhook", self, {"webhook_id": webhook_id, "sync_type": "async"})
         return response.json()
 
     @api_error_handler
     async def delete_webhook(self, webhook_id: int) -> Dict[str, str]:
         response = await self.async_client.delete(f"api/v1/webhooks/{webhook_id}/")
         response.raise_for_status()
-        capture_client_event("client.delete_webhook", self.sync_client, {"webhook_id": webhook_id, "sync_type": "async"})
+        capture_client_event("client.delete_webhook", self, {"webhook_id": webhook_id, "sync_type": "async"})
         return response.json()
 
     @api_error_handler
@@ -1042,5 +1172,6 @@ class AsyncMemoryClient:
 
         response = await self.async_client.post("/v1/feedback/", json=data)
         response.raise_for_status()
-        capture_client_event("client.feedback", self.sync_client, data, {"sync_type": "async"})
+        capture_client_event("client.feedback", self, data, {"sync_type": "async"})
         return response.json()
+
diff --git a/mem0/graphs/configs.py b/mem0/graphs/configs.py
index e0e98274..50c585a1 100644
--- a/mem0/graphs/configs.py
+++ b/mem0/graphs/configs.py
@@ -9,6 +9,7 @@ class Neo4jConfig(BaseModel):
     url: Optional[str] = Field(None, description="Host address for the graph database")
     username: Optional[str] = Field(None, description="Username for the graph database")
     password: Optional[str] = Field(None, description="Password for the graph database")
+    database: Optional[str] = Field(None, description="Database for the graph database")
 
     @model_validator(mode="before")
     def check_host_port_or_path(cls, values):
diff --git a/mem0/memory/graph_memory.py b/mem0/memory/graph_memory.py
index 18aaf41b..760d864a 100644
--- a/mem0/memory/graph_memory.py
+++ b/mem0/memory/graph_memory.py
@@ -33,6 +33,9 @@ class MemoryGraph:
             self.config.graph_store.config.url,
             self.config.graph_store.config.username,
             self.config.graph_store.config.password,
+            self.config.graph_store.config.database,
+            refresh_schema=False,
+            driver_config={"notifications_min_severity":"OFF"},
         )
         self.embedding_model = EmbedderFactory.create(
             self.config.embedder.provider, self.config.embedder.config, self.config.vector_store.config
diff --git a/mem0/memory/main.py b/mem0/memory/main.py
index d409be75..fd02cf57 100644
--- a/mem0/memory/main.py
+++ b/mem0/memory/main.py
@@ -7,6 +7,7 @@ import logging
 import os
 import uuid
 import warnings
+from copy import deepcopy
 from datetime import datetime
 from typing import Any, Dict
 
@@ -66,6 +67,8 @@ class Memory(MemoryBase):
 
             self.graph = MemoryGraph(self.config)
             self.enable_graph = True
+        else:
+            self.graph = None
 
         self.config.vector_store.config.collection_name = "mem0migrations"
         if self.config.vector_store.provider in ["faiss", "qdrant"]:
@@ -263,20 +266,20 @@ class Memory(MemoryBase):
         )
 
         try:
-            new_memories_with_actions = self.llm.generate_response(
+            response: str = self.llm.generate_response(
                 messages=[{"role": "user", "content": function_calling_prompt}],
                 response_format={"type": "json_object"},
             )
         except Exception as e:
-            logging.error(f"Error in new_memories_with_actions: {e}")
-            new_memories_with_actions = []
+            logging.error(f"Error in new memory actions response: {e}")
+            response = ""
 
         try:
-            new_memories_with_actions = remove_code_blocks(new_memories_with_actions)
-            new_memories_with_actions = json.loads(new_memories_with_actions)
+            response = remove_code_blocks(response)
+            new_memories_with_actions = json.loads(response)
         except Exception as e:
             logging.error(f"Invalid JSON response: {e}")
-            new_memories_with_actions = []
+            new_memories_with_actions = {}
 
         returned_memories = []
         try:
@@ -290,7 +293,7 @@ class Memory(MemoryBase):
                         memory_id = self._create_memory(
                             data=resp.get("text"),
                             existing_embeddings=new_message_embeddings,
-                            metadata=metadata,
+                            metadata=deepcopy(metadata),
                         )
                         returned_memories.append(
                             {
@@ -304,7 +307,7 @@ class Memory(MemoryBase):
                             memory_id=temp_uuid_mapping[resp["id"]],
                             data=resp.get("text"),
                             existing_embeddings=new_message_embeddings,
-                            metadata=metadata,
+                            metadata=deepcopy(metadata),
                         )
                         returned_memories.append(
                             {
@@ -767,13 +770,13 @@ class Memory(MemoryBase):
         logger.warning("Resetting all memories")
 
         # Close the old connection if possible
-        if hasattr(self.db, 'connection') and self.db.connection:
-                self.db.connection.execute("DROP TABLE IF EXISTS history")
-                self.db.connection.close()
+        if hasattr(self.db, "connection") and self.db.connection:
+            self.db.connection.execute("DROP TABLE IF EXISTS history")
+            self.db.connection.close()
 
         self.db = SQLiteManager(self.config.history_db_path)
 
-        if hasattr(self.vector_store, 'reset'):
+        if hasattr(self.vector_store, "reset"):
             self.vector_store = VectorStoreFactory.reset(self.vector_store)
         else:
             logger.warning("Vector store does not support reset. Skipping.")
@@ -811,6 +814,8 @@ class AsyncMemory(MemoryBase):
 
             self.graph = MemoryGraph(self.config)
             self.enable_graph = True
+        else:
+            self.graph = None
 
         capture_event("mem0.init", self, {"sync_type": "async"})
 
@@ -1007,21 +1012,21 @@ class AsyncMemory(MemoryBase):
         )
 
         try:
-            new_memories_with_actions = await asyncio.to_thread(
+            response: str = await asyncio.to_thread(
                 self.llm.generate_response,
                 messages=[{"role": "user", "content": function_calling_prompt}],
                 response_format={"type": "json_object"},
             )
         except Exception as e:
-            logging.error(f"Error in new_memories_with_actions: {e}")
-            new_memories_with_actions = []
+            logging.error(f"Error in new memory actions response: {e}")
+            response = ""
 
         try:
-            new_memories_with_actions = remove_code_blocks(new_memories_with_actions)
-            new_memories_with_actions = json.loads(new_memories_with_actions)
+            response = remove_code_blocks(response)
+            new_memories_with_actions = json.loads(response)
         except Exception as e:
             logging.error(f"Invalid JSON response: {e}")
-            new_memories_with_actions = []
+            new_memories_with_actions = {}
 
         returned_memories = []
         try:
@@ -1035,7 +1040,9 @@ class AsyncMemory(MemoryBase):
                     elif resp.get("event") == "ADD":
                         task = asyncio.create_task(
                             self._create_memory(
-                                data=resp.get("text"), existing_embeddings=new_message_embeddings, metadata=metadata
+                                data=resp.get("text"),
+                                existing_embeddings=new_message_embeddings,
+                                metadata=deepcopy(metadata),
                             )
                         )
                         memory_tasks.append((task, resp, "ADD", None))
@@ -1045,7 +1052,7 @@ class AsyncMemory(MemoryBase):
                                 memory_id=temp_uuid_mapping[resp["id"]],
                                 data=resp.get("text"),
                                 existing_embeddings=new_message_embeddings,
-                                metadata=metadata,
+                                metadata=deepcopy(metadata),
                             )
                         )
                         memory_tasks.append((task, resp, "UPDATE", temp_uuid_mapping[resp["id"]]))
@@ -1092,7 +1099,9 @@ class AsyncMemory(MemoryBase):
         except Exception as e:
             logging.error(f"Error in new_memories_with_actions: {e}")
 
-        capture_event("mem0.add", self, {"version": self.api_version, "keys": list(filters.keys()), "sync_type": "async"})
+        capture_event(
+            "mem0.add", self, {"version": self.api_version, "keys": list(filters.keys()), "sync_type": "async"}
+        )
 
         return returned_memories
 
@@ -1547,10 +1556,10 @@ class AsyncMemory(MemoryBase):
 
         gc.collect()
 
-        if hasattr(self.vector_store, 'client') and hasattr(self.vector_store.client, 'close'):
+        if hasattr(self.vector_store, "client") and hasattr(self.vector_store.client, "close"):
             await asyncio.to_thread(self.vector_store.client.close)
 
-        if hasattr(self.db, 'connection') and self.db.connection:
+        if hasattr(self.db, "connection") and self.db.connection:
             await asyncio.to_thread(lambda: self.db.connection.execute("DROP TABLE IF EXISTS history"))
             await asyncio.to_thread(self.db.connection.close)
 
diff --git a/poetry.lock b/poetry.lock
index d8c3d6d2..fd05427e 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1544,6 +1544,43 @@ tomli = {version = ">=1", markers = "python_version < \"3.11\""}
 [package.extras]
 dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"]
 
+[[package]]
+name = "pytest-asyncio"
+version = "0.23.8"
+description = "Pytest support for asyncio"
+optional = false
+python-versions = ">=3.8"
+groups = ["test"]
+files = [
+    {file = "pytest_asyncio-0.23.8-py3-none-any.whl", hash = "sha256:50265d892689a5faefb84df80819d1ecef566eb3549cf915dfb33569359d1ce2"},
+    {file = "pytest_asyncio-0.23.8.tar.gz", hash = "sha256:759b10b33a6dc61cce40a8bd5205e302978bbbcc00e279a8b61d9a6a3c82e4d3"},
+]
+
+[package.dependencies]
+pytest = ">=7.0.0,<9"
+
+[package.extras]
+docs = ["sphinx (>=5.3)", "sphinx-rtd-theme (>=1.0)"]
+testing = ["coverage (>=6.2)", "hypothesis (>=5.7.1)"]
+
+[[package]]
+name = "pytest-mock"
+version = "3.14.0"
+description = "Thin-wrapper around the mock package for easier use with pytest"
+optional = false
+python-versions = ">=3.8"
+groups = ["test"]
+files = [
+    {file = "pytest-mock-3.14.0.tar.gz", hash = "sha256:2719255a1efeceadbc056d6bf3df3d1c5015530fb40cf347c0f9afac88410bd0"},
+    {file = "pytest_mock-3.14.0-py3-none-any.whl", hash = "sha256:0b72c38033392a5f4621342fe11e9219ac11ec9d375f8e2a0c164539e0d70f6f"},
+]
+
+[package.dependencies]
+pytest = ">=6.2.5"
+
+[package.extras]
+dev = ["pre-commit", "pytest-asyncio", "tox"]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -2163,4 +2200,4 @@ graph = ["langchain-neo4j", "neo4j", "rank-bm25"]
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.9,<4.0"
-content-hash = "6d54994c8286378de8dc348b8b11c57ce922dd8104a26cd63eb171563d1bbf5f"
+content-hash = "07f2aee9c596c2d2470df085b92551b7b7e3c19cabe61ae5bee7505395601417"
diff --git a/pyproject.toml b/pyproject.toml
index 4f6d624d..658e118c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -36,6 +36,8 @@ graph = ["langchain-neo4j", "neo4j", "rank-bm25"]
 
 [tool.poetry.group.test.dependencies]
 pytest = "^8.2.2"
+pytest-mock = "^3.14.0"
+pytest-asyncio = "^0.23.7"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.6.5"
@@ -49,4 +51,3 @@ build-backend = "poetry.core.masonry.api"
 [tool.ruff]
 line-length = 120
 exclude = ["embedchain/"]
-


Test patch:
diff --git a/tests/memory/test_main.py b/tests/memory/test_main.py
new file mode 100644
index 00000000..18966e44
--- /dev/null
+++ b/tests/memory/test_main.py
@@ -0,0 +1,135 @@
+import logging
+from unittest.mock import MagicMock
+
+import pytest
+
+from mem0.memory.main import AsyncMemory, Memory
+
+
+def _setup_mocks(mocker):
+    """Helper to setup common mocks for both sync and async fixtures"""
+    mock_embedder = mocker.MagicMock()
+    mock_embedder.return_value.embed.return_value = [0.1, 0.2, 0.3]
+    mocker.patch('mem0.utils.factory.EmbedderFactory.create', mock_embedder)
+    
+    mock_vector_store = mocker.MagicMock()
+    mock_vector_store.return_value.search.return_value = []
+    mocker.patch('mem0.utils.factory.VectorStoreFactory.create', 
+                side_effect=[mock_vector_store.return_value, mocker.MagicMock()])
+    
+    mock_llm = mocker.MagicMock()
+    mocker.patch('mem0.utils.factory.LlmFactory.create', mock_llm)
+    
+    mocker.patch('mem0.memory.storage.SQLiteManager', mocker.MagicMock())
+    
+    return mock_llm, mock_vector_store
+
+
+class TestAddToVectorStoreErrors:
+    @pytest.fixture
+    def mock_memory(self, mocker):
+        """Fixture that returns a Memory instance with mocker-based mocks"""
+        mock_llm, _ = _setup_mocks(mocker)
+        
+        memory = Memory()
+        memory.config = mocker.MagicMock()
+        memory.config.custom_fact_extraction_prompt = None
+        memory.config.custom_update_memory_prompt = None
+        memory.api_version = "v1.1"
+        
+        return memory
+
+    def test_empty_llm_response_fact_extraction(self, mock_memory, caplog):
+        """Test empty response from LLM during fact extraction"""
+        # Setup
+        mock_memory.llm.generate_response.return_value = ""
+        
+        # Execute
+        with caplog.at_level(logging.ERROR):
+            result = mock_memory._add_to_vector_store(
+                messages=[{"role": "user", "content": "test"}],
+                metadata={},
+                filters={},
+                infer=True
+            )
+        
+        # Verify
+        assert mock_memory.llm.generate_response.call_count == 2
+        assert result == []  # Should return empty list when no memories processed
+        assert "Error in new_retrieved_facts" in caplog.text
+
+    def test_empty_llm_response_memory_actions(self, mock_memory, caplog):
+        """Test empty response from LLM during memory actions"""
+        # Setup
+        # First call returns valid JSON, second call returns empty string
+        mock_memory.llm.generate_response.side_effect = [
+            '{"facts": ["test fact"]}',
+            ""
+        ]
+        
+        # Execute
+        with caplog.at_level(logging.ERROR):
+            result = mock_memory._add_to_vector_store(
+                messages=[{"role": "user", "content": "test"}],
+                metadata={},
+                filters={},
+                infer=True
+            )
+        
+        # Verify
+        assert mock_memory.llm.generate_response.call_count == 2
+        assert result == []  # Should return empty list when no memories processed
+        assert "Invalid JSON response" in caplog.text
+
+
+@pytest.mark.asyncio
+class TestAsyncAddToVectorStoreErrors:
+    @pytest.fixture
+    def mock_async_memory(self, mocker):
+        """Fixture for AsyncMemory with mocker-based mocks"""
+        mock_llm, _ = _setup_mocks(mocker)
+        
+        memory = AsyncMemory()
+        memory.config = mocker.MagicMock()
+        memory.config.custom_fact_extraction_prompt = None
+        memory.config.custom_update_memory_prompt = None
+        memory.api_version = "v1.1"
+        
+        return memory
+
+    @pytest.mark.asyncio
+    async def test_async_empty_llm_response_fact_extraction(self, mock_async_memory, caplog, mocker):
+        """Test empty response in AsyncMemory._add_to_vector_store"""
+        mocker.patch('mem0.utils.factory.EmbedderFactory.create', return_value=MagicMock())
+        mock_async_memory.llm.generate_response.return_value = ""
+        
+        with caplog.at_level(logging.ERROR):
+            result = await mock_async_memory._add_to_vector_store(
+                messages=[{"role": "user", "content": "test"}],
+                metadata={},
+                filters={},
+                infer=True
+            )
+        
+        assert result == []
+        assert "Error in new_retrieved_facts" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_async_empty_llm_response_memory_actions(self, mock_async_memory, caplog, mocker):
+        """Test empty response in AsyncMemory._add_to_vector_store"""
+        mocker.patch('mem0.utils.factory.EmbedderFactory.create', return_value=MagicMock())
+        mock_async_memory.llm.generate_response.side_effect = [
+            '{"facts": ["test fact"]}',
+            ""
+        ]
+        
+        with caplog.at_level(logging.ERROR):
+            result = await mock_async_memory._add_to_vector_store(
+                messages=[{"role": "user", "content": "test"}],
+                metadata={},
+                filters={},
+                infer=True
+            )
+        
+        assert result == []
+        assert "Invalid JSON response" in caplog.text