feat: implement memory compression and deduplication (issue #141) (#216)

* feat: implement memory compression and deduplication (issue #141)

* fix: resolve mock vector store issue in test_memory_optimize.py

- Use Mock(spec=Memory) instead of instantiating Memory to avoid
  vector store initialization that requires real provider configuration
- This fixes the ValueError: Unsupported VectorStore provider: mock

Closes #216

* build: drop python 3.10 support to fix pyseekdb dependency conflict

* chore: remove unnecessary files per maintainer review

- Remove docs/plans/2026-01-31-memory-compression-deduplication.md
  (AI-generated descriptive file, not needed in formal PR)
- Remove uv.lock (project uses plain venv workflow, uv.lock
  adds no value and causes confusion)

Reviewed-by: Teingi

---------

Co-authored-by: NTLx <lx@openclaw.ai>

Author: NTLx

.github/workflows/test.yml

@@ -26,7 +26,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.11", "3.12"]
       fail-fast: false
 
     steps:
@@ -64,7 +64,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11", "3.12"]
+        python-version: ["3.11", "3.12"]
       fail-fast: false
 
     steps:

pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
     "Topic :: Software Development :: Libraries :: Python Modules",
     "Topic :: Scientific/Engineering :: Artificial Intelligence",
 ]
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 dependencies = [
     "pydantic>=2.0.0",
     "pydantic-settings>=2.0.0",

src/powermem/core/memory.py

@@ -26,6 +26,7 @@
 from ..integrations.rerank.factory import RerankFactory
 from .telemetry import TelemetryManager
 from .audit import AuditLogger
+from ..intelligence.memory_optimizer import MemoryOptimizer
 from ..intelligence.plugin import IntelligentMemoryPlugin, EbbinghausIntelligencePlugin
 from ..utils.utils import remove_code_blocks, convert_config_object_to_dict, parse_vision_messages, set_timezone
 from ..prompts.intelligent_memory_prompts import (
@@ -309,6 +310,9 @@ def __init__(
             audit_config = self.config
         self.audit = AuditLogger(audit_config)
 
+        # Initialize memory optimizer
+        self.optimizer = MemoryOptimizer(self.storage, self.llm)
+
         # Save custom prompts from config
         if self.memory_config:
             self.custom_fact_extraction_prompt = self.memory_config.custom_fact_extraction_prompt
@@ -1586,6 +1590,33 @@ def get_all(
             logger.error(f"Failed to get all memories: {e}")
             raise
 
+    def optimize(self, strategy: str = "deduplicate", **kwargs) -> Dict[str, Any]:
+        """
+        Optimize memory storage.
+
+        Args:
+            strategy: "deduplicate" or "compress"
+            **kwargs: Additional args like threshold, user_id, dedup_strategy
+
+        Returns:
+            Optimization stats
+        """
+        if strategy == "deduplicate":
+            # Extract specific args
+            sub_strategy = kwargs.get("dedup_strategy", "exact")
+            return self.optimizer.deduplicate(
+                user_id=kwargs.get("user_id"),
+                strategy=sub_strategy,
+                threshold=kwargs.get("threshold", 0.95)
+            )
+        elif strategy == "compress":
+            return self.optimizer.compress(
+                user_id=kwargs.get("user_id"),
+                threshold=kwargs.get("threshold", 0.85)
+            )
+        else:
+            raise ValueError(f"Unknown optimization strategy: {strategy}")
+
     def get_statistics(
         self, user_id: Optional[str] = None, agent_id: Optional[str] = None
     ) -> Dict[str, Any]: