doczyai-pipelines/src/qc_qa/confidence/signals.py

"""Rule-based confidence signals for one-to-one fields.

Each signal returns a float in [0.0, 1.0] or None when not applicable.
None means "this signal could not produce evidence for this (value, context)
pair" -- the scorer skips it and renormalizes weights over the remaining
signals.

Signals:
    exact          - normalized value is a substring of the context.
    token_overlap  - share of value tokens present in the context.
    fuzzy          - SequenceMatcher best-substring similarity.
    number_match   - all numbers in the value appear in the context.
    regex          - field-type-aware format check (date, money, TIN, etc.).
    grounding      - same family as exact/token but scoped to the *retrieved
                     chunks the LLM actually saw* (much stronger anti-
                     hallucination signal than scanning the whole contract).
    na_check       - special-case scoring when the value is N/A: rewards a
                     correctly-missing field, penalizes a suspicious miss.
"""

from __future__ import annotations

import re
from difflib import SequenceMatcher
from typing import Optional

from src.qc_qa.confidence.field_types import FieldType


_NA_TOKENS = {"", "n/a", "na", "none", "null", "not found"}
_TOKEN_RE = re.compile(r"[A-Za-z0-9]+")
_NUMBER_RE = re.compile(r"\d+(?:[\.,]\d+)?")

# Field-type-aware format regex. Matches the *value*, not the contract.
_FORMAT_PATTERNS = {
    FieldType.DATE: re.compile(
        r"\b\d{1,2}[/\-.]\d{1,2}[/\-.]\d{2,4}\b"
        r"|\b\d{4}[/\-.]\d{1,2}[/\-.]\d{1,2}\b"
        r"|\b(?:january|february|march|april|may|june|july|august|"
        r"september|october|november|december)\b",
        re.IGNORECASE,
    ),
    FieldType.MONEY: re.compile(
        r"\$\s*[\d,]+(?:\.\d{1,2})?"
        r"|\b\d{1,3}(?:,\d{3})+(?:\.\d+)?\b"
        r"|\b\d+(?:\.\d+)?\s*(?:USD|dollars?)\b",
        re.IGNORECASE,
    ),
    FieldType.TIN: re.compile(r"^\s*\d{2}-?\d{7}\s*$"),
    FieldType.NPI: re.compile(r"^\s*\d{10}\s*$"),
    FieldType.CODE: re.compile(r"^[A-Z0-9.\-]{1,12}$"),
    FieldType.BOOLEAN: re.compile(
        r"^\s*(?:y|n|yes|no|true|false|0|1)\s*$", re.IGNORECASE
    ),
}


def _normalize(text: str) -> str:
    """Lowercase + collapse whitespace. Used for substring/match comparisons."""
    return re.sub(r"\s+", " ", text.strip().lower())


def is_na(value: object) -> bool:
    """True when the extracted value should be treated as 'no answer'.

    Catches all the shapes a "missing" value can take coming out of the
    pipeline: Python None, the standard N/A string tokens, AND pandas /
    float NaN (which arrives as a float when the value column was empty
    in the dataframe). Float NaN is the only float that is not equal to
    itself, which is the cheapest portable check.
    """
    if value is None:
        return True
    # Float NaN check before any str() coercion (str(float('nan')) == 'nan',
    # which would otherwise slip past the token check below).
    if isinstance(value, float) and value != value:
        return True
    if isinstance(value, (list, tuple, set)):
        return all(is_na(item) for item in value)
    return _normalize(str(value)) in _NA_TOKENS


def _stringify(value: object) -> str:
    """Best-effort coercion of a value to a single string for matching."""
    if isinstance(value, (list, tuple, set)):
        return " ".join(_stringify(v) for v in value)
    return str(value) if value is not None else ""


# ----------------------------- signals -------------------------------------


def signal_exact(value: object, context: str) -> Optional[float]:
    """1.0 if the normalized value appears verbatim in the context, else 0.0."""
    if not context or is_na(value):
        return None
    needle = _normalize(_stringify(value))
    if not needle:
        return None
    return 1.0 if needle in _normalize(context) else 0.0


def signal_token_overlap(value: object, context: str) -> Optional[float]:
    """Share of value tokens (alphanumeric) present in the context."""
    if not context or is_na(value):
        return None
    value_tokens = [t.lower() for t in _TOKEN_RE.findall(_stringify(value))]
    if not value_tokens:
        return None
    haystack_tokens = {t.lower() for t in _TOKEN_RE.findall(context)}
    hits = sum(1 for t in value_tokens if t in haystack_tokens)
    return hits / len(value_tokens)


def signal_fuzzy(value: object, context: str) -> Optional[float]:
    """Best fuzzy substring similarity between the value and the context.

    Uses SequenceMatcher on the lowercased strings. Inexpensive on the
    capped chunk text we feed it. For very long contracts we'd want a
    sliding window, but for retrieved-chunks blobs the full match is fine.
    """
    if not context or is_na(value):
        return None
    needle = _normalize(_stringify(value))
    if not needle:
        return None
    haystack = _normalize(context)
    return SequenceMatcher(None, needle, haystack).ratio()


def signal_number_match(value: object, context: str) -> Optional[float]:
    """Share of numbers in the value that also appear in the context.

    Useful for dates / money / TINs / NPIs where the "right answer" is a
    set of digits. Returns None for values that don't contain any numbers
    (the signal isn't applicable).
    """
    if not context or is_na(value):
        return None
    value_numbers = _NUMBER_RE.findall(_stringify(value))
    if not value_numbers:
        return None
    haystack = context
    hits = sum(1 for n in value_numbers if n in haystack)
    return hits / len(value_numbers)


def signal_regex(value: object, field_type: FieldType) -> Optional[float]:
    """Does the value match the format expected for its field type?

    Returns None for FREE_TEXT and NAME (no meaningful format check).
    """
    if is_na(value):
        return None
    pattern = _FORMAT_PATTERNS.get(field_type)
    if pattern is None:
        return None
    candidate = _stringify(value).strip()
    return 1.0 if pattern.search(candidate) else 0.0


def signal_grounding(value: object, retrieved_chunks_text: str) -> Optional[float]:
    """How well does the value match the chunks the LLM actually saw?

    This is the strongest anti-hallucination signal: if the LLM produced
    a value that isn't in its retrieved context, it likely made it up.
    Combines exact substring (worth most) with fuzzy similarity (forgiving
    on minor edits / capitalization / spacing).
    """
    if not retrieved_chunks_text or is_na(value):
        return None
    needle = _normalize(_stringify(value))
    if not needle:
        return None
    haystack = _normalize(retrieved_chunks_text)
    if needle in haystack:
        return 1.0
    # Fall back to fuzzy: long haystack penalizes raw ratio, so use
    # find_longest_match against a heuristic window.
    fuzzy = SequenceMatcher(None, needle, haystack).ratio()
    # Boost slightly when most value tokens are present even if not in order.
    value_tokens = [t.lower() for t in _TOKEN_RE.findall(_stringify(value))]
    if value_tokens:
        haystack_tokens = {t.lower() for t in _TOKEN_RE.findall(haystack)}
        token_hit_rate = sum(1 for t in value_tokens if t in haystack_tokens) / len(
            value_tokens
        )
        return max(fuzzy, 0.7 * token_hit_rate)
    return fuzzy


def signal_na_check(
    value: object,
    contract_text: str,
    field_keywords: Optional[list[str]] = None,
) -> Optional[float]:
    """Score legitimate vs suspicious 'N/A' answers.

    Returns None for values that are NOT N/A (other signals cover those).

    For N/A values, counts total occurrences of the field's keywords in
    the contract. A doc that genuinely contains the field's concept will
    mention it many times; a doc that doesn't will have zero or near-zero
    mentions. The score is continuous, not bucketed, so an amendment doc
    that mentions "amendment" four times scores meaningfully lower (more
    suspicious) than one that mentions it once in passing.

    Score ramp:
        0 hits          -> 0.85   (the absence looks correct)
        saturation      -> 0.20   (the contract clearly discusses this;
                                   the LLM's N/A is almost certainly wrong)
    """
    if not is_na(value):
        return None
    if not contract_text or not field_keywords:
        # Without keywords we can't tell suspicious from legitimate; stay
        # neutral rather than guess.
        return 0.6
    haystack = _normalize(contract_text)
    # Count total occurrences (not just presence) so frequency carries
    # signal. Saturation has a meaningful floor (~30 mentions) because
    # contracts often contain a few mentions of any concept in boilerplate
    # ("may be amended by written amendment", etc.) without actually being
    # about it. Only docs that mention the concept tens of times are
    # clearly "about" it; those are the ones whose N/A is suspicious.
    total_hits = sum(haystack.count(kw.lower()) for kw in field_keywords if kw)
    saturation = max(30, 5 * len(field_keywords))
    intensity = min(total_hits / saturation, 1.0)
    return 0.85 - 0.65 * intensity