Spaces:
Sleeping
Sleeping
| """ | |
| FALSIFY orchestration — the belief-revision copilot's public API. | |
| Three verbs tie the engine together: | |
| build_graph() -> seed the Session-1 investigation graph (clean slate). | |
| revise(new_fact) -> run the full revision pipeline for an incoming fact: | |
| detect -> propagate -> promote -> record supersede -> forget. | |
| scoreboard(question) -> the money shot: FALSIFY's revised answer (reads truth | |
| state, skips dead branches) vs a plain-RAG baseline | |
| (raw vector search, no truth filter) that still cites the | |
| refuted fact. | |
| Everything is persisted on the graph (truth-state on nodes), so a fresh process / | |
| second session sees the revised beliefs — the cross-session guarantee. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from dataclasses import dataclass, field | |
| from typing import Dict, List, Optional | |
| from falsify import graph_ops | |
| from falsify.edges import SUPERSEDES, SUPPORTS | |
| from falsify.models import Evidence, TruthState | |
| from falsify.seed import SeededGraph, build_investigation | |
| from falsify.tasks import ( | |
| Contradiction, | |
| cascade_forget, | |
| detect_contradictions, | |
| promote_competing_hypothesis, | |
| propagate_refutation, | |
| ) | |
| logger = logging.getLogger("falsify.orchestrator") | |
| _ALIVE = TruthState.ALIVE.value | |
| _EVIDENCE_COLLECTION = "Evidence_claim" | |
| class RevisionReport: | |
| """Full record of one :func:`revise` run, for demo output and tests.""" | |
| new_fact: str | |
| contradictions: List[Contradiction] = field(default_factory=list) | |
| refuted: List[str] = field(default_factory=list) | |
| invalidated: List[str] = field(default_factory=list) | |
| hypothesis_actions: Dict[str, str] = field(default_factory=dict) | |
| forgotten: List[str] = field(default_factory=list) | |
| forgotten_labels: Dict[str, str] = field(default_factory=dict) | |
| retained_provenance: List[str] = field(default_factory=list) | |
| new_evidence_id: Optional[str] = None | |
| epoch: int = 0 | |
| def revised(self) -> bool: | |
| """True if the fact actually triggered a belief change.""" | |
| return bool(self.refuted or self.invalidated) | |
| async def build_graph() -> SeededGraph: | |
| """Prune everything and build the Session-1 investigation graph. | |
| Returns the :class:`SeededGraph` with stable handles to the seeded nodes. | |
| """ | |
| import cognee | |
| from cognee.low_level import setup | |
| logger.info("build_graph: pruning and seeding fresh investigation") | |
| await cognee.forget(everything=True) | |
| await setup() # (re)create relational tables the storage pipeline needs | |
| seeded = await build_investigation() | |
| return seeded | |
| async def revise( | |
| new_fact: str, | |
| *, | |
| pinned_target_id: Optional[str] = None, | |
| source_id: str = "session2_fact", | |
| ) -> RevisionReport: | |
| """Run the full belief-revision pipeline for an incoming fact. | |
| Pipeline (REQUIREMENTS §1.3-§1.5): | |
| 1. detect_contradictions -> which evidence does the fact contradict? | |
| 2. propagate_refutation -> refute it; cascade invalidation forward. | |
| 3. promote_competing_hypothesis -> demote the losing hypothesis, ignite the rival. | |
| 4. record the new fact as Evidence + a ``supersedes`` edge to the refuted node | |
| (so the refuted node is retained as a provenance tombstone). | |
| 5. cascade_forget -> hard-delete orphaned dead-ends from graph + vector. | |
| Args: | |
| new_fact: the incoming claim. | |
| pinned_target_id: demo override — refute this evidence id deterministically. | |
| source_id: provenance id for the materialized new-fact Evidence node. | |
| Returns: | |
| A :class:`RevisionReport` describing everything that changed. | |
| """ | |
| report = RevisionReport(new_fact=new_fact) | |
| # 1) detect | |
| contradictions = await detect_contradictions(new_fact, pinned_target_id=pinned_target_id) | |
| report.contradictions = contradictions | |
| if not contradictions: | |
| logger.info("revise: no contradiction found; graph unchanged") | |
| return report | |
| target_ids = [c.target_id for c in contradictions] | |
| # 2) propagate | |
| prop = await propagate_refutation(target_ids) | |
| report.refuted = prop.refuted | |
| report.invalidated = prop.invalidated | |
| report.epoch = prop.epoch | |
| # 3) promote competing hypothesis | |
| report.hypothesis_actions = await promote_competing_hypothesis(target_ids, prop.epoch) | |
| # 4) record the new fact + supersedes edge (keeps the refuted node as provenance) | |
| report.new_evidence_id = await _record_new_fact(new_fact, contradictions, source_id) | |
| # 5) forget orphaned dead-ends | |
| forget_res = await cascade_forget(prop.affected) | |
| report.forgotten = forget_res.forgotten | |
| report.forgotten_labels = forget_res.labels | |
| report.retained_provenance = forget_res.retained_provenance | |
| logger.info( | |
| "revise complete: refuted=%d invalidated=%d forgotten=%d", | |
| len(report.refuted), len(report.invalidated), len(report.forgotten), | |
| ) | |
| return report | |
| async def _record_new_fact( | |
| new_fact: str, | |
| contradictions: List[Contradiction], | |
| source_id: str, | |
| ) -> Optional[str]: | |
| """Materialize the new fact as an Evidence node and link supersedes edges. | |
| The new (alive) evidence ``supersedes`` each refuted evidence node. This both | |
| records provenance and pins the refuted node as a retained tombstone (an alive | |
| supersedes-source protects its target from forget — REQUIREMENTS §1.5c). | |
| """ | |
| from cognee.tasks.storage import add_data_points | |
| try: | |
| new_ev = Evidence( | |
| claim=new_fact, | |
| source_id=source_id, | |
| stance="refutes", | |
| confidence=max((c.confidence for c in contradictions), default=0.9), | |
| ) | |
| await add_data_points([new_ev]) | |
| for c in contradictions: | |
| await graph_ops.add_edge( | |
| str(new_ev.id), str(c.target_id), SUPERSEDES, {"confidence": c.confidence} | |
| ) | |
| logger.info("recorded new fact %s superseding %d node(s)", new_ev.id, len(contradictions)) | |
| return str(new_ev.id) | |
| except Exception as exc: | |
| logger.error("failed to record new fact: %s", exc) | |
| return None | |
| class Scoreboard: | |
| """The FALSIFY-vs-RAG comparison shown every run.""" | |
| question: str | |
| falsify_answer: str | |
| falsify_support: List[str] = field(default_factory=list) | |
| rag_answer: str = "" | |
| rag_citations: List[str] = field(default_factory=list) | |
| stale: bool = False # True if RAG still cites a refuted node FALSIFY dropped | |
| async def scoreboard(question: str, seeded: Optional[SeededGraph] = None) -> Scoreboard: | |
| """Compare FALSIFY's revised answer against a plain-RAG baseline. | |
| FALSIFY answer: derived from the graph, reading truth-state and using only | |
| hypotheses/evidence still ``alive`` (the promoted frontier hypothesis). | |
| RAG baseline: a raw vector search over ``Evidence_claim`` with **no** truth | |
| filter — so it still returns evidence FALSIFY has refuted, and cites the stale | |
| fact. This asymmetry is the demo's whole point. | |
| """ | |
| board = Scoreboard(question=question, falsify_answer="(no surviving hypothesis)") | |
| # ---- FALSIFY: alive-filtered graph answer ---- | |
| nodes, edges = await graph_ops.load_graph() | |
| node_ids = [nid for nid, _p in nodes] | |
| truth = await graph_ops.get_truth(node_ids) | |
| props_by_id = {str(nid): (p or {}) for nid, p in nodes} | |
| # The winning hypothesis = alive hypothesis with the most alive supporting evidence weight. | |
| best_hyp, best_score = None, -1.0 | |
| support_edges = [(s, d, p) for (s, d, r, p) in edges if r == SUPPORTS] | |
| for nid, props in nodes: | |
| nid = str(nid) | |
| if "statement" not in props: # crude: hypotheses/conclusions carry 'statement' | |
| continue | |
| if _ALIVE not in truth.get(nid, [_ALIVE]): | |
| continue | |
| score = 0.0 | |
| alive_support = [] | |
| for (src, dst, ep) in support_edges: | |
| if str(dst) != nid: | |
| continue | |
| if _ALIVE in truth.get(str(src), [_ALIVE]): | |
| score += float(ep.get("weight", 0.5)) | |
| alive_support.append(graph_ops.node_label(props_by_id.get(str(src), {}))) | |
| if alive_support and score > best_score: | |
| best_hyp, best_score = nid, score | |
| board.falsify_answer = graph_ops.node_label(props) | |
| board.falsify_support = alive_support | |
| # ---- RAG baseline: raw vector search, no truth filter ---- | |
| ve = graph_ops.get_vector_engine() | |
| try: | |
| hits = await ve.search(_EVIDENCE_COLLECTION, query_text=question, limit=5, include_payload=True) | |
| except Exception as exc: | |
| logger.warning("RAG baseline search failed: %s", exc) | |
| hits = [] | |
| refuted_ids = {nid for nid in node_ids if TruthState.REFUTED.value in truth.get(str(nid), [])} | |
| for h in (hits or []): | |
| payload = getattr(h, "payload", {}) or {} | |
| text = payload.get("claim") or payload.get("text") or graph_ops.node_label(payload) | |
| board.rag_citations.append(str(text)) | |
| if str(h.id) in refuted_ids: | |
| board.stale = True | |
| board.rag_answer = board.rag_citations[0] if board.rag_citations else "(no vector hits)" | |
| logger.info("scoreboard: falsify=%r stale_rag=%s", board.falsify_answer, board.stale) | |
| return board | |