File size: 6,191 Bytes
87b48de | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | """Typed knowledge graph with governed, versioned facts (decisions 18, 38, 47; section 11).
Entities: project, mission, hypothesis, candidate, composition, atom/site, structure, property, method,
solver_version, run, dataset, publication, evidence_claim, agent, software_version.
Facts are append-only versioned events with status proposed | provisional | verified | contradicted | deprecated.
Automatic ingestion never promotes to 'verified': see knowledge/governance.py.
"""
from __future__ import annotations
from typing import Any, Optional
from core.errors import ErrorCode, MatterOSError
from core.provenance import utcnow
from core.schemas import KnowledgeStatus, new_id
from knowledge import governance
from storage.db import Database
ENTITY_TYPES = ["project", "mission", "hypothesis", "candidate", "composition", "atom_site", "structure", "property",
"method", "solver_version", "run", "dataset", "publication", "evidence_claim", "agent", "software_version"]
class KnowledgeGraph:
def __init__(self, db: Database):
self.db = db
# ---------- entities / edges ----------
def upsert_entity(self, entity_id: str, etype: str, label: str = "", props: dict | None = None) -> str:
if etype not in ENTITY_TYPES:
raise MatterOSError(ErrorCode.INVALID_INPUT, f"unknown entity type {etype}")
if not self.db.one("SELECT 1 FROM kg_entities WHERE entity_id=?", (entity_id,)):
self.db.execute("INSERT INTO kg_entities VALUES (?,?,?,?,?)", (entity_id, etype, label, self.db.dumps(props or {}), utcnow()))
return entity_id
def link(self, src: str, dst: str, etype: str, props: dict | None = None, status: str = "provisional") -> str:
eid = new_id("edge")
self.db.execute("INSERT INTO kg_edges VALUES (?,?,?,?,?,?,?)", (eid, src, dst, etype, self.db.dumps(props or {}), status, utcnow()))
return eid
# ---------- governed facts ----------
def propose_event(self, subject: str, predicate: str, obj: Any, *, actor: str, evidence_refs: list[str] | None = None,
provenance: dict | None = None, note: str = "") -> dict:
status = governance.initial_status(evidence_refs or [], actor)
fact_id = new_id("fact")
return self._append(fact_id, 1, subject, predicate, obj, status, evidence_refs or [], provenance or {}, actor, note)
def _append(self, fact_id, version, subject, predicate, obj, status, evidence_refs, provenance, actor, note) -> dict:
ev = {"event_id": new_id("kge"), "fact_id": fact_id, "version": version, "subject": subject, "predicate": predicate,
"object": obj, "status": status, "evidence_refs": evidence_refs, "provenance": provenance, "actor": actor,
"note": note, "created_at": utcnow()}
self.db.execute("INSERT INTO kg_events VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
(ev["event_id"], fact_id, version, subject, predicate, self.db.dumps(obj), status,
self.db.dumps(evidence_refs), self.db.dumps(provenance), actor, note, ev["created_at"]))
return ev
def current(self, fact_id: str) -> dict:
r = self.db.one("SELECT * FROM kg_events WHERE fact_id=? ORDER BY version DESC LIMIT 1", (fact_id,))
if not r:
raise MatterOSError(ErrorCode.INVALID_INPUT, f"fact not found {fact_id}")
return self._row(r)
def history(self, fact_id: str) -> list[dict]:
return [self._row(r) for r in self.db.query("SELECT * FROM kg_events WHERE fact_id=? ORDER BY version", (fact_id,))]
def _row(self, r):
r = dict(r)
for k in ("object", "evidence_refs", "provenance"):
r[k] = self.db.loads(r[k])
return r
def transition(self, fact_id: str, new_status: str, *, actor: str, evidence_refs: list[str] | None = None,
evidence_check=None, note: str = "") -> dict:
cur = self.current(fact_id)
refs = list(dict.fromkeys((cur["evidence_refs"] or []) + (evidence_refs or [])))
governance.check_transition(cur["status"], new_status, refs, actor, evidence_check)
return self._append(fact_id, cur["version"] + 1, cur["subject"], cur["predicate"], cur["object"], new_status, refs,
cur["provenance"], actor, note)
def verify_event(self, fact_id: str, *, actor: str, evidence_refs: list[str], evidence_check=None) -> dict:
return self.transition(fact_id, KnowledgeStatus.VERIFIED.value, actor=actor, evidence_refs=evidence_refs,
evidence_check=evidence_check, note="verification requested")
def search_graph(self, text: str = "", status: Optional[str] = None, etype: Optional[str] = None, limit: int = 200) -> dict:
ents = self.db.query("SELECT * FROM kg_entities WHERE (label LIKE ? OR entity_id LIKE ?)" + (" AND type=?" if etype else "") +
" LIMIT ?", (f"%{text}%", f"%{text}%", *( [etype] if etype else []), limit))
# latest version per fact
facts = self.db.query("SELECT e.* FROM kg_events e JOIN (SELECT fact_id, MAX(version) v FROM kg_events GROUP BY fact_id) m "
"ON e.fact_id=m.fact_id AND e.version=m.v WHERE (e.subject LIKE ? OR e.predicate LIKE ? OR e.object LIKE ?)"
+ (" AND e.status=?" if status else "") + " ORDER BY e.created_at DESC LIMIT ?",
(f"%{text}%", f"%{text}%", f"%{text}%", *([status] if status else []), limit))
return {"entities": ents, "facts": [self._row(f) for f in facts]}
def status_counts(self) -> dict[str, int]:
rows = self.db.query("SELECT e.status, COUNT(*) n FROM kg_events e JOIN (SELECT fact_id, MAX(version) v FROM kg_events GROUP BY fact_id) m "
"ON e.fact_id=m.fact_id AND e.version=m.v GROUP BY e.status")
return {r["status"]: r["n"] for r in rows}
def edges(self, node: Optional[str] = None) -> list[dict]:
if node:
return self.db.query("SELECT * FROM kg_edges WHERE src=? OR dst=?", (node, node))
return self.db.query("SELECT * FROM kg_edges")
|