enterprise-memory-skill

SKILL.md

import os
import logging
from pathlib import Path
from typing import Dict, Any, Optional

from openclaw.core.skill import BaseSkill
from .vectorstorage import VectorStorage

logger = logging.getLogger(__name__)

class EnterpriseMemorySkill(BaseSkill):
    """
    Enterprise Memory Skill - OpenClaw 企业级长期记忆插件 (v1.1.1)
    """

    def __init__(self):
        super().__init__()
        self.vector_storage: Optional[VectorStorage] = None
        self.plugin_dir = Path(__file__).parent.absolute()
        self.config = {}

    def load_config(self):
        yaml_path = self.plugin_dir / "memory_config.yaml"
        try:
            import yaml
            with open(yaml_path, 'r', encoding='utf-8') as f:
                self.config = yaml.safe_load(f) or {}
            logger.info(f"✅ Loaded memory config from {yaml_path}")
        except Exception as e:
            logger.warning(f"⚠️ Failed to load memory_config.yaml: {e}")
            self.config = {}

    def _init_storage(self):
        if not self.vector_storage:
            try:
                yaml_path = str(self.plugin_dir / "memory_config.yaml")
                self.vector_storage = VectorStorage(config_path=yaml_path)
                logger.info("✅ VectorStorage initialized successfully.")
            except Exception as e:
                logger.error(f"❌ VectorStorage init failed: {e}")
                self.vector_storage = None

    def on_startup(self):
        logger.info("🚀 Enterprise Memory Skill Starting...")
        self.load_config()
        self._init_storage()
        if self.vector_storage:
            self.vector_storage.initialize_model()

    def on_shutdown(self):
        logger.info("🛑 Enterprise Memory Skill Shutting down...")
        if self.vector_storage:
            self.vector_storage._save_db()
            del self.vector_storage

    def get_context(self, query: str, context_limit: int = 2000) -> str:
        if not self.vector_storage:
            return ""
        try:
            top_k = self.config.get('retrieval_top_k', 5)
            results = self.vector_storage.retrieve_similar(query, top_k=top_k)
            if not results:
                return ""

            chunks = []
            total = 0
            threshold = self.config.get('retrieval_threshold', 0.82)
            for uuid_str, text, score in results:
                if total >= context_limit:
                    break
                if score < threshold:
                    continue
                chunk = f"[Memory {uuid_str[:8]} | Score: {score:.3f}] {text}"
                chunks.append(chunk)
                total += len(chunk)
            return "\n".join(chunks)
        except Exception as e:
            logger.error(f"get_context error: {e}")
            return ""

    def execute_action(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        if not self.vector_storage:
            return {"status": "error", "message": "Vector storage not initialized"}

        try:
            if action in ("remember", "ADD_MEMORY"):
                text = params.get("content") or params.get("text", "")
                confidence = float(params.get("confidence", 0.7))
                metadata = params.get("metadata", {})

                if not text or confidence < self.config.get("storage_confidence", 0.7):
                    return {"status": "skipped", "reason": "low confidence"}

                uuid_obj = self.vector_storage.add_text(text, metadata=metadata, confidence=confidence)
                return {"status": "success", "message": "Memory stored", "id": uuid_obj}

            elif action == "recall":
                query = params.get("query", "")
                top_k = params.get("top_k", self.config.get("retrieval_top_k", 5))
                results = self.vector_storage.retrieve_similar(query, top_k=top_k)
                return {
                    "status": "success",
                    "results": [{"id": u, "text": t, "score": float(s)} for u, t, s in results],
                    "count": len(results)
                }

            elif action == "REJECT_MEMORY":
                content = params.get("content", "")
                logger.info(f"Memory rejected: {content}")
                # TODO: 可扩展实现按内容或 metadata 删除
                return {"status": "success", "action": "rejected"}

            return {"status": "error", "message": f"Unknown action: {action}"}
        except Exception as e:
            logger.error(f"execute_action error: {e}")
            return {"status": "error", "message": str(e)}