Upload manas/core/semantic.py with huggingface_hub
Browse files- manas/core/semantic.py +65 -0
manas/core/semantic.py
ADDED
|
@@ -0,0 +1,65 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""Semantic memory — an editable, verifiable knowledge graph (DOC 4 §5.3, GraphRAG).
|
| 2 |
+
|
| 3 |
+
The model can READ and WRITE facts. Only verifier-confirmed facts are written
|
| 4 |
+
as 'verified'. Supports lookup for grounding and contradiction checks.
|
| 5 |
+
"""
|
| 6 |
+
import re
|
| 7 |
+
import time
|
| 8 |
+
|
| 9 |
+
|
| 10 |
+
def _norm(s):
|
| 11 |
+
return re.sub(r"\s+", " ", s.strip().lower())
|
| 12 |
+
|
| 13 |
+
|
| 14 |
+
class SemanticMemory:
|
| 15 |
+
def __init__(self):
|
| 16 |
+
# subject -> list of fact dicts
|
| 17 |
+
self.facts = {}
|
| 18 |
+
|
| 19 |
+
@staticmethod
|
| 20 |
+
def _subject(text):
|
| 21 |
+
# crude subject extraction: first content chunk before 'is/are/was'
|
| 22 |
+
m = re.split(r"\b(is|are|was|were|has|have)\b", text, maxsplit=1)
|
| 23 |
+
return _norm(m[0]) if m else _norm(text)[:48]
|
| 24 |
+
|
| 25 |
+
def write(self, fact: str, verified: bool = False, source: str = "model"):
|
| 26 |
+
subj = self._subject(fact)
|
| 27 |
+
entry = {"fact": fact.strip(), "verified": bool(verified),
|
| 28 |
+
"source": source, "ts": time.time()}
|
| 29 |
+
self.facts.setdefault(subj, [])
|
| 30 |
+
# avoid exact dupes
|
| 31 |
+
if not any(_norm(e["fact"]) == _norm(fact) for e in self.facts[subj]):
|
| 32 |
+
self.facts[subj].append(entry)
|
| 33 |
+
return subj
|
| 34 |
+
|
| 35 |
+
def read(self, query: str, k: int = 4):
|
| 36 |
+
q = set(re.findall(r"[a-z]{3,}", query.lower()))
|
| 37 |
+
scored = []
|
| 38 |
+
for subj, entries in self.facts.items():
|
| 39 |
+
for e in entries:
|
| 40 |
+
fw = set(re.findall(r"[a-z]{3,}", e["fact"].lower()))
|
| 41 |
+
ov = len(q & fw)
|
| 42 |
+
if ov:
|
| 43 |
+
scored.append((ov, e))
|
| 44 |
+
scored.sort(key=lambda x: (-x[0], not x[1]["verified"]))
|
| 45 |
+
return [e for _, e in scored[:k]]
|
| 46 |
+
|
| 47 |
+
def verify_fact(self, fact: str):
|
| 48 |
+
"""Mark a stored fact verified (called by the symbolic verifier)."""
|
| 49 |
+
subj = self._subject(fact)
|
| 50 |
+
for e in self.facts.get(subj, []):
|
| 51 |
+
if _norm(e["fact"]) == _norm(fact):
|
| 52 |
+
e["verified"] = True
|
| 53 |
+
return True
|
| 54 |
+
return False
|
| 55 |
+
|
| 56 |
+
def contradicts(self, fact: str) -> bool:
|
| 57 |
+
"""True if a stored verified fact on the same subject says something different."""
|
| 58 |
+
subj = self._subject(fact)
|
| 59 |
+
for e in self.facts.get(subj, []):
|
| 60 |
+
if e["verified"] and _norm(e["fact"]) != _norm(fact):
|
| 61 |
+
return True
|
| 62 |
+
return False
|
| 63 |
+
|
| 64 |
+
def __len__(self):
|
| 65 |
+
return sum(len(v) for v in self.facts.values())
|