Lancedb Memory

#!/usr/bin/env python3 """ LanceDB integration for long-term memory management. Provides vector search and semantic memory capabilities. """

import os import json import lancedb from datetime import datetime from typing import List, Dict, Any, Optional from pathlib import Path

class LanceMemoryDB: """LanceDB wrapper for long-term memory storage and retrieval."""

def __init__(self, db_path: str = "/Users/prerak/clawd/memory/lancedb"):
    self.db_path = Path(db_path)
    self.db_path.mkdir(parents=True, exist_ok=True)
    self.db = lancedb.connect(self.db_path)
    
    # Ensure memory table exists
    if "memory" not in self.db.table_names():
        self._create_memory_table()

def _create_memory_table(self):
    """Create the memory table with appropriate schema."""
    schema = [
        {"name": "id", "type": "int", "nullable": False},
        {"name": "timestamp", "type": "timestamp", "nullable": False},
        {"name": "content", "type": "str", "nullable": False},
        {"name": "category", "type": "str", "nullable": True},
        {"name": "tags", "type": "str[]", "nullable": True},
        {"name": "importance", "type": "int", "nullable": True},
        {"name": "metadata", "type": "json", "nullable": True},
    ]
    
    self.db.create_table("memory", schema=schema)

def add_memory(self, content: str, category: str = "general", tags: List[str] = None, 
               importance: int = 5, metadata: Dict[str, Any] = None) -> int:
    """Add a new memory entry."""
    table = self.db.open_table("memory")
    
    # Get next ID
    max_id = table.to_pandas()["id"].max() if len(table) > 0 else 0
    new_id = max_id + 1
    
    # Insert new memory
    memory_data = {
        "id": new_id,
        "timestamp": datetime.now(),
        "content": content,
        "category": category,
        "tags": tags or [],
        "importance": importance,
        "metadata": metadata or {}
    }
    
    table.add([memory_data])
    return new_id

def search_memories(self, query: str, category: str = None, limit: int = 10) -> List[Dict]:
    """Search memories using vector similarity."""
    table = self.db.open_table("memory")
    
    # Build filter
    where_clause = []
    if category:
        where_clause.append(f"category = '{category}'")
    
    filter_expr = " AND ".join(where_clause) if where_clause else None
    
    # Vector search
    results = table.vector_search(query).limit(limit).where(filter_expr).to_list()
    
    return results

def get_memories_by_category(self, category: str, limit: int = 50) -> List[Dict]:
    """Get memories by category."""
    table = self.db.open_table("memory")
    df = table.to_pandas()
    filtered = df[df["category"] == category]....