jarvis-memory/skills/qdrant-memory/scripts/auto_memory.py

#!/usr/bin/env python3
"""
Auto-memory management with proactive context retrieval
Usage: auto_memory.py store "text" [--importance medium] [--tags tag1,tag2]
       auto_memory.py search "query" [--limit 3]
       auto_memory.py should_store "conversation_snippet"
       auto_memory.py context "current_topic" [--min-score 0.6]
       auto_memory.py proactive "user_message" [--auto-include]
"""

import argparse
import json
import subprocess
import sys

WORKSPACE = "/root/.openclaw/workspace"
QDRANT_SKILL = f"{WORKSPACE}/skills/qdrant-memory/scripts"

def store_memory(text, importance="medium", tags=None, confidence="high", 
                 source_type="user", verified=True, expires=None):
    """Store a memory automatically with full metadata"""
    cmd = [
        "python3", f"{QDRANT_SKILL}/store_memory.py",
        text,
        "--importance", importance,
        "--confidence", confidence,
        "--source-type", source_type,
    ]
    if verified:
        cmd.append("--verified")
    if tags:
        cmd.extend(["--tags", ",".join(tags)])
    if expires:
        cmd.extend(["--expires", expires])
    
    result = subprocess.run(cmd, capture_output=True, text=True)
    return result.returncode == 0

def search_memories(query, limit=3, min_score=0.0):
    """Search memories for relevant context"""
    cmd = [
        "python3", f"{QDRANT_SKILL}/search_memories.py",
        query,
        "--limit", str(limit),
        "--json"
    ]
    
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
    if result.returncode == 0:
        try:
            memories = json.loads(result.stdout)
            # Filter by score if specified
            if min_score > 0:
                memories = [m for m in memories if m.get("score", 0) >= min_score]
            return memories
        except:
            return []
    return []

def should_store_memory(text):
    """Determine if a memory should be stored based on content"""
    text_lower = text.lower()
    
    # Explicit store markers (highest priority)
    explicit_markers = ["remember this", "note this", "save this", "log this", "record this"]
    if any(marker in text_lower for marker in explicit_markers):
        return True, "explicit_store", "high"
    
    # Permanent markers (never expire)
    permanent_markers = [
        "my name is", "i am ", "i'm ", "call me", "i live in", "my address",
        "my phone", "my email", "my birthday", "i work at", "my job"
    ]
    if any(marker in text_lower for marker in permanent_markers):
        return True, "permanent_fact", "high"
    
    # Preference/decision indicators
    pref_markers = ["i prefer", "i like", "i want", "my favorite", "i need", "i use", "i choose"]
    if any(marker in text_lower for marker in pref_markers):
        return True, "preference", "high"
    
    # Setup/achievement markers
    setup_markers = ["setup", "installed", "configured", "working", "completed", "finished", "created"]
    if any(marker in text_lower for marker in setup_markers):
        return True, "setup_complete", "medium"
    
    # Rule/policy markers
    rule_markers = ["rule", "policy", "always", "never", "every", "schedule", "deadline"]
    if any(marker in text_lower for marker in rule_markers):
        return True, "rule_policy", "high"
    
    # Temporary markers (should expire)
    temp_markers = ["for today", "for now", "temporarily", "this time only", "just for"]
    if any(marker in text_lower for marker in temp_markers):
        return True, "temporary", "low", "7d"  # 7 day expiration
    
    # Important keywords (check density)
    important_keywords = [
        "important", "critical", "essential", "key", "main", "primary",
        "password", "api key", "token", "secret", "backup", "restore",
        "decision", "choice", "selected", "chose", "picked"
    ]
    matches = sum(1 for kw in important_keywords if kw in text_lower)
    if matches >= 2:
        return True, "keyword_match", "medium"
    
    # Error/lesson learned markers
    lesson_markers = ["error", "mistake", "fixed", "solved", "lesson", "learned", "solution"]
    if any(marker in text_lower for marker in lesson_markers):
        return True, "lesson", "high"
    
    return False, "not_important", None

def get_relevant_context(query, min_score=0.6, limit=5):
    """Get relevant memories for current context with smart filtering"""
    memories = search_memories(query, limit=limit, min_score=min_score)
    
    # Sort by importance and score
    importance_order = {"high": 0, "medium": 1, "low": 2}
    memories.sort(key=lambda m: (
        importance_order.get(m.get("importance", "medium"), 1),
        -m.get("score", 0)
    ))
    
    return memories

def proactive_retrieval(user_message, auto_include=False):
    """
    Proactively retrieve relevant memories based on user message.
    Returns relevant memories that might be helpful context.
    """
    # Extract key concepts from the message
    # Simple approach: use the whole message as query
    # Better approach: extract noun phrases (could be enhanced)
    
    memories = get_relevant_context(user_message, min_score=0.5, limit=5)
    
    if not memories:
        return []
    
    # Filter for highly relevant or important memories
    proactive_memories = []
    for m in memories:
        score = m.get("score", 0)
        importance = m.get("importance", "medium")
        
        # Include if:
        # - High score (0.7+) regardless of importance
        # - Medium score (0.5+) AND high importance
        if score >= 0.7 or (score >= 0.5 and importance == "high"):
            proactive_memories.append(m)
    
    return proactive_memories

def format_context_for_prompt(memories):
    """Format memories as context for the LLM prompt"""
    if not memories:
        return ""
    
    context = "\n[Relevant context from previous conversations]:\n"
    for i, m in enumerate(memories, 1):
        text = m.get("text", "")
        date = m.get("date", "unknown")
        importance = m.get("importance", "medium")
        
        prefix = "🔴" if importance == "high" else "🟡" if importance == "medium" else "🟢"
        context += f"{prefix} [{date}] {text}\n"
    
    return context

def auto_tag(text, reason):
    """Automatically generate tags based on content"""
    tags = []
    
    # Add tag based on reason
    reason_tags = {
        "explicit_store": "recorded",
        "permanent_fact": "identity",
        "preference": "preference",
        "setup_complete": "setup",
        "rule_policy": "policy",
        "temporary": "temporary",
        "keyword_match": "important",
        "lesson": "lesson"
    }
    if reason in reason_tags:
        tags.append(reason_tags[reason])
    
    # Content-based tags
    text_lower = text.lower()
    content_tags = {
        "voice": ["voice", "tts", "stt", "whisper", "audio", "speak"],
        "tools": ["tool", "script", "command", "cli", "error"],
        "config": ["config", "setting", "setup", "install"],
        "memory": ["memory", "remember", "recall", "search"],
        "web": ["search", "web", "online", "internet"],
        "security": ["password", "token", "secret", "key", "auth"]
    }
    
    for tag, keywords in content_tags.items():
        if any(kw in text_lower for kw in keywords):
            tags.append(tag)
    
    return list(set(tags))  # Remove duplicates

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Auto-memory management")
    parser.add_argument("action", choices=[
        "store", "search", "should_store", "context", 
        "proactive", "auto_process"
    ])
    parser.add_argument("text", help="Text to process")
    parser.add_argument("--importance", default="medium", choices=["low", "medium", "high"])
    parser.add_argument("--tags", help="Comma-separated tags")
    parser.add_argument("--limit", type=int, default=3)
    parser.add_argument("--min-score", type=float, default=0.6)
    parser.add_argument("--auto-include", action="store_true", help="Auto-include context in response")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    
    args = parser.parse_args()
    
    if args.action == "store":
        tags = [t.strip() for t in args.tags.split(",")] if args.tags else []
        if store_memory(args.text, args.importance, tags):
            result = {"stored": True, "importance": args.importance, "tags": tags}
            print(json.dumps(result) if args.json else f"✅ Stored: {args.text[:50]}...")
        else:
            result = {"stored": False, "error": "Failed to store"}
            print(json.dumps(result) if args.json else "❌ Failed to store")
            sys.exit(1)
    
    elif args.action == "search":
        results = search_memories(args.text, args.limit, args.min_score)
        if args.json:
            print(json.dumps(results))
        else:
            print(f"Found {len(results)} memories:")
            for r in results:
                print(f"  [{r.get('score', 0):.2f}] {r.get('text', '')[:60]}...")
    
    elif args.action == "should_store":
        should_store, reason, importance = should_store_memory(args.text)
        result = {"should_store": should_store, "reason": reason, "importance": importance}
        print(json.dumps(result) if args.json else f"Store? {should_store} ({reason}, {importance})")
    
    elif args.action == "context":
        context = get_relevant_context(args.text, args.min_score, args.limit)
        if args.json:
            print(json.dumps(context))
        else:
            print(format_context_for_prompt(context))
    
    elif args.action == "proactive":
        memories = proactive_retrieval(args.text, args.auto_include)
        if args.json:
            print(json.dumps(memories))
        else:
            if memories:
                print(f"🔍 Found {len(memories)} relevant memories:")
                for m in memories:
                    score = m.get("score", 0)
                    text = m.get("text", "")[:60]
                    print(f"   [{score:.2f}] {text}...")
            else:
                print("ℹ️ No highly relevant memories found")
    
    elif args.action == "auto_process":
        # Full pipeline: check if should store, auto-tag, store, and return context
        should_store, reason, importance = should_store_memory(args.text)
        
        result = {
            "should_store": should_store,
            "reason": reason,
            "stored": False
        }
        
        if should_store:
            # Auto-generate tags
            tags = auto_tag(args.text, reason)
            if args.tags:
                tags.extend([t.strip() for t in args.tags.split(",")])
            tags = list(set(tags))
            
            # Determine expiration for temporary memories
            expires = None
            if reason == "temporary":
                from datetime import datetime, timedelta
                expires = (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d")
            
            # Store it
            stored = store_memory(args.text, importance or "medium", tags, 
                                  expires=expires)
            result["stored"] = stored
            result["tags"] = tags
            result["importance"] = importance
        
        # Also get relevant context
        context = get_relevant_context(args.text, args.min_score, args.limit)
        result["context"] = context
        
        print(json.dumps(result) if args.json else result)