my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Mnemosyne Temporal Triples
#3	Time-aware knowledge graph on top of SQLite.
#4	Tracks when facts were true, enabling contradiction detection and historical queries.
#5	"""
#6
#7	import os
#8	import sqlite3
#9	import tempfile
#10	from datetime import datetime
#11	from pathlib import Path
#12	from typing import List, Dict, Optional
#13
#14	LEGACY_DATA_DIR = Path.home() / ".hermes" / "mnemosyne" / "data"
#15	DEFAULT_DATA_DIR = Path(os.environ.get("MNEMOSYNE_DATA_DIR", LEGACY_DATA_DIR))
#16	DEFAULT_DB = DEFAULT_DATA_DIR / "triples.db"
#17	LEGACY_DB = LEGACY_DATA_DIR / "triples.db"
#18
#19
#20	def _copy_legacy_db(source: Path, destination: Path) -> None:
#21	"""Copy a SQLite DB using SQLite's backup API for a consistent snapshot."""
#22	destination.parent.mkdir(parents=True, exist_ok=True)
#23	with tempfile.NamedTemporaryFile(
#24	prefix=f".{destination.name}.",
#25	suffix=".tmp",
#26	dir=destination.parent,
#27	delete=False,
#28	) as temp_file:
#29	temp_path = Path(temp_file.name)
#30
#31	try:
#32	source_conn = sqlite3.connect(f"file:{source}?mode=ro", uri=True)
#33	try:
#34	dest_conn = sqlite3.connect(str(temp_path))
#35	try:
#36	source_conn.backup(dest_conn)
#37	finally:
#38	dest_conn.close()
#39	finally:
#40	source_conn.close()
#41
#42	if not destination.exists():
#43	temp_path.replace(destination)
#44	else:
#45	temp_path.unlink(missing_ok=True)
#46	except Exception:
#47	temp_path.unlink(missing_ok=True)
#48	raise
#49
#50
#51	def _resolve_default_db() -> Path:
#52	"""Return the default triples DB, copying legacy data into place if needed."""
#53	if DEFAULT_DATA_DIR != LEGACY_DATA_DIR and not DEFAULT_DB.exists() and LEGACY_DB.exists():
#54	_copy_legacy_db(LEGACY_DB, DEFAULT_DB)
#55	return DEFAULT_DB
#56
#57
#58	def _get_conn(db_path = None) -> sqlite3.Connection:
#59	path = Path(db_path) if db_path else _resolve_default_db()
#60	path.parent.mkdir(parents=True, exist_ok=True)
#61	conn = sqlite3.connect(str(path), check_same_thread=False)
#62	conn.row_factory = sqlite3.Row
#63	return conn
#64
#65
#66	def init_triples(db_path: Path = None):
#67	conn = _get_conn(db_path)
#68	cursor = conn.cursor()
#69
#70	cursor.execute("""
#71	CREATE TABLE IF NOT EXISTS triples (
#72	id INTEGER PRIMARY KEY AUTOINCREMENT,
#73	subject TEXT NOT NULL,
#74	predicate TEXT NOT NULL,
#75	object TEXT NOT NULL,
#76	valid_from TEXT NOT NULL,
#77	valid_until TEXT,
#78	source TEXT,
#79	confidence REAL DEFAULT 1.0,
#80	created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
#81	)
#82	""")
#83
#84	cursor.execute("CREATE INDEX IF NOT EXISTS idx_triples_subject ON triples(subject)")
#85	cursor.execute("CREATE INDEX IF NOT EXISTS idx_triples_predicate ON triples(predicate)")
#86	cursor.execute("CREATE INDEX IF NOT EXISTS idx_triples_object ON triples(object)")
#87	cursor.execute("CREATE INDEX IF NOT EXISTS idx_triples_valid_from ON triples(valid_from)")
#88
#89	conn.commit()
#90
#91
#92	class TripleStore:
#93	"""
#94	Temporal knowledge graph for Mnemosyne.
#95
#96	Example:
#97	>>> kg = TripleStore()
#98	>>> kg.add("Maya", "assigned_to", "auth-migration", valid_from="2026-01-15")
#99	>>> kg.query("Maya", as_of="2026-01-20")
#100	"""
#101
#102	def __init__(self, db_path: Path = None):
#103	self.db_path = Path(db_path) if db_path else _resolve_default_db()
#104	init_triples(self.db_path)
#105	self.conn = _get_conn(self.db_path)
#106
#107	def add(self, subject: str, predicate: str, object: str,
#108	valid_from: str = None, source: str = "inferred",
#109	confidence: float = 1.0) -> int:
#110	"""
#111	Add a temporal triple. Automatically closes previous matching triples.
#112	"""
#113	valid_from = valid_from or datetime.now().isoformat()[:10]
#114
#115	# Invalidate previous triples for same (subject, predicate)
#116	cursor = self.conn.cursor()
#117	cursor.execute("""
#118	UPDATE triples
#119	SET valid_until = ?
#120	WHERE subject = ? AND predicate = ? AND valid_until IS NULL
#121	""", (valid_from, subject, predicate))
#122
#123	# Insert new triple
#124	cursor.execute("""
#125	INSERT INTO triples (subject, predicate, object, valid_from, source, confidence)
#126	VALUES (?, ?, ?, ?, ?, ?)
#127	""", (subject, predicate, object, valid_from, source, confidence))
#128
#129	self.conn.commit()
#130	return cursor.lastrowid
#131
#132	def query(self, subject: str = None, predicate: str = None,
#133	object: str = None, as_of: str = None) -> List[Dict]:
#134	"""
#135	Query triples, optionally as of a specific date.
#136	"""
#137	cursor = self.conn.cursor()
#138	as_of = as_of or datetime.now().isoformat()[:10]
#139
#140	conditions = []
#141	params = []
#142
#143	if subject:
#144	conditions.append("subject = ?")
#145	params.append(subject)
#146	if predicate:
#147	conditions.append("predicate = ?")
#148	params.append(predicate)
#149	if object:
#150	conditions.append("object = ?")
#151	params.append(object)
#152
#153	# Temporal filter: valid at as_of date
#154	conditions.append("valid_from <= ?")
#155	params.append(as_of)
#156	conditions.append("(valid_until IS NULL OR valid_until > ?)")
#157	params.append(as_of)
#158
#159	where_clause = " AND ".join(conditions)
#160	cursor.execute(f"SELECT * FROM triples WHERE {where_clause} ORDER BY valid_from DESC", params)
#161
#162	return [dict(row) for row in cursor.fetchall()]
#163
#164	def query_by_predicate(self, predicate: str, object: str = None, subject: str = None) -> List[Dict]:
#165	"""
#166	Query triples by predicate, optionally filtering by object or subject.
#167
#168	Useful for entity queries: find all memories that mention a specific entity.
#169
#170	Examples:
#171	>>> kg.query_by_predicate("mentions", "Abdias")
#172	# Returns all triples where someone/something mentions Abdias
#173
#174	>>> kg.query_by_predicate("mentions", subject="memory_123")
#175	# Returns entities mentioned by memory_123
#176	"""
#177	cursor = self.conn.cursor()
#178
#179	conditions = ["predicate = ?"]
#180	params = [predicate]
#181
#182	if object:
#183	conditions.append("object = ?")
#184	params.append(object)
#185	if subject:
#186	conditions.append("subject = ?")
#187	params.append(subject)
#188
#189	where_clause = " AND ".join(conditions)
#190	cursor.execute(f"SELECT * FROM triples WHERE {where_clause} ORDER BY created_at DESC", params)
#191
#192	return [dict(row) for row in cursor.fetchall()]
#193
#194	def get_distinct_objects(self, predicate: str) -> List[str]:
#195	"""
#196	Get all distinct object values for a given predicate.
#197
#198	Useful for building entity lists: get all known entities that have been mentioned.
#199	"""
#200	cursor = self.conn.cursor()
#201	cursor.execute(
#202	"SELECT DISTINCT object FROM triples WHERE predicate = ? ORDER BY object",
#203	(predicate,)
#204	)
#205	return [row["object"] for row in cursor.fetchall()]
#206
#207	def add_facts(self, memory_id: str, facts: List[str], source: str = "", confidence: float = 0.7) -> int:
#208	"""
#209	Batch-store extracted facts as triples.
#210
#211	Args:
#212	memory_id: The subject memory ID
#213	facts: List of fact strings to store
#214	source: Source identifier
#215	confidence: Confidence score for extracted facts (default 0.7)
#216
#217	Returns:
#218	Number of facts stored
#219	"""
#220	if not facts:
#221	return 0
#222
#223	stored = 0
#224	for fact in facts:
#225	if fact and len(fact) > 10:
#226	self.add(
#227	subject=memory_id,
#228	predicate="fact",
#229	object=fact,
#230	source=source,
#231	confidence=confidence
#232	)
#233	stored += 1
#234
#235	return stored
#236
#237	def export_all(self) -> List[Dict]:
#238	"""Export all triples to a list of dictionaries."""
#239	cursor = self.conn.cursor()
#240	cursor.execute("""
#241	SELECT id, subject, predicate, object, valid_from, valid_until,
#242	source, confidence, created_at
#243	FROM triples
#244	ORDER BY id
#245	""")
#246	return [dict(row) for row in cursor.fetchall()]
#247
#248	def import_all(self, triples: List[Dict], force: bool = False) -> Dict:
#249	"""
#250	Import triples from a list of dictionaries.
#251	Idempotent by default: skips records whose id already exists.
#252	Set force=True to overwrite.
#253	Returns import statistics.
#254	"""
#255	stats = {"inserted": 0, "skipped": 0, "overwritten": 0}
#256	cursor = self.conn.cursor()
#257	for item in triples:
#258	tid = item.get("id")
#259	cursor.execute("SELECT 1 FROM triples WHERE id = ?", (tid,))
#260	exists = cursor.fetchone() is not None
#261	if exists and not force:
#262	stats["skipped"] += 1
#263	continue
#264	if exists and force:
#265	cursor.execute("DELETE FROM triples WHERE id = ?", (tid,))
#266	stats["overwritten"] += 1
#267	else:
#268	stats["inserted"] += 1
#269	cursor.execute("""
#270	INSERT INTO triples (id, subject, predicate, object, valid_from,
#271	valid_until, source, confidence, created_at)
#272	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
#273	""", (
#274	tid, item.get("subject"), item.get("predicate"), item.get("object"),
#275	item.get("valid_from"), item.get("valid_until"),
#276	item.get("source", "imported"), item.get("confidence", 1.0),
#277	item.get("created_at")
#278	))
#279	self.conn.commit()
#280	return stats
#281
#282
#283	# ---------------------------------------------------------------------------
#284	# Module-level convenience functions
#285	# ---------------------------------------------------------------------------
#286
#287	def add_triple(subject: str, predicate: str, object: str,
#288	valid_from: str = None, source: str = "inferred",
#289	confidence: float = 1.0, db_path: Path = None) -> int:
#290	"""
#291	Add a temporal triple without instantiating TripleStore manually.
#292	Optional db_path aligns with BEAM memory database when used from Hermes.
#293	"""
#294	store = TripleStore(db_path=db_path)
#295	return store.add(subject, predicate, object,
#296	valid_from=valid_from, source=source, confidence=confidence)
#297
#298
#299	def query_triples(subject: str = None, predicate: str = None,
#300	object: str = None, as_of: str = None,
#301	db_path: Path = None) -> List[Dict]:
#302	"""
#303	Query temporal triples without instantiating TripleStore manually.
#304	Optional db_path aligns with BEAM memory database when used from Hermes.
#305	"""
#306	store = TripleStore(db_path=db_path)
#307	return store.query(subject=subject, predicate=predicate,
#308	object=object, as_of=as_of)
#309

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public