wehub-resource-sync
c3749daf48
Tests / test-linux (3.13) (push) Failing after 0s
Tests / test-linux (3.11) (push) Failing after 1s
Tests / lint (push) Failing after 0s
Tests / test-linux (3.9) (push) Failing after 1s
Docker / build (push) Failing after 1s
Docker / build-gpu (push) Failing after 2s
Tests / test-windows (push) Has been cancelled
Tests / test-macos (push) Has been cancelled
705 lines
27 KiB
Python
705 lines
27 KiB
Python
"""
|
|
knowledge_graph.py — Temporal Entity-Relationship Graph for MemPalace
|
|
=====================================================================
|
|
|
|
Real knowledge graph with:
|
|
- Entity nodes (people, projects, tools, concepts)
|
|
- Typed relationship edges (daughter_of, does, loves, works_on, etc.)
|
|
- Temporal validity (valid_from → valid_to — knows WHEN facts are true)
|
|
- Closet references (links back to the verbatim memory)
|
|
|
|
Storage: SQLite (local, no dependencies, no subscriptions)
|
|
Query: entity-first traversal with time filtering
|
|
|
|
This is what competes with Zep's temporal knowledge graph.
|
|
Zep uses Neo4j in the cloud ($25/mo+). We use SQLite locally (free).
|
|
|
|
Usage:
|
|
from mempalace.knowledge_graph import KnowledgeGraph
|
|
|
|
kg = KnowledgeGraph()
|
|
kg.add_triple("Max", "child_of", "Alice", valid_from="2015-04-01")
|
|
kg.add_triple("Max", "does", "swimming", valid_from="2025-01-01")
|
|
kg.add_triple("Max", "loves", "chess", valid_from="2025-10-01")
|
|
|
|
# Query: everything about Max
|
|
kg.query_entity("Max")
|
|
|
|
# Query: what was true about Max in January 2026?
|
|
kg.query_entity("Max", as_of="2026-01-15")
|
|
|
|
# Query: who is connected to Alice?
|
|
kg.query_entity("Alice", direction="both")
|
|
|
|
# Invalidate: Max's sports injury resolved
|
|
kg.invalidate("Max", "has_issue", "sports_injury", ended="2026-02-15")
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from .config import sanitize_iso_temporal
|
|
from .ids import make_triple_id
|
|
|
|
|
|
DEFAULT_KG_PATH = os.path.expanduser("~/.mempalace/knowledge_graph.sqlite3")
|
|
|
|
|
|
def _is_date_only_temporal(value: str) -> bool:
|
|
return isinstance(value, str) and len(value) == 10 and value[4] == "-" and value[7] == "-"
|
|
|
|
|
|
def _temporal_start_key(value: Optional[str]) -> Optional[str]:
|
|
"""Return the comparable instant for a valid_from/as_of value."""
|
|
|
|
if value is None:
|
|
return None
|
|
|
|
if _is_date_only_temporal(value):
|
|
return f"{value}T00:00:00Z"
|
|
|
|
return value
|
|
|
|
|
|
def _temporal_end_key(value: Optional[str]) -> Optional[str]:
|
|
"""Return the comparable instant for a valid_to value.
|
|
|
|
Date-only valid_to values represent the whole day for backward
|
|
compatibility with existing KG facts.
|
|
"""
|
|
|
|
if value is None:
|
|
return None
|
|
|
|
if _is_date_only_temporal(value):
|
|
return f"{value}T23:59:59Z"
|
|
|
|
return value
|
|
|
|
|
|
def _sql_temporal_start_expr(column: str) -> str:
|
|
"""SQLite expression for comparing valid_from-style temporal values."""
|
|
|
|
return (
|
|
f"CASE WHEN length({column}) = 10 "
|
|
f"AND substr({column}, 5, 1) = '-' "
|
|
f"AND substr({column}, 8, 1) = '-' "
|
|
f"THEN {column} || 'T00:00:00Z' ELSE {column} END"
|
|
)
|
|
|
|
|
|
def _sql_temporal_end_expr(column: str) -> str:
|
|
"""SQLite expression for comparing valid_to-style temporal values."""
|
|
|
|
return (
|
|
f"CASE WHEN length({column}) = 10 "
|
|
f"AND substr({column}, 5, 1) = '-' "
|
|
f"AND substr({column}, 8, 1) = '-' "
|
|
f"THEN {column} || 'T23:59:59Z' ELSE {column} END"
|
|
)
|
|
|
|
|
|
def _temporal_filter_sql(as_of: str) -> tuple[str, list[str]]:
|
|
"""Return SQL and parameters for an as-of temporal filter.
|
|
|
|
Date-only KG values are normalized for comparison:
|
|
|
|
- valid_from='2026-05-06' compares as '2026-05-06T00:00:00Z'
|
|
- valid_to='2026-05-06' compares as '2026-05-06T23:59:59Z'
|
|
|
|
This keeps legacy date-only facts working when callers query with
|
|
canonical UTC datetimes such as '2026-05-06T15:00:00Z'.
|
|
|
|
The upper bound is *strict* (``valid_to > as_of``): a fact whose
|
|
``valid_to`` equals the query instant has already ended at that instant,
|
|
so the interval is treated as half-open ``[valid_from, valid_to)``. This
|
|
is what lets a fact and its successor share a boundary instant without an
|
|
as-of query returning both. Date-only ``valid_to`` still expands to the
|
|
end of that day (``T23:59:59Z``), so a standalone date-only fact stays
|
|
valid through its whole final day exactly as before.
|
|
"""
|
|
|
|
as_of_key = _temporal_start_key(as_of)
|
|
valid_from_expr = _sql_temporal_start_expr("t.valid_from")
|
|
valid_to_expr = _sql_temporal_end_expr("t.valid_to")
|
|
|
|
return (
|
|
f" AND (t.valid_from IS NULL OR {valid_from_expr} <= ?) "
|
|
f"AND (t.valid_to IS NULL OR {valid_to_expr} > ?)",
|
|
[as_of_key, as_of_key],
|
|
)
|
|
|
|
|
|
class KnowledgeGraph:
|
|
def __init__(self, db_path: str = None):
|
|
self.db_path = db_path or DEFAULT_KG_PATH
|
|
db_parent = Path(self.db_path).parent
|
|
db_parent.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
db_parent.chmod(0o700)
|
|
except (OSError, NotImplementedError):
|
|
pass
|
|
self._connection = None
|
|
self._lock = threading.Lock()
|
|
self._init_db()
|
|
|
|
def _init_db(self):
|
|
conn = self._conn()
|
|
conn.executescript("""
|
|
PRAGMA journal_mode=WAL;
|
|
|
|
CREATE TABLE IF NOT EXISTS entities (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
type TEXT DEFAULT 'unknown',
|
|
properties TEXT DEFAULT '{}',
|
|
created_at TEXT DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS triples (
|
|
id TEXT PRIMARY KEY,
|
|
subject TEXT NOT NULL,
|
|
predicate TEXT NOT NULL,
|
|
object TEXT NOT NULL,
|
|
valid_from TEXT,
|
|
valid_to TEXT,
|
|
confidence REAL DEFAULT 1.0,
|
|
source_closet TEXT,
|
|
source_file TEXT,
|
|
source_drawer_id TEXT,
|
|
adapter_name TEXT,
|
|
extracted_at TEXT DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (subject) REFERENCES entities(id),
|
|
FOREIGN KEY (object) REFERENCES entities(id)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_triples_subject ON triples(subject);
|
|
CREATE INDEX IF NOT EXISTS idx_triples_object ON triples(object);
|
|
CREATE INDEX IF NOT EXISTS idx_triples_predicate ON triples(predicate);
|
|
CREATE INDEX IF NOT EXISTS idx_triples_valid ON triples(valid_from, valid_to);
|
|
""")
|
|
self._migrate_schema(conn)
|
|
conn.commit()
|
|
|
|
def _migrate_schema(self, conn):
|
|
"""Backwards-compatible schema migration for older triples tables.
|
|
|
|
Fresh palaces get ``source_drawer_id`` / ``adapter_name`` (RFC 002 §5.5)
|
|
directly from the canonical ``CREATE TABLE`` above, so this path is a
|
|
no-op on new installs. It exists for palaces that were created before
|
|
those columns were added: SQLite has no ``ADD COLUMN IF NOT EXISTS``,
|
|
so we introspect the schema and only issue the ALTER when the column
|
|
is missing.
|
|
"""
|
|
existing = {row["name"] for row in conn.execute("PRAGMA table_info(triples)")}
|
|
if "source_drawer_id" not in existing:
|
|
conn.execute("ALTER TABLE triples ADD COLUMN source_drawer_id TEXT")
|
|
if "adapter_name" not in existing:
|
|
conn.execute("ALTER TABLE triples ADD COLUMN adapter_name TEXT")
|
|
|
|
def _conn(self):
|
|
if self._connection is None:
|
|
self._connection = sqlite3.connect(self.db_path, timeout=10, check_same_thread=False)
|
|
self._connection.execute("PRAGMA journal_mode=WAL")
|
|
self._connection.row_factory = sqlite3.Row
|
|
return self._connection
|
|
|
|
def close(self):
|
|
"""Close the database connection."""
|
|
with self._lock:
|
|
if self._connection is not None:
|
|
self._connection.close()
|
|
self._connection = None
|
|
|
|
def __enter__(self):
|
|
"""Allow KnowledgeGraph to be used as a context manager."""
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
"""Close the SQLite connection when leaving a context manager block."""
|
|
self.close()
|
|
return False
|
|
|
|
def _entity_id(self, name: str) -> str:
|
|
return name.lower().replace(" ", "_").replace("'", "")
|
|
|
|
# ── Write operations ──────────────────────────────────────────────────
|
|
|
|
def add_entity(self, name: str, entity_type: str = "unknown", properties: dict = None):
|
|
"""Add or update an entity node."""
|
|
eid = self._entity_id(name)
|
|
props = json.dumps(properties or {})
|
|
with self._lock:
|
|
conn = self._conn()
|
|
with conn:
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO entities (id, name, type, properties) VALUES (?, ?, ?, ?)",
|
|
(eid, name, entity_type, props),
|
|
)
|
|
return eid
|
|
|
|
def add_triple(
|
|
self,
|
|
subject: str,
|
|
predicate: str,
|
|
obj: str,
|
|
valid_from: str = None,
|
|
valid_to: str = None,
|
|
confidence: float = 1.0,
|
|
source_closet: str = None,
|
|
source_file: str = None,
|
|
source_drawer_id: str = None,
|
|
adapter_name: str = None,
|
|
):
|
|
"""
|
|
Add a relationship triple: subject → predicate → object.
|
|
|
|
``source_drawer_id`` and ``adapter_name`` are RFC 002 §5.5 provenance
|
|
fields populated by adapters that advertise ``supports_kg_triples``;
|
|
they default to ``None`` so every existing caller stays
|
|
source-compatible.
|
|
|
|
Examples:
|
|
add_triple("Max", "child_of", "Alice", valid_from="2015-04-01")
|
|
add_triple("Max", "does", "swimming", valid_from="2025-01-01")
|
|
add_triple("Alice", "worried_about", "Max injury", valid_from="2026-01-01")
|
|
"""
|
|
|
|
valid_from = sanitize_iso_temporal(valid_from, "valid_from")
|
|
valid_to = sanitize_iso_temporal(valid_to, "valid_to")
|
|
|
|
# Reject inverted intervals. Use temporal comparison keys rather than
|
|
# raw string comparison so legacy date-only values and canonical UTC
|
|
# datetimes can safely coexist.
|
|
if (
|
|
valid_from is not None
|
|
and valid_to is not None
|
|
and _temporal_end_key(valid_to) < _temporal_start_key(valid_from)
|
|
):
|
|
raise ValueError(
|
|
f"valid_to={valid_to!r} is before valid_from={valid_from!r}; "
|
|
"an inverted interval would be invisible to every KG query"
|
|
)
|
|
|
|
sub_id = self._entity_id(subject)
|
|
obj_id = self._entity_id(obj)
|
|
pred = predicate.lower().replace(" ", "_")
|
|
|
|
# Auto-create entities if they don't exist
|
|
with self._lock:
|
|
conn = self._conn()
|
|
with conn:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO entities (id, name) VALUES (?, ?)",
|
|
(sub_id, subject),
|
|
)
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO entities (id, name) VALUES (?, ?)",
|
|
(obj_id, obj),
|
|
)
|
|
|
|
# Check for existing identical triple
|
|
existing = conn.execute(
|
|
"SELECT id FROM triples WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
|
|
(sub_id, pred, obj_id),
|
|
).fetchone()
|
|
if existing:
|
|
return existing["id"] # Already exists and still valid
|
|
|
|
triple_id = make_triple_id(
|
|
sub_id, pred, obj_id, valid_from, datetime.now().isoformat()
|
|
)
|
|
conn.execute(
|
|
"""INSERT INTO triples (
|
|
id, subject, predicate, object, valid_from, valid_to,
|
|
confidence, source_closet, source_file,
|
|
source_drawer_id, adapter_name
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
triple_id,
|
|
sub_id,
|
|
pred,
|
|
obj_id,
|
|
valid_from,
|
|
valid_to,
|
|
confidence,
|
|
source_closet,
|
|
source_file,
|
|
source_drawer_id,
|
|
adapter_name,
|
|
),
|
|
)
|
|
return triple_id
|
|
|
|
def invalidate(self, subject: str, predicate: str, obj: str, ended: str = None):
|
|
"""Mark a relationship as no longer valid (set valid_to date/time)."""
|
|
sub_id = self._entity_id(subject)
|
|
obj_id = self._entity_id(obj)
|
|
pred = predicate.lower().replace(" ", "_")
|
|
ended = sanitize_iso_temporal(ended or date.today().isoformat(), "ended")
|
|
|
|
with self._lock:
|
|
conn = self._conn()
|
|
with conn:
|
|
rows = conn.execute(
|
|
"SELECT id, valid_from FROM triples "
|
|
"WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
|
|
(sub_id, pred, obj_id),
|
|
).fetchall()
|
|
|
|
for row in rows:
|
|
valid_from = row["valid_from"]
|
|
if valid_from is not None and _temporal_end_key(ended) < _temporal_start_key(
|
|
valid_from
|
|
):
|
|
raise ValueError(
|
|
f"valid_to={ended!r} is before valid_from={valid_from!r}; "
|
|
"an inverted interval would be invisible to every KG query"
|
|
)
|
|
|
|
conn.execute(
|
|
"UPDATE triples SET valid_to=? "
|
|
"WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
|
|
(ended, sub_id, pred, obj_id),
|
|
)
|
|
|
|
def supersede(
|
|
self,
|
|
subject: str,
|
|
predicate: str,
|
|
old_obj: str,
|
|
new_obj: str,
|
|
at: str = None,
|
|
confidence: float = 1.0,
|
|
source_closet: str = None,
|
|
source_file: str = None,
|
|
source_drawer_id: str = None,
|
|
adapter_name: str = None,
|
|
):
|
|
"""Atomically replace one fact with another at a single shared boundary.
|
|
|
|
Closes the currently-open ``(subject, predicate, old_obj)`` triple with
|
|
``valid_to = at`` and opens ``(subject, predicate, new_obj)`` with
|
|
``valid_from = at`` in one transaction, at a single shared instant.
|
|
Paired with the half-open upper bound in ``_temporal_filter_sql``, an
|
|
as-of query at that instant returns only the successor.
|
|
|
|
This is the primitive for a value change. Hand-rolling a handover as
|
|
``invalidate(ended=D)`` + ``add_triple(valid_from=D)`` with date-only
|
|
``D`` leaves two facts sharing the whole day ``D`` (``valid_to`` expands
|
|
to ``T23:59:59Z`` while ``valid_from`` expands to ``T00:00:00Z``), so an
|
|
as-of query on ``D`` returns both. ``supersede`` avoids this by writing
|
|
one identical precise instant to both sides.
|
|
|
|
``at`` defaults to the current UTC instant. A date-only ``at`` is
|
|
normalized to ``<date>T00:00:00Z`` so both sides carry the same precise
|
|
value rather than the asymmetric whole-day expansion.
|
|
|
|
Returns the new triple's id. If no open ``old_obj`` triple exists the
|
|
successor is still opened, so ``supersede`` degrades to ``add_triple``.
|
|
"""
|
|
if at is None:
|
|
boundary = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
elif _is_date_only_temporal(at):
|
|
boundary = f"{at}T00:00:00Z"
|
|
else:
|
|
boundary = at
|
|
boundary = sanitize_iso_temporal(boundary, "at")
|
|
|
|
sub_id = self._entity_id(subject)
|
|
old_id = self._entity_id(old_obj)
|
|
new_id = self._entity_id(new_obj)
|
|
pred = predicate.lower().replace(" ", "_")
|
|
|
|
with self._lock:
|
|
conn = self._conn()
|
|
with conn:
|
|
# Only create entities we actually open a fact for. old_obj is
|
|
# matched by id in the UPDATE below whether or not its row
|
|
# exists, so inserting it would just orphan an entity when no
|
|
# open old fact is present (the degrade-to-add path).
|
|
for name, eid in ((subject, sub_id), (new_obj, new_id)):
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO entities (id, name) VALUES (?, ?)",
|
|
(eid, name),
|
|
)
|
|
|
|
# Reject a boundary that precedes the old fact's start — an
|
|
# inverted interval would be invisible to every KG query.
|
|
rows = conn.execute(
|
|
"SELECT valid_from FROM triples "
|
|
"WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
|
|
(sub_id, pred, old_id),
|
|
).fetchall()
|
|
for row in rows:
|
|
valid_from = row["valid_from"]
|
|
if valid_from is not None and _temporal_end_key(boundary) < _temporal_start_key(
|
|
valid_from
|
|
):
|
|
raise ValueError(
|
|
f"at={boundary!r} is before valid_from={valid_from!r}; "
|
|
"an inverted interval would be invisible to every KG query"
|
|
)
|
|
|
|
# Close the open old fact at the shared boundary.
|
|
conn.execute(
|
|
"UPDATE triples SET valid_to=? "
|
|
"WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
|
|
(boundary, sub_id, pred, old_id),
|
|
)
|
|
|
|
# Open the successor at the same instant (idempotent if already open).
|
|
existing = conn.execute(
|
|
"SELECT id FROM triples "
|
|
"WHERE subject=? AND predicate=? AND object=? AND valid_to IS NULL",
|
|
(sub_id, pred, new_id),
|
|
).fetchone()
|
|
if existing:
|
|
return existing["id"]
|
|
|
|
triple_id = make_triple_id(
|
|
sub_id, pred, new_id, boundary, datetime.now().isoformat()
|
|
)
|
|
conn.execute(
|
|
"""INSERT INTO triples (
|
|
id, subject, predicate, object, valid_from, valid_to,
|
|
confidence, source_closet, source_file,
|
|
source_drawer_id, adapter_name
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
triple_id,
|
|
sub_id,
|
|
pred,
|
|
new_id,
|
|
boundary,
|
|
None,
|
|
confidence,
|
|
source_closet,
|
|
source_file,
|
|
source_drawer_id,
|
|
adapter_name,
|
|
),
|
|
)
|
|
return triple_id
|
|
|
|
# ── Query operations ──────────────────────────────────────────────────
|
|
|
|
def query_entity(self, name: str, as_of: str = None, direction: str = "outgoing"):
|
|
"""
|
|
Get all relationships for an entity.
|
|
|
|
direction: "outgoing" (entity → ?), "incoming" (? → entity), "both"
|
|
as_of: ISO date or canonical UTC datetime — only return facts valid then
|
|
"""
|
|
as_of = sanitize_iso_temporal(as_of, "as_of")
|
|
eid = self._entity_id(name)
|
|
results = []
|
|
|
|
temporal_sql = ""
|
|
temporal_params = []
|
|
if as_of:
|
|
temporal_sql, temporal_params = _temporal_filter_sql(as_of)
|
|
|
|
with self._lock:
|
|
conn = self._conn()
|
|
|
|
if direction in ("outgoing", "both"):
|
|
query = (
|
|
"SELECT t.*, e.name as obj_name FROM triples t "
|
|
"JOIN entities e ON t.object = e.id WHERE t.subject = ?" + temporal_sql
|
|
)
|
|
params = [eid] + temporal_params
|
|
|
|
for row in conn.execute(query, params).fetchall():
|
|
results.append(
|
|
{
|
|
"direction": "outgoing",
|
|
"subject": name,
|
|
"predicate": row["predicate"],
|
|
"object": row["obj_name"],
|
|
"valid_from": row["valid_from"],
|
|
"valid_to": row["valid_to"],
|
|
"confidence": row["confidence"],
|
|
"source_closet": row["source_closet"],
|
|
"current": row["valid_to"] is None,
|
|
}
|
|
)
|
|
|
|
if direction in ("incoming", "both"):
|
|
query = (
|
|
"SELECT t.*, e.name as sub_name FROM triples t "
|
|
"JOIN entities e ON t.subject = e.id WHERE t.object = ?" + temporal_sql
|
|
)
|
|
params = [eid] + temporal_params
|
|
|
|
for row in conn.execute(query, params).fetchall():
|
|
results.append(
|
|
{
|
|
"direction": "incoming",
|
|
"subject": row["sub_name"],
|
|
"predicate": row["predicate"],
|
|
"object": name,
|
|
"valid_from": row["valid_from"],
|
|
"valid_to": row["valid_to"],
|
|
"confidence": row["confidence"],
|
|
"source_closet": row["source_closet"],
|
|
"current": row["valid_to"] is None,
|
|
}
|
|
)
|
|
|
|
return results
|
|
|
|
def query_relationship(self, predicate: str, as_of: str = None):
|
|
"""Get all triples with a given relationship type."""
|
|
as_of = sanitize_iso_temporal(as_of, "as_of")
|
|
pred = predicate.lower().replace(" ", "_")
|
|
|
|
query = """
|
|
SELECT t.*, s.name as sub_name, o.name as obj_name
|
|
FROM triples t
|
|
JOIN entities s ON t.subject = s.id
|
|
JOIN entities o ON t.object = o.id
|
|
WHERE t.predicate = ?
|
|
"""
|
|
params = [pred]
|
|
|
|
if as_of:
|
|
temporal_sql, temporal_params = _temporal_filter_sql(as_of)
|
|
query += temporal_sql
|
|
params.extend(temporal_params)
|
|
|
|
results = []
|
|
with self._lock:
|
|
conn = self._conn()
|
|
for row in conn.execute(query, params).fetchall():
|
|
results.append(
|
|
{
|
|
"subject": row["sub_name"],
|
|
"predicate": pred,
|
|
"object": row["obj_name"],
|
|
"valid_from": row["valid_from"],
|
|
"valid_to": row["valid_to"],
|
|
"current": row["valid_to"] is None,
|
|
}
|
|
)
|
|
return results
|
|
|
|
def timeline(self, entity_name: str = None):
|
|
"""Get all facts in chronological order, optionally filtered by entity."""
|
|
with self._lock:
|
|
conn = self._conn()
|
|
if entity_name:
|
|
eid = self._entity_id(entity_name)
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT t.*, s.name as sub_name, o.name as obj_name
|
|
FROM triples t
|
|
JOIN entities s ON t.subject = s.id
|
|
JOIN entities o ON t.object = o.id
|
|
WHERE (t.subject = ? OR t.object = ?)
|
|
ORDER BY t.valid_from ASC NULLS LAST
|
|
LIMIT 100
|
|
""",
|
|
(eid, eid),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute("""
|
|
SELECT t.*, s.name as sub_name, o.name as obj_name
|
|
FROM triples t
|
|
JOIN entities s ON t.subject = s.id
|
|
JOIN entities o ON t.object = o.id
|
|
ORDER BY t.valid_from ASC NULLS LAST
|
|
LIMIT 100
|
|
""").fetchall()
|
|
|
|
return [
|
|
{
|
|
"subject": r["sub_name"],
|
|
"predicate": r["predicate"],
|
|
"object": r["obj_name"],
|
|
"valid_from": r["valid_from"],
|
|
"valid_to": r["valid_to"],
|
|
"current": r["valid_to"] is None,
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
# ── Stats ─────────────────────────────────────────────────────────────
|
|
|
|
def stats(self):
|
|
with self._lock:
|
|
conn = self._conn()
|
|
entities = conn.execute("SELECT COUNT(*) as cnt FROM entities").fetchone()["cnt"]
|
|
triples = conn.execute("SELECT COUNT(*) as cnt FROM triples").fetchone()["cnt"]
|
|
current = conn.execute(
|
|
"SELECT COUNT(*) as cnt FROM triples WHERE valid_to IS NULL"
|
|
).fetchone()["cnt"]
|
|
expired = triples - current
|
|
predicates = [
|
|
r["predicate"]
|
|
for r in conn.execute(
|
|
"SELECT DISTINCT predicate FROM triples ORDER BY predicate"
|
|
).fetchall()
|
|
]
|
|
return {
|
|
"entities": entities,
|
|
"triples": triples,
|
|
"current_facts": current,
|
|
"expired_facts": expired,
|
|
"relationship_types": predicates,
|
|
}
|
|
|
|
# ── Seed from known facts ─────────────────────────────────────────────
|
|
|
|
def seed_from_entity_facts(self, entity_facts: dict):
|
|
"""
|
|
Seed the knowledge graph from fact_checker.py ENTITY_FACTS.
|
|
This bootstraps the graph with known ground truth.
|
|
"""
|
|
for key, facts in entity_facts.items():
|
|
name = facts.get("full_name", key.capitalize())
|
|
etype = facts.get("type", "person")
|
|
self.add_entity(
|
|
name,
|
|
etype,
|
|
{
|
|
"gender": facts.get("gender", ""),
|
|
"birthday": facts.get("birthday", ""),
|
|
},
|
|
)
|
|
|
|
# Relationships
|
|
parent = facts.get("parent")
|
|
if parent:
|
|
self.add_triple(
|
|
name, "child_of", parent.capitalize(), valid_from=facts.get("birthday")
|
|
)
|
|
|
|
partner = facts.get("partner")
|
|
if partner:
|
|
self.add_triple(name, "married_to", partner.capitalize())
|
|
|
|
relationship = facts.get("relationship", "")
|
|
if relationship == "daughter":
|
|
self.add_triple(
|
|
name,
|
|
"is_child_of",
|
|
facts.get("parent", "").capitalize() or name,
|
|
valid_from=facts.get("birthday"),
|
|
)
|
|
elif relationship == "husband":
|
|
self.add_triple(name, "is_partner_of", facts.get("partner", name).capitalize())
|
|
elif relationship == "brother":
|
|
self.add_triple(name, "is_sibling_of", facts.get("sibling", name).capitalize())
|
|
elif relationship == "dog":
|
|
self.add_triple(name, "is_pet_of", facts.get("owner", name).capitalize())
|
|
self.add_entity(name, "animal")
|
|
|
|
# Interests
|
|
for interest in facts.get("interests", []):
|
|
self.add_triple(name, "loves", interest.capitalize(), valid_from="2025-01-01")
|