chore(gateway): rimuove la copia duplicata del gateway dal package (dedup)
La cartella gateway/ era un duplicato byte-identico (sha256 su 14 file) del repository canonico privato enne2/qmem-gateway (GATEWAY_VERSION 2.11.0, guardrail similarity-v2): due fonti dello stesso codice, rischio di divergenza. - la fonte unica del gateway + deploy (docker-compose.yml, .env, test, README operativo) è git:git.enne2.net/enne2/qmem-gateway, clonata in ~/dev/qmem-gateway - gli artefatti presenti SOLO qui (README.md, requirements-dev.txt, tests/) sono stati spostati in quella repo prima della rimozione (commit locale 15cc9d3, nessun push): nessuna perdita di contenuto - backup integrale della cartella rimossa: ~/archive/backups/pi-qmem-gateway-copy-20260913-173728.tar.gz - README aggiornato con il puntatore al repo canonico Il package pi-qmem ora contiene solo estensione, skill, tool CLI e test dell'indice locale (il gateway si deploya dal repo dedicato).
This commit is contained in:
Matteo Benedetto
@@ -116,9 +116,21 @@ Le regole vincolanti (obbligo `project_id`, punteggi, correzione/supersede, disc
|
||||
|
||||
## Gateway (componente server)
|
||||
|
||||
La cartella `gateway/` contiene il Memory Gateway FastAPI da deployare sul
|
||||
server (Docker Compose con Qdrant 1.19 + Ollama BGE-M3). Vedi
|
||||
`gateway/README.md` per il deploy.
|
||||
Il Memory Gateway FastAPI + Qdrant **non è più duplicato in questo package**:
|
||||
la fonte unica è il repository dedicato
|
||||
|
||||
```
|
||||
git:git.enne2.net/enne2/qmem-gateway (privato)
|
||||
```
|
||||
|
||||
che contiene il codice (`gateway/`), il deploy (`docker-compose.yml` con Qdrant
|
||||
1.19 + gateway, `.env`, `.gitignore`), la suite di test e il README operativo.
|
||||
Su questa macchina è clonato in `~/dev/qmem-gateway`.
|
||||
|
||||
Motivo della dedup: la copia qui dentro era byte-identica al repo canonico
|
||||
(GATEWAY_VERSION 2.11.0 / guardrail similarity-v2) e manteneva due fonti
|
||||
potenzialmente divergenti. Storia completa del codice rimossa:
|
||||
`git log -- gateway/` (ultimo commit prima della rimozione).
|
||||
|
||||
## Architettura
|
||||
|
||||
|
||||
@@ -1,23 +0,0 @@
|
||||
FROM python:3.12-slim
|
||||
|
||||
WORKDIR /app
|
||||
COPY requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
# Hash del commit Git da cui è costruita l'immagine (iniettato al build:
|
||||
# docker compose build --build-arg GIT_COMMIT=$(git rev-parse HEAD) gateway
|
||||
# o come args nel compose). Esposto da GET /v1/version e /v1/status.
|
||||
ARG GIT_COMMIT=unknown
|
||||
ENV GIT_COMMIT=$GIT_COMMIT
|
||||
|
||||
COPY . .
|
||||
|
||||
# Utente non-root con privilegi minimi (best practice container)
|
||||
RUN useradd --create-home --uid 10001 appuser
|
||||
USER appuser
|
||||
|
||||
# Pre-download del modello sparso BM25 (cache in /home/appuser/.cache/fastembed)
|
||||
RUN python -c "from fastembed import SparseTextEmbedding; SparseTextEmbedding(model_name='Qdrant/bm25')"
|
||||
|
||||
EXPOSE 8080
|
||||
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
|
||||
@@ -1,90 +0,0 @@
|
||||
# Memory Gateway — deploy
|
||||
|
||||
Componente server di **pi-qmem**: FastAPI + Qdrant 1.19 + Ollama (BGE-M3).
|
||||
Nessun LLM in scrittura: l'agente salva record deliberati e strutturati.
|
||||
|
||||
```
|
||||
pi (estensione pi-qmem) ──HTTPS/VPN──▶ Memory Gateway (FastAPI:8082) ──▶ Qdrant 1.19 (6333)
|
||||
│
|
||||
└──▶ Ollama BGE-M3 (11434, nativo host)
|
||||
```
|
||||
|
||||
## Deploy (Docker Compose)
|
||||
|
||||
```bash
|
||||
# 1. Prepara l'ambiente (vedi qmem-gateway/docker-compose.yml come riferimento)
|
||||
cp .env.example .env
|
||||
chmod 600 .env
|
||||
# genera le chiavi:
|
||||
# QDRANT_ADMIN_API_KEY=$(openssl rand -hex 32)
|
||||
# QDRANT_READ_ONLY_API_KEY=$(openssl rand -hex 32)
|
||||
# API_KEYS=$(openssl rand -hex 32) # chiave condivisa per gli agenti
|
||||
|
||||
# 2. Avvia
|
||||
docker compose up -d --build
|
||||
|
||||
# 3. Verifica
|
||||
curl http://127.0.0.1:8082/v1/status
|
||||
```
|
||||
|
||||
Requisiti: Docker + Compose v2, Ollama con modello `bge-m3` sul host
|
||||
(`ollama pull bge-m3`), porta 8082 libera sull'interfaccia VPN.
|
||||
|
||||
## API
|
||||
|
||||
| Endpoint | Descrizione |
|
||||
|---|---|
|
||||
| `POST /v1/memories` | Crea record (text, kind, agent_id, scope, **project_id obbligatorio**, source, expires_at, supersedes_id, supersede_reason). Applica il guardrail di similarità pre-scrittura |
|
||||
| `POST /v1/memories:search` | Ricerca semantica (query, kind, project_id, scope, top_k, include_superseded, min_score) |
|
||||
| `GET /v1/memories/{id}` | Recupera per UUID |
|
||||
| `DELETE /v1/memories/{id}` | Elimina per UUID |
|
||||
| `GET /v1/meta/overview` | Discovery: scope×kind, progetti, agenti, superseduti (cache 60s) |
|
||||
| `GET /v1/status` | Health + statistiche |
|
||||
|
||||
Auth: header `X-API-Key` (chiave condivisa, accesso completo). Rate limit 120 req/min per chiave. Audit log in JSON lines (docker logs).
|
||||
|
||||
## Versione del codice
|
||||
|
||||
`GET /v1/version` (pubblico) espone la versione del codice in esecuzione, inclusa l'hash del commit Git da cui è stato costruito il container:
|
||||
|
||||
```json
|
||||
{"version": "2.7.0", "git_commit": "eccb2cb...", "guardrail_version": "similarity-v1", ...}
|
||||
```
|
||||
|
||||
Anche `GET /v1/status` include `version`, `git_commit` e `guardrail_version`. L'hash è iniettato al build via `ARG GIT_COMMIT`/`ENV GIT_COMMIT` nel Dockerfile (default `unknown`). Per costruire con l'hash:
|
||||
|
||||
```bash
|
||||
docker compose build --build-arg GIT_COMMIT=$(git rev-parse HEAD) gateway
|
||||
# o nel compose: build: { context: ./gateway, args: { GIT_COMMIT: ${GIT_COMMIT:-unknown} } }
|
||||
```
|
||||
|
||||
## Guardrail di similarità (v1)
|
||||
|
||||
Enforcement deterministico FUORI dall'LLM, prima di ogni scrittura su `POST /v1/memories`:
|
||||
|
||||
1. **Strato 1 — hash esatto**: SHA-256 del testo normalizzato (`text_hash` nel payload). Se esiste un record attivo con lo stesso hash → `409 BLOCK (EXACT_DUPLICATE)`.
|
||||
2. **Strato 2 — similarità semantica top-3**: embedding BGE-M3 cosine sui record attivi (esclusi i superseded).
|
||||
- top-1 ≥ `GUARDRAIL_BLOCK_THRESHOLD` (default 0.85) → `409 BLOCK (KNOWN_SOLUTION)`
|
||||
- top-1 ≥ `GUARDRAIL_WARN_THRESHOLD` (default 0.70) → `WARN`: salva con flag `guardrail` nel payload
|
||||
- altrimenti → `ALLOW`
|
||||
|
||||
Il **supersede esplicito** (`supersedes_id`) è una correzione intenzionale: bypassa il guardrail.
|
||||
|
||||
Configurazione (env): `GUARDRAIL_ENABLED` (default true), `GUARDRAIL_BLOCK_THRESHOLD`, `GUARDRAIL_WARN_THRESHOLD`. Soglie di partenza da calibrare sul corpus reale.
|
||||
|
||||
Risposta BLOCK (409):
|
||||
```json
|
||||
{"detail": {"error": "duplicate_memory", "reason": "KNOWN_SOLUTION", "matches": [{"memory_id": "...", "score": 0.92}], "message": "..."}}
|
||||
```
|
||||
|
||||
## Sicurezza
|
||||
|
||||
- Qdrant bindato su 127.0.0.1; gateway solo su interfaccia VPN
|
||||
- Chiavi in `.env` (0600), mai committate
|
||||
- JWT RBAC su Qdrant (admin + read-only)
|
||||
- Backup: snapshot Qdrant + rotazione 7 giorni (cron: `0 3 * * * /opt/memory/backup.sh`)
|
||||
|
||||
## Dettagli operativi
|
||||
|
||||
Procedure complete (teardown, restore, nginx, troubleshooting): vedi
|
||||
`docs/playbook.md` nel repo pi-qmem.
|
||||
@@ -1,38 +0,0 @@
|
||||
"""Audit e autenticazione del gateway."""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from fastapi import Header, HTTPException
|
||||
|
||||
import config
|
||||
import state
|
||||
|
||||
|
||||
def require_auth(x_api_key: str = Header(...)) -> str:
|
||||
if x_api_key not in config.API_KEYS:
|
||||
raise HTTPException(status_code=401, detail="API key non valida")
|
||||
now = time.monotonic()
|
||||
window = state.ratelimit.setdefault(x_api_key, [])
|
||||
window[:] = [t for t in window if now - t < 60]
|
||||
if len(window) >= config.RATE_LIMIT_PER_MIN:
|
||||
raise HTTPException(status_code=429, detail="Rate limit superato")
|
||||
window.append(now)
|
||||
return x_api_key
|
||||
|
||||
|
||||
def audit(key: str, action: str, **extra: Any) -> None:
|
||||
entry = {
|
||||
"ts": datetime.now(timezone.utc).isoformat(),
|
||||
"key": key[:8] + "...",
|
||||
"action": action,
|
||||
"request_id": state.request_id.get(),
|
||||
**extra,
|
||||
}
|
||||
config.log.info(__import__("json").dumps(entry, default=str))
|
||||
|
||||
|
||||
def now_iso() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
@@ -1,28 +0,0 @@
|
||||
"""Pulizia periodica dei record scaduti."""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any, Callable
|
||||
|
||||
from qdrant_client.http import models as qm
|
||||
|
||||
from config import log
|
||||
|
||||
|
||||
async def loop(qdrant: Any, collection: str, invalidate_meta: Callable[[], None]) -> None:
|
||||
while True:
|
||||
try:
|
||||
scroll = qdrant.scroll(
|
||||
collection_name=collection,
|
||||
scroll_filter=qm.Filter(must=[qm.FieldCondition(key="expires_at", range=qm.Range(lt=time.time()))]),
|
||||
limit=100,
|
||||
with_payload=False,
|
||||
)
|
||||
ids = [point.id for point in scroll[0]]
|
||||
if ids:
|
||||
qdrant.delete(collection_name=collection, points_selector=ids)
|
||||
invalidate_meta()
|
||||
log.info("cleanup: rimossi %d record scaduti", len(ids))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning("cleanup error: %s", exc)
|
||||
await __import__("asyncio").sleep(3600)
|
||||
@@ -1,78 +0,0 @@
|
||||
"""Configurazione statica del Memory Gateway letta dall'ambiente."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from collections import Counter
|
||||
from typing import Any
|
||||
|
||||
QDRANT_URL = os.environ.get("QDRANT_URL", "http://127.0.0.1:6333")
|
||||
QDRANT_API_KEY = os.environ.get("QDRANT_API_KEY", "")
|
||||
EMBED_API = os.environ.get("EMBED_API", "ollama")
|
||||
EMBED_URL = os.environ.get("EMBED_URL", os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434"))
|
||||
EMBED_MODEL = os.environ.get("EMBED_MODEL", "bge-m3")
|
||||
EMBED_API_KEY = os.environ.get("EMBED_API_KEY", "")
|
||||
EMBED_DIM = int(os.environ.get("EMBED_DIM", "1024"))
|
||||
# Catena di fallback per gli embedding (JSON, formato RERANK_CHAIN + campo "api").
|
||||
# Vuota → comportamento legacy: endpoint singolo da EMBED_API/EMBED_URL/EMBED_API_KEY.
|
||||
EMBED_CHAIN = os.environ.get("EMBED_CHAIN", "")
|
||||
EMBED_TIMEOUT_MS = int(os.environ.get("EMBED_TIMEOUT_MS", "30000"))
|
||||
EMBED_RETRY_COOLDOWN_S = int(os.environ.get("EMBED_RETRY_COOLDOWN_S", "60"))
|
||||
# Retry transiente per le chiamate Qdrant (store/search inclusi)
|
||||
QDRANT_RETRIES = int(os.environ.get("QDRANT_RETRIES", "3"))
|
||||
COLLECTION = os.environ.get("COLLECTION", "memories")
|
||||
API_KEYS: set[str] = {k.strip() for k in os.environ.get("API_KEYS", "").split(",") if k.strip()}
|
||||
RATE_LIMIT_PER_MIN = int(os.environ.get("RATE_LIMIT_PER_MIN", "120"))
|
||||
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-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.90"))
|
||||
GUARDRAIL_RERANK_SUGGEST = float(os.environ.get("GUARDRAIL_RERANK_SUGGEST", "0.85"))
|
||||
# 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.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"))
|
||||
METRICS_ENABLED = os.environ.get("METRICS_ENABLED", "true").lower() == "true"
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
||||
log = logging.getLogger("memory-gateway")
|
||||
|
||||
SPARSE_VECTOR_NAME = "bm25"
|
||||
|
||||
# Re-ranking: catena di fallback resiliente (frigate → brain locale).
|
||||
# Il default nel codice è OFF; il deploy imposta RERANK_ENABLED=true e la catena.
|
||||
RERANK_ENABLED = os.environ.get("RERANK_ENABLED", "false").lower() == "true"
|
||||
RERANK_MODEL = os.environ.get("RERANK_MODEL", "bge-reranker-v2-m3")
|
||||
RERANK_CANDIDATES = int(os.environ.get("RERANK_CANDIDATES", "16"))
|
||||
RERANK_MAX_DOC_CHARS = int(os.environ.get("RERANK_MAX_DOC_CHARS", "800"))
|
||||
RERANK_TIMEOUT_MS = int(os.environ.get("RERANK_TIMEOUT_MS", "10000"))
|
||||
RERANK_RETRY_COOLDOWN_S = int(os.environ.get("RERANK_RETRY_COOLDOWN_S", "60"))
|
||||
RERANK_CHAIN = os.environ.get("RERANK_CHAIN", "")
|
||||
_metrics: dict[str, Any] = {
|
||||
"requests": Counter(),
|
||||
"duration_sum": Counter(),
|
||||
"duration_count": Counter(),
|
||||
"errors": Counter(),
|
||||
"search_queries": 0,
|
||||
"search_hits": 0,
|
||||
"rerank_calls": Counter(),
|
||||
"rerank_duration_sum": Counter(),
|
||||
"embed_calls": Counter(),
|
||||
"embed_duration_sum": Counter(),
|
||||
"qdrant_retries": 0,
|
||||
}
|
||||
@@ -1,222 +0,0 @@
|
||||
"""Embedding denso (Ollama/llama.cpp) e sparse BM25, con catena di fallback resiliente.
|
||||
|
||||
Catena da EMBED_CHAIN (JSON, stesso formato di RERANK_CHAIN + campo "api"):
|
||||
il primo nodo raggiungibile vince, i nodi falliti entrano in cooldown. Se
|
||||
EMBED_CHAIN è vuota si usa il comportamento legacy (endpoint singolo da
|
||||
EMBED_API/EMBED_URL/EMBED_API_KEY). L'ultimo errore viene rilanciato al client
|
||||
come gli endpoint precedenti: nessuna degradazione silenziosa della scrittura.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Optional
|
||||
|
||||
import httpx
|
||||
from qdrant_client.http import models as qm
|
||||
|
||||
import metrics
|
||||
from config import (
|
||||
EMBED_API,
|
||||
EMBED_API_KEY,
|
||||
EMBED_CHAIN,
|
||||
EMBED_DIM,
|
||||
EMBED_MODEL,
|
||||
EMBED_RETRY_COOLDOWN_S,
|
||||
EMBED_TIMEOUT_MS,
|
||||
EMBED_URL,
|
||||
SPARSE_VECTOR_NAME,
|
||||
log,
|
||||
)
|
||||
|
||||
try:
|
||||
from fastembed import SparseTextEmbedding
|
||||
_sparse_model: Optional[Any] = None
|
||||
SPARSE_AVAILABLE = True
|
||||
except Exception: # noqa: BLE001
|
||||
_sparse_model = None
|
||||
SPARSE_AVAILABLE = False
|
||||
log.warning("fastembed non disponibile: hybrid retrieval disattivato")
|
||||
|
||||
_http: Optional[httpx.AsyncClient] = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EmbedNode:
|
||||
"""Un endpoint embedding nella catena di fallback."""
|
||||
|
||||
name: str
|
||||
url: str
|
||||
api: str # "llamacpp" (/v1/embeddings) | "ollama" (/api/embed)
|
||||
key: str
|
||||
timeout_ms: int
|
||||
|
||||
|
||||
def parse_chain(raw: str, default_api: str, default_url: str, default_key: str) -> list[EmbedNode]:
|
||||
"""Parsa EMBED_CHAIN (JSON); vuota o invalida → endpoint legacy singolo."""
|
||||
nodes: list[EmbedNode] = []
|
||||
if raw:
|
||||
try:
|
||||
entries = json.loads(raw)
|
||||
for entry in entries if isinstance(entries, list) else []:
|
||||
if not isinstance(entry, dict) or not entry.get("url"):
|
||||
continue
|
||||
url = str(entry["url"]).rstrip("/")
|
||||
api = str(entry.get("api") or "llamacpp")
|
||||
if api not in ("llamacpp", "ollama") or not url.startswith(("http://", "https://")):
|
||||
continue
|
||||
nodes.append(
|
||||
EmbedNode(
|
||||
name=str(entry.get("name") or url),
|
||||
url=url,
|
||||
api=api,
|
||||
key=str(entry.get("key") or ""),
|
||||
timeout_ms=int(entry.get("timeout_ms", EMBED_TIMEOUT_MS)),
|
||||
)
|
||||
)
|
||||
except (json.JSONDecodeError, TypeError, ValueError):
|
||||
log.error("EMBED_CHAIN non è JSON valido: uso l'endpoint legacy")
|
||||
if not nodes and default_url:
|
||||
# Compatibilità legacy: endpoint singolo dagli env EMBED_*
|
||||
nodes = [EmbedNode(name="embed", url=default_url.rstrip("/"), api=default_api, key=default_key, timeout_ms=EMBED_TIMEOUT_MS)]
|
||||
return nodes
|
||||
|
||||
|
||||
_chain: Optional[list[EmbedNode]] = None
|
||||
_down_until: dict[str, float] = {} # url → monotonic deadline del cooldown
|
||||
_http: Optional[httpx.AsyncClient] = None
|
||||
|
||||
|
||||
def _get_chain() -> list[EmbedNode]:
|
||||
global _chain
|
||||
if _chain is None:
|
||||
_chain = parse_chain(EMBED_CHAIN, EMBED_API, EMBED_URL, EMBED_API_KEY)
|
||||
return _chain
|
||||
|
||||
|
||||
def reset_chain_cache() -> None:
|
||||
"""Forza il re-parse della catena (usato dai test)."""
|
||||
global _chain
|
||||
_chain = None
|
||||
_down_until.clear()
|
||||
|
||||
|
||||
def get_http() -> httpx.AsyncClient:
|
||||
global _http
|
||||
if _http is None:
|
||||
_http = httpx.AsyncClient(timeout=30)
|
||||
return _http
|
||||
|
||||
|
||||
async def close_http() -> None:
|
||||
global _http
|
||||
if _http is not None:
|
||||
await _http.aclose()
|
||||
_http = None
|
||||
|
||||
|
||||
def chain_nodes() -> list[EmbedNode]:
|
||||
return _get_chain()
|
||||
|
||||
|
||||
async def embed(text: str) -> list[float]:
|
||||
"""Embedding con catena di fallback: ritorna il vettore o rilancia dopo l'ultimo fallimento."""
|
||||
chain = _get_chain()
|
||||
if not chain:
|
||||
raise RuntimeError("nessun endpoint embedding configurato")
|
||||
now = time.monotonic()
|
||||
live = [n for n in chain if _down_until.get(n.url, 0) <= now]
|
||||
if not live:
|
||||
# tutti in cooldown: ritenta comunque il primo (meglio di un fallimento immediato)
|
||||
live = [chain[0]]
|
||||
payload = {"model": EMBED_MODEL, "input": text}
|
||||
started = time.monotonic()
|
||||
last_exc: Optional[Exception] = None
|
||||
for node in live:
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if node.key:
|
||||
headers["Authorization"] = f"Bearer {node.key}"
|
||||
path = "/v1/embeddings" if node.api == "llamacpp" else "/api/embed"
|
||||
try:
|
||||
t0 = time.monotonic()
|
||||
response = await get_http().post(
|
||||
f"{node.url}{path}",
|
||||
json=payload,
|
||||
headers=headers,
|
||||
timeout=httpx.Timeout(node.timeout_ms / 1000.0),
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
vector = data["data"][0]["embedding"] if node.api == "llamacpp" else data["embeddings"][0]
|
||||
if len(vector) != EMBED_DIM:
|
||||
raise ValueError(f"dimensione vettore {len(vector)} != EMBED_DIM {EMBED_DIM}")
|
||||
took = int((time.monotonic() - started) * 1000)
|
||||
metrics.record_embed(node.name, True, took)
|
||||
return vector
|
||||
except (httpx.HTTPError, ValueError, KeyError, IndexError, TypeError) as exc:
|
||||
took = int((time.monotonic() - t0) * 1000)
|
||||
_down_until[node.url] = time.monotonic() + EMBED_RETRY_COOLDOWN_S
|
||||
last_exc = exc
|
||||
metrics.record_embed(node.name, False, took)
|
||||
log.warning(
|
||||
"embed: nodo '%s' fallito dopo %dms (%s: %s) → cooldown %ds",
|
||||
node.name,
|
||||
took,
|
||||
exc.__class__.__name__,
|
||||
exc,
|
||||
EMBED_RETRY_COOLDOWN_S,
|
||||
)
|
||||
raise RuntimeError(f"tutti i nodi embedding falliti ({len(live)} tentativi)") from last_exc
|
||||
|
||||
|
||||
def get_sparse_model():
|
||||
global _sparse_model
|
||||
if _sparse_model is None and SPARSE_AVAILABLE:
|
||||
_sparse_model = SparseTextEmbedding(model_name="Qdrant/bm25")
|
||||
return _sparse_model
|
||||
|
||||
|
||||
def sparse_encode(text: str) -> Optional[qm.SparseVector]:
|
||||
model = get_sparse_model()
|
||||
if model is None:
|
||||
return None
|
||||
emb = next(model.embed(text))
|
||||
return qm.SparseVector(indices=emb.indices.tolist(), values=emb.values.tolist())
|
||||
|
||||
|
||||
def backfill_sparse(qdrant: Any, collection: str) -> None:
|
||||
if not SPARSE_AVAILABLE:
|
||||
return
|
||||
offset: Any = None
|
||||
updated = 0
|
||||
while True:
|
||||
points, next_offset = qdrant.scroll(
|
||||
collection_name=collection,
|
||||
limit=100,
|
||||
with_payload=["text"],
|
||||
with_vectors=True,
|
||||
offset=offset,
|
||||
)
|
||||
batch: list[qm.PointStruct] = []
|
||||
for p in points:
|
||||
vecs = p.vector or {}
|
||||
if SPARSE_VECTOR_NAME in vecs:
|
||||
continue
|
||||
text = (p.payload or {}).get("text", "")
|
||||
if not text:
|
||||
continue
|
||||
sparse = sparse_encode(text)
|
||||
if sparse is None:
|
||||
continue
|
||||
batch.append(qm.PointStruct(id=p.id, vector={SPARSE_VECTOR_NAME: sparse}))
|
||||
if batch:
|
||||
# update_vectors: aggiorna SOLO il vettore sparso, preservando payload e vettore denso
|
||||
# (upsert parziale sostituirebbe l'intero punto — incidente 2026-08-16)
|
||||
qdrant.update_vectors(collection_name=collection, points=batch)
|
||||
updated += len(batch)
|
||||
if not next_offset:
|
||||
break
|
||||
offset = next_offset
|
||||
if updated:
|
||||
log.info("backfill sparse: %d record aggiornati", updated)
|
||||
@@ -1,135 +0,0 @@
|
||||
"""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 = 5) -> 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
|
||||
-111
@@ -1,111 +0,0 @@
|
||||
"""Memory Gateway — bootstrap FastAPI, lifecycle e middleware.
|
||||
|
||||
Gli endpoint e la logica di dominio sono separati in moduli:
|
||||
config, models, state, audit, guardrail, embed, store, metrics, cleanup,
|
||||
routes. Il contratto HTTP resta invariato.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import uuid
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
import uvicorn
|
||||
from fastapi import FastAPI, Request
|
||||
from qdrant_client.http import models as qm
|
||||
|
||||
import cleanup
|
||||
import embed as embedding
|
||||
import metrics
|
||||
import rerank
|
||||
import state
|
||||
from config import (
|
||||
COLLECTION,
|
||||
EMBED_DIM,
|
||||
METRICS_ENABLED,
|
||||
SPARSE_VECTOR_NAME,
|
||||
GATEWAY_VERSION,
|
||||
log,
|
||||
)
|
||||
from routes import router
|
||||
|
||||
# Alias utili per compatibilità con import/debug locali; lo stato effettivo è in state.py.
|
||||
qdrant = state.qdrant
|
||||
embed = embedding.embed
|
||||
state.embed = embedding.embed
|
||||
state.sparse_encode = embedding.sparse_encode
|
||||
|
||||
|
||||
async def _lifespan(_app: FastAPI):
|
||||
"""Crea collection/indici e avvia i loop periodici."""
|
||||
collections = state.qdrant.get_collections().collections
|
||||
if not any(c.name == COLLECTION for c in collections):
|
||||
state.qdrant.create_collection(
|
||||
collection_name=COLLECTION,
|
||||
vectors_config=qm.VectorParams(size=EMBED_DIM, distance=qm.Distance.COSINE),
|
||||
sparse_vectors_config={SPARSE_VECTOR_NAME: qm.SparseVectorParams(modifier=qm.Modifier.IDF)},
|
||||
)
|
||||
for field in ("agent_id", "project_id", "scope", "kind", "supersedes_id", "superseded_by", "text_hash", "parent_id", "level", "topic"):
|
||||
state.qdrant.create_payload_index(collection_name=COLLECTION, field_name=field, field_schema=qm.PayloadSchemaType.KEYWORD)
|
||||
state.qdrant.create_payload_index(collection_name=COLLECTION, field_name="text", field_schema=qm.PayloadSchemaType.TEXT)
|
||||
log.info("collection %s creata con indici (dense + sparse %s)", COLLECTION, SPARSE_VECTOR_NAME)
|
||||
else:
|
||||
log.info("collection %s già esistente", COLLECTION)
|
||||
for field in ("parent_id", "level", "topic"):
|
||||
try:
|
||||
state.qdrant.create_payload_index(collection_name=COLLECTION, field_name=field, field_schema=qm.PayloadSchemaType.KEYWORD)
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
info = state.qdrant.get_collection(COLLECTION)
|
||||
sparse_vectors = (info.config.params.sparse_vectors or {}) if info.config and info.config.params else {}
|
||||
if SPARSE_VECTOR_NAME not in sparse_vectors:
|
||||
state.qdrant.create_vector_name(COLLECTION, SPARSE_VECTOR_NAME, qm.SparseVectorNameConfig(sparse=qm.SparseVectorConfig(modifier=qm.Modifier.IDF)))
|
||||
log.info("sparse vector %s aggiunto alla collection esistente", SPARSE_VECTOR_NAME)
|
||||
embedding.backfill_sparse(state.qdrant, COLLECTION)
|
||||
|
||||
cleanup_task = asyncio.create_task(cleanup.loop(state.qdrant, COLLECTION, state.invalidate_meta))
|
||||
metrics_task = asyncio.create_task(metrics.push_loop(state.qdrant, COLLECTION, embedding.get_http)) if METRICS_ENABLED else None
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
cleanup_task.cancel()
|
||||
try:
|
||||
await cleanup_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
if metrics_task is not None:
|
||||
metrics_task.cancel()
|
||||
try:
|
||||
await metrics_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
await embedding.close_http()
|
||||
await rerank.close_http()
|
||||
|
||||
|
||||
app = FastAPI(title="Memory Gateway", version=GATEWAY_VERSION, lifespan=_lifespan)
|
||||
app.include_router(router)
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def request_id_middleware(request: Request, call_next):
|
||||
rid = request.headers.get("X-Request-ID") or str(uuid.uuid4())
|
||||
state.request_id.set(rid)
|
||||
response = await call_next(request)
|
||||
response.headers["X-Request-ID"] = rid
|
||||
return response
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def metrics_middleware(request: Request, call_next):
|
||||
start = time.monotonic()
|
||||
response = await call_next(request)
|
||||
route = request.scope.get("route")
|
||||
endpoint = route.path if route else request.url.path
|
||||
metrics.record_request(endpoint, time.monotonic() - start, response.status_code)
|
||||
return response
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=8080)
|
||||
@@ -1,94 +0,0 @@
|
||||
"""Metriche in-memory e push Prometheus/VictoriaMetrics."""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from config import VM_PUSH_INTERVAL, VM_PUSH_URL, _metrics, log
|
||||
|
||||
|
||||
def record_request(endpoint: str, duration: float, status_code: int) -> None:
|
||||
_metrics["requests"][endpoint] += 1
|
||||
_metrics["duration_sum"][endpoint] += duration
|
||||
_metrics["duration_count"][endpoint] += 1
|
||||
if status_code >= 400:
|
||||
_metrics["errors"][(endpoint, status_code)] += 1
|
||||
|
||||
|
||||
def record_search(hits: int) -> None:
|
||||
_metrics["search_queries"] += 1
|
||||
_metrics["search_hits"] += hits
|
||||
|
||||
|
||||
def record_rerank(backend: str, ok: bool, took_ms: int) -> None:
|
||||
_metrics["rerank_calls"][(backend, "ok" if ok else "fail")] += 1
|
||||
_metrics["rerank_duration_sum"][backend] += took_ms
|
||||
|
||||
|
||||
def record_embed(backend: str, ok: bool, took_ms: int) -> None:
|
||||
_metrics["embed_calls"][(backend, "ok" if ok else "fail")] += 1
|
||||
_metrics["embed_duration_sum"][backend] += took_ms
|
||||
|
||||
|
||||
def snapshot(qdrant: Any, collection: str) -> dict:
|
||||
try:
|
||||
points = qdrant.get_collection(collection).points_count
|
||||
except Exception: # noqa: BLE001
|
||||
points = None
|
||||
return {
|
||||
"requests": dict(_metrics["requests"]),
|
||||
"avg_duration_ms": {
|
||||
endpoint: round(_metrics["duration_sum"][endpoint] / _metrics["duration_count"][endpoint] * 1000, 2)
|
||||
for endpoint in _metrics["duration_count"]
|
||||
},
|
||||
"errors": {f"{endpoint}:{status}": count for (endpoint, status), count in _metrics["errors"].items()},
|
||||
"search_queries": _metrics["search_queries"],
|
||||
"search_hits": _metrics["search_hits"],
|
||||
"rerank_calls": {f"{backend}:{outcome}": count for (backend, outcome), count in _metrics["rerank_calls"].items()},
|
||||
"rerank_avg_ms": {backend: round(total / _metrics["rerank_calls"][(backend, "ok")], 2) for backend, total in _metrics["rerank_duration_sum"].items() if _metrics["rerank_calls"][(backend, "ok")]},
|
||||
"embed_calls": {f"{backend}:{outcome}": count for (backend, outcome), count in _metrics["embed_calls"].items()},
|
||||
"embed_avg_ms": {backend: round(total / _metrics["embed_calls"][(backend, "ok")], 2) for backend, total in _metrics["embed_duration_sum"].items() if _metrics["embed_calls"][(backend, "ok")]},
|
||||
"qdrant_retries": _metrics["qdrant_retries"],
|
||||
"points": points,
|
||||
}
|
||||
|
||||
|
||||
def prometheus_lines(qdrant: Any, collection: str) -> list[str]:
|
||||
lines: list[str] = []
|
||||
for endpoint, count in _metrics["requests"].items():
|
||||
lines.append(f'qmem_requests_total{{endpoint="{endpoint}"}} {count}')
|
||||
for endpoint, total in _metrics["duration_sum"].items():
|
||||
count = _metrics["duration_count"][endpoint]
|
||||
lines.append(f'qmem_request_duration_seconds_sum{{endpoint="{endpoint}"}} {total:.6f}')
|
||||
lines.append(f'qmem_request_duration_seconds_count{{endpoint="{endpoint}"}} {count}')
|
||||
for (endpoint, status), count in _metrics["errors"].items():
|
||||
lines.append(f'qmem_errors_total{{endpoint="{endpoint}",status="{status}"}} {count}')
|
||||
lines.append(f"qmem_search_queries_total {_metrics['search_queries']}")
|
||||
lines.append(f"qmem_search_hits_total {_metrics['search_hits']}")
|
||||
for (backend, outcome), count in _metrics["rerank_calls"].items():
|
||||
lines.append(f'qmem_rerank_calls_total{{backend="{backend}",outcome="{outcome}"}} {count}')
|
||||
for backend, s in _metrics["rerank_duration_sum"].items():
|
||||
lines.append(f'qmem_rerank_duration_seconds_sum{{backend="{backend}"}} {s / 1000:.6f}')
|
||||
for (backend, outcome), count in _metrics["embed_calls"].items():
|
||||
lines.append(f'qmem_embed_calls_total{{backend="{backend}",outcome="{outcome}"}} {count}')
|
||||
for backend, s in _metrics["embed_duration_sum"].items():
|
||||
lines.append(f'qmem_embed_duration_seconds_sum{{backend="{backend}"}} {s / 1000:.6f}')
|
||||
lines.append(f"qmem_qdrant_retries_total {_metrics['qdrant_retries']}")
|
||||
try:
|
||||
lines.append(f"qmem_points {qdrant.get_collection(collection).points_count}")
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
return lines
|
||||
|
||||
|
||||
async def push_loop(qdrant: Any, collection: str, get_http) -> None:
|
||||
while True:
|
||||
try:
|
||||
now_ms = int(time.time() * 1000)
|
||||
body = "\n".join(f"{line} {now_ms}" for line in prometheus_lines(qdrant, collection)) + "\n"
|
||||
response = await get_http().post(VM_PUSH_URL, content=body, headers={"Content-Type": "text/plain"})
|
||||
if response.status_code >= 300:
|
||||
log.warning("metrics push: HTTP %s", response.status_code)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning("metrics push error: %s", exc)
|
||||
await __import__("asyncio").sleep(VM_PUSH_INTERVAL)
|
||||
@@ -1,80 +0,0 @@
|
||||
"""Schemi Pydantic del Memory Gateway."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Literal, Optional
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
from config import MAX_TEXT_LEN
|
||||
|
||||
|
||||
class MemoryLink(BaseModel):
|
||||
target_id: str = Field(..., description="UUID del record target collegato")
|
||||
predicate: str = Field(default="part_of", max_length=64, description="Tipo di relazione: parent_of, part_of, relates_to, supersedes...")
|
||||
weight: float = Field(default=1.0, ge=0.0, le=1.0)
|
||||
|
||||
|
||||
class MemoryIn(BaseModel):
|
||||
text: str = Field(min_length=1, max_length=MAX_TEXT_LEN)
|
||||
kind: Literal["decision", "fact", "episode", "preference"] = "fact"
|
||||
agent_id: Optional[str] = Field(default=None, max_length=64, description="Solo provenienza, nessun isolamento")
|
||||
project_id: str = Field(min_length=1, max_length=64, description="OBBLIGATORIO: progetto/dominio di appartenenza (kebab-case)")
|
||||
scope: Literal["agent", "project", "org"] = "agent"
|
||||
source: Optional[str] = Field(default=None, max_length=256)
|
||||
confidence: Literal["high", "medium", "low"] = Field(default="medium", description="Affidabilità del record")
|
||||
expires_at: Optional[str] = None
|
||||
supersedes_id: Optional[str] = None
|
||||
supersede_reason: Optional[str] = Field(default=None, max_length=512)
|
||||
parent_id: Optional[str] = Field(default=None, description="UUID del record genitore per gerarchia/subtopic")
|
||||
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")
|
||||
@classmethod
|
||||
def _validate_expires_at(cls, v: Optional[str]) -> Optional[str]:
|
||||
if v is None:
|
||||
return v
|
||||
try:
|
||||
datetime.fromisoformat(v.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
raise ValueError("expires_at deve essere una data ISO 8601 valida (es. 2026-09-01T00:00:00Z)")
|
||||
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
|
||||
project_id: Optional[str] = None
|
||||
scope: Optional[Literal["agent", "project", "org"]] = None
|
||||
include_superseded: bool = False
|
||||
min_score: Optional[float] = Field(default=None, ge=0.0, le=1.0)
|
||||
top_k: int = Field(default=5, ge=1, le=20)
|
||||
hybrid: bool = False
|
||||
parent_id: Optional[str] = None
|
||||
level: Optional[Literal["L1_ROOT", "L2_SUBTOPIC", "L3_DETAIL"]] = None
|
||||
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")
|
||||
@@ -1,2 +0,0 @@
|
||||
# Dipendenze di sviluppo (test): installare con pip install -r requirements-dev.txt
|
||||
pytest==8.3.4
|
||||
@@ -1,6 +0,0 @@
|
||||
fastapi==0.115.6
|
||||
uvicorn[standard]==0.34.0
|
||||
qdrant-client==1.19.0
|
||||
httpx==0.28.1
|
||||
pydantic==2.10.4
|
||||
fastembed==0.5.1
|
||||
@@ -1,179 +0,0 @@
|
||||
"""Stadio di re-ranking (cross-encoder) con catena di fallback resiliente.
|
||||
|
||||
La catena è definita da RERANK_CHAIN (JSON): il primo nodo raggiungibile vince.
|
||||
Dopo un fallimento il nodo entra in cooldown (RERANK_RETRY_COOLDOWN_S) e la
|
||||
richiesta passa al successivo; se tutti i nodi sono in cooldown si ritenta
|
||||
comunque il primo (meglio di un fallimento immediato). Se nessun nodo risponde
|
||||
la ricerca degrada con grazia all'ordine di fusione ibrida (nessun errore al
|
||||
client): il reranking è un miglioramento, non una dipendenza.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
import httpx
|
||||
|
||||
import metrics
|
||||
from config import (
|
||||
RERANK_CHAIN,
|
||||
RERANK_ENABLED,
|
||||
RERANK_MAX_DOC_CHARS,
|
||||
RERANK_MODEL,
|
||||
RERANK_RETRY_COOLDOWN_S,
|
||||
RERANK_TIMEOUT_MS,
|
||||
log,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RerankNode:
|
||||
"""Un endpoint reranker nella catena di fallback."""
|
||||
|
||||
name: str
|
||||
url: str
|
||||
key: str
|
||||
timeout_ms: int
|
||||
|
||||
|
||||
def parse_chain(raw: str) -> list[RerankNode]:
|
||||
"""Parsa RERANK_CHAIN: JSON [{name, url, key, timeout_ms}]. URL senza schema → scartato."""
|
||||
try:
|
||||
entries = json.loads(raw) if raw else []
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
log.error("RERANK_CHAIN non è JSON valido: reranking disattivato")
|
||||
return []
|
||||
if not isinstance(entries, list):
|
||||
log.error("RERANK_CHAIN non è una lista: reranking disattivato")
|
||||
return []
|
||||
nodes: list[RerankNode] = []
|
||||
for entry in entries:
|
||||
if not isinstance(entry, dict) or not entry.get("url"):
|
||||
continue
|
||||
url = str(entry["url"]).rstrip("/")
|
||||
if not url.startswith(("http://", "https://")):
|
||||
continue
|
||||
nodes.append(
|
||||
RerankNode(
|
||||
name=str(entry.get("name") or url),
|
||||
url=url,
|
||||
key=str(entry.get("key") or ""),
|
||||
timeout_ms=int(entry.get("timeout_ms", RERANK_TIMEOUT_MS)),
|
||||
)
|
||||
)
|
||||
return nodes
|
||||
|
||||
|
||||
_chain: Optional[list[RerankNode]] = None
|
||||
_down_until: dict[str, float] = {} # url → monotonic deadline del cooldown
|
||||
_http: Optional[httpx.AsyncClient] = None
|
||||
|
||||
|
||||
def _get_chain() -> list[RerankNode]:
|
||||
global _chain
|
||||
if _chain is None:
|
||||
_chain = parse_chain(RERANK_CHAIN)
|
||||
return _chain
|
||||
|
||||
|
||||
def reset_chain_cache() -> None:
|
||||
"""Forza il re-parse della catena (usato dai test)."""
|
||||
global _chain
|
||||
_chain = None
|
||||
_down_until.clear()
|
||||
|
||||
|
||||
def get_http() -> httpx.AsyncClient:
|
||||
global _http
|
||||
if _http is None:
|
||||
_http = httpx.AsyncClient(timeout=30)
|
||||
return _http
|
||||
|
||||
|
||||
async def close_http() -> None:
|
||||
global _http
|
||||
if _http is not None:
|
||||
await _http.aclose()
|
||||
_http = None
|
||||
|
||||
|
||||
def enabled() -> bool:
|
||||
"""Reranking attivo: flag env + catena configurata non vuota."""
|
||||
return RERANK_ENABLED and bool(_get_chain())
|
||||
|
||||
|
||||
def live_nodes() -> tuple[list[RerankNode], bool]:
|
||||
"""Nodi fuori cooldown; all_down=True se nessun nodo è live (forza retry totale)."""
|
||||
chain = _get_chain()
|
||||
now = time.monotonic()
|
||||
live = [n for n in chain if _down_until.get(n.url, 0) <= now]
|
||||
return live, bool(chain) and len(live) < len(chain)
|
||||
|
||||
|
||||
async def rerank(query: str, docs: list[str]) -> Optional[tuple[list[float], str, int]]:
|
||||
"""Reranka i documenti rispetto alla query tramite la catena di fallback.
|
||||
|
||||
Ritorna (scores, backend_name, took_ms) dove scores è allineato a docs
|
||||
(logit sigmoide in [0,1]), oppure None se tutti i nodi falliscono.
|
||||
"""
|
||||
chain = _get_chain()
|
||||
if not chain or not docs:
|
||||
return None
|
||||
# Troncamento dei documenti: limita il costo di inferenza (i cross-encoder
|
||||
# scala con la lunghezza della coppia query+doc) e evita input oltre il ctx.
|
||||
docs = [d[:RERANK_MAX_DOC_CHARS] for d in docs]
|
||||
live, all_down = live_nodes()
|
||||
if not live:
|
||||
# tutti in cooldown: ritenta comunque il primo (meglio di un fallimento immediato)
|
||||
live = [chain[0]]
|
||||
payload = {"model": RERANK_MODEL, "query": query, "documents": docs, "top_n": len(docs)}
|
||||
started = time.monotonic()
|
||||
for node in live:
|
||||
headers = {"Content-Type": "application/json"}
|
||||
if node.key:
|
||||
headers["Authorization"] = f"Bearer {node.key}"
|
||||
try:
|
||||
t0 = time.monotonic()
|
||||
response = await get_http().post(
|
||||
f"{node.url}/v1/rerank",
|
||||
json=payload,
|
||||
headers=headers,
|
||||
timeout=httpx.Timeout(node.timeout_ms / 1000.0),
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
# Il risultato è [{index, relevance_score}] ordinato per rilevanza:
|
||||
# riportiamo ogni score alla posizione originaria del documento.
|
||||
scores = [0.0] * len(docs)
|
||||
for item in data.get("results", []):
|
||||
idx = int(item["index"])
|
||||
if 0 <= idx < len(docs):
|
||||
scores[idx] = float(item.get("relevance_score", 0.0))
|
||||
took = int((time.monotonic() - started) * 1000)
|
||||
metrics.record_rerank(node.name, True, took)
|
||||
return scores, node.name, took
|
||||
except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc:
|
||||
took = int((time.monotonic() - t0) * 1000)
|
||||
_down_until[node.url] = time.monotonic() + RERANK_RETRY_COOLDOWN_S
|
||||
metrics.record_rerank(node.name, False, took)
|
||||
log.warning(
|
||||
"rerank: nodo '%s' fallito dopo %dms (%s: %s) → cooldown %ds",
|
||||
node.name,
|
||||
took,
|
||||
exc.__class__.__name__,
|
||||
exc,
|
||||
RERANK_RETRY_COOLDOWN_S,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def normalize_score(logit: float) -> float:
|
||||
"""Sigmoide: logit di rilevanza → punteggio [0,1] leggibile nei risultati."""
|
||||
if logit >= 0:
|
||||
z = math.exp(-logit)
|
||||
return 1.0 / (1.0 + z)
|
||||
z = math.exp(logit)
|
||||
return z / (1.0 + z)
|
||||
@@ -1,333 +0,0 @@
|
||||
"""Endpoint HTTP del Memory Gateway."""
|
||||
from __future__ import annotations
|
||||
|
||||
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
|
||||
from qdrant_client.http import models as qm
|
||||
|
||||
import config
|
||||
import embed
|
||||
import guardrail
|
||||
import metrics
|
||||
import rerank
|
||||
import state
|
||||
import store
|
||||
from audit import audit, now_iso, require_auth
|
||||
from config import (
|
||||
API_KEYS,
|
||||
COLLECTION,
|
||||
EMBED_DIM,
|
||||
EMBED_MODEL,
|
||||
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, ScoreIn, SearchIn
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.post("/v1/memories")
|
||||
async def add_memory(
|
||||
body: MemoryIn,
|
||||
key: str = Depends(require_auth),
|
||||
idempotency_key: Optional[str] = Header(default=None, alias="Idempotency-Key"),
|
||||
) -> dict:
|
||||
idem_key = f"{key}:{idempotency_key}" if idempotency_key else None
|
||||
if idem_key:
|
||||
state.idempotency_cleanup()
|
||||
existing = state.idempotency.get(idem_key)
|
||||
if existing:
|
||||
if existing["hash"] != state.payload_hash(body):
|
||||
raise HTTPException(status_code=409, detail="Idempotency-Key già usata con payload diverso")
|
||||
audit(key, "create_replay", idempotency_key=idempotency_key[:16])
|
||||
return existing["response"]
|
||||
|
||||
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:
|
||||
raise HTTPException(status_code=404, detail="Memoria da supersedere non trovata")
|
||||
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 = 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(
|
||||
status_code=409,
|
||||
detail={
|
||||
"error": "duplicate_memory",
|
||||
"reason": similarity_guardrail["reason"],
|
||||
"matches": similarity_guardrail["matches"],
|
||||
"message": "Memoria già presente o quasi identica: usa supersedes_id per correggere la versione attiva, oppure riformula il contenuto.",
|
||||
},
|
||||
)
|
||||
|
||||
payload: dict[str, Any] = {
|
||||
"text": body.text,
|
||||
"kind": body.kind,
|
||||
"agent_id": body.agent_id or "shared",
|
||||
"project_id": body.project_id,
|
||||
"scope": body.scope,
|
||||
"source": body.source,
|
||||
"confidence": body.confidence,
|
||||
"private": body.private,
|
||||
"created_at": now_iso(),
|
||||
"expires_at": guardrail.parse_ts(body.expires_at),
|
||||
"supersedes_id": superseded_id,
|
||||
"supersede_reason": body.supersede_reason,
|
||||
"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),
|
||||
}
|
||||
if similarity_guardrail:
|
||||
payload["guardrail"] = {
|
||||
"version": GUARDRAIL_VERSION,
|
||||
"decision": similarity_guardrail["decision"],
|
||||
"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
|
||||
state.qdrant.upsert(collection_name=COLLECTION, points=[qm.PointStruct(id=memory_id, vector=point_vector, payload=payload)])
|
||||
state.invalidate_meta()
|
||||
|
||||
reparented_count = 0
|
||||
if superseded_id:
|
||||
state.qdrant.set_payload(
|
||||
collection_name=COLLECTION,
|
||||
payload={"superseded_by": memory_id, "superseded_at": now_iso(), "supersede_reason": body.supersede_reason},
|
||||
points=[superseded_id],
|
||||
)
|
||||
audit(key, "supersede", old_id=superseded_id, new_id=memory_id, kind=body.kind, agent_id=payload["agent_id"])
|
||||
reparented_count = store.reparent_active_children(state.qdrant, COLLECTION, superseded_id, memory_id)
|
||||
if reparented_count:
|
||||
audit(key, "reparent", old_id=superseded_id, new_id=memory_id, count=reparented_count)
|
||||
else:
|
||||
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
|
||||
|
||||
# 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)
|
||||
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)"
|
||||
elif use_rerank:
|
||||
rerank_info["reason"] = "candidati insufficienti"
|
||||
|
||||
results = results[: body.top_k]
|
||||
audit(
|
||||
key,
|
||||
"search",
|
||||
query_hash=hashlib.sha256(body.query.encode()).hexdigest()[:16],
|
||||
top_k=body.top_k,
|
||||
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:
|
||||
try:
|
||||
point = state.qdrant.retrieve(collection_name=COLLECTION, ids=[memory_id], with_payload=True)
|
||||
except Exception: # noqa: BLE001 — id non-UUID o payload malformato → non trovato, non 500
|
||||
point = []
|
||||
if not point:
|
||||
raise HTTPException(status_code=404, detail="Memoria non trovata")
|
||||
audit(key, "get", memory_id=memory_id)
|
||||
return {"memory_id": memory_id, **point[0].payload}
|
||||
|
||||
|
||||
@router.delete("/v1/memories/{memory_id}")
|
||||
async def delete_memory(memory_id: str, key: str = Depends(require_auth)) -> dict:
|
||||
point = state.qdrant.retrieve(collection_name=COLLECTION, ids=[memory_id], with_payload=True)
|
||||
if not point:
|
||||
raise HTTPException(status_code=404, detail="Memoria non trovata")
|
||||
state.qdrant.delete(collection_name=COLLECTION, points_selector=[memory_id])
|
||||
state.invalidate_meta()
|
||||
audit(key, "delete", memory_id=memory_id)
|
||||
return {"deleted": memory_id}
|
||||
|
||||
|
||||
@router.get("/v1/meta/overview")
|
||||
async def meta_overview(key: str = Depends(require_auth)) -> dict:
|
||||
now = time.time()
|
||||
cached = state.meta_cache.get("overview")
|
||||
if cached and now - cached["ts"] < 60:
|
||||
audit(key, "meta", cached=True)
|
||||
return {**cached["data"], "cached": True}
|
||||
|
||||
scope_kinds: dict[str, Counter] = {}
|
||||
projects: Counter = Counter()
|
||||
agents: Counter = Counter()
|
||||
total = 0
|
||||
superseded = 0
|
||||
offset: Any = None
|
||||
while True:
|
||||
points, next_offset = state.qdrant.scroll(collection_name=COLLECTION, limit=1000, with_payload=["scope", "kind", "project_id", "agent_id", "superseded_by"], with_vectors=False, offset=offset)
|
||||
for point in points:
|
||||
payload = point.payload
|
||||
total += 1
|
||||
scope = payload.get("scope", "agent")
|
||||
kind = payload.get("kind", "fact")
|
||||
scope_kinds.setdefault(scope, Counter())[kind] += 1
|
||||
if payload.get("project_id"):
|
||||
projects[payload["project_id"]] += 1
|
||||
agents[payload.get("agent_id", "shared")] += 1
|
||||
if payload.get("superseded_by"):
|
||||
superseded += 1
|
||||
if not next_offset:
|
||||
break
|
||||
offset = next_offset
|
||||
|
||||
data = {
|
||||
"scopes": [{"scope": scope, "count": sum(counts.values()), "kinds": [{"kind": kind, "count": count} for kind, count in sorted(counts.items())]} for scope, counts in sorted(scope_kinds.items())],
|
||||
"projects": [{"project_id": project, "count": count} for project, count in projects.most_common()],
|
||||
"agents": [{"agent_id": agent, "count": count} for agent, count in agents.most_common()],
|
||||
"superseded": superseded,
|
||||
"total": total,
|
||||
}
|
||||
state.meta_cache["overview"] = {"ts": now, "data": data}
|
||||
audit(key, "meta", cached=False, total=total)
|
||||
return {**data, "cached": False}
|
||||
|
||||
|
||||
@router.get("/v1/status")
|
||||
async def status(request: Request) -> dict:
|
||||
ip = request.client.host if request.client else "unknown"
|
||||
now = time.monotonic()
|
||||
window = state.status_ratelimit.setdefault(ip, [])
|
||||
window[:] = [t for t in window if now - t < 60]
|
||||
if len(window) >= state.STATUS_RATE_LIMIT_PER_MIN:
|
||||
raise HTTPException(status_code=429, detail="Rate limit superato")
|
||||
window.append(now)
|
||||
info = state.qdrant.get_collection(COLLECTION)
|
||||
return {"status": "ok", "collection": COLLECTION, "points": info.points_count, "embedding_model": EMBED_MODEL, "embedding_dim": EMBED_DIM, "access": "shared", "api_keys": len(API_KEYS), "version": GATEWAY_VERSION, "git_commit": __import__("config").GIT_COMMIT, "guardrail_version": GUARDRAIL_VERSION}
|
||||
|
||||
|
||||
@router.get("/v1/version")
|
||||
async def version() -> dict:
|
||||
return {"version": GATEWAY_VERSION, "git_commit": __import__("config").GIT_COMMIT, "guardrail_version": GUARDRAIL_VERSION, "guardrail_enabled": GUARDRAIL_ENABLED, "guardrail_block_threshold": GUARDRAIL_BLOCK_THRESHOLD, "guardrail_warn_threshold": GUARDRAIL_WARN_THRESHOLD, "embedding_model": EMBED_MODEL, "collection": COLLECTION, "embed_nodes": [n.name for n in embed.chain_nodes()], "rerank_enabled": rerank.enabled(), "rerank_model": config.RERANK_MODEL, "rerank_nodes": [n.name for n in rerank._get_chain()]}
|
||||
|
||||
|
||||
@router.get("/v1/metrics")
|
||||
async def metrics_endpoint(key: str = Depends(require_auth)) -> dict:
|
||||
return metrics.snapshot(state.qdrant, COLLECTION)
|
||||
@@ -1,83 +0,0 @@
|
||||
"""Stato runtime condiviso tra bootstrap e route."""
|
||||
from __future__ import annotations
|
||||
|
||||
import contextvars
|
||||
import hashlib
|
||||
import json
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from qdrant_client import QdrantClient
|
||||
|
||||
from config import QDRANT_API_KEY, QDRANT_URL, QDRANT_RETRIES, log, _metrics
|
||||
|
||||
|
||||
class ResilientQdrant:
|
||||
"""Proxy del client Qdrant che ritenta i metodi su errori di transport
|
||||
(connessione/timeout transienti, es. riavvio del container Qdrant).
|
||||
Gli errori applicativi (404, validazione) non vengono ritentati."""
|
||||
|
||||
def __init__(self, client: Any, attempts: int = QDRANT_RETRIES, backoff_s: float = 0.4):
|
||||
self._client = client
|
||||
self._attempts = max(1, attempts)
|
||||
self._backoff = backoff_s
|
||||
|
||||
@staticmethod
|
||||
def _transient(exc: Exception) -> bool:
|
||||
if isinstance(exc, httpx.TransportError):
|
||||
return True
|
||||
return type(exc).__name__ in ("ConnectionError", "TimeoutError")
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
attr = getattr(self._client, name)
|
||||
if not callable(attr):
|
||||
return attr
|
||||
|
||||
def wrapped(*args: Any, **kwargs: Any) -> Any:
|
||||
for attempt in range(self._attempts):
|
||||
try:
|
||||
return attr(*args, **kwargs)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
if attempt == self._attempts - 1 or not self._transient(exc):
|
||||
raise
|
||||
delay = self._backoff * (2**attempt)
|
||||
_metrics["qdrant_retries"] += 1
|
||||
log.warning(
|
||||
"qdrant.%s: errore transiente (%s: %s) → retry %d/%d tra %.1fs",
|
||||
name,
|
||||
exc.__class__.__name__,
|
||||
exc,
|
||||
attempt + 2,
|
||||
self._attempts,
|
||||
delay,
|
||||
)
|
||||
time.sleep(delay)
|
||||
|
||||
return wrapped
|
||||
|
||||
|
||||
qdrant = ResilientQdrant(QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY))
|
||||
request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
|
||||
ratelimit: dict[str, list[float]] = {}
|
||||
status_ratelimit: dict[str, list[float]] = {}
|
||||
idempotency: dict[str, dict[str, Any]] = {}
|
||||
meta_cache: dict[str, Any] = {}
|
||||
STATUS_RATE_LIMIT_PER_MIN = 30
|
||||
IDEMPOTENCY_TTL_SECONDS = 24 * 3600
|
||||
|
||||
|
||||
def payload_hash(body: Any) -> str:
|
||||
canonical = json.dumps(body.model_dump(), sort_keys=True, default=str)
|
||||
return hashlib.sha256(canonical.encode()).hexdigest()
|
||||
|
||||
|
||||
def idempotency_cleanup() -> None:
|
||||
now = time.time()
|
||||
expired = [k for k, v in idempotency.items() if now - v["ts"] > IDEMPOTENCY_TTL_SECONDS]
|
||||
for key in expired:
|
||||
idempotency.pop(key, None)
|
||||
|
||||
|
||||
def invalidate_meta() -> None:
|
||||
meta_cache.clear()
|
||||
@@ -1,96 +0,0 @@
|
||||
"""Operazioni Qdrant condivise dalle route."""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from typing import Any, Optional
|
||||
|
||||
from qdrant_client.http import models as qm
|
||||
|
||||
from config import SPARSE_VECTOR_NAME
|
||||
from models import SearchIn
|
||||
|
||||
|
||||
def search_filter(body: SearchIn) -> qm.Filter | None:
|
||||
must: list[Any] = []
|
||||
for key in ("kind", "project_id", "scope", "parent_id", "level", "topic"):
|
||||
value = getattr(body, key)
|
||||
if value:
|
||||
must.append(qm.FieldCondition(key=key, match=qm.MatchValue(value=value)))
|
||||
if not body.include_superseded:
|
||||
must.append(qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")))
|
||||
must_not: list[Any] = []
|
||||
if not body.include_private:
|
||||
# default: esclude i record riservati (private=true) dalle ricerche standard
|
||||
must_not.append(qm.FieldCondition(key="private", match=qm.MatchValue(value=True)))
|
||||
if must or must_not:
|
||||
return qm.Filter(must=must or None, must_not=must_not or None)
|
||||
return None
|
||||
|
||||
|
||||
def search(qdrant: Any, collection: str, body: SearchIn, vector: list[float], sparse: Any, limit: Optional[int] = None) -> list[Any]:
|
||||
"""Ricerca ibrida o densa. Con reranking attivo limit > top_k per dare candidati extra allo stadio di rerank."""
|
||||
eff_limit = limit if limit is not None else body.top_k
|
||||
qfilter = search_filter(body)
|
||||
if body.hybrid and sparse is not None:
|
||||
return qdrant.query_points(
|
||||
collection_name=collection,
|
||||
prefetch=[
|
||||
qm.Prefetch(query=vector, using="", limit=max(body.top_k * 4, eff_limit), score_threshold=body.min_score),
|
||||
qm.Prefetch(query=sparse, using=SPARSE_VECTOR_NAME, limit=max(body.top_k * 4, eff_limit)),
|
||||
],
|
||||
query=qm.FusionQuery(fusion=qm.Fusion.RRF),
|
||||
query_filter=qfilter,
|
||||
limit=eff_limit,
|
||||
with_payload=True,
|
||||
).points
|
||||
return qdrant.query_points(
|
||||
collection_name=collection,
|
||||
query=vector,
|
||||
query_filter=qfilter,
|
||||
limit=eff_limit,
|
||||
score_threshold=body.min_score,
|
||||
with_payload=True,
|
||||
).points
|
||||
|
||||
|
||||
def format_results(hits: list[Any]) -> list[dict]:
|
||||
return [
|
||||
{
|
||||
"memory_id": h.id,
|
||||
"score": round(h.score, 4),
|
||||
"text": h.payload.get("text"),
|
||||
"kind": h.payload.get("kind"),
|
||||
"agent_id": h.payload.get("agent_id"),
|
||||
"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"),
|
||||
"superseded_by": h.payload.get("superseded_by"),
|
||||
"supersede_reason": h.payload.get("supersede_reason"),
|
||||
"parent_id": h.payload.get("parent_id"),
|
||||
"level": h.payload.get("level"),
|
||||
"topic": h.payload.get("topic"),
|
||||
"private": h.payload.get("private", False),
|
||||
"links": h.payload.get("links"),
|
||||
}
|
||||
for h in hits
|
||||
]
|
||||
|
||||
|
||||
def reparent_active_children(qdrant: Any, collection: str, old_id: str, new_id: str) -> int:
|
||||
children, _ = qdrant.scroll(
|
||||
collection_name=collection,
|
||||
scroll_filter=qm.Filter(must=[
|
||||
qm.FieldCondition(key="parent_id", match=qm.MatchValue(value=old_id)),
|
||||
qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")),
|
||||
]),
|
||||
limit=1000,
|
||||
with_payload=False,
|
||||
)
|
||||
if not children:
|
||||
return 0
|
||||
qdrant.set_payload(collection_name=collection, payload={"parent_id": new_id}, points=[p.id for p in children])
|
||||
return len(children)
|
||||
@@ -1,154 +0,0 @@
|
||||
"""Fixtures pytest per il Memory Gateway: FakeQdrant in-memory + mock di embed.
|
||||
|
||||
I test non richiedono Qdrant né Ollama: il gateway viene importato con le
|
||||
dipendenze reali (fastapi/pydantic/qdrant-client/httpx) ma qdrant e embed
|
||||
sono sostituiti da fake deterministici.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Disabilita il push metriche nei test
|
||||
os.environ["METRICS_ENABLED"] = "false"
|
||||
os.environ["API_KEYS"] = "test-key"
|
||||
# Guardrail disabilitato di default nei test esistenti (abilitato nei test del guardrail)
|
||||
os.environ["GUARDRAIL_ENABLED"] = "false"
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import pytest # noqa: E402
|
||||
import main as gateway # noqa: E402
|
||||
|
||||
|
||||
class FakePoint:
|
||||
def __init__(self, point_id, vector=None, payload=None):
|
||||
self.id = point_id
|
||||
self.vector = vector or {}
|
||||
self.payload = payload or {}
|
||||
|
||||
|
||||
class FakeQdrant:
|
||||
"""Implementazione in-memory dei metodi Qdrant usati dal gateway."""
|
||||
|
||||
def __init__(self):
|
||||
self.points: dict[str, FakePoint] = {}
|
||||
self.collection_exists = False
|
||||
self.upsert_calls = 0
|
||||
self.query_score = 0.9 # score di default per query_points (configurabile nei test)
|
||||
|
||||
def get_collections(self):
|
||||
class _C:
|
||||
def __init__(self, names):
|
||||
self.collections = [type("X", (), {"name": n})() for n in names]
|
||||
|
||||
return _C(["memories"] if self.collection_exists else [])
|
||||
|
||||
def create_collection(self, **kw):
|
||||
self.collection_exists = True
|
||||
|
||||
def create_payload_index(self, **kw):
|
||||
pass
|
||||
|
||||
def create_vector_name(self, *a, **kw):
|
||||
pass
|
||||
|
||||
def upsert(self, collection_name, points, **kw):
|
||||
self.upsert_calls += 1
|
||||
for p in points:
|
||||
self.points[p.id] = FakePoint(p.id, p.vector, p.payload)
|
||||
|
||||
def retrieve(self, collection_name, ids, with_payload=True):
|
||||
return [self.points[i] for i in ids if i in self.points]
|
||||
|
||||
def set_payload(self, collection_name, payload, points, **kw):
|
||||
for pid in points:
|
||||
if pid in self.points:
|
||||
self.points[pid].payload.update(payload)
|
||||
|
||||
def delete(self, collection_name, points_selector, **kw):
|
||||
for pid in points_selector:
|
||||
self.points.pop(pid, None)
|
||||
|
||||
def _matches(self, pl, query_filter):
|
||||
"""Applica i filtri metadata (FieldCondition match / IsEmptyCondition)."""
|
||||
if not query_filter or not query_filter.must:
|
||||
return True
|
||||
for cond in query_filter.must:
|
||||
if hasattr(cond, "key") and hasattr(cond, "match"):
|
||||
if pl.get(cond.key) != cond.match.value:
|
||||
return False
|
||||
elif hasattr(cond, "is_empty"):
|
||||
if pl.get(cond.is_empty.key):
|
||||
return False
|
||||
return True
|
||||
|
||||
def scroll(self, collection_name, scroll_filter=None, limit=None, with_payload=True, **kw):
|
||||
points = [p for p in self.points.values() if self._matches(p.payload, scroll_filter)]
|
||||
if limit:
|
||||
points = points[:limit]
|
||||
return points, None
|
||||
|
||||
def get_collection(self, collection_name):
|
||||
class _Info:
|
||||
points_count = len(self.points)
|
||||
|
||||
return _Info()
|
||||
|
||||
def query_points(self, collection_name, query=None, query_filter=None, limit=5,
|
||||
score_threshold=None, with_payload=True, prefetch=None, **kw):
|
||||
# Ritorna tutti i punti (score fisso); i filtri metadata sono applicati
|
||||
# in modo semplice per testare kind/project_id/scope/superseded.
|
||||
results = []
|
||||
for p in self.points.values():
|
||||
pl = p.payload
|
||||
if query_filter and query_filter.must:
|
||||
ok = True
|
||||
for cond in query_filter.must:
|
||||
if hasattr(cond, "key") and hasattr(cond, "match"):
|
||||
if pl.get(cond.key) != cond.match.value:
|
||||
ok = False
|
||||
elif hasattr(cond, "is_empty"):
|
||||
if pl.get(cond.is_empty.key):
|
||||
ok = False
|
||||
if not ok:
|
||||
continue
|
||||
results.append(type("H", (), {"id": p.id, "score": self.query_score, "payload": pl})())
|
||||
return type("R", (), {"points": results[:limit]})()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch):
|
||||
"""TestClient con qdrant e embed finti."""
|
||||
fake = FakeQdrant()
|
||||
monkeypatch.setattr(gateway.state, "qdrant", fake)
|
||||
monkeypatch.setattr(gateway, "qdrant", fake)
|
||||
monkeypatch.setattr(gateway, "API_KEYS", {"test-key"}, raising=False)
|
||||
monkeypatch.setattr(gateway.state, "ratelimit", {}) # rate limit pulito per test
|
||||
|
||||
async def fake_embed(text):
|
||||
return [0.0] * 1024
|
||||
|
||||
monkeypatch.setattr(gateway.state, "embed", fake_embed)
|
||||
monkeypatch.setattr(gateway, "embed", fake_embed)
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
with TestClient(gateway.app) as c:
|
||||
c.fake_qdrant = fake
|
||||
yield c
|
||||
|
||||
|
||||
def auth_headers():
|
||||
return {"X-API-Key": "test-key"}
|
||||
|
||||
|
||||
def make_record(**overrides):
|
||||
base = {
|
||||
"text": "record di test",
|
||||
"kind": "fact",
|
||||
"project_id": "test-proj",
|
||||
"scope": "agent",
|
||||
}
|
||||
base.update(overrides)
|
||||
return base
|
||||
@@ -1,399 +0,0 @@
|
||||
"""Test API del Memory Gateway: validazione, auth, idempotency, supersede, ricerca."""
|
||||
|
||||
import pytest
|
||||
from conftest import auth_headers, make_record
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Validazione input
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_project_id_obbligatorio(client):
|
||||
body = make_record()
|
||||
del body["project_id"]
|
||||
r = client.post("/v1/memories", json=body, headers=auth_headers())
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_kind_invalido(client):
|
||||
r = client.post("/v1/memories", json=make_record(kind="boh"), headers=auth_headers())
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_expires_at_invalido(client):
|
||||
r = client.post("/v1/memories", json=make_record(expires_at="non-una-data"), headers=auth_headers())
|
||||
assert r.status_code == 422
|
||||
assert "ISO 8601" in r.text
|
||||
|
||||
|
||||
def test_expires_at_valido(client):
|
||||
r = client.post("/v1/memories", json=make_record(expires_at="2026-09-01T00:00:00Z"), headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
def test_confidence_invalido(client):
|
||||
r = client.post("/v1/memories", json=make_record(confidence="super"), headers=auth_headers())
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_confidence_default_medium(client):
|
||||
r = client.post("/v1/memories", json=make_record(), headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
mid = r.json()["memory_id"]
|
||||
g = client.get(f"/v1/memories/{mid}", headers=auth_headers())
|
||||
assert g.json()["confidence"] == "medium"
|
||||
|
||||
|
||||
def test_text_troppo_lungo(client):
|
||||
r = client.post("/v1/memories", json=make_record(text="x" * 9000), headers=auth_headers())
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Auth e rate limit
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_senza_chiave_422(client):
|
||||
# Header X-API-Key mancante → 422 (header richiesto da FastAPI)
|
||||
r = client.post("/v1/memories", json=make_record())
|
||||
assert r.status_code == 422
|
||||
|
||||
|
||||
def test_chiave_invalida_401(client):
|
||||
r = client.post("/v1/memories", json=make_record(), headers={"X-API-Key": "sbagliata"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_rate_limit_429(client, monkeypatch):
|
||||
import config
|
||||
|
||||
monkeypatch.setattr(config, "RATE_LIMIT_PER_MIN", 3)
|
||||
for _ in range(3):
|
||||
r = client.post("/v1/memories", json=make_record(), headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
r = client.post("/v1/memories", json=make_record(), headers=auth_headers())
|
||||
assert r.status_code == 429
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Idempotency
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_idempotency_replay_stessa_risposta(client):
|
||||
h = {**auth_headers(), "Idempotency-Key": "k-1"}
|
||||
r1 = client.post("/v1/memories", json=make_record(), headers=h)
|
||||
r2 = client.post("/v1/memories", json=make_record(), headers=h)
|
||||
assert r1.status_code == 200 and r2.status_code == 200
|
||||
assert r1.json()["memory_id"] == r2.json()["memory_id"]
|
||||
assert client.fake_qdrant.upsert_calls == 1
|
||||
|
||||
|
||||
def test_idempotency_payload_diverso_409(client):
|
||||
h = {**auth_headers(), "Idempotency-Key": "k-2"}
|
||||
client.post("/v1/memories", json=make_record(), headers=h)
|
||||
r = client.post("/v1/memories", json=make_record(text="diverso"), headers=h)
|
||||
assert r.status_code == 409
|
||||
|
||||
|
||||
def test_idempotency_key_diverse_record_distinti(client):
|
||||
r1 = client.post("/v1/memories", json=make_record(), headers={**auth_headers(), "Idempotency-Key": "k-a"})
|
||||
r2 = client.post("/v1/memories", json=make_record(), headers={**auth_headers(), "Idempotency-Key": "k-b"})
|
||||
assert r1.json()["memory_id"] != r2.json()["memory_id"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Supersede
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_supersede_target_inesistente_404(client):
|
||||
r = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(supersedes_id="00000000-0000-0000-0000-000000000000"),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_supersede_ok_e_lineage(client):
|
||||
r1 = client.post("/v1/memories", json=make_record(text="fatto falso"), headers=auth_headers())
|
||||
old_id = r1.json()["memory_id"]
|
||||
r2 = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="fatto corretto", supersedes_id=old_id, supersede_reason="evidenza"),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert r2.status_code == 200
|
||||
new_id = r2.json()["memory_id"]
|
||||
# il vecchio è marcato superseded_by
|
||||
old = client.get(f"/v1/memories/{old_id}", headers=auth_headers()).json()
|
||||
assert old["superseded_by"] == new_id
|
||||
# la ricerca di default esclude i superseduti
|
||||
s = client.post("/v1/memories:search", json={"query": "fatto", "top_k": 10, "min_score": 0.0}, headers=auth_headers())
|
||||
ids = [x["memory_id"] for x in s.json()["results"]]
|
||||
assert old_id not in ids
|
||||
# include_superseded li mostra
|
||||
s2 = client.post(
|
||||
"/v1/memories:search",
|
||||
json={"query": "fatto", "top_k": 10, "min_score": 0.0, "include_superseded": True},
|
||||
headers=auth_headers(),
|
||||
)
|
||||
ids2 = [x["memory_id"] for x in s2.json()["results"]]
|
||||
assert old_id in ids2
|
||||
|
||||
|
||||
def test_supersede_doppio_409(client):
|
||||
r1 = client.post("/v1/memories", json=make_record(text="falso"), headers=auth_headers())
|
||||
old_id = r1.json()["memory_id"]
|
||||
client.post("/v1/memories", json=make_record(text="corretto", supersedes_id=old_id), headers=auth_headers())
|
||||
r = client.post("/v1/memories", json=make_record(text="ancora", supersedes_id=old_id), headers=auth_headers())
|
||||
assert r.status_code == 409
|
||||
|
||||
|
||||
def test_supersede_root_ri_parenta_figli_attivi(client):
|
||||
# L1 root + figlio L2
|
||||
r1 = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="root L1", level="L1_ROOT", topic="TEST-TOPIC/ROOT"),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
root_id = r1.json()["memory_id"]
|
||||
r2 = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="figlio L2", level="L2_SUBTOPIC", topic="TEST-TOPIC/SUB", parent_id=root_id),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
child_id = r2.json()["memory_id"]
|
||||
|
||||
# supersede il root
|
||||
r3 = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(
|
||||
text="root L1 corretto",
|
||||
level="L1_ROOT",
|
||||
topic="TEST-TOPIC/ROOT",
|
||||
supersedes_id=root_id,
|
||||
supersede_reason="aggiornamento",
|
||||
),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert r3.status_code == 200
|
||||
new_root_id = r3.json()["memory_id"]
|
||||
assert r3.json()["reparented"] == 1
|
||||
|
||||
# il figlio attivo ora punta al nuovo root
|
||||
child = client.get(f"/v1/memories/{child_id}", headers=auth_headers()).json()
|
||||
assert child["parent_id"] == new_root_id
|
||||
|
||||
# search per parent_id sul nuovo root trova il figlio
|
||||
s = client.post(
|
||||
"/v1/memories:search",
|
||||
json={"query": "*", "parent_id": new_root_id, "top_k": 10, "min_score": 0.0},
|
||||
headers=auth_headers(),
|
||||
)
|
||||
ids = [x["memory_id"] for x in s.json()["results"]]
|
||||
assert child_id in ids
|
||||
|
||||
|
||||
def test_supersede_root_ri_parenta_solo_figli_attivi(client):
|
||||
# L1 root + figlio L2 + figlio L2 già superseduto (versione attiva C1')
|
||||
r1 = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="root", level="L1_ROOT", topic="T2/ROOT"),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
root_id = r1.json()["memory_id"]
|
||||
c1 = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="figlio vecchio", level="L2_SUBTOPIC", topic="T2/SUB", parent_id=root_id),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
c1_id = c1.json()["memory_id"]
|
||||
c1p = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(
|
||||
text="figlio nuovo",
|
||||
level="L2_SUBTOPIC",
|
||||
topic="T2/SUB",
|
||||
parent_id=root_id,
|
||||
supersedes_id=c1_id,
|
||||
),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
c1p_id = c1p.json()["memory_id"]
|
||||
|
||||
r2 = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="root corretto", level="L1_ROOT", topic="T2/ROOT", supersedes_id=root_id),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
new_root_id = r2.json()["memory_id"]
|
||||
assert r2.json()["reparented"] == 1 # solo C1' (attivo)
|
||||
|
||||
# C1' ri-parentato al nuovo root; C1 storico resta ancorato al vecchio
|
||||
assert client.get(f"/v1/memories/{c1p_id}", headers=auth_headers()).json()["parent_id"] == new_root_id
|
||||
assert client.get(f"/v1/memories/{c1_id}", headers=auth_headers()).json()["parent_id"] == root_id
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Ricerca e filtri
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_search_filtro_project_id(client):
|
||||
client.post("/v1/memories", json=make_record(text="uno", project_id="proj-a"), headers=auth_headers())
|
||||
client.post("/v1/memories", json=make_record(text="due", project_id="proj-b"), headers=auth_headers())
|
||||
s = client.post(
|
||||
"/v1/memories:search",
|
||||
json={"query": "test", "project_id": "proj-a", "top_k": 10, "min_score": 0.0},
|
||||
headers=auth_headers(),
|
||||
)
|
||||
results = s.json()["results"]
|
||||
assert len(results) == 1
|
||||
assert results[0]["project_id"] == "proj-a"
|
||||
|
||||
|
||||
def test_search_hybrid_param_accettato(client):
|
||||
client.post("/v1/memories", json=make_record(text="codice XYZ-123"), headers=auth_headers())
|
||||
s = client.post(
|
||||
"/v1/memories:search",
|
||||
json={"query": "XYZ-123", "top_k": 5, "min_score": 0.0, "hybrid": True},
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert s.status_code == 200
|
||||
assert "results" in s.json()
|
||||
|
||||
|
||||
def test_meta_overview(client):
|
||||
client.post("/v1/memories", json=make_record(project_id="proj-a"), headers=auth_headers())
|
||||
r = client.get("/v1/meta/overview", headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert data["total"] >= 1
|
||||
assert any(p["project_id"] == "proj-a" for p in data["projects"])
|
||||
|
||||
|
||||
def test_status_pubblico(client):
|
||||
r = client.get("/v1/status")
|
||||
assert r.status_code == 200
|
||||
assert r.json()["status"] == "ok"
|
||||
|
||||
|
||||
def test_metrics_auth(client):
|
||||
# Header mancante → 422; chiave invalida → 401; chiave valida → 200
|
||||
assert client.get("/v1/metrics").status_code == 422
|
||||
assert client.get("/v1/metrics", headers={"X-API-Key": "sbagliata"}).status_code == 401
|
||||
r2 = client.get("/v1/metrics", headers=auth_headers())
|
||||
assert r2.status_code == 200
|
||||
assert "requests" in r2.json()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Versione / metadata del codice
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_version_endpoint_pubblico(client):
|
||||
"""GET /v1/version è pubblico e espone git_commit e guardrail_version."""
|
||||
r = client.get("/v1/version")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert "git_commit" in data
|
||||
assert "version" in data
|
||||
assert "guardrail_version" in data
|
||||
assert data["guardrail_version"] == "similarity-v2"
|
||||
|
||||
|
||||
def test_status_espone_git_commit(client):
|
||||
"""/v1/status include version, git_commit e guardrail_version."""
|
||||
r = client.get("/v1/status")
|
||||
assert r.status_code == 200
|
||||
data = r.json()
|
||||
assert "git_commit" in data
|
||||
assert "version" in data
|
||||
assert "guardrail_version" in data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Struttura Gerarchica e Relazionale
|
||||
# ---------------------------------------------------------------------------
|
||||
def test_create_and_retrieve_hierarchical_record(client):
|
||||
"""Crea un nodo Root L1 e un nodo Figlio L2 con links, parent_id, level, topic."""
|
||||
# 1. Crea Root L1
|
||||
r_root = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(
|
||||
text="Master Topic Alfa Romeo",
|
||||
level="L1_ROOT",
|
||||
topic="ALFA-ROMEO/ROOT",
|
||||
),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert r_root.status_code == 200
|
||||
root_id = r_root.json()["memory_id"]
|
||||
|
||||
# 2. Crea Figlio L2 collegato
|
||||
r_child = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(
|
||||
text="Scheda Tecnica Bialbero 1.3",
|
||||
parent_id=root_id,
|
||||
level="L2_SUBTOPIC",
|
||||
topic="ALFA-ROMEO/SPECS",
|
||||
links=[{"target_id": root_id, "predicate": "part_of", "weight": 1.0}],
|
||||
),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert r_child.status_code == 200
|
||||
child_id = r_child.json()["memory_id"]
|
||||
|
||||
# 3. Recupera e verifica payload strutturato
|
||||
g = client.get(f"/v1/memories/{child_id}", headers=auth_headers())
|
||||
assert g.status_code == 200
|
||||
data = g.json()
|
||||
assert data["parent_id"] == root_id
|
||||
assert data["level"] == "L2_SUBTOPIC"
|
||||
assert data["topic"] == "ALFA-ROMEO/SPECS"
|
||||
assert len(data["links"]) == 1
|
||||
assert data["links"][0]["target_id"] == root_id
|
||||
|
||||
|
||||
def test_search_filters_hierarchical(client):
|
||||
"""Filtra per parent_id, level e topic."""
|
||||
r_root = client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="Root doc", level="L1_ROOT", topic="TOPIC/ROOT"),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
root_id = r_root.json()["memory_id"]
|
||||
|
||||
client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="Child A", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/A"),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
client.post(
|
||||
"/v1/memories",
|
||||
json=make_record(text="Child B", parent_id=root_id, level="L2_SUBTOPIC", topic="TOPIC/B"),
|
||||
headers=auth_headers(),
|
||||
)
|
||||
|
||||
# Cerca solo L1_ROOT
|
||||
s1 = client.post(
|
||||
"/v1/memories:search",
|
||||
json={"query": "doc", "level": "L1_ROOT", "top_k": 5, "min_score": 0.0},
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert s1.status_code == 200
|
||||
assert len(s1.json()["results"]) == 1
|
||||
assert s1.json()["results"][0]["level"] == "L1_ROOT"
|
||||
|
||||
# Cerca per parent_id
|
||||
s2 = client.post(
|
||||
"/v1/memories:search",
|
||||
json={"query": "Child", "parent_id": root_id, "top_k": 5, "min_score": 0.0},
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert s2.status_code == 200
|
||||
assert len(s2.json()["results"]) == 2
|
||||
|
||||
# Cerca per topic specifico
|
||||
s3 = client.post(
|
||||
"/v1/memories:search",
|
||||
json={"query": "Child", "topic": "TOPIC/A", "top_k": 5, "min_score": 0.0},
|
||||
headers=auth_headers(),
|
||||
)
|
||||
assert s3.status_code == 200
|
||||
assert len(s3.json()["results"]) == 1
|
||||
assert s3.json()["results"][0]["topic"] == "TOPIC/A"
|
||||
@@ -1,189 +0,0 @@
|
||||
"""Test della catena di fallback per gli embedding e del wrapper Qdrant resilient."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import embed as embed_mod
|
||||
import state
|
||||
from test_api import auth_headers, make_record
|
||||
|
||||
CHAIN = json.dumps(
|
||||
[
|
||||
{"name": "primario", "url": "http://primario:9001", "api": "llamacpp", "key": "k1", "timeout_ms": 500},
|
||||
{"name": "fallback", "url": "http://fallback:9002", "api": "ollama", "key": "k2", "timeout_ms": 5000},
|
||||
]
|
||||
)
|
||||
|
||||
VEC_1024 = [0.01] * 1024
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_chain(monkeypatch):
|
||||
embed_mod.reset_chain_cache()
|
||||
embed_mod._http = None
|
||||
yield
|
||||
embed_mod.reset_chain_cache()
|
||||
embed_mod._http = None
|
||||
|
||||
|
||||
def _mock_client(handler) -> list[str]:
|
||||
calls: list[str] = []
|
||||
|
||||
def tracking_handler(request):
|
||||
calls.append(f"{request.url.host}{request.url.path}")
|
||||
return handler(request)
|
||||
|
||||
embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking_handler))
|
||||
return calls
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse della catena
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parse_chain_valida(monkeypatch):
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN)
|
||||
embed_mod.reset_chain_cache()
|
||||
nodes = embed_mod.chain_nodes()
|
||||
assert [n.name for n in nodes] == ["primario", "fallback"]
|
||||
assert [n.api for n in nodes] == ["llamacpp", "ollama"]
|
||||
|
||||
|
||||
def test_parse_legacy_quando_catena_vuota(monkeypatch):
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", "")
|
||||
monkeypatch.setattr(embed_mod, "EMBED_API", "ollama")
|
||||
monkeypatch.setattr(embed_mod, "EMBED_URL", "http://legacy:11434")
|
||||
monkeypatch.setattr(embed_mod, "EMBED_API_KEY", "lk")
|
||||
embed_mod.reset_chain_cache()
|
||||
nodes = embed_mod.chain_nodes()
|
||||
assert len(nodes) == 1
|
||||
assert nodes[0].name == "embed"
|
||||
assert nodes[0].api == "ollama"
|
||||
assert nodes[0].url == "http://legacy:11434"
|
||||
assert nodes[0].key == "lk"
|
||||
|
||||
|
||||
def test_parse_chain_json_invalido_cade_su_legacy(monkeypatch):
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", "non-json")
|
||||
monkeypatch.setattr(embed_mod, "EMBED_URL", "http://legacy:11434")
|
||||
embed_mod.reset_chain_cache()
|
||||
assert [n.url for n in embed_mod.chain_nodes()] == ["http://legacy:11434"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# fallback e cooldown
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_fallback_primario_llamacpp_fallito(monkeypatch):
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN)
|
||||
embed_mod.reset_chain_cache()
|
||||
|
||||
def handler(request):
|
||||
if request.url.host == "primario":
|
||||
return httpx.Response(500)
|
||||
# nodo ollama-style: risposta con campo "embeddings"
|
||||
return httpx.Response(200, json={"embeddings": [VEC_1024]})
|
||||
|
||||
embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||||
vector = asyncio.run(embed_mod.embed("test"))
|
||||
assert vector == VEC_1024
|
||||
|
||||
|
||||
def test_cooldown_salta_primario(monkeypatch):
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN)
|
||||
embed_mod.reset_chain_cache()
|
||||
calls: list[str] = []
|
||||
|
||||
def tracking(request):
|
||||
calls.append(request.url.host)
|
||||
if request.url.host == "primario":
|
||||
return httpx.Response(500)
|
||||
return httpx.Response(200, json={"embeddings": [VEC_1024]})
|
||||
|
||||
embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking))
|
||||
asyncio.run(embed_mod.embed("test"))
|
||||
asyncio.run(embed_mod.embed("test"))
|
||||
# il primario fallito entra in cooldown: la seconda chiamata lo salta
|
||||
assert calls == ["primario", "fallback", "fallback"]
|
||||
|
||||
|
||||
def test_dimensione_errata_salta_nodo(monkeypatch):
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN)
|
||||
embed_mod.reset_chain_cache()
|
||||
|
||||
def handler(request):
|
||||
if request.url.host == "primario":
|
||||
return httpx.Response(200, json={"data": [{"embedding": [0.0] * 512}]}) # dim sbagliata
|
||||
return httpx.Response(200, json={"embeddings": [VEC_1024]})
|
||||
|
||||
embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||||
vector = asyncio.run(embed_mod.embed("test"))
|
||||
assert len(vector) == 1024
|
||||
|
||||
|
||||
def test_tutti_nodi_falliti_rilancia(monkeypatch):
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN)
|
||||
embed_mod.reset_chain_cache()
|
||||
embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(500)))
|
||||
with pytest.raises(RuntimeError, match="tutti i nodi embedding falliti"):
|
||||
asyncio.run(embed_mod.embed("test"))
|
||||
|
||||
|
||||
def test_empty_text_comunque_chiamata(monkeypatch):
|
||||
# il modello pydantic valida già la query; qui verifichiamo il passthrough
|
||||
monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN)
|
||||
embed_mod.reset_chain_cache()
|
||||
|
||||
def handler(request):
|
||||
body = json.loads(request.content)
|
||||
return httpx.Response(200, json={"embeddings": [VEC_1024] if body["input"] else []})
|
||||
|
||||
embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||||
assert asyncio.run(embed_mod.embed("ok")) == VEC_1024
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ResilientQdrant (retry transiente)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FlakyQdrant:
|
||||
def __init__(self, failures: int, exc: Exception):
|
||||
self.calls = 0
|
||||
self.failures = failures
|
||||
self.exc = exc
|
||||
|
||||
def upsert(self, **kw):
|
||||
self.calls += 1
|
||||
if self.calls <= self.failures:
|
||||
raise self.exc
|
||||
return "ok"
|
||||
|
||||
|
||||
def test_retry_su_errore_transiente(monkeypatch):
|
||||
fake = FlakyQdrant(2, httpx.ConnectError("conn"))
|
||||
client = state.ResilientQdrant(fake, attempts=3, backoff_s=0.01)
|
||||
assert client.upsert(x=1) == "ok"
|
||||
assert fake.calls == 3
|
||||
|
||||
|
||||
def test_niente_retry_su_errore_applicativo():
|
||||
fake = FlakyQdrant(2, ValueError("404 logico"))
|
||||
client = state.ResilientQdrant(fake, attempts=3, backoff_s=0.01)
|
||||
with pytest.raises(ValueError):
|
||||
client.upsert(x=1)
|
||||
assert fake.calls == 1
|
||||
|
||||
|
||||
def test_retry_esaurito_rilancia():
|
||||
fake = FlakyQdrant(99, httpx.ReadTimeout("t"))
|
||||
client = state.ResilientQdrant(fake, attempts=2, backoff_s=0.01)
|
||||
with pytest.raises(httpx.ReadTimeout):
|
||||
client.upsert(x=1)
|
||||
assert fake.calls == 2
|
||||
@@ -1,111 +0,0 @@
|
||||
"""Test del guardrail di similarità pre-scrittura (Memory Gateway).
|
||||
|
||||
Casi: duplicato esatto (hash) -> BLOCK 409; similarità alta -> BLOCK 409;
|
||||
similarità moderata -> WARN (salva con flag); nessun candidato -> ALLOW;
|
||||
supersede esplicito bypassa il guardrail.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from conftest import auth_headers, make_record
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def enable_guardrail(monkeypatch):
|
||||
"""Abilita il guardrail per i test di questa suite (il conftest lo disabilita di default)."""
|
||||
import config
|
||||
|
||||
monkeypatch.setattr(config, "GUARDRAIL_ENABLED", True)
|
||||
|
||||
|
||||
def test_duplicato_esatto_bloccato_409(client):
|
||||
"""Stesso testo normalizzato -> hash uguale -> BLOCK (409)."""
|
||||
body = make_record(text="Il cliente non accede al portale COSMO-SkyMed")
|
||||
r1 = client.post("/v1/memories", json=body, headers=auth_headers())
|
||||
assert r1.status_code == 200
|
||||
|
||||
# Stesso testo con maiuscole/spazi diversi -> stesso hash normalizzato
|
||||
body2 = make_record(text=" IL CLIENTE NON ACCEDE al portale COSMO-SkyMed ")
|
||||
r2 = client.post("/v1/memories", json=body2, headers=auth_headers())
|
||||
assert r2.status_code == 409
|
||||
detail = r2.json()["detail"]
|
||||
assert detail["error"] == "duplicate_memory"
|
||||
assert detail["reason"] == "EXACT_DUPLICATE"
|
||||
|
||||
|
||||
def test_similarita_alta_bloccata_409(client):
|
||||
"""Score top-1 >= soglia BLOCK (0.85) -> 409 KNOWN_SOLUTION."""
|
||||
client.fake_qdrant.query_score = 0.9
|
||||
r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers())
|
||||
assert r1.status_code == 200
|
||||
|
||||
r2 = client.post("/v1/memories", json=make_record(text="secondo record simile"), headers=auth_headers())
|
||||
assert r2.status_code == 409
|
||||
assert r2.json()["detail"]["reason"] == "KNOWN_SOLUTION"
|
||||
|
||||
|
||||
def test_similarita_moderata_warn_salva(client):
|
||||
"""Score top-1 tra 0.70 e 0.85 -> WARN: salva con flag guardrail."""
|
||||
client.fake_qdrant.query_score = 0.75
|
||||
r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers())
|
||||
assert r1.status_code == 200
|
||||
|
||||
r2 = client.post("/v1/memories", json=make_record(text="secondo record simile"), headers=auth_headers())
|
||||
assert r2.status_code == 200
|
||||
memory_id = r2.json()["memory_id"]
|
||||
# Il record salvato deve avere il flag guardrail WARN
|
||||
saved = client.fake_qdrant.points[memory_id].payload
|
||||
assert saved["guardrail"]["decision"] == "WARN"
|
||||
assert saved["guardrail"]["reason"] == "MODERATE_SIMILARITY"
|
||||
|
||||
|
||||
def test_nessun_candidato_allow(client):
|
||||
"""Score top-1 sotto soglia WARN -> ALLOW, nessun flag guardrail."""
|
||||
client.fake_qdrant.query_score = 0.5
|
||||
r1 = client.post("/v1/memories", json=make_record(text="primo record"), headers=auth_headers())
|
||||
assert r1.status_code == 200
|
||||
|
||||
r2 = client.post("/v1/memories", json=make_record(text="secondo record"), headers=auth_headers())
|
||||
assert r2.status_code == 200
|
||||
memory_id = r2.json()["memory_id"]
|
||||
saved = client.fake_qdrant.points[memory_id].payload
|
||||
assert saved["guardrail"]["decision"] == "ALLOW"
|
||||
assert saved["guardrail"]["reason"] == "NEW_SOLUTION"
|
||||
|
||||
|
||||
def test_supersede_bypassa_guardrail(client):
|
||||
"""Il supersede esplicito è una correzione intenzionale: bypassa il guardrail."""
|
||||
client.fake_qdrant.query_score = 0.9
|
||||
r1 = client.post("/v1/memories", json=make_record(text="record originale"), headers=auth_headers())
|
||||
assert r1.status_code == 200
|
||||
old_id = r1.json()["memory_id"]
|
||||
|
||||
# Supersede con testo molto simile -> deve passare (correzione intenzionale)
|
||||
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 r2.json()["supersedes_id"] == old_id
|
||||
|
||||
|
||||
def test_guardrail_disabilitato_salva_sempre(client, monkeypatch):
|
||||
"""Con GUARDRAIL_ENABLED=false non si blocca nulla."""
|
||||
import config
|
||||
|
||||
monkeypatch.setattr(config, "GUARDRAIL_ENABLED", False)
|
||||
client.fake_qdrant.query_score = 0.9
|
||||
r1 = client.post("/v1/memories", json=make_record(text="primo"), headers=auth_headers())
|
||||
assert r1.status_code == 200
|
||||
r2 = client.post("/v1/memories", json=make_record(text="primo"), headers=auth_headers())
|
||||
assert r2.status_code == 200
|
||||
|
||||
|
||||
def test_text_hash_salvato_nel_payload(client):
|
||||
"""Ogni record salvato deve avere text_hash (per lo strato 1 del guardrail)."""
|
||||
r = client.post("/v1/memories", json=make_record(text="record con hash"), headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
memory_id = r.json()["memory_id"]
|
||||
saved = client.fake_qdrant.points[memory_id].payload
|
||||
assert "text_hash" in saved
|
||||
assert len(saved["text_hash"]) == 64 # SHA-256 hex
|
||||
@@ -1,219 +0,0 @@
|
||||
"""Test dello stadio rerank: catena di fallback, cooldown, degrada con grazia."""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import rerank
|
||||
from test_api import auth_headers, make_record
|
||||
|
||||
CHAIN = json.dumps(
|
||||
[
|
||||
{"name": "primario", "url": "http://primario:9002", "key": "k1", "timeout_ms": 500},
|
||||
{"name": "fallback", "url": "http://fallback:9003", "key": "k2", "timeout_ms": 5000},
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_chain(monkeypatch):
|
||||
rerank.reset_chain_cache()
|
||||
rerank._http = None
|
||||
yield
|
||||
rerank.reset_chain_cache()
|
||||
rerank._http = None
|
||||
|
||||
|
||||
def _use_chain(monkeypatch, raw=CHAIN):
|
||||
# rerank.py importa i valori di config con `from config import`: si patchano
|
||||
# gli attributi del modulo rerank, non config.
|
||||
monkeypatch.setattr(rerank, "RERANK_CHAIN", raw)
|
||||
rerank.reset_chain_cache()
|
||||
|
||||
|
||||
def _mock_client(handler) -> list[str]:
|
||||
"""Client con transport mockato; ritorna la lista in cui registrare le chiamate."""
|
||||
calls: list[str] = []
|
||||
|
||||
def tracking_handler(request):
|
||||
calls.append(f"{request.url.host}{request.url.path}")
|
||||
return handler(request)
|
||||
|
||||
rerank._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking_handler))
|
||||
return calls
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# parse della catena
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_parse_chain_valida(monkeypatch):
|
||||
_use_chain(monkeypatch)
|
||||
nodes = rerank._get_chain()
|
||||
assert [n.name for n in nodes] == ["primario", "fallback"]
|
||||
assert nodes[0].key == "k1"
|
||||
assert nodes[0].timeout_ms == 500
|
||||
assert nodes[1].timeout_ms == 5000
|
||||
|
||||
|
||||
def test_parse_chain_json_invalido(monkeypatch):
|
||||
_use_chain(monkeypatch, raw="non-json")
|
||||
assert rerank._get_chain() == []
|
||||
assert not rerank.enabled()
|
||||
|
||||
|
||||
def test_parse_chain_scarta_url_senza_schema(monkeypatch):
|
||||
_use_chain(monkeypatch, raw=json.dumps([{"name": "x", "url": "primario:9002"}]))
|
||||
assert rerank._get_chain() == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# fallback e cooldown
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_fallback_primario_500(monkeypatch):
|
||||
_use_chain(monkeypatch)
|
||||
|
||||
def handler(request):
|
||||
if request.url.host == "primario":
|
||||
return httpx.Response(500)
|
||||
return httpx.Response(200, json={"results": [{"index": 1, "relevance_score": 2.0}, {"index": 0, "relevance_score": -1.0}]})
|
||||
|
||||
_mock_client(handler)
|
||||
scores, backend, _took = asyncio.run(rerank.rerank("q", ["docA", "docB"]))
|
||||
assert backend == "fallback"
|
||||
# gli score tornano allineati alla posizione originaria dei documenti
|
||||
assert scores == pytest.approx([-1.0, 2.0])
|
||||
|
||||
|
||||
def test_cooldown_salta_nodo_fallito(monkeypatch):
|
||||
_use_chain(monkeypatch)
|
||||
|
||||
def handler(request):
|
||||
if request.url.host == "primario":
|
||||
return httpx.Response(500)
|
||||
return httpx.Response(200, json={"results": [{"index": 0, "relevance_score": 1.0}]})
|
||||
|
||||
_mock_client(handler)
|
||||
asyncio.run(rerank.rerank("q", ["a", "b"]))
|
||||
scores, backend, _ = asyncio.run(rerank.rerank("q", ["a", "b"]))
|
||||
assert backend == "fallback" # il primario è in cooldown e non viene richiamato
|
||||
|
||||
|
||||
def test_cooldown_non_blocca_per_sempre(monkeypatch):
|
||||
_use_chain(monkeypatch)
|
||||
calls: list[str] = []
|
||||
|
||||
def handler(request):
|
||||
calls.append(request.url.host)
|
||||
if request.url.host == "primario":
|
||||
return httpx.Response(500)
|
||||
return httpx.Response(200, json={"results": [{"index": 0, "relevance_score": 1.0}]})
|
||||
|
||||
_mock_client(handler)
|
||||
asyncio.run(rerank.rerank("q", ["a"]))
|
||||
# svuota il cooldown: il primario torna eleggibile
|
||||
rerank._down_until.clear()
|
||||
asyncio.run(rerank.rerank("q", ["a"]))
|
||||
assert calls == ["primario", "fallback", "primario", "fallback"]
|
||||
|
||||
|
||||
def test_tutti_nodi_falliti_restifica_none(monkeypatch):
|
||||
_use_chain(monkeypatch)
|
||||
_mock_client(lambda request: httpx.Response(500))
|
||||
assert asyncio.run(rerank.rerank("q", ["a", "b"])) is None
|
||||
|
||||
|
||||
def test_enabled_richiede_catena(monkeypatch):
|
||||
_use_chain(monkeypatch, raw="")
|
||||
assert not rerank.enabled()
|
||||
_use_chain(monkeypatch)
|
||||
monkeypatch.setattr(rerank, "RERANK_ENABLED", True)
|
||||
assert rerank.enabled()
|
||||
monkeypatch.setattr(rerank, "RERANK_ENABLED", False)
|
||||
assert not rerank.enabled()
|
||||
|
||||
|
||||
def test_normalize_score():
|
||||
assert rerank.normalize_score(0.0) == pytest.approx(0.5)
|
||||
assert rerank.normalize_score(10.0) > 0.99
|
||||
assert rerank.normalize_score(-10.0) < 0.01
|
||||
|
||||
|
||||
def test_troncamento_documenti(monkeypatch):
|
||||
_use_chain(monkeypatch)
|
||||
seen: dict = {}
|
||||
|
||||
def handler(request):
|
||||
seen["docs"] = json.loads(request.content)["documents"]
|
||||
return httpx.Response(200, json={"results": [{"index": 0, "relevance_score": 1.0}]})
|
||||
|
||||
_mock_client(handler)
|
||||
asyncio.run(rerank.rerank("q", ["x" * 5000, "corto"]))
|
||||
assert len(seen["docs"][0]) == 800 # default RERANK_MAX_DOC_CHARS
|
||||
assert seen["docs"][1] == "corto"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# integrazione endpoint search
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_search_rerank_riordina(client, monkeypatch):
|
||||
for text in ["alpha", "beta", "gamma"]:
|
||||
r = client.post("/v1/memories", json=make_record(text=text), headers=auth_headers())
|
||||
assert r.status_code == 200
|
||||
|
||||
async def fake_rerank(query, docs):
|
||||
# inverte: beta (index 1) primo, poi gamma/alpha
|
||||
return [0.1, 0.9, 0.5], "finto", 12
|
||||
|
||||
monkeypatch.setattr(rerank, "enabled", lambda: True)
|
||||
monkeypatch.setattr(rerank, "rerank", fake_rerank)
|
||||
|
||||
resp = client.post("/v1/memories:search", json={"query": "q", "top_k": 3}, headers=auth_headers())
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["rerank"]["used"] is True
|
||||
assert data["rerank"]["backend"] == "finto"
|
||||
texts = [r["text"] for r in data["results"]]
|
||||
assert texts == ["beta", "gamma", "alpha"]
|
||||
assert data["results"][0]["rerank_score"] == pytest.approx(rerank.normalize_score(0.9), abs=0.01)
|
||||
|
||||
|
||||
def test_search_rerank_disattivato_per_query(client, monkeypatch):
|
||||
for text in ["alpha", "beta"]:
|
||||
client.post("/v1/memories", json=make_record(text=text), headers=auth_headers())
|
||||
|
||||
async def fail_rerank(query, docs):
|
||||
raise AssertionError("rerank non deve essere chiamato con rerank=false")
|
||||
|
||||
monkeypatch.setattr(rerank, "enabled", lambda: True)
|
||||
monkeypatch.setattr(rerank, "rerank", fail_rerank)
|
||||
|
||||
resp = client.post("/v1/memories:search", json={"query": "q", "top_k": 2, "rerank": False}, headers=auth_headers())
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["rerank"]["enabled"] is False
|
||||
|
||||
|
||||
def test_search_rerank_fallito_degrada_con_grazia(client, monkeypatch):
|
||||
for text in ["alpha", "beta"]:
|
||||
client.post("/v1/memories", json=make_record(text=text), headers=auth_headers())
|
||||
|
||||
async def fail_rerank(query, docs):
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(rerank, "enabled", lambda: True)
|
||||
monkeypatch.setattr(rerank, "rerank", fail_rerank)
|
||||
|
||||
resp = client.post("/v1/memories:search", json={"query": "q", "top_k": 2}, headers=auth_headers())
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["rerank"]["used"] is False
|
||||
assert "non raggiungibili" in data["rerank"]["reason"]
|
||||
assert len(data["results"]) == 2 # ordine di fusione preservato
|
||||
@@ -1,252 +0,0 @@
|
||||
"""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.8], "finto", 5 # sigmoid ≈ 0.858: ≥ 0.85 (suggest), < 0.90 (block)
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user