56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
"""Semantic memory: facts, policies, domain knowledge for AGI."""
|
|
|
|
from typing import Any
|
|
|
|
from fusionagi._logger import logger
|
|
|
|
|
|
class SemanticMemory:
|
|
"""Stores facts, policies, domain knowledge. Queryable and updatable."""
|
|
|
|
def __init__(self, max_facts: int = 50000) -> None:
|
|
self._facts: dict[str, dict[str, Any]] = {}
|
|
self._by_key: dict[str, list[str]] = {}
|
|
self._max_facts = max_facts
|
|
|
|
def add_fact(self, fact_id: str, statement: str, source: str = "", domain: str = "", metadata: dict[str, Any] | None = None) -> None:
|
|
if len(self._facts) >= self._max_facts and fact_id not in self._facts:
|
|
self._evict_one()
|
|
entry = {"statement": statement, "source": source, "domain": domain, "metadata": metadata or {}}
|
|
self._facts[fact_id] = entry
|
|
if domain:
|
|
self._by_key.setdefault(domain, []).append(fact_id)
|
|
logger.debug("Semantic memory: fact added", extra={"fact_id": fact_id, "domain": domain})
|
|
|
|
def get_fact(self, fact_id: str) -> dict[str, Any] | None:
|
|
return self._facts.get(fact_id)
|
|
|
|
def query(self, domain: str | None = None, limit: int = 100) -> list[dict[str, Any]]:
|
|
if domain:
|
|
ids = self._by_key.get(domain, [])[-limit:]
|
|
return [self._facts[id] for id in ids if id in self._facts]
|
|
return list(self._facts.values())[-limit:]
|
|
|
|
def update_fact(self, fact_id: str, **kwargs: Any) -> bool:
|
|
if fact_id not in self._facts:
|
|
return False
|
|
for k, v in kwargs.items():
|
|
if k in self._facts[fact_id]:
|
|
self._facts[fact_id][k] = v
|
|
return True
|
|
|
|
def forget(self, fact_id: str) -> bool:
|
|
if fact_id not in self._facts:
|
|
return False
|
|
entry = self._facts.pop(fact_id)
|
|
domain = entry.get("domain", "")
|
|
if domain and fact_id in self._by_key.get(domain, []):
|
|
self._by_key[domain] = [x for x in self._by_key[domain] if x != fact_id]
|
|
return True
|
|
|
|
def _evict_one(self) -> None:
|
|
if not self._facts:
|
|
return
|
|
rid = next(iter(self._facts))
|
|
self.forget(rid)
|