Spaces:
Sleeping
Sleeping
File size: 8,329 Bytes
1605cbb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 | """
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
|