""" 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 = "" @dataclass 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