Matteo Benedetto
4776b4b496
- gateway: separate config, models, state, audit, guardrail, embeddings, store, metrics, cleanup and routes; keep main.py as FastAPI bootstrap - extension: split client/config, six tools, config command and rules; preserve jiti entrypoint and registrations - Dockerfile copies the complete gateway module set - tests: update monkeypatch boundaries for modular config/state
68 lines
2.8 KiB
Python
68 lines
2.8 KiB
Python
"""Guardrail anti-duplicati e similarità pre-scrittura."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import unicodedata
|
|
from typing import Any, Optional
|
|
|
|
from qdrant_client.http import models as qm
|
|
|
|
from config import GUARDRAIL_BLOCK_THRESHOLD, GUARDRAIL_WARN_THRESHOLD
|
|
|
|
|
|
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
|
|
]
|
|
|
|
|
|
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": []}
|
|
top1 = matches[0]["score"]
|
|
if top1 >= GUARDRAIL_BLOCK_THRESHOLD:
|
|
if (topic or parent_id) and any(m.get("memory_id") != parent_id for m in matches):
|
|
return {"decision": "WARN", "reason": "HIERARCHICAL_SUBTOPIC", "matches": matches}
|
|
return {"decision": "BLOCK", "reason": "KNOWN_SOLUTION", "matches": matches}
|
|
if top1 >= GUARDRAIL_WARN_THRESHOLD:
|
|
return {"decision": "WARN", "reason": "MODERATE_SIMILARITY", "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
|