pi-qmem/gateway/guardrail.py

"""Guardrail anti-duplicati e similarità pre-scrittura.

Strato 1: hash esatto. Strato 2: cosine (bi-encoder). Strato 3 (opzionale,
GUARDRAIL_RERANK): cross-encoder che conferma o scarta il "quasi-duplicato" —
il cosine confonde "stesso argomento" con "stesso fatto", il cross-encoder
legge le coppie e giudica se il nuovo testo sia davvero lo stesso contenuto.
Il reranker è un miglioramento: se non raggiungibile si degrada alla sola
similarità (niente fallimenti di scrittura per un reranker giù).
"""
from __future__ import annotations

import hashlib
import unicodedata
from typing import Any, Optional

from qdrant_client.http import models as qm

import rerank
from config import (
    GUARDRAIL_BLOCK_THRESHOLD,
    GUARDRAIL_RERANK,
    GUARDRAIL_RERANK_BLOCK,
    GUARDRAIL_RERANK_SUGGEST,
    GUARDRAIL_WARN_THRESHOLD,
    log,
)


def normalize_text(text: str) -> str:
    s = unicodedata.normalize("NFD", text.lower())
    s = "".join(c for c in s if not unicodedata.combining(c))
    return " ".join(s.split())


def text_hash(text: str) -> str:
    return hashlib.sha256(normalize_text(text).encode("utf-8")).hexdigest()


def find_similar(qdrant: Any, collection: str, text: str, vector: list[float], top_k: int = 3) -> list[dict]:
    qfilter = qm.Filter(must=[qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by"))])
    hits = qdrant.query_points(collection_name=collection, query=vector, query_filter=qfilter, limit=top_k, with_payload=True).points
    return [
        {
            "memory_id": h.id,
            "score": round(float(h.score), 4),
            "text": (h.payload or {}).get("text", ""),
            "kind": (h.payload or {}).get("kind", ""),
            "project_id": (h.payload or {}).get("project_id", ""),
        }
        for h in hits
    ]


async def _cross_scores(text: str, matches: list[dict]) -> Optional[list[float]]:
    """Cross-score (0,1) di (nuovo testo, candidato) per ogni match; None se non disponibile."""
    if not GUARDRAIL_RERANK or not rerank.enabled():
        return None
    docs = [m["text"] or " " for m in matches]
    try:
        rr = await rerank.rerank(text, docs)
    except Exception as exc:  # noqa: BLE001
        log.warning("guardrail: rerank non disponibile (%s) → decisione solo cosine", exc.__class__.__name__)
        return None
    if rr is None:
        log.warning("guardrail: tutti i nodi rerank non raggiungibili → decisione solo cosine")
        return None
    scores, _backend, _took = rr
    return [rerank.normalize_score(s) for s in scores]


async def decide(qdrant: Any, collection: str, text: str, vector: list[float], topic: Optional[str] = None, parent_id: Optional[str] = None) -> dict:
    exact_filter = qm.Filter(must=[
        qm.FieldCondition(key="text_hash", match=qm.MatchValue(value=text_hash(text))),
        qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")),
    ])
    exact = qdrant.query_points(collection_name=collection, query=vector, query_filter=exact_filter, limit=1, with_payload=True).points
    if exact:
        return {"decision": "BLOCK", "reason": "EXACT_DUPLICATE", "matches": [{"memory_id": exact[0].id, "score": 1.0}]}

    matches = find_similar(qdrant, collection, text, vector, top_k=3)
    if not matches:
        return {"decision": "ALLOW", "reason": "NO_CANDIDATE", "matches": []}

    # Strato 3: cross-encoder sulla short-list (giudice "è lo stesso fatto?")
    cross = await _cross_scores(text, matches)
    if cross is not None:
        for m, c in zip(matches, cross):
            m["cross_score"] = round(c, 4)
        best_cross = max(cross)
        best_match = matches[cross.index(best_cross)]
    else:
        best_cross = None
        best_match = matches[0]

    top1 = matches[0]["score"]
    hierarchical = (topic or parent_id) and any(m.get("memory_id") != parent_id for m in matches)

    if top1 >= GUARDRAIL_BLOCK_THRESHOLD:
        if hierarchical:
            return {"decision": "WARN", "reason": "HIERARCHICAL_SUBTOPIC", "matches": matches}
        if best_cross is not None:
            if best_cross >= GUARDRAIL_RERANK_BLOCK:
                return {"decision": "BLOCK", "reason": "CROSS_DUP_CONFIRMED", "matches": matches}
            return {"decision": "WARN", "reason": "CROSS_DUP_WEAK", "matches": matches,
                    "message": "Similarità alta ma il cross-encoder non conferma lo stesso fatto: probabilmente correlati, non duplicati."}
        return {"decision": "BLOCK", "reason": "KNOWN_SOLUTION", "matches": matches}

    if top1 >= GUARDRAIL_WARN_THRESHOLD:
        if best_cross is not None and best_cross >= GUARDRAIL_RERANK_BLOCK:
            return {"decision": "BLOCK", "reason": "CROSS_DUP_CONFIRMED", "matches": matches}
        d = {"decision": "WARN", "reason": "MODERATE_SIMILARITY", "matches": matches}
        if best_cross is not None and best_cross >= GUARDRAIL_RERANK_SUGGEST:
            d["suggestion"] = {
                "supersedes_id": best_match["memory_id"],
                "cross_score": round(best_cross, 4),
                "message": "Sembra un aggiornamento del record indicato: valuta supersedes_id.",
            }
        return d

    # Cosine sotto la soglia WARN, ma cross-encoder che conferma un duplicato
    # parafrasato sfuggito al bi-encoder.
    if best_cross is not None and best_cross >= GUARDRAIL_RERANK_BLOCK:
        return {"decision": "BLOCK", "reason": "CROSS_DUP_LOW_COSINE", "matches": matches}

    return {"decision": "ALLOW", "reason": "NEW_SOLUTION", "matches": matches}


def parse_ts(value: Optional[str]) -> Optional[float]:
    from datetime import datetime
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None