enterprise-memory-skill

import os\r import logging\r from pathlib import Path\r from typing import Dict, Any, Optional\r \r from openclaw.core.skill import BaseSkill\r from .vectorstorage import VectorStorage\r \r logger = logging.getLogger(name)\r \r class EnterpriseMemorySkill(BaseSkill):\r """\r Enterprise Memory Skill - OpenClaw 企业级长期记忆插件 (v1.1.1)\r """\r \r def init(self):\r super().init()\r self.vector_storage: Optional[VectorStorage] = None\r self.plugin_dir = Path(file).parent.absolute()\r self.config = {}\r \r def load_config(self):\r yaml_path = self.plugin_dir / "memory_config.yaml"\r try:\r import yaml\r with open(yaml_path, 'r', encoding='utf-8') as f:\r self.config = yaml.safe_load(f) or {}\r logger.info(f"✅ Loaded memory config from {yaml_path}")\r except Exception as e:\r logger.warning(f"⚠️ Failed to load memory_config.yaml: {e}")\r self.config = {}\r \r def _init_storage(self):\r if not self.vector_storage:\r try:\r yaml_path = str(self.plugin_dir / "memory_config.yaml")\r self.vector_storage = VectorStorage(config_path=yaml_path)\r logger.info("✅ VectorStorage initialized successfully.")\r except Exception as e:\r logger.error(f"❌ VectorStorage init failed: {e}")\r self.vector_storage = None\r \r def on_startup(self):\r logger.info("🚀 Enterprise Memory Skill Starting...")\r self.load_config()\r self._init_storage()\r if self.vector_storage:\r self.vector_storage.initialize_model()\r \r def on_shutdown(self):\r logger.info("🛑 Enterprise Memory Skill Shutting down...")\r if self.vector_storage:\r self.vector_storage._save_db()\r del self.vector_storage\r \r def get_context(self, query: str, context_limit: int = 2000) -> str:\r if not self.vector_storage:\r return ""\r try:\r top_k = self.config.get('retrieval_top_k', 5)\r results = self.vector_storage.retrieve_similar(query, top_k=top_k)\r if not results:\r return ""\r \r chunks = []\r total = 0\r threshold = self.config.get('retrieval_threshold', 0.82)\r for uuid_str, text, score in results:\r if total >= context_limit:\r break\r if score \x3C threshold:\r continue\r chunk = f"[Memory {uuid_str[:8]} | Score: {score:.3f}] {text}"\r chunks.append(chunk)\r total += len(chunk)\r return "
".join(chunks)\r except Exception as e:\r logger.error(f"get_context error: {e}")\r return ""\r \r def execute_action(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:\r if not self.vector_storage:\r return {"status": "error", "message": "Vector storage not initialized"}\r \r try:\r if action in ("remember", "ADD_MEMORY"):\r text = params.get("content") or params.get("text", "")\r confidence = float(params.get("confidence", 0.7))\r metadata = params.get("metadata", {})\r \r if not text or confidence \x3C self.config.get("storage_confidence", 0.7):\r return {"status": "skipped", "reason": "low confidence"}\r \r uuid_obj = self.vector_storage.add_text(text, metadata=metadata, confidence=confidence)\r return {"status": "success", "message": "Memory stored", "id": uuid_obj}\r \r elif action == "recall":\r query = params.get("query", "")\r top_k = params.get("top_k", self.config.get("retrieval_top_k", 5))\r results = self.vector_storage.retrieve_similar(query, top_k=top_k)\r return {\r "status": "success",\r "results": [{"id": u, "text": t, "score": float(s)} for u, t, s in results],\r "count": len(results)\r }\r \r elif action == "REJECT_MEMORY":\r content = params.get("content", "")\r logger.info(f"Memory rejected: {content}")\r # TODO: 可扩展实现按内容或 metadata 删除\r return {"status": "success", "action": "rejected"}\r \r return {"status": "error", "message": f"Unknown action: {action}"}\r except Exception as e:\r logger.error(f"execute_action error: {e}")\r return {"status": "error", "message": str(e)}