Spaces:
Sleeping
Sleeping
| """ | |
| Two-gate contradiction detection — the trigger for belief revision. | |
| Deciding that a new fact *contradicts* an existing belief is the one genuinely | |
| fuzzy step in FALSIFY. A false positive nukes a valid conclusion; a false negative | |
| lets a stale fact survive. We therefore gate it twice, cheap-deterministic first, | |
| expensive-semantic second, with a hard deterministic override for the live demo. | |
| Gate 1 — vector prefilter (deterministic). | |
| Embed the new fact and search the ``Evidence_claim`` collection. Only | |
| evidence within a cosine-distance threshold (``< 0.35`` by default, i.e. | |
| clearly on-topic) proceeds. This narrows the LLM to plausibly-conflicting | |
| claims and keeps cost + nondeterminism bounded. | |
| Gate 2 — LLM adjudication (semantic). | |
| For each surviving candidate, ask an LLM acting as a skeptical analyst to | |
| classify the relation as contradicts / supersedes / supports / unrelated with | |
| a confidence. Only ``contradicts`` or ``supersedes`` at confidence >= 0.6 | |
| triggers refutation. This separates a genuine contradiction ("the report was | |
| back-dated") from mere topical overlap ("also mentions the report"). | |
| Demo override (``pinned_target_id`` / ``DEMO_MODE``). | |
| When set, gates are bypassed and a fixed high-confidence ``contradicts`` | |
| verdict is returned for the pinned evidence id, so the on-stage cascade runs | |
| on real graph APIs even if the LLM is slow, rate-limited, or the key is | |
| absent. This is FALSIFY's demo safety net (REQUIREMENTS §1.3). | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from dataclasses import dataclass, field | |
| from typing import List, Literal, Optional | |
| from pydantic import BaseModel, Field | |
| from falsify import graph_ops | |
| logger = logging.getLogger("falsify.detect") | |
| # Cosine distance below which two claims are "about the same thing" (Gate 1). | |
| DEFAULT_DISTANCE_THRESHOLD = 0.35 | |
| # Minimum LLM confidence for a contradiction/supersede to count (Gate 2). | |
| DEFAULT_CONFIDENCE_THRESHOLD = 0.6 | |
| # Vector collection holding Evidence claims. | |
| _EVIDENCE_COLLECTION = "Evidence_claim" | |
| _SYSTEM_PROMPT = ( | |
| "You are a skeptical forensic analyst. You are given an EXISTING evidence claim " | |
| "and a NEW fact. Decide the logical relation of the NEW fact to the EXISTING " | |
| "claim. Answer 'contradicts' only if the new fact makes the existing claim false " | |
| "or untrustworthy (e.g. it was fabricated, back-dated, retracted, or refuted). " | |
| "Answer 'supersedes' if the new fact replaces the existing claim with a newer, " | |
| "more authoritative version of the same fact. Answer 'supports' if it corroborates " | |
| "the claim, and 'unrelated' otherwise. Be conservative: when unsure, prefer " | |
| "'unrelated'. Provide a calibrated confidence in [0,1] and a one-sentence rationale." | |
| ) | |
| class ContradictionJudgement(BaseModel): | |
| """Structured verdict returned by the Gate-2 LLM adjudication.""" | |
| relation: Literal["contradicts", "supersedes", "supports", "unrelated"] | |
| confidence: float = Field(ge=0.0, le=1.0) | |
| rationale: str = "" | |
| class Contradiction: | |
| """A confirmed conflict between the new fact and an existing evidence node. | |
| Attributes: | |
| target_id: the existing Evidence node id that is contradicted/superseded. | |
| relation: ``contradicts`` or ``supersedes``. | |
| confidence: adjudicated confidence. | |
| rationale: short human explanation (shown in the demo). | |
| distance: Gate-1 cosine distance (lower = more on-topic). | |
| """ | |
| target_id: str | |
| relation: str | |
| confidence: float | |
| rationale: str = "" | |
| distance: float = 0.0 | |
| async def detect_contradictions( | |
| new_fact: str, | |
| *, | |
| pinned_target_id: Optional[str] = None, | |
| distance_threshold: float = DEFAULT_DISTANCE_THRESHOLD, | |
| confidence_threshold: float = DEFAULT_CONFIDENCE_THRESHOLD, | |
| max_candidates: int = 5, | |
| ) -> List[Contradiction]: | |
| """Return the existing evidence nodes that ``new_fact`` contradicts or supersedes. | |
| Args: | |
| new_fact: the incoming claim (e.g. the Session-2 forensic finding). | |
| pinned_target_id: demo/deterministic override; if given, gates are skipped and | |
| a single high-confidence ``contradicts`` verdict is returned for this id. | |
| distance_threshold: Gate-1 cosine-distance cutoff (lower = stricter on-topic). | |
| confidence_threshold: Gate-2 minimum confidence to accept a verdict. | |
| max_candidates: cap on Gate-1 candidates sent to the LLM. | |
| Returns: | |
| A list of :class:`Contradiction` (possibly empty). Callers feed the target | |
| ids into :func:`falsify.tasks.propagate_refutation.propagate_refutation`. | |
| """ | |
| # -------- Demo / deterministic override ------------------------------- | |
| if pinned_target_id: | |
| logger.info("detect_contradictions: pinned target %s (demo mode)", pinned_target_id) | |
| return [ | |
| Contradiction( | |
| target_id=str(pinned_target_id), | |
| relation="contradicts", | |
| confidence=0.9, | |
| rationale="Pinned contradiction (demo mode): new fact invalidates the target evidence.", | |
| distance=0.0, | |
| ) | |
| ] | |
| # -------- Gate 1: vector prefilter ------------------------------------ | |
| ve = graph_ops.get_vector_engine() | |
| try: | |
| hits = await ve.search( | |
| _EVIDENCE_COLLECTION, | |
| query_text=new_fact, | |
| limit=max_candidates, | |
| include_payload=True, | |
| ) | |
| except Exception as exc: | |
| logger.warning("Gate-1 vector search failed (%s); no contradictions detected", exc) | |
| return [] | |
| candidates = [(str(h.id), float(getattr(h, "score", 1.0)), getattr(h, "payload", {}) or {}) | |
| for h in (hits or [])] | |
| on_topic = [c for c in candidates if c[1] < distance_threshold] | |
| logger.info( | |
| "Gate-1: %d hit(s), %d within distance %.2f", len(candidates), len(on_topic), distance_threshold | |
| ) | |
| if not on_topic: | |
| return [] | |
| # -------- Gate 2: LLM adjudication ------------------------------------ | |
| from cognee.infrastructure.llm.LLMGateway import LLMGateway | |
| confirmed: List[Contradiction] = [] | |
| for target_id, distance, payload in on_topic: | |
| existing_claim = _payload_text(payload) or "(existing evidence claim)" | |
| text_input = ( | |
| f"EXISTING claim:\n{existing_claim}\n\nNEW fact:\n{new_fact}\n\n" | |
| "Classify the relation of the NEW fact to the EXISTING claim." | |
| ) | |
| try: | |
| verdict: ContradictionJudgement = await LLMGateway.acreate_structured_output( | |
| text_input=text_input, | |
| system_prompt=_SYSTEM_PROMPT, | |
| response_model=ContradictionJudgement, | |
| ) | |
| except Exception as exc: | |
| logger.warning("Gate-2 LLM judge failed for %s (%s); skipping candidate", target_id, exc) | |
| continue | |
| if verdict.relation in ("contradicts", "supersedes") and verdict.confidence >= confidence_threshold: | |
| confirmed.append( | |
| Contradiction( | |
| target_id=target_id, | |
| relation=verdict.relation, | |
| confidence=verdict.confidence, | |
| rationale=verdict.rationale, | |
| distance=distance, | |
| ) | |
| ) | |
| logger.info( | |
| "Gate-2: CONFIRMED %s on %s (conf=%.2f)", verdict.relation, target_id, verdict.confidence | |
| ) | |
| else: | |
| logger.info( | |
| "Gate-2: rejected %s (relation=%s conf=%.2f)", | |
| target_id, verdict.relation, verdict.confidence, | |
| ) | |
| return confirmed | |
| def _payload_text(payload: dict) -> Optional[str]: | |
| """Extract the human-readable claim text from a vector-hit payload.""" | |
| if not payload: | |
| return None | |
| for key in ("claim", "text", "statement", "content"): | |
| if payload.get(key): | |
| return str(payload[key]) | |
| # cognee payloads sometimes nest the original properties | |
| props = payload.get("properties") or payload.get("metadata") | |
| if isinstance(props, dict): | |
| for key in ("claim", "text", "statement"): | |
| if props.get(key): | |
| return str(props[key]) | |
| return None | |