Hot func main/delete memory

src/memos/api/handlers/memory_handler.py

@@ -6,8 +6,9 @@
 
 from typing import Any, Literal
 
-from memos.api.product_models import MemoryResponse
+from memos.api.product_models import DeleteMemoryRequest, DeleteMemoryResponse, MemoryResponse
 from memos.log import get_logger
+from memos.mem_cube.navie import NaiveMemCube
 from memos.mem_os.utils.format_utils import (
     convert_graph_to_tree_forworkmem,
     ensure_unique_tree_ids,
@@ -149,3 +150,50 @@ def handle_get_subgraph(
     except Exception as e:
         logger.error(f"Failed to get subgraph: {e}", exc_info=True)
         raise
+
+
+def handle_delete_memories(delete_mem_req: DeleteMemoryRequest, naive_mem_cube: NaiveMemCube):
+    # Validate that only one of memory_ids, file_ids, or filter is provided
+    provided_params = [
+        delete_mem_req.memory_ids is not None,
+        delete_mem_req.file_ids is not None,
+        delete_mem_req.filter is not None,
+    ]
+    if sum(provided_params) != 1:
+        return DeleteMemoryResponse(
+            message="Exactly one of memory_ids, file_ids, or filter must be provided",
+            data={"status": "failure"},
+        )
+
+    try:
+        if delete_mem_req.memory_ids is not None:
+            for cube_id in delete_mem_req.writable_cube_ids:
+                naive_mem_cube.text_mem.delete(delete_mem_req.memory_ids, user_name=cube_id)
+            if naive_mem_cube.pref_mem is not None:
+                naive_mem_cube.pref_mem.delete(delete_mem_req.memory_ids)
+        elif delete_mem_req.file_ids is not None:
+            # TODO: Implement deletion by file_ids
+            # Need to find memory_ids associated with file_ids and delete them
+            logger.warning("Deletion by file_ids not implemented yet")
+            return DeleteMemoryResponse(
+                message="Deletion by file_ids not implemented yet",
+                data={"status": "failure"},
+            )
+        elif delete_mem_req.filter is not None:
+            # TODO: Implement deletion by filter
+            # Need to find memories matching filter and delete them
+            logger.warning("Deletion by filter not implemented yet")
+            return DeleteMemoryResponse(
+                message="Deletion by filter not implemented yet",
+                data={"status": "failure"},
+            )
+    except Exception as e:
+        logger.error(f"Failed to delete memories: {e}", exc_info=True)
+        return DeleteMemoryResponse(
+            message="Failed to delete memories",
+            data={"status": "failure"},
+        )
+    return DeleteMemoryResponse(
+        message="Memories deleted successfully",
+        data={"status": "success"},
+    )

src/memos/api/product_models.py

@@ -1,7 +1,7 @@
 import os
 import uuid
 
-from typing import Generic, Literal, TypeVar
+from typing import Any, Generic, Literal, TypeVar
 
 from pydantic import BaseModel, Field
 
@@ -224,6 +224,15 @@ class APIChatCompleteRequest(BaseRequest):
     )
 
 
+class DeleteMemoryRequest(BaseRequest):
+    """Request model for deleting memories."""
+
+    writable_cube_ids: list[str] = Field(..., description="Writable cube IDs")
+    memory_ids: list[str] | None = Field(None, description="Memory IDs")
+    file_ids: list[str] | None = Field(None, description="File IDs")
+    filter: dict[str, Any] | None = Field(None, description="Filter for the memory")
+
+
 class SuggestionRequest(BaseRequest):
     """Request model for getting suggestion queries."""
 
@@ -313,3 +322,7 @@ class MemOSAddResponse(BaseModel):
     def success(self) -> bool:
         """Convenient access to success status."""
         return self.data.success
+
+
+class DeleteMemoryResponse(BaseResponse[dict]):
+    """Response model for deleting memories."""

src/memos/api/routers/server_router.py

@@ -27,6 +27,8 @@
     APIChatCompleteRequest,
     APISearchRequest,
     ChatRequest,
+    DeleteMemoryRequest,
+    DeleteMemoryResponse,
     GetMemoryRequest,
     MemoryResponse,
     SearchResponse,
@@ -219,3 +221,12 @@ def get_all_memories(memory_req: GetMemoryRequest):
             memory_type=memory_req.memory_type or "text_mem",
             naive_mem_cube=naive_mem_cube,
         )
+
+
+@router.post(
+    "/delete_memory", summary="Delete memories for user", response_model=DeleteMemoryResponse
+)
+def delete_memories(memory_req: DeleteMemoryRequest):
+    return handlers.memory_handler.handle_delete_memories(
+        delete_mem_req=memory_req, naive_mem_cube=naive_mem_cube
+    )

src/memos/memories/textual/preference.py

@@ -260,7 +260,9 @@ def delete(self, memory_ids: list[str]) -> None:
         Args:
             memory_ids (list[str]): List of memory IDs to delete.
         """
-        raise NotImplementedError
+        collection_list = self.vector_db.config.collection_name
+        for collection_name in collection_list:
+            self.vector_db.delete(collection_name, memory_ids)
 
     def delete_with_collection_name(self, collection_name: str, memory_ids: list[str]) -> None:
         """Delete memories by their IDs and collection name.