Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| FALSIFY — the AI research copilot that *revises*, not forgets. | |
| Run this to watch belief revision happen on a real Cognee knowledge graph: | |
| python main.py # full run (uses your LLM to judge the contradiction) | |
| python main.py --demo # deterministic: pins the contradiction so the cascade | |
| # always runs, even without/with a flaky LLM key | |
| python main.py --keep # don't prune existing memory first (advanced) | |
| The story (Company X recall investigation) | |
| ------------------------------------------ | |
| Session 1 builds a belief graph with two competing hypotheses: | |
| A — "X knew via the March 2021 QA report" (supported by evidence E_qa) | |
| B — "X knew via a January 2021 supplier email" (supported by evidence E_email) | |
| and a Conclusion K that *depends on* E_qa. | |
| Session 2 drops ONE contradicting fact: "the March QA report was back-dated." | |
| FALSIFY refutes E_qa, cascades the refutation forward (K collapses), promotes B as | |
| the new frontier, and surgically forgets the orphaned conclusion — writing the | |
| disbelief onto the graph so it survives a restart. A plain-RAG baseline, which has no | |
| notion of truth-state, keeps citing the refuted March report. That contrast is the | |
| whole point: *AI revised, not forgot.* | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import os | |
| import sys | |
| # Load .env before importing cognee/falsify so provider config is in place. | |
| try: | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| except Exception: # python-dotenv is optional; env may be set another way | |
| pass | |
| # Importing falsify sets single-user Cognee env defaults (access control off, cache on). | |
| import falsify # noqa: F401 (import-time side effects) | |
| from falsify.seed import NEW_FACT, QUESTION_TEXT | |
| C = { | |
| "b": "\033[1m", "dim": "\033[2m", "g": "\033[92m", "r": "\033[91m", | |
| "y": "\033[93m", "c": "\033[96m", "x": "\033[0m", | |
| } | |
| if os.environ.get("NO_COLOR") or not sys.stdout.isatty(): | |
| C = {k: "" for k in C} | |
| def banner(text: str) -> None: | |
| print(f"\n{C['b']}{C['c']}{'━' * 64}{C['x']}") | |
| print(f"{C['b']}{C['c']} {text}{C['x']}") | |
| print(f"{C['b']}{C['c']}{'━' * 64}{C['x']}") | |
| def _has_llm_key() -> bool: | |
| """True if some LLM API key is configured (OpenAI-compatible or otherwise).""" | |
| for var in ("LLM_API_KEY", "OPENAI_API_KEY"): | |
| if os.environ.get(var): | |
| return True | |
| return False | |
| def _print_no_key_help() -> None: | |
| print( | |
| f""" | |
| {C['y']}{C['b']}No LLM API key found.{C['x']} | |
| FALSIFY needs an OpenAI-compatible API key to (a) embed claims for the vector | |
| prefilter and (b) judge contradictions. Set it up: | |
| 1. cp .env.template .env | |
| 2. edit .env and set: | |
| LLM_API_KEY="your_key_here" | |
| LLM_MODEL="gpt-4o-mini" # or any OpenAI-compatible model | |
| # For a non-OpenAI endpoint (OpenRouter, vLLM, LM Studio, Groq): | |
| # LLM_PROVIDER="custom" | |
| # LLM_ENDPOINT="https://your-endpoint/v1" | |
| 3. re-run: python main.py | |
| {C['dim']}Tip: `python main.py --demo` runs FULLY OFFLINE — fastembed handles | |
| embeddings locally and the contradiction is pinned, so no API key is needed at all. | |
| A key is only required for LIVE mode, where the LLM judges the contradiction.{C['x']} | |
| """ | |
| ) | |
| async def run(demo: bool, keep: bool) -> int: | |
| from falsify.falsify import build_graph, revise, scoreboard | |
| from falsify.utils import get_belief_summary, print_graph_state, visualize_belief_graph | |
| banner("FALSIFY — belief-revision research copilot") | |
| print(f" Research question: {C['b']}{QUESTION_TEXT}{C['x']}") | |
| print(f" Mode: {'DEMO (deterministic contradiction pin)' if demo else 'LIVE (LLM judge)'}") | |
| # ---- Session 1: build the belief graph ------------------------------- | |
| banner("SESSION 1 — build the investigation") | |
| if keep: | |
| from cognee.low_level import setup | |
| from falsify.seed import build_investigation | |
| await setup() | |
| seeded = await build_investigation() | |
| else: | |
| seeded = await build_graph() | |
| await print_graph_state("BEFORE — both hypotheses stand, Conclusion K rests on E_qa") | |
| # ---- Session 2: drop the contradicting fact -------------------------- | |
| banner("SESSION 2 — a new fact arrives") | |
| print(f" {C['y']}New fact:{C['x']} {NEW_FACT}\n") | |
| pinned = seeded.refuted_target_id if demo else None | |
| report = await revise(NEW_FACT, pinned_target_id=pinned) | |
| if not report.revised: | |
| print(f" {C['y']}No contradiction was confirmed — graph unchanged.{C['x']}") | |
| print(f" {C['dim']}(Try `python main.py --demo` to force the cascade deterministically.){C['x']}") | |
| else: | |
| print(f" {C['r']}✗ refuted:{C['x']} {len(report.refuted)} evidence node(s)") | |
| print(f" {C['r']}✗ invalidated:{C['x']} {len(report.invalidated)} conclusion(s)") | |
| for hid, action in report.hypothesis_actions.items(): | |
| if action == "superseded": | |
| mark = f"{C['r']}↓ superseded{C['x']}" | |
| else: | |
| mark = f"{C['g']}↑ promoted (new frontier){C['x']}" | |
| print(f" hypothesis {hid[:8]} → {mark}") | |
| for fid in report.forgotten: | |
| label = report.forgotten_labels.get(fid, fid) | |
| print(f" {C['dim']}🗑 forgotten (deleted from graph + vector):{C['x']} {label}") | |
| if report.retained_provenance: | |
| print(f" {C['dim']}⚑ kept as red provenance:{C['x']} {len(report.retained_provenance)} node(s)") | |
| await print_graph_state("AFTER — refutation cascaded, B ignites, orphan forgotten") | |
| # ---- The scoreboard: FALSIFY vs plain RAG ---------------------------- | |
| banner("SCOREBOARD — FALSIFY (revised) vs plain RAG (stale)") | |
| board = await scoreboard(QUESTION_TEXT, seeded) | |
| print(f" {C['g']}{C['b']}FALSIFY :{C['x']} {board.falsify_answer}") | |
| if board.falsify_support: | |
| print(f" {C['dim']}supported by: {', '.join(board.falsify_support)}{C['x']}") | |
| rag_tag = f"{C['r']}[STALE — still cites a refuted fact]{C['x']}" if board.stale else "" | |
| print(f" {C['y']}{C['b']}RAG :{C['x']} {board.rag_answer} {rag_tag}") | |
| print(f"\n {C['b']}→ AI revised, not forgot.{C['x']}") | |
| # ---- Cross-session proof: reload belief state fresh ------------------ | |
| banner("CROSS-SESSION PROOF — reopen memory, beliefs stay revised") | |
| summary = await get_belief_summary() | |
| print(f" Persisted belief state (re-read from graph): {summary}") | |
| print(f" {C['dim']}Truth-state lives on the graph nodes, so a brand-new process sees the") | |
| print(f" revised graph — the refuted branch never comes back.{C['x']}") | |
| # ---- Visualization --------------------------------------------------- | |
| out = await visualize_belief_graph("output/graph.html", title="FALSIFY — Company X investigation") | |
| if out: | |
| banner("VISUALIZATION") | |
| print(f" Interactive belief graph written to: {C['b']}{os.path.abspath(out)}{C['x']}") | |
| print(f" {C['dim']}Open it in a browser — red = refuted, grey = invalidated, green = alive.{C['x']}") | |
| return 0 | |
| def main() -> int: | |
| parser = argparse.ArgumentParser(description="FALSIFY belief-revision demo") | |
| parser.add_argument("--demo", action="store_true", | |
| help="pin the contradiction deterministically (LLM-independent cascade)") | |
| parser.add_argument("--keep", action="store_true", | |
| help="do not prune existing memory before building") | |
| args = parser.parse_args() | |
| if not _has_llm_key() and not args.demo: | |
| _print_no_key_help() | |
| # Live mode needs an LLM key to judge the contradiction. Exit cleanly (0) with | |
| # setup help rather than a stack trace. (`--demo` runs fully offline below.) | |
| return 0 | |
| try: | |
| return asyncio.run(run(demo=args.demo, keep=args.keep)) | |
| except KeyboardInterrupt: | |
| print("\ninterrupted.") | |
| return 130 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |