Cómo construir un sistema operativo de agente de IA persistente estilo EverMem con memoria jerárquica, recuperación de vectores FAISS, almacenamiento SQLite y consolidación de memoria automatizada

clase EverMemAgentOS: def __init__( self, workdir: str = “/content/evermem_agent_os”, db_name: str = “evermem.sqlite”, embedding_model: str = “sentence-transformers/all-MiniLM-L6-v2”, gen_model: str = “google/flan-t5-small”, stm_max_turns: int = 10, ltm_topk: int = 6, consolidate_every: int = 8, consolidate_trigger_tokens: int = 1400, compress_target_chars: int = 420, semilla: int = 7,): self.workdir = workdir _ensure_dir(self.workdir) self.db_path = os.path.join(self.workdir, db_name) self.embedder = SentenceTransformer(embedding_model) self.embed_dim = self.embedder.get_sentence_embedding_dimension() self.tokenizer = AutoTokenizer.from_pretrained(gen_model) self.model = AutoModelForSeq2SeqLM.from_pretrained(gen_model) self.model.to(self.device) self.model.eval() self.stm_max_turns = stm_max_turns self.ltm_topk = ltm_topk self.consolidate_every = consolidar_every self.consolidate_trigger_tokens = consolidar_trigger_tokens self.compress_target_chars = compress_target_chars np.random.seed(seed) self._init_db() self._init_faiss() self.stm: Lista[Dict[str, str]]= []
self.turns = 0 def _init_db(self): conn = sqlite3.connect(self.db_path) cur = conn.cursor() cur.execute( “”” CREAR TABLA SI NO EXISTE memorias ( TEXTO medio CLAVE PRIMARIA, TEXTO de rol, TEXTO de texto, TEXTO creado_ts INTEGER, importancia REAL, tokens_est INTEGER, TEXTO meta_json ) “”” ) cur.execute( “”” CREAR TABLA SI NO EXISTE kv_store ( k TEXTO CLAVE PRIMARIA, v_json TEXTO, actualizado_ts INTEGER ) “”” ) cur.execute( “”” CREAR TABLA SI NO EXISTE consolidaciones ( cid TEXTO CLAVE PRIMARIA, creado_ts INTEGER, TEXTO resumen, source_mids_json TEXTO ) “”” ) conn.commit() conn.close() def _init_faiss(self): self.faiss_index_path = os.path.join(self.workdir, “faiss.index”) self.faiss_map_path = os.path.join(self.workdir, “faiss_map.json”) si os.path.exists(self.faiss_index_path) y os.path.exists(self.faiss_map_path): self.index = faiss.read_index(self.faiss_index_path) con open(self.faiss_map_path, “r”, encoding=”utf-8″) como f: self.id_map = json.load(f) self.id_map = {int(k): v para k, v en self.id_map.items()} self.next_faiss_id = (max(self.id_map.keys()) + 1) if self.id_map else 0 return self.index = faiss.IndexFlatIP(self.embed_dim) self.id_map: Dict[int, str] = {} self.next_faiss_id = 0 self._persist_faiss() def _persist_faiss(self): faiss.write_index(self.index, self.faiss_index_path) con open(self.faiss_map_path, “w”, encoding=”utf-8″) como f: json.dump({str(k): v para k, v in self.id_map.items()}, f) def _embed(self, textos: Lista[str]) -> np.ndarray: vecs = self.embedder.encode(texts, convert_to_numpy=True, normalize_embeddings=True) if vecs.ndim == 1: vecs = vecs.reshape(1, -1) return vecs.astype(“float32”) def _tokens_est(self, text: str) -> int: text = text o “” return max(1, int(len(text.split()) * 1.25)) def _importance_score(self, rol: str, text: str, meta: Dict[str, Any]) -> float: base = 0.35 length_bonus = min(0.45, math.log1p(len(text)) / 20.0) role_bonus = 0.08 if role == “user” else 0.03 pin = 0.35 if meta.get(“fijado”) else 0.0 signal = meta.get(“signal”, “”) signal_bonus = 0.18 if signal in {“decisión”, “preferencia”, “hecho”, “tarea”} else 0.0 q_bonus = 0.06 if “?” en texto else 0.0 number_bonus = 0.05 si hay alguno(ch.isdigit() para ch en texto) else 0.0 return float(min(1.0, base + length_bonus + role_bonus + pin + signal_bonus + q_bonus + number_bonus)) def upsert_kv(self, k: str, v: Any): conn = sqlite3.connect(self.db_path) cur = conn.cursor() cur.execute( “INSERTAR EN kv_store (k, v_json, actualizado_ts) VALORES (?, ?, ?) EN CONFLICTO(k) ACTUALIZAR ESTABLECER v_json=excluded.v_json, actualizado_ts=excluido.updated_ts”, (k, json.dumps(v, asegurar_ascii=False), _now_ts()), ) conn.commit() conn.close() def get_kv(self, k: str, default=None): conn = sqlite3.connect(self.db_path) cur = conn.cursor() cur.execute(“SELECCIONE v_json DESDE kv_store DONDE k=?”, (k,)) fila = cur.fetchone() conn.close() si no es fila: devuelve el valor predeterminado intente: devuelve json.loads(fila[0]) excepto Excepción: devolver valor predeterminado def add_memory(self, rol: str, text: str, meta: Opcional[Dict[str, Any]]= Ninguno) -> str: meta = meta o {} texto = (texto o “”).strip() mid = meta.get(“mid”) o f”m:{_sha(f'{_now_ts()}::{role}::{texto[:80]}::{np.random.randint(0, 10**9)}’)}” creado_ts = _now_ts() tokens_est = self._tokens_est(texto) importancia = float(meta.get(“importancia”)) si meta.get(“importancia”) no es Ninguno más self._importance_score(role, text, meta) conn = sqlite3.connect(self.db_path) cur = conn.cursor() cur.execute( “INSERTAR O REEMPLAZAR EN memorias (mid, rol, texto, creado_ts, importancia, tokens_est, meta_json) VALORES (?, ?, ?, ?, ?, ?, ?)”, (mid, rol, texto, creado_ts, importancia, tokens_est, json.dumps(meta, asegurar_ascii=False)), ) conn.commit() conn.close() vec = self._embed([text]) fid = self.next_faiss_id self.next_faiss_id += 1 self.index.add(vec) self.id_map[fid] = medio self._persist_faiss() regresar medio