diff --git a/.gitignore b/.gitignore index 743946b..c330c52 100644 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,6 @@ build/ spell.mpedb* *.mpedb *.mpedb.wlock + +# The graph tables the app records finished runs into. +knowledge.graph.db* diff --git a/dev/tests/test_graphdb.py b/dev/tests/test_graphdb.py new file mode 100644 index 0000000..05180a0 --- /dev/null +++ b/dev/tests/test_graphdb.py @@ -0,0 +1,151 @@ +"""The graph as tables: recorded once per run, queryable in SQL, in memory too. + +The RDF is the graph as a document; this is the graph as a database. Both draw +from the same job events, so the test that matters most here is the one that +feeds a settle-shaped event stream in and asks the questions a consumer would: +which findings did the run make, how was agreement reached, what did the vote +say. If SQL can answer those, the tables carry the structure. +""" + +from __future__ import annotations + +import json +import time + +import pytest +from conftest import normal_script, step + +from mpe_lkg.backends import DeterministicEmbedding, ScriptedChat +from mpe_lkg.graphdb import GraphDB +from mpe_lkg.jobs import Job + + +def settled_job() -> Job: + """A job holding the events a settle run produces, without running a model.""" + job = Job("How many minutes are there in a fortnight?") + job.mode = "settle" + job.session = "clocks" + for event in [ + {"type": "round", "round": 1, "of": 3}, + {"type": "finding", "question": "How many days are in a fortnight?", + "answer": "A fortnight has 14 days."}, + {"type": "finding", "question": "How many minutes are in a day?", + "answer": "There are 1440 minutes in a day."}, + {"type": "finding", "level": 1, "question": "a sub-run's own finding", + "answer": "must not be recorded at the top level"}, + {"type": "step", "step": 1, "title": "Convert", "content": "Ask for it."}, + {"type": "calc", "step": 1, "expression": "14*1440", "value": "20,160"}, + {"type": "convert", "step": 1, "request": "1 fortnight to minutes", + "result": "1 fortnight = 20,160 minute"}, + {"type": "vote", "agree": 2, "disagree": 1, "about": "the total"}, + {"type": "agreed", "by": "vote", "round": 2, "answer": "20160"}, + {"type": "final", "content": "There are 20160 minutes in a fortnight.", + "graph": {"nodes": [], "edges": [ + {"from": "Step1", "to": "Step2", "value": 0.83}]}}, + ]: + job.append(event) + job.answer = "There are 20160 minutes in a fortnight." + job.done.set() + return job + + +class TestRecording: + def test_a_settle_run_is_answerable_in_sql(self): + graph = GraphDB(":memory:") + graph.record(settled_job()) + + # Qualified columns: both tables carry `question` and `answer`, and + # mpedb refuses the ambiguity where sqlite silently picked one. The + # refusal is the better behaviour. + findings = graph.query( + "SELECT findings.question, findings.answer FROM findings" + " JOIN runs ON findings.run = runs.id" + " WHERE runs.session = 'clocks' ORDER BY findings.question") + assert len(findings) == 2 + assert findings[0][1] == "A fortnight has 14 days." + + how = graph.query( + "SELECT agreement.reached_by, votes.agree, votes.disagree FROM agreement" + " JOIN votes ON votes.run = agreement.run") + assert how == [("vote", 2, 1)] + + facts = graph.query("SELECT kind, statement FROM facts ORDER BY kind") + assert ("calculation", "14*1440 = 20,160") in facts + assert ("conversion", "1 fortnight = 20,160 minute") in facts + + def test_sub_run_findings_stay_out_of_the_top_level(self): + graph = GraphDB(":memory:") + graph.record(settled_job()) + texts = [row[0] for row in graph.query("SELECT question FROM findings")] + assert "a sub-run's own finding" not in texts + + def test_recording_twice_records_once(self): + graph = GraphDB(":memory:") + job = settled_job() + graph.record(job) + graph.record(job) + assert graph.query("SELECT COUNT(*) FROM runs")[0][0] == 1 + assert graph.query("SELECT COUNT(*) FROM findings")[0][0] == 2 + + def test_session_stats_count_what_a_session_did(self): + graph = GraphDB(":memory:") + graph.record(settled_job()) + stats = graph.session_stats("clocks") + assert stats == {"runs": 1, "agreed": 1, "exact_facts": 2} + assert graph.session_stats("elsewhere") == { + "runs": 0, "agreed": 0, "exact_facts": 0} + + def test_edges_carry_their_similarity(self): + graph = GraphDB(":memory:") + graph.record(settled_job()) + assert graph.query("SELECT a, b, sim FROM edges") == [("Step1", "Step2", 0.83)] + + +class TestThroughTheApi: + @pytest.fixture + def client(self, tmp_path): + import mpe_lkg.app as app_module + + app_module.app.config["DB_PATH"] = str(tmp_path / "app.db") + app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db") + app_module.app.config["BACKENDS_FACTORY"] = lambda: ( + ScriptedChat(normal_script(4), repeat_last=True), + DeterministicEmbedding(24)) + app_module.JOBS = app_module.Registry() + with app_module.app.test_client() as c: + yield c, str(tmp_path / "graph.db") + + def test_a_finished_job_lands_in_the_tables(self, client): + c, graph_path = client + job_id = c.post("/jobs", json={"query": "What is the capital of France?", + "session": "geo"}).get_json()["id"] + deadline = time.time() + 20 + while time.time() < deadline: + if c.get(f"/jobs/{job_id}").get_json()["state"] == "done": + break + time.sleep(0.05) + + graph = GraphDB(graph_path) + runs = graph.query("SELECT id, session, question FROM runs") + assert (job_id, "geo", "What is the capital of France?") in runs + assert graph.query("SELECT COUNT(*) FROM steps WHERE run = ?", + (job_id,))[0][0] > 0 + + def test_the_sessions_listing_carries_the_graphs_numbers(self, client): + c, _ = client + job_id = c.post("/jobs", json={"query": "q", "session": "geo"}).get_json()["id"] + deadline = time.time() + 20 + while time.time() < deadline: + if c.get(f"/jobs/{job_id}").get_json()["state"] == "done": + break + time.sleep(0.05) + listed = {s["name"]: s for s in c.get("/sessions").get_json()["sessions"]} + assert listed["geo"]["runs"] == 1 + assert "exact_facts" in listed["geo"] + + +def test_the_scripted_step_helper_still_matches_the_schema(): + # The settle-shaped stream above is hand-built; this pins that the helper + # the rest of the suite uses produces steps record() can read. + parsed = json.loads(step("T", "C")) + assert {"title", "content", "next_action"} <= set(parsed) diff --git a/dev/tests/test_jobs.py b/dev/tests/test_jobs.py index 48bca83..a83d8fa 100644 --- a/dev/tests/test_jobs.py +++ b/dev/tests/test_jobs.py @@ -27,6 +27,7 @@ def client(tmp_path): import mpe_lkg.app as app_module app_module.app.config["DB_PATH"] = str(tmp_path / "jobs.db") + app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db") app_module.app.config["BACKENDS_FACTORY"] = lambda: ( ScriptedChat(normal_script(5)), DeterministicEmbedding(32)) app_module.JOBS = app_module.Registry() diff --git a/dev/tests/test_render.py b/dev/tests/test_render.py index 28ec12f..252ca41 100644 --- a/dev/tests/test_render.py +++ b/dev/tests/test_render.py @@ -34,6 +34,7 @@ def __init__(self, script, tmp_path, *, repeat_last=False, dim=48, delay=0.0): chat = ScriptedChat(script, repeat_last=repeat_last, delay=delay) app_module.app.config["BACKENDS_FACTORY"] = lambda: (chat, DeterministicEmbedding(dim)) app_module.app.config["DB_PATH"] = str(tmp_path / "render.db") + app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db") self.port = free_port() self._server = make_server("127.0.0.1", self.port, app_module.app, threaded=True) self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) diff --git a/dev/tests/test_sessions.py b/dev/tests/test_sessions.py index da680ec..aa0ff5e 100644 --- a/dev/tests/test_sessions.py +++ b/dev/tests/test_sessions.py @@ -107,6 +107,9 @@ def client(tmp_path): import mpe_lkg.app as app_module app_module.app.config["DB_PATH"] = str(tmp_path / "app.db") + # Without this the sessions endpoint records into the repo root -- the + # stray-file mistake this project has already committed once. + app_module.app.config["GRAPH_DB_PATH"] = str(tmp_path / "graph.db") app_module.app.config["BACKENDS_FACTORY"] = lambda: ( ScriptedChat(normal_script(4), repeat_last=True), DeterministicEmbedding(24)) app_module.JOBS = app_module.Registry() diff --git a/pyproject.toml b/pyproject.toml index e185ef8..add246d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,11 @@ dependencies = [ # would leave the conversions silently missing instead of failing loudly, # which is the worst way for a dependency floor to be wrong. "mpeqs>=0.3.0", + # The store engine. A drop-in for sqlite3 that also opens existing + # embeddings.db files in place -- probed before adoption, tested after -- + # plus the in-memory mode the run-graph tables use. Wheels cover the same + # matrix this package supports: 3.10+, Intel and ARM Mac, Linux, Windows. + "mpedb>=0.3.0", ] [project.optional-dependencies] diff --git a/src/mpe_lkg/app.py b/src/mpe_lkg/app.py index f194521..bc8c2c1 100644 --- a/src/mpe_lkg/app.py +++ b/src/mpe_lkg/app.py @@ -17,6 +17,7 @@ from flask import Flask, Response, jsonify, render_template, request from . import backends, rdf +from .graphdb import GraphDB from .jobs import Job, Registry from .reasoning import explore, reason, settle from .store import EmbeddingStore @@ -163,8 +164,24 @@ def favicon(): JOBS = Registry() +GRAPH_DB = "knowledge.graph.db" + + +def _record(job) -> None: + """One finished job into the graph tables. Failures are logged, not fatal: + the answer has already been delivered, and recording is bookkeeping.""" + try: + graph = GraphDB(app.config.get("GRAPH_DB_PATH", GRAPH_DB)) + try: + graph.record(job) + finally: + graph.close() + except Exception: # noqa: BLE001 + app.logger.exception("could not record run %s", job.id) + + def _job_events(user_query: str, store_path: str, mode: str = "reason", - session: str = ""): + session: str = "", job=None): """The reasoning events for a headless run, with its own backends and store.""" chat, embedder = app.config.get("BACKENDS_FACTORY", make_backends)() store = EmbeddingStore(store_path) @@ -184,6 +201,8 @@ def _job_events(user_query: str, store_path: str, mode: str = "reason", decompose=decompose) finally: store.close() + if job is not None: + _record(job) @app.route("/jobs", methods=["GET", "POST"]) @@ -209,7 +228,8 @@ def jobs(): job.mode = mode job.session = session JOBS.add(job) - job.start(_job_events(user_query, app.config.get("DB_PATH", DB_PATH), mode, session)) + job.start(_job_events(user_query, app.config.get("DB_PATH", DB_PATH), mode, + session, job=job)) # 202: accepted and still running. The Location header is where to look. return jsonify(job.status()), 202, {"Location": f"/jobs/{job.id}"} @@ -218,9 +238,19 @@ def jobs(): def sessions(): store = EmbeddingStore(app.config.get("DB_PATH", DB_PATH)) try: - return jsonify({"sessions": store.sessions()}) + listed = store.sessions() finally: store.close() + try: + graph = GraphDB(app.config.get("GRAPH_DB_PATH", GRAPH_DB)) + try: + for entry in listed: + entry.update(graph.session_stats(entry["name"])) + finally: + graph.close() + except Exception: # noqa: BLE001 -- stats are additive, never blocking + app.logger.exception("could not read graph stats") + return jsonify({"sessions": listed}) @app.route("/sessions/", methods=["DELETE"]) diff --git a/src/mpe_lkg/graphdb.py b/src/mpe_lkg/graphdb.py new file mode 100644 index 0000000..3f35c67 --- /dev/null +++ b/src/mpe_lkg/graphdb.py @@ -0,0 +1,142 @@ +"""The knowledge graph as tables something can query. + +The RDF export is the graph as a *document* -- for handing to another tool. This +is the graph as a *database*: the same runs, steps, edges and facts, in mpedb +tables that SQL can join, filter and aggregate across every run a session has +ever made. ``GraphDB(":memory:")`` gives the same schema for a single run's +analysis without touching disk. + +What gets recorded is exactly what the RDF serialises, drawn from the same job +events by the same extraction -- one definition of what is worth keeping, two +representations of it. The vote rows carry their tally and what the dispute was +about, because "how was this answer reached" is the query this table exists for. +""" + +from __future__ import annotations + +import time + +import mpedb + +SCHEMA = [ + """CREATE TABLE IF NOT EXISTS runs ( + id TEXT PRIMARY KEY, + session TEXT NOT NULL DEFAULT '', + question TEXT NOT NULL, + answer TEXT NOT NULL DEFAULT '', + mode TEXT NOT NULL DEFAULT 'reason', + created REAL NOT NULL + )""", + """CREATE TABLE IF NOT EXISTS steps ( + run TEXT NOT NULL, n INTEGER NOT NULL, + title TEXT NOT NULL DEFAULT '', content TEXT NOT NULL DEFAULT '' + )""", + """CREATE TABLE IF NOT EXISTS edges ( + run TEXT NOT NULL, a TEXT NOT NULL, b TEXT NOT NULL, sim REAL NOT NULL + )""", + """CREATE TABLE IF NOT EXISTS facts ( + run TEXT NOT NULL, kind TEXT NOT NULL, statement TEXT NOT NULL + )""", + """CREATE TABLE IF NOT EXISTS findings ( + run TEXT NOT NULL, question TEXT NOT NULL, answer TEXT NOT NULL + )""", + """CREATE TABLE IF NOT EXISTS votes ( + run TEXT NOT NULL, agree INTEGER NOT NULL, disagree INTEGER NOT NULL, + about TEXT NOT NULL DEFAULT '' + )""", + # reached_by, not "by": BY is a reserved word and mpedb's parser refuses + # it as a column name where sqlite3 tolerated it. "round" survives both. + """CREATE TABLE IF NOT EXISTS agreement ( + run TEXT NOT NULL, reached_by TEXT NOT NULL, round INTEGER NOT NULL DEFAULT 0 + )""", +] + + +class GraphDB: + def __init__(self, path: str) -> None: + self.path = path + self.conn = mpedb.connect(path, check_same_thread=False) + for statement in SCHEMA: + self.conn.execute(statement) + self.conn.commit() + + def close(self) -> None: + self.conn.close() + + def record(self, job) -> None: + """One finished job into the tables. Idempotent per job id.""" + already = self.conn.execute( + "SELECT COUNT(*) FROM runs WHERE id = ?", (job.id,)).fetchone()[0] + if already: + return + self.conn.execute( + "INSERT INTO runs (id, session, question, answer, mode, created)" + " VALUES (?, ?, ?, ?, ?, ?)", + (job.id, getattr(job, "session", ""), job.question, job.answer, + getattr(job, "mode", "reason"), getattr(job, "created", time.time()))) + + for event in job.of_type("step"): + self.conn.execute( + "INSERT INTO steps (run, n, title, content) VALUES (?, ?, ?, ?)", + (job.id, int(event.get("step") or 0), + str(event.get("title", "")), str(event.get("content", "")))) + + for edge in job.graph().get("edges", []): + if edge.get("from") and edge.get("to"): + self.conn.execute( + "INSERT INTO edges (run, a, b, sim) VALUES (?, ?, ?, ?)", + (job.id, str(edge["from"]), str(edge["to"]), + float(edge.get("value", 0.0)))) + + for event in job.of_type("convert"): + self.conn.execute( + "INSERT INTO facts (run, kind, statement) VALUES (?, 'conversion', ?)", + (job.id, str(event.get("result", "")))) + for event in job.of_type("calc"): + self.conn.execute( + "INSERT INTO facts (run, kind, statement) VALUES (?, 'calculation', ?)", + (job.id, f"{event.get('expression', '')} = {event.get('value', '')}")) + + # Top-level findings only, as in the RDF: a sub-run's findings describe + # its own question, and flattening the levels loses which is which. + for event in job.of_type("finding"): + if not event.get("level"): + self.conn.execute( + "INSERT INTO findings (run, question, answer) VALUES (?, ?, ?)", + (job.id, str(event.get("question", "")), + str(event.get("answer", "")))) + + for event in job.of_type("vote"): + self.conn.execute( + "INSERT INTO votes (run, agree, disagree, about) VALUES (?, ?, ?, ?)", + (job.id, int(event.get("agree", 0)), int(event.get("disagree", 0)), + str(event.get("about", "")))) + + for event in job.of_type("agreed"): + self.conn.execute( + "INSERT INTO agreement (run, reached_by, round) VALUES (?, ?, ?)", + (job.id, str(event.get("by", "vote")), int(event.get("round") or 0))) + break # one agreement finishes a run; a second is noise + + self.conn.commit() + + def query(self, sql: str, params: tuple = ()) -> list[tuple]: + """Programmatic SQL over the recorded graph. Not exposed over HTTP.""" + return list(self.conn.execute(sql, params)) + + def session_stats(self, session: str = "") -> dict: + """The canned query the sessions endpoint reads: what a session has done.""" + # Two plain queries, not SUM(EXISTS(...)): mpedb answers the correlated + # form with nothing at all, and a missing row here took the sessions + # endpoint down with it. + runs = self.conn.execute( + "SELECT COUNT(*) FROM runs WHERE session = ?", (session,)).fetchone()[0] + agreed = self.conn.execute( + "SELECT COUNT(DISTINCT agreement.run) FROM agreement" + " JOIN runs ON agreement.run = runs.id WHERE runs.session = ?", + (session,)).fetchone()[0] + facts = self.conn.execute( + "SELECT COUNT(*) FROM facts JOIN runs ON facts.run = runs.id" + " WHERE runs.session = ?", (session,)).fetchone()[0] + return {"runs": int(runs or 0), "agreed": int(agreed or 0), + "exact_facts": int(facts or 0)} diff --git a/src/mpe_lkg/store.py b/src/mpe_lkg/store.py index 59c8a28..1a57989 100644 --- a/src/mpe_lkg/store.py +++ b/src/mpe_lkg/store.py @@ -1,4 +1,8 @@ -"""SQLite-backed embedding store with brute-force similarity search. +"""MPEdb-backed embedding store with brute-force similarity search. + +The engine is mpedb, which speaks the sqlite3 DB-API and opens this project's +existing embeddings.db files in place -- verified by test, not assumed -- so the +swap changed an import and nothing a caller can see. This replaces the Annoy index the project used to carry. An approximate nearest-neighbour index earns its keep somewhere around a hundred thousand vectors; @@ -15,8 +19,7 @@ from __future__ import annotations -import sqlite3 - +import mpedb import numpy as np DEFAULT_PATH = "embeddings.db" @@ -39,7 +42,7 @@ def __init__(self, path: str = DEFAULT_PATH) -> None: self.path = path # The rows are produced inside a streaming response, which Flask may run on # a different thread than the one that opened the connection. - self.conn = sqlite3.connect(path, check_same_thread=False) + self.conn = mpedb.connect(path, check_same_thread=False) self.conn.execute(SCHEMA) self._migrate() self.conn.commit() @@ -59,15 +62,26 @@ def _migrate(self) -> None: vector from a 384-dimensional one, so switching embedding model silently corrupted every search against the old rows. """ - existing = {row[1] for row in self.conn.execute("PRAGMA table_info(embeddings)")} + # cursor.description, not PRAGMA table_info: the pragma is a sqlite3 + # extension that mpedb answers with nothing, and an empty answer made + # this method try to re-add every column. The DB-API way works on both. + cursor = self.conn.execute("SELECT * FROM embeddings LIMIT 0") + existing = {column[0] for column in cursor.description} + # Each ADD COLUMN is followed by an explicit backfill. sqlite3 writes + # the DEFAULT into existing rows; mpedb leaves them NULL, so without the + # UPDATE every pre-migration row silently vanishes from queries that + # filter on the new column -- which is all of them. if "dim" not in existing: self.conn.execute("ALTER TABLE embeddings ADD COLUMN dim INTEGER NOT NULL DEFAULT 0") + self.conn.execute("UPDATE embeddings SET dim = 0 WHERE dim IS NULL") if "model" not in existing: self.conn.execute("ALTER TABLE embeddings ADD COLUMN model TEXT NOT NULL DEFAULT ''") + self.conn.execute("UPDATE embeddings SET model = '' WHERE model IS NULL") if "session" not in existing: # Rows from before sessions existed land in the default session, # which is where a caller that never names one still works. self.conn.execute("ALTER TABLE embeddings ADD COLUMN session TEXT NOT NULL DEFAULT ''") + self.conn.execute("UPDATE embeddings SET session = '' WHERE session IS NULL") def close(self) -> None: self.conn.close() @@ -83,17 +97,20 @@ def count(self) -> int: def add(self, text: str, embedding: np.ndarray, *, is_question: bool = False, model: str = "", session: str = "") -> int: vector = np.asarray(embedding, dtype=np.float32).ravel() - cursor = self.conn.execute( + # RETURNING, not lastrowid: mpedb's file engine reports lastrowid as + # None (the in-memory engine reports it fine, which is how this hid + # from the first probes). RETURNING is answered by both. + row = self.conn.execute( "INSERT INTO embeddings (text, embedding, is_question, dim, model, session)" - " VALUES (?, ?, ?, ?, ?, ?)", - (text, sqlite3.Binary(vector.tobytes()), int(is_question), + " VALUES (?, ?, ?, ?, ?, ?) RETURNING id", + (text, vector.tobytes(), int(is_question), int(vector.size), model, session or self.default_session), - ) + ).fetchone() self.conn.commit() # Both this session's matrix and the every-other-session matrices are # stale now; dropping by prefix is simpler than tracking which. self._cache.clear() - return int(cursor.lastrowid) + return int(row[0]) def _matrix(self, dim: int, model: str, session: str | None = "", exclude: str | None = None) -> tuple[np.ndarray, list]: