From 128059dd4be202b743e1e6acf87c2c81fb334d12 Mon Sep 17 00:00:00 2001 From: enne2 Date: Tue, 8 Sep 2026 12:57:33 +0200 Subject: [PATCH] feat(gateway): strategie rerank oltre la search (A-F) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A: gate store con cross-encoder — guardrail.decide async, conferma/scarta quasi-duplicati (CROSS_DUP_CONFIRMED/WEAK/LOW_COSINE), suggerimento supersedes in WARN, degrada a cosine-only se il reranker è giù - B: verifica supersede — cross-score (nuovo,vecchio) sotto soglia → supersede_warning non bloccante + audit - C: score composito in search — rerank + importance (nuovo campo payload) + recency decay (180gg) + authority, pesi SCORE_W_* da env - E: multi-query — SearchIn.queries (max 3), pool unito con dedup, rerank unico; endpoint POST /v1/score come primitiva cross-encoder (F-lite) - extension search.ts: param queries + rerank_score/composite in output - D: scripts/consolidate.py — dedup periodico a coppie via cross-encoder con report ntfy e --apply via gateway - test: 69 pass (+11 strategie); guardrail_version similarity-v2 --- extensions/tools/search.ts | 9 +- gateway/config.py | 17 ++- gateway/guardrail.py | 80 +++++++++- gateway/models.py | 20 +++ gateway/routes.py | 90 ++++++++++- gateway/store.py | 1 + gateway/tests/test_api.py | 2 +- gateway/tests/test_strategies.py | 252 +++++++++++++++++++++++++++++++ scripts/consolidate.py | 218 ++++++++++++++++++++++++++ 9 files changed, 671 insertions(+), 18 deletions(-) create mode 100644 gateway/tests/test_strategies.py create mode 100644 scripts/consolidate.py diff --git a/extensions/tools/search.ts b/extensions/tools/search.ts index 0b80f51..b6a9ee2 100644 --- a/extensions/tools/search.ts +++ b/extensions/tools/search.ts @@ -47,6 +47,12 @@ export function registerQmemSearch(pi: ExtensionAPI) { description: "Include private records only for explicit sensitive-data lookup.", }), ), + queries: Type.Optional( + Type.Array(Type.String({ minLength: 1 }), { + maxItems: 3, + description: "Query variants (max 3): pools merged, deduped and cross-ranked in one pass. Improves recall on long-tail queries.", + }), + ), }), async execute(toolCallId, params, signal, onUpdate, ctx) { const cfg = loadConfig(); @@ -75,6 +81,7 @@ export function registerQmemSearch(pi: ExtensionAPI) { parent_id: p.parent_id, level: p.level, topic: p.topic, + ...(p.queries && p.queries.length > 0 ? { queries: p.queries } : {}), }, signal, ); @@ -102,7 +109,7 @@ export function registerQmemSearch(pi: ExtensionAPI) { const top = r.topic ? ` (${r.topic})` : ""; const parent = r.parent_id ? `, parent: ${r.parent_id}` : ""; const links = r.links && r.links.length > 0 ? `, links: ${r.links.length}` : ""; - return `${i + 1}. [${r.kind}/${r.scope}${lvl}${top} score=${r.score}${r.score < 0.6 ? " ⚠️" : ""}${r.confidence ? ` conf=${r.confidence}` : ""}] ${r.text}\n (id: ${r.memory_id}${parent}${links}, agente: ${r.agent_id ?? "?"}, creato: ${r.created_at ?? "?"}${r.source ? `, fonte: ${r.source}` : ""}${r.supersedes_id ? `, supersede ${r.supersedes_id}` : ""}${r.superseded_by ? `, ⚠️ superseduto da ${r.superseded_by}` : ""})`; + return `${i + 1}. [${r.kind}/${r.scope}${lvl}${top} score=${r.score}${r.score < 0.6 ? " ⚠️" : ""}${r.rerank_score != null ? ` rerank=${r.rerank_score}` : ""}${r.composite_score != null ? ` composite=${r.composite_score}` : ""}${r.confidence ? ` conf=${r.confidence}` : ""}] ${r.text}\n (id: ${r.memory_id}${parent}${links}, agente: ${r.agent_id ?? "?"}, creato: ${r.created_at ?? "?"}${r.importance != null && r.importance !== 0.5 ? `, importanza: ${r.importance}` : ""}${r.source ? `, fonte: ${r.source}` : ""}${r.supersedes_id ? `, supersede ${r.supersedes_id}` : ""}${r.superseded_by ? `, ⚠️ superseduto da ${r.superseded_by}` : ""})`; }, ); return { diff --git a/gateway/config.py b/gateway/config.py index a66f817..08c3d8b 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -28,9 +28,22 @@ MAX_TEXT_LEN = int(os.environ.get("MAX_TEXT_LEN", "8000")) GUARDRAIL_ENABLED = os.environ.get("GUARDRAIL_ENABLED", "true").lower() == "true" GUARDRAIL_BLOCK_THRESHOLD = float(os.environ.get("GUARDRAIL_BLOCK_THRESHOLD", "0.85")) GUARDRAIL_WARN_THRESHOLD = float(os.environ.get("GUARDRAIL_WARN_THRESHOLD", "0.70")) -GUARDRAIL_VERSION = "similarity-v1" +GUARDRAIL_VERSION = "similarity-v2" +# Strato 3 del guardrail: cross-encoder (richiede catena rerank attiva) +GUARDRAIL_RERANK = os.environ.get("GUARDRAIL_RERANK", "false").lower() == "true" +GUARDRAIL_RERANK_BLOCK = float(os.environ.get("GUARDRAIL_RERANK_BLOCK", "0.88")) +GUARDRAIL_RERANK_SUGGEST = float(os.environ.get("GUARDRAIL_RERANK_SUGGEST", "0.80")) +# Verifica supersede: cross-score (nuovo, vecchio) sotto soglia → warning non bloccante +GUARDRAIL_SUPERSEDE_CHECK = os.environ.get("GUARDRAIL_SUPERSEDE_CHECK", "false").lower() == "true" +GUARDRAIL_SUPERSEDE_MIN = float(os.environ.get("GUARDRAIL_SUPERSEDE_MIN", "0.50")) +# Score composito: rerank + importance + recency + authority (post-rerank) +SCORE_W_RELEVANCE = float(os.environ.get("SCORE_W_RELEVANCE", "0.55")) +SCORE_W_IMPORTANCE = float(os.environ.get("SCORE_W_IMPORTANCE", "0.20")) +SCORE_W_RECENCY = float(os.environ.get("SCORE_W_RECENCY", "0.15")) +SCORE_W_AUTHORITY = float(os.environ.get("SCORE_W_AUTHORITY", "0.10")) +SCORE_DECAY_HALF_LIFE_DAYS = float(os.environ.get("SCORE_DECAY_HALF_LIFE_DAYS", "180")) GIT_COMMIT = os.environ.get("GIT_COMMIT", "unknown").strip() -GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.10.0").strip() +GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.11.0").strip() VM_PUSH_URL = os.environ.get("VM_PUSH_URL", "http://host.docker.internal:8428/api/v1/import/prometheus") VM_PUSH_INTERVAL = int(os.environ.get("VM_PUSH_INTERVAL", "30")) diff --git a/gateway/guardrail.py b/gateway/guardrail.py index ee0420d..ac2b841 100644 --- a/gateway/guardrail.py +++ b/gateway/guardrail.py @@ -1,4 +1,12 @@ -"""Guardrail anti-duplicati e similarità pre-scrittura.""" +"""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 @@ -7,7 +15,15 @@ from typing import Any, Optional from qdrant_client.http import models as qm -from config import GUARDRAIL_BLOCK_THRESHOLD, GUARDRAIL_WARN_THRESHOLD +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: @@ -35,7 +51,24 @@ def find_similar(qdrant: Any, collection: str, text: str, vector: list[float], t ] -def decide(qdrant: Any, collection: str, text: str, vector: list[float], topic: Optional[str] = None, parent_id: Optional[str] = None) -> dict: +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")), @@ -47,13 +80,48 @@ def decide(qdrant: Any, collection: str, text: str, vector: list[float], topic: 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 (topic or parent_id) and any(m.get("memory_id") != parent_id for m in matches): + 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: - return {"decision": "WARN", "reason": "MODERATE_SIMILARITY", "matches": matches} + 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} @@ -64,4 +132,4 @@ def parse_ts(value: Optional[str]) -> Optional[float]: try: return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() except ValueError: - return None + return None \ No newline at end of file diff --git a/gateway/models.py b/gateway/models.py index 5287a7d..75b5957 100644 --- a/gateway/models.py +++ b/gateway/models.py @@ -30,6 +30,7 @@ class MemoryIn(BaseModel): level: Optional[Literal["L1_ROOT", "L2_SUBTOPIC", "L3_DETAIL"]] = Field(default=None, description="Livello gerarchico") topic: Optional[str] = Field(default=None, max_length=128, description="Topic gerarchico") links: Optional[list[MemoryLink]] = Field(default=None, description="Collegamenti semantici e relazionali") + importance: float = Field(default=0.5, ge=0.0, le=1.0, description="Importanza stabile del record (usata nello score composito)") private: bool = Field(default=False, description="Riservato: escluso dalle ricerche standard, visibile solo con include_private o topic esplicito") @field_validator("expires_at") @@ -44,6 +45,13 @@ class MemoryIn(BaseModel): return v +class ScoreIn(BaseModel): + """Primitiva di scoring cross-encoder (usata da estensione e job di consolidamento).""" + + query: str = Field(min_length=1, max_length=512) + documents: list[str] = Field(min_length=1, max_length=32) + + class SearchIn(BaseModel): query: str = Field(min_length=1, max_length=512) kind: Optional[Literal["decision", "fact", "episode", "preference"]] = None @@ -58,3 +66,15 @@ class SearchIn(BaseModel): topic: Optional[str] = None include_private: bool = Field(default=False, description="Includi i record privati (solo ricerche esplicite)") rerank: Optional[bool] = Field(default=None, description="Override per-query dello stadio rerank (None = default server)") + queries: Optional[list[str]] = Field(default=None, max_length=3, description="Varianti di query (max 3): pool unito, dedup e rerank unico") + + @field_validator("queries") + @classmethod + def _validate_queries(cls, v: Optional[list[str]]) -> Optional[list[str]]: + if v is None: + return v + cleaned = [q.strip() for q in v if q and q.strip()] + if len(cleaned) != len(v): + raise ValueError("le query non devono essere vuote") + return cleaned + queries: Optional[list[str]] = Field(default=None, max_length=3, description="Varianti di query (max 3): pool unito, dedup e rerank unico") diff --git a/gateway/routes.py b/gateway/routes.py index 7ecfa1e..75a7fb3 100644 --- a/gateway/routes.py +++ b/gateway/routes.py @@ -5,6 +5,7 @@ import hashlib import time import uuid from collections import Counter +from datetime import datetime from typing import Any, Optional from fastapi import APIRouter, Depends, Header, HTTPException, Request @@ -26,12 +27,20 @@ from config import ( GATEWAY_VERSION, GUARDRAIL_BLOCK_THRESHOLD, GUARDRAIL_ENABLED, + GUARDRAIL_RERANK_BLOCK, + GUARDRAIL_SUPERSEDE_CHECK, + GUARDRAIL_SUPERSEDE_MIN, GUARDRAIL_VERSION, GUARDRAIL_WARN_THRESHOLD, MAX_TEXT_LEN, RERANK_CANDIDATES, + SCORE_DECAY_HALF_LIFE_DAYS, + SCORE_W_AUTHORITY, + SCORE_W_IMPORTANCE, + SCORE_W_RECENCY, + SCORE_W_RELEVANCE, ) -from models import MemoryIn, SearchIn +from models import MemoryIn, ScoreIn, SearchIn router = APIRouter() @@ -54,6 +63,7 @@ async def add_memory( memory_id = str(uuid.uuid4()) superseded_id: Optional[str] = None + supersede_warning: Optional[dict] = None if body.supersedes_id: old = state.qdrant.retrieve(collection_name=COLLECTION, ids=[body.supersedes_id], with_payload=True) if not old: @@ -61,12 +71,23 @@ async def add_memory( if old[0].payload.get("superseded_by"): raise HTTPException(status_code=409, detail="La memoria è già stata superseduta: correggi la versione attiva") superseded_id = body.supersedes_id + # Verifica lineage (B): la correzione deve parlare dello stesso fatto del record vecchio + if config.GUARDRAIL_SUPERSEDE_CHECK and rerank.enabled() and (old[0].payload or {}).get("text"): + rr = await rerank.rerank(body.text, [(old[0].payload or {}).get("text", "")]) + if rr: + cross = rerank.normalize_score(rr[0][0]) + if cross < config.GUARDRAIL_SUPERSEDE_MIN: + supersede_warning = { + "cross_score": round(cross, 4), + "message": "La correzione non sembra riguardare lo stesso fatto del record originale: verifica il lineage.", + } + audit(key, "supersede_weak_cross", old_id=superseded_id, cross_score=round(cross, 4)) vector = await state.embed(body.text) sparse = state.sparse_encode(body.text) similarity_guardrail: Optional[dict] = None if config.GUARDRAIL_ENABLED and not body.supersedes_id: - similarity_guardrail = guardrail.decide(state.qdrant, COLLECTION, body.text, vector, topic=body.topic, parent_id=body.parent_id) + similarity_guardrail = await guardrail.decide(state.qdrant, COLLECTION, body.text, vector, topic=body.topic, parent_id=body.parent_id) if similarity_guardrail["decision"] == "BLOCK": audit(key, "create_blocked", kind=body.kind, agent_id=body.agent_id or "shared", reason=similarity_guardrail["reason"], matches=[m["memory_id"] for m in similarity_guardrail["matches"]]) raise HTTPException( @@ -95,6 +116,7 @@ async def add_memory( "parent_id": body.parent_id, "level": body.level, "topic": body.topic, + "importance": body.importance, "links": [link.model_dump() for link in body.links] if body.links else None, "embedding_model": EMBED_MODEL, "text_hash": guardrail.text_hash(body.text), @@ -106,6 +128,8 @@ async def add_memory( "reason": similarity_guardrail["reason"], "matches": similarity_guardrail["matches"], } + if similarity_guardrail.get("suggestion"): + payload["guardrail"]["suggestion"] = similarity_guardrail["suggestion"] point_vector: dict[str, Any] = {"": vector} if sparse is not None: point_vector["bm25"] = sparse @@ -127,29 +151,60 @@ async def add_memory( audit(key, "create", memory_id=memory_id, kind=body.kind, agent_id=payload["agent_id"], guardrail=payload.get("guardrail", {}).get("decision", "ALLOW")) response = {"memory_id": memory_id, "created_at": payload["created_at"], "supersedes_id": superseded_id, "reparented": reparented_count} + if supersede_warning: + response["supersede_warning"] = supersede_warning if idem_key: state.idempotency[idem_key] = {"hash": state.payload_hash(body), "response": response, "ts": time.time()} return response +def _composite_score(r: dict, now: float) -> float: + """Score composito (C): rerank + importance + recency-decay + authority, pesi normalizzati.""" + try: + age_days = max(0.0, (now - datetime.fromisoformat(str(r.get("created_at")).replace("Z", "+00:00")).timestamp()) / 86400.0) + except (ValueError, TypeError, AttributeError): + age_days = 0.0 + recency = pow(0.5, age_days / SCORE_DECAY_HALF_LIFE_DAYS) + authority = {"high": 1.0, "medium": 0.7, "low": 0.4}.get(r.get("confidence"), 0.7) + importance = float(r.get("importance", 0.5) or 0.5) + total_w = SCORE_W_RELEVANCE + SCORE_W_IMPORTANCE + SCORE_W_RECENCY + SCORE_W_AUTHORITY + raw = ( + SCORE_W_RELEVANCE * float(r["rerank_score"]) + + SCORE_W_IMPORTANCE * importance + + SCORE_W_RECENCY * recency + + SCORE_W_AUTHORITY * authority + ) + return raw / total_w if total_w else raw + + @router.post("/v1/memories:search") async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> dict: use_rerank = rerank.enabled() and body.rerank is not False # Con reranking attivo recuperiamo più candidati di top_k per dare margine allo stadio di rerank limit = max(body.top_k, RERANK_CANDIDATES) if use_rerank else body.top_k - vector = await state.embed(body.query) - sparse = state.sparse_encode(body.query) if body.hybrid else None - hits = store.search(state.qdrant, COLLECTION, body, vector, sparse, limit=limit) - results = store.format_results(hits) - rerank_info: dict = {"enabled": use_rerank, "used": False} + # Multi-query (E): varianti della stessa query, pool unito con dedup (la prima ha priorità) + queries = list(dict.fromkeys([body.query] + [q for q in (body.queries or []) if q]))[:3] + + merged: dict[str, Any] = {} + for q in queries: + vector = await state.embed(q) + sparse = state.sparse_encode(q) if body.hybrid else None + hits = store.search(state.qdrant, COLLECTION, body, vector, sparse, limit=limit) + for h in hits: + merged.setdefault(h.id, h) + results = store.format_results(list(merged.values())[:limit]) + + rerank_info: dict = {"enabled": use_rerank, "used": False, "queries_used": len(queries)} if use_rerank and len(results) >= 2: rr = await rerank.rerank(body.query, [r["text"] or "" for r in results]) if rr: scores, backend, took_ms = rr + now = time.time() for r, s in zip(results, scores): r["rerank_score"] = round(rerank.normalize_score(s), 4) - results.sort(key=lambda r: r["rerank_score"], reverse=True) + r["composite_score"] = round(_composite_score(r, now), 4) + results.sort(key=lambda r: (r["composite_score"], r["rerank_score"]), reverse=True) rerank_info.update(used=True, backend=backend, took_ms=took_ms, candidates=len(results)) else: rerank_info["reason"] = "tutti i nodi rerank non raggiungibili (ordine di fusione preservato)" @@ -165,11 +220,30 @@ async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> d min_score=body.min_score, hits=len(results), rerank_backend=rerank_info.get("backend"), + queries_used=len(queries), ) metrics.record_search(len(results)) return {"results": results, "min_score": body.min_score, "total_hits": len(results), "rerank": rerank_info} +@router.post("/v1/score") +async def score(body: ScoreIn, key: str = Depends(require_auth)) -> dict: + """Primitiva cross-encoder: rilevanza (query, documento) in [0,1] via catena rerank. + + Building block per estensione (validazione estrattore, lineage check) e job + di consolidamento; 503 se tutti i nodi della catena non raggiungibili.""" + rr = await rerank.rerank(body.query, body.documents) + if rr is None: + raise HTTPException(status_code=503, detail="nessun nodo rerank raggiungibile") + scores, backend, took_ms = rr + return { + "scores": [round(rerank.normalize_score(s), 4) for s in scores], + "raw": [round(s, 4) for s in scores], + "backend": backend, + "took_ms": took_ms, + } + + @router.get("/v1/memories/{memory_id}") async def get_memory(memory_id: str, key: str = Depends(require_auth)) -> dict: point = state.qdrant.retrieve(collection_name=COLLECTION, ids=[memory_id], with_payload=True) diff --git a/gateway/store.py b/gateway/store.py index e54ae1d..8063661 100644 --- a/gateway/store.py +++ b/gateway/store.py @@ -64,6 +64,7 @@ def format_results(hits: list[Any]) -> list[dict]: "scope": h.payload.get("scope"), "project_id": h.payload.get("project_id"), "confidence": h.payload.get("confidence"), + "importance": h.payload.get("importance", 0.5), "created_at": h.payload.get("created_at"), "source": h.payload.get("source"), "supersedes_id": h.payload.get("supersedes_id"), diff --git a/gateway/tests/test_api.py b/gateway/tests/test_api.py index b41132f..030bd6f 100644 --- a/gateway/tests/test_api.py +++ b/gateway/tests/test_api.py @@ -292,7 +292,7 @@ def test_version_endpoint_pubblico(client): assert "git_commit" in data assert "version" in data assert "guardrail_version" in data - assert data["guardrail_version"] == "similarity-v1" + assert data["guardrail_version"] == "similarity-v2" def test_status_espone_git_commit(client): diff --git a/gateway/tests/test_strategies.py b/gateway/tests/test_strategies.py new file mode 100644 index 0000000..d8aaae6 --- /dev/null +++ b/gateway/tests/test_strategies.py @@ -0,0 +1,252 @@ +"""Test strategie rerank oltre la search: gate store (A), supersede verify (B), +score composito (C), multi-query (E), primitiva /v1/score.""" +from __future__ import annotations + +import pytest +from conftest import auth_headers, make_record + + +@pytest.fixture(autouse=True) +def enable_guardrail(monkeypatch): + import config + + monkeypatch.setattr(config, "GUARDRAIL_ENABLED", True) + + +def _patch_rerank(monkeypatch, scores, backend="finto", took=10): + """Abilita il rerank anche nel guardrail (che importa i valori da config).""" + monkeypatch.setattr("rerank.enabled", lambda: True) + monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) + recorded: dict = {} + + async def fake_rerank(query, docs): + recorded["query"] = query + recorded["docs"] = list(docs) + return scores, backend, took + + monkeypatch.setattr("rerank.rerank", fake_rerank) + return recorded + + +# --------------------------------------------------------------------------- +# A: gate store con cross-encoder +# --------------------------------------------------------------------------- + + +def test_cosine_alto_cross_basso_downgrade_a_warn(client, monkeypatch): + """Cosine 0.9 (zona BLOCK) ma cross-score basso → WARN CROSS_DUP_WEAK, salva.""" + _patch_rerank(monkeypatch, scores=[-6.0]) # sigmoid ≈ 0.0024 + client.fake_qdrant.query_score = 0.9 + client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) + r2 = client.post("/v1/memories", json=make_record(text="secondo simile"), headers=auth_headers()) + assert r2.status_code == 200 + saved = client.fake_qdrant.points[r2.json()["memory_id"]].payload + assert saved["guardrail"]["decision"] == "WARN" + assert saved["guardrail"]["reason"] == "CROSS_DUP_WEAK" + + +def test_cosine_warn_cross_alto_upgrade_a_block(client, monkeypatch): + """Cosine in zona WARN (0.75) ma cross-score altissimo → BLOCK parafrasato catturato.""" + import rerank as rr_mod + monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) + monkeypatch.setattr("rerank.enabled", lambda: True) + + async def fake_rerank(query, docs): + return [3.0], "finto", 5 # sigmoid ≈ 0.953 ≥ 0.88 + + monkeypatch.setattr("rerank.rerank", fake_rerank) + client.fake_qdrant.query_score = 0.75 + client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) + r2 = client.post("/v1/memories", json=make_record(text="stesso fatto riformulato"), headers=auth_headers()) + assert r2.status_code == 409 + assert r2.json()["detail"]["reason"] == "CROSS_DUP_CONFIRMED" + + +def test_cosine_basso_cross_alto_block_low_cosine(client, monkeypatch): + """Cosine sotto soglia WARN (0.5) ma cross altissimo → CROSS_DUP_LOW_COSINE.""" + monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) + monkeypatch.setattr("rerank.enabled", lambda: True) + + async def fake_rerank(query, docs): + return [4.0], "finto", 5 # sigmoid ≈ 0.982 + + monkeypatch.setattr("rerank.rerank", fake_rerank) + client.fake_qdrant.query_score = 0.5 + client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) + r2 = client.post("/v1/memories", json=make_record(text="secondo riformulato"), headers=auth_headers()) + assert r2.status_code == 409 + assert r2.json()["detail"]["reason"] == "CROSS_DUP_LOW_COSINE" + + +def test_warn_con_suggerimento_supersedes(client, monkeypatch): + """Cosine in zona WARN, cross ≥ soglia suggest → WARN con suggestion.supersedes_id.""" + monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) + monkeypatch.setattr("rerank.enabled", lambda: True) + + async def fake_rerank(query, docs): + return [1.5], "finto", 5 # sigmoid ≈ 0.818: ≥ 0.80, < 0.88 + + monkeypatch.setattr("rerank.rerank", fake_rerank) + client.fake_qdrant.query_score = 0.75 + r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) + old_id = r1.json()["memory_id"] + r2 = client.post("/v1/memories", json=make_record(text="secondo simile"), headers=auth_headers()) + assert r2.status_code == 200 + saved = client.fake_qdrant.points[r2.json()["memory_id"]].payload + assert saved["guardrail"]["reason"] == "MODERATE_SIMILARITY" + assert saved["guardrail"]["suggestion"]["supersedes_id"] == old_id + + +def test_rerank_giu_degrada_a_solo_cosine(client, monkeypatch): + """Reranker irraggiungibile → decisione legacy per cosine (BLOCK a 0.9).""" + monkeypatch.setattr("guardrail.GUARDRAIL_RERANK", True) + monkeypatch.setattr("rerank.enabled", lambda: True) + + async def fail_rerank(query, docs): + return None + + monkeypatch.setattr("rerank.rerank", fail_rerank) + client.fake_qdrant.query_score = 0.9 + client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers()) + r2 = client.post("/v1/memories", json=make_record(text="secondo simile"), headers=auth_headers()) + assert r2.status_code == 409 + assert r2.json()["detail"]["reason"] == "KNOWN_SOLUTION" + + +# --------------------------------------------------------------------------- +# B: verifica supersede +# --------------------------------------------------------------------------- + + +def test_supersede_cross_basso_warning(client, monkeypatch): + monkeypatch.setattr("config.GUARDRAIL_SUPERSEDE_CHECK", True) + monkeypatch.setattr("rerank.enabled", lambda: True) + + async def fake_rerank(query, docs): + return [-8.0], "finto", 5 # sigmoid ≈ 0.0003 < 0.50 + + monkeypatch.setattr("rerank.rerank", fake_rerank) + r1 = client.post("/v1/memories", json=make_record(text="record originale"), headers=auth_headers()) + old_id = r1.json()["memory_id"] + r2 = client.post( + "/v1/memories", + json=make_record(text="contenuto del tutto diverso", supersedes_id=old_id, supersede_reason="fix"), + headers=auth_headers(), + ) + assert r2.status_code == 200 + data = r2.json() + assert data["supersede_warning"]["cross_score"] < 0.1 + assert "lineage" in data["supersede_warning"]["message"] + + +def test_supersede_cross_alto_nessun_warning(client, monkeypatch): + monkeypatch.setattr("config.GUARDRAIL_SUPERSEDE_CHECK", True) + monkeypatch.setattr("rerank.enabled", lambda: True) + + async def fake_rerank(query, docs): + return [3.0], "finto", 5 # sigmoid ≈ 0.95 ≥ 0.50 + + monkeypatch.setattr("rerank.rerank", fake_rerank) + r1 = client.post("/v1/memories", json=make_record(text="record originale"), headers=auth_headers()) + old_id = r1.json()["memory_id"] + r2 = client.post( + "/v1/memories", + json=make_record(text="record originale corretto", supersedes_id=old_id, supersede_reason="correzione"), + headers=auth_headers(), + ) + assert r2.status_code == 200 + assert "supersede_warning" not in r2.json() + + +# --------------------------------------------------------------------------- +# C: score composito +# --------------------------------------------------------------------------- + + +def test_composite_score_in_risultati(client, monkeypatch): + client.fake_qdrant.query_score = 0.5 # il guardrail cosine non blocca il seeding + client.post("/v1/memories", json=make_record(text="record alpha", importance=1.0, confidence="high"), headers=auth_headers()) + client.post("/v1/memories", json=make_record(text="record beta", importance=0.0, confidence="low"), headers=auth_headers()) + + async def fake_rerank(query, docs): + return [2.0, 2.0], "finto", 5 # rerank in parità → il composito decide + + monkeypatch.setattr("rerank.enabled", lambda: True) + monkeypatch.setattr("rerank.rerank", fake_rerank) + + resp = client.post("/v1/memories:search", json={"query": "record", "top_k": 2}, headers=auth_headers()) + assert resp.status_code == 200 + data = resp.json() + results = data["results"] + assert all("composite_score" in r for r in results) + assert results[0]["text"] == "record alpha" + assert results[0]["composite_score"] > results[1]["composite_score"] + + +# --------------------------------------------------------------------------- +# E: multi-query +# --------------------------------------------------------------------------- + + +def test_multi_query_pool_unito(client, monkeypatch): + client.fake_qdrant.query_score = 0.5 # il guardrail cosine non blocca il seeding + for text in ["alpha", "beta", "gamma"]: + client.post("/v1/memories", json=make_record(text=text), headers=auth_headers()) + + embed_calls: list[str] = [] + + async def fake_embed(text): + embed_calls.append(text) + return [0.0] * 1024 + + monkeypatch.setattr("state.embed", fake_embed) + monkeypatch.setattr("state.sparse_encode", lambda text: None) + + async def fake_rerank(query, docs): + assert query == "alpha" # il rerank usa la query principale + return [0.9, 0.5, 0.1], "finto", 5 + + monkeypatch.setattr("rerank.enabled", lambda: True) + monkeypatch.setattr("rerank.rerank", fake_rerank) + + resp = client.post( + "/v1/memories:search", + json={"query": "alpha", "queries": ["gamma", "alpha "], "top_k": 3}, + headers=auth_headers(), + ) + assert resp.status_code == 200 + data = resp.json() + assert data["rerank"]["queries_used"] == 2 # "alpha " normalizzata e deduplicata + assert len(embed_calls) == 2 + assert set(r["text"] for r in data["results"]) == {"alpha", "beta", "gamma"} + assert data["results"][0]["text"] == "alpha" + + +# --------------------------------------------------------------------------- +# Primitiva /v1/score +# --------------------------------------------------------------------------- + + +def test_score_endpoint_ok(client, monkeypatch): + async def fake_rerank(query, docs): + assert query == "q" + return [2.0, -3.0], "finto", 7 + + monkeypatch.setattr("rerank.enabled", lambda: True) + monkeypatch.setattr("rerank.rerank", fake_rerank) + resp = client.post("/v1/score", json={"query": "q", "documents": ["a", "b"]}, headers=auth_headers()) + assert resp.status_code == 200 + data = resp.json() + assert data["scores"][0] == pytest.approx(0.88, abs=0.01) + assert data["scores"][1] < 0.1 + assert data["backend"] == "finto" + + +def test_score_endpoint_503_se_catena_giu(client, monkeypatch): + async def fail_rerank(query, docs): + return None + + monkeypatch.setattr("rerank.enabled", lambda: True) + monkeypatch.setattr("rerank.rerank", fail_rerank) + resp = client.post("/v1/score", json={"query": "q", "documents": ["a"]}, headers=auth_headers()) + assert resp.status_code == 503 \ No newline at end of file diff --git a/scripts/consolidate.py b/scripts/consolidate.py new file mode 100644 index 0000000..09a21d3 --- /dev/null +++ b/scripts/consolidate.py @@ -0,0 +1,218 @@ +#!/usr/bin/env python3 +"""Consolidamento assistito da cross-encoder (strategia D). + +Trova i veri duplicati nella collection 'memories': candidatos per cosine +(bi-encoder) → cross-score a coppie col reranker (giudice "è lo stesso fatto?") +→ cluster di duplicati confermati → report (e, con --apply, rimozione dei +duplicati perdenti via API gateway, con audit). + +Uso (su brain): + python3 consolidate.py # report su stdout (+ ntfy se configurato) + python3 consolidate.py --apply # applica le rimozioni suggerite + python3 consolidate.py --limit 300 # limita il numero di record scansionati + +Env (da /opt/memory/.env se presente): QDRANT_URL, QDRANT_API_KEY, API_KEYS, +RERANK_CHAIN, NTFY_CONSOLIDAMENTO (opzionale: URL completo del topic ntfy). +""" +from __future__ import annotations + +import argparse +import json +import math +import os +import sys +import time +from datetime import datetime + +import httpx + +DEFAULT_ENV_FILE = "/opt/memory/.env" + + +def load_env(path: str) -> dict: + env = {} + if os.path.exists(path): + for line in open(path): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + k, v = line.split("=", 1) + env[k] = v + return env + + +def sigmoid(x: float) -> float: + if x >= 0: + z = math.exp(-x) + return 1.0 / (1.0 + z) + z = math.exp(x) + return z / (1.0 + z) + + +def parse_chain(raw: str) -> list[dict]: + try: + entries = json.loads(raw) if raw else [] + except json.JSONDecodeError: + return [] + return [e for e in entries if isinstance(e, dict) and e.get("url")] + + +def cross_score(http, chain: list[dict], query: str, docs: list[str], timeout_default: float) -> tuple[list[float], str] | None: + payload = {"model": "bge-reranker-v2-m3", "query": query, "documents": docs, "top_n": len(docs)} + for node in chain: + headers = {"Content-Type": "application/json"} + if node.get("key"): + headers["Authorization"] = f"Bearer {node['key']}" + try: + r = http.post( + f"{node['url'].rstrip('/')}/v1/rerank", + json=payload, + headers=headers, + timeout=node.get("timeout_ms", 10000) / 1000.0, + ) + r.raise_for_status() + scores = [0.0] * len(docs) + for item in r.json().get("results", []): + idx = int(item["index"]) + if 0 <= idx < len(docs): + scores[idx] = sigmoid(float(item.get("relevance_score", 0.0))) + return scores, node.get("name", node["url"]) + except Exception: + continue + return None + + +def recency_of(created_at: str, half_life_days: float = 180.0) -> float: + try: + age = (time.time() - datetime.fromisoformat(str(created_at).replace("Z", "+00:00")).timestamp()) / 86400.0 + except Exception: + return 0.5 + return pow(0.5, max(0.0, age) / half_life_days) + + +import time # noqa: E402 (dopo i docstring per leggibilità dell'ordine di import) + + +def priority(rec: dict, recency: float) -> float: + """Chi resta nel cluster: confidence + importance + recency.""" + conf = {"high": 1.0, "medium": 0.7, "low": 0.4}.get(rec.get("confidence"), 0.7) + return conf * 0.5 + float(rec.get("importance", 0.5) or 0.5) * 0.3 + recency * 0.2 + + +def main() -> int: + ap = argparse.ArgumentParser(description="Consolidamento duplicati via cross-encoder") + ap.add_argument("--env-file", default=DEFAULT_ENV_FILE) + ap.add_argument("--limit", type=int, default=0, help="max record da scansionare (0 = tutti)") + ap.add_argument("--cosine", type=float, default=0.70, help="soglia cosine per i candidati") + ap.add_argument("--cross", type=float, default=0.88, help="soglia cross-encoder per duplicato confermato") + ap.add_argument("--apply", action="store_true", help="rimuove i duplicati perdenti via API gateway") + ap.add_argument("--gateway-url", default=os.environ.get("GATEWAY_URL", "http://127.0.0.1:8082")) + ap.add_argument("--ntfy", default="", help="URL topic ntfy per il report (es. http://127.0.0.1:8091/qmem-consolidamento)") + args = ap.parse_args() + + env = load_env(args.env_file) + env.update({k: v for k, v in os.environ.items() if k in ("QDRANT_URL", "QDRANT_API_KEY", "RERANK_CHAIN", "GATEWAY_URL")}) + qdrant_url = env.get("QDRANT_URL", "http://127.0.0.1:6333").rstrip("/") + api_key = env.get("QDRANT_API_KEY", "") + chain = parse_chain(env.get("RERANK_CHAIN", "")) + if not chain: + print("RERANK_CHAIN vuota: niente cross-scoring, esco", file=sys.stderr) + return 2 + + import httpx + + headers = {"api-key": api_key} if api_key else {} + with httpx.Client(timeout=60) as http: + # 1) scroll record attivi (id, testo, metadata, vettore denso) + records: dict[str, dict] = {} + offset = None + while True: + body: dict = { + "filter": {"must": [{"key": "superseded_by", "match": None}]}, + "limit": 256, + "with_payload": True, + "with_vector": True, + } + if offset: + body["offset"] = offset + r = http.post(f"{qdrant_url}/collections/memories/points/scroll", json=body, headers=headers) + r.raise_for_status() + data = r.json() + for p in data.get("points", []): + vec = (p.get("vector") or {}).get("") if isinstance(p.get("vector"), dict) else p.get("vector") + if not vec: + continue + records[p["id"]] = { + "text": (p.get("payload") or {}).get("text", ""), + "confidence": (p.get("payload") or {}).get("confidence", "medium"), + "importance": (p.get("payload") or {}).get("importance", 0.5), + "created_at": (p.get("payload") or {}).get("created_at", ""), + "vector": vec, + } + offset = data.get("next_page_offset") + if not offset: + break + if args.limit: + records = dict(list(records.items())[: args.limit]) + print(f"scansionati {len(records)} record attivi") + + # 2) candidati per cosine (il vettore del record stesso come query) + pairs: set[tuple[str, str]] = set() + for rid, rec in records.items(): + r = http.post( + f"{qdrant_url}/collections/memories/points/query", + json={"query": rec["vector"], "limit": 4, "with_payload": False}, + headers=headers, + ) + r.raise_for_status() + for h in r.json().get("points", []): + oid = h["id"] + if oid == rid or oid not in records or h["score"] < args.cosine: + continue + pairs.add((min(rid, oid), max(rid, oid))) + print(f"coppie candidate (cosine ≥ {args.cosine}): {len(pairs)}") + + # 3) cross-score a coppie + confirmed: list[dict] = [] + for a, b in sorted(pairs): + rr = cross_score(http, chain, records[a]["text"][:800], [records[b]["text"][:800]], 10.0) + if rr is None: + print("catena rerank irraggiungibile: interrompo il cross-scoring", file=sys.stderr) + return 3 + cross = rr[0][0] + if cross >= args.cross: + keep, drop = (a, b) if priority(records[a], recency_of(records[a]["created_at"])) >= priority(records[b], recency_of(records[b]["created_at"])) else (b, a) + confirmed.append({"keep": keep, "drop": drop, "cross": round(cross, 4)}) + + # 4) report + print(f"duplicati confermati (cross ≥ {args.cross}): {len(confirmed)}") + for c in confirmed: + keep_txt = records[c["keep"]]["text"][:70].replace("\n", " ") + drop_txt = records[c["drop"]]["text"][:60].replace("\n", " ") + print(f" KEEP {c['keep']} DROP {c['drop']} cross={c['cross']} | drop: {drop_txt}") + + if args.apply and confirmed: + gw_headers = {"Content-Type": "application/json", "X-API-Key": env.get("API_KEYS", "").split(",")[0]} + removed = 0 + for c in confirmed: + try: + r = http.delete(f"{args.gateway_url.rstrip('/')}/v1/memories/{c['drop']}", headers=gw_headers) + if r.status_code == 200: + removed += 1 + else: + print(f" delete {c['drop']}: HTTP {r.status_code}", file=sys.stderr) + except Exception as exc: + print(f" delete {c['drop']}: {exc}", file=sys.stderr) + print(f"rimossi {removed}/{len(confirmed)} duplicati") + + if args.ntfy and confirmed: + lines = [f"qmem consolidamento: {len(confirmed)} duplicati confermati"] + lines += [f"• {c['cross']} — {records[c['drop']]['text'][:60]}" for c in confirmed[:5]] + try: + http.post(args.ntfy, data="\n".join(lines).encode(), headers={"Title": "qmem consolidamento"}) + except Exception as exc: + print(f"ntfy: {exc}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file