""" Memory Gateway — memoria centralizzata condivisa per agenti AI. Stack snello: FastAPI + Qdrant + Ollama (BGE-M3). Nessun LLM in scrittura. Accesso: UNA o più API key condivise con accesso COMPLETO in lettura e scrittura all'intera conoscenza. Nessun isolamento per agente: qualsiasi agente (attuale o futuro) con la chiave può consultare e aggiungere informazioni liberamente. L'agent_id è solo metadata di provenienza. Endpoints: POST /v1/memories → crea un record (con supersedes_id corregge un record esistente) POST /v1/memories:search → ricerca semantica con filtri (include_superseded per la lineage) GET /v1/memories/{id} → recupera per UUID DELETE /v1/memories/{id} → elimina per UUID GET /v1/meta/overview → discovery: scope×kind, progetti, agenti (cache 60s) GET /v1/status → health + statistiche """ from __future__ import annotations import asyncio import contextvars import hashlib import json import logging import os import time import uuid from collections import Counter from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import Any, Literal, Optional import httpx import uvicorn from fastapi import Depends, FastAPI, Header, HTTPException, Request from pydantic import BaseModel, Field, field_validator from qdrant_client import QdrantClient from qdrant_client.http import models as qm # --------------------------------------------------------------------------- # Configurazione (env) # --------------------------------------------------------------------------- QDRANT_URL = os.environ.get("QDRANT_URL", "http://127.0.0.1:6333") QDRANT_API_KEY = os.environ.get("QDRANT_API_KEY", "") # Backend embedding: ollama (default) | llamacpp (OpenAI-compatible /v1/embeddings) 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")) COLLECTION = os.environ.get("COLLECTION", "memories") # Chiavi condivise (separate da virgola): accesso completo in lettura/scrittura 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 di similarità pre-scrittura (v1, 2026-08-17) # Approccio a strati: hash esatto SHA-256 -> BLOCK; similarità semantica top-3 # (BGE-M3 cosine) -> BLOCK / WARN / ALLOW. Enforcement FUORI dall'LLM. GUARDRAIL_ENABLED = os.environ.get("GUARDRAIL_ENABLED", "true").lower() == "true" # Soglie cosine BGE-M3 (valori di partenza da calibrare sul corpus) GUARDRAIL_BLOCK_THRESHOLD = float(os.environ.get("GUARDRAIL_BLOCK_THRESHOLD", "0.85")) GUARDRAIL_WARN_THRESHOLD = float(os.environ.get("GUARDRAIL_WARN_THRESHOLD", "0.70")) GUARDRAIL_VERSION = "similarity-v1" # Versione del codice: hash del commit Git da cui è stato costruito il container # (iniettato come build arg nel Dockerfile: ARG GIT_COMMIT / ENV GIT_COMMIT) GIT_COMMIT = os.environ.get("GIT_COMMIT", "unknown").strip() GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.7.0").strip() # Metriche: push a VictoriaMetrics (stesso pattern dell'energy engine domotics) 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" _metrics: dict[str, Any] = { "requests": Counter(), "duration_sum": Counter(), "duration_count": Counter(), "errors": Counter(), "search_queries": 0, "search_hits": 0, } # Hybrid retrieval: sparse vector BM25 (Qdrant/bm25 via fastembed, modifier IDF) SPARSE_VECTOR_NAME = "bm25" try: from fastembed import SparseTextEmbedding _sparse_model: Optional[SparseTextEmbedding] = None SPARSE_AVAILABLE = True except Exception: # noqa: BLE001 _sparse_model = None SPARSE_AVAILABLE = False log = logging.getLogger("memory-gateway") log.warning("fastembed non disponibile: hybrid retrieval disattivato") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") log = logging.getLogger("memory-gateway") @asynccontextmanager async def lifespan(_app: FastAPI): """Startup/shutdown: crea collection e indici, avvia il cleanup periodico.""" collections = qdrant.get_collections().collections if not any(c.name == COLLECTION for c in collections): 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"): qdrant.create_payload_index( collection_name=COLLECTION, field_name=field, field_schema=qm.PayloadSchemaType.KEYWORD, ) 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) # Migrazione: aggiunge lo sparse vector se manca (collection pre-ibrida) info = 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: 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) _backfill_sparse() cleanup_task = asyncio.create_task(_cleanup_loop()) metrics_task = asyncio.create_task(_metrics_push_loop()) 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 global _http if _http is not None: await _http.aclose() _http = None app = FastAPI(title="Memory Gateway", version="2.7.0", lifespan=lifespan) qdrant = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY) # Request ID: generato per richiesta, loggato nell'audit e restituito in header _request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-") @app.middleware("http") async def request_id_middleware(request: Request, call_next): rid = request.headers.get("X-Request-ID") or str(uuid.uuid4()) _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): """Raccoglie conteggi, latenza ed errori per endpoint.""" start = time.monotonic() response = await call_next(request) dur = time.monotonic() - start route = request.scope.get("route") endpoint = route.path if route else request.url.path _metrics["requests"][endpoint] += 1 _metrics["duration_sum"][endpoint] += dur _metrics["duration_count"][endpoint] += 1 if response.status_code >= 400: _metrics["errors"][(endpoint, response.status_code)] += 1 return response # Rate limit in-memory: {key: [timestamps]} _ratelimit: dict[str, list[float]] = {} # Rate limit /v1/status (pubblico, per IP): {ip: [timestamps]} _STATUS_RATE_LIMIT_PER_MIN = 30 _status_ratelimit: dict[str, list[float]] = {} # Idempotency in-memory: {api_key:key: {hash, response, ts}} (TTL 24h) _IDEMPOTENCY_TTL_SECONDS = 24 * 3600 _idempotency: dict[str, dict[str, Any]] = {} def _payload_hash(body: MemoryIn) -> str: """Hash canonico del payload per il confronto idempotenza.""" canonical = json.dumps(body.model_dump(), sort_keys=True, default=str) return hashlib.sha256(canonical.encode()).hexdigest() def _idempotency_cleanup() -> None: """Rimuove le entry idempotenza scadute (lazy, chiamato a ogni write).""" now = time.time() expired = [k for k, v in _idempotency.items() if now - v["ts"] > _IDEMPOTENCY_TTL_SECONDS] for k in expired: _idempotency.pop(k, None) # Cache overview metadati (TTL 60s, invalidata su scrittura) _META_TTL_SECONDS = 60 _meta_cache: dict[str, Any] = {} def _invalidate_meta() -> None: _meta_cache.clear() # --------------------------------------------------------------------------- # Modelli # --------------------------------------------------------------------------- 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: high = verificato, medium = probabile, low = osservazione non confermata") expires_at: Optional[str] = None # ISO 8601 supersedes_id: Optional[str] = None supersede_reason: Optional[str] = Field(default=None, max_length=512) @field_validator("expires_at") @classmethod def _validate_expires_at(cls, v: Optional[str]) -> Optional[str]: """Valida il formato ISO 8601: invalida → 422 (niente fallback silenzioso).""" 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 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 = Field(default=False, description="True = hybrid retrieval (BM25 + vettoriale, RRF). I punteggi risultanti sono RRF, non cosine.") # --------------------------------------------------------------------------- # Auth: chiave condivisa → accesso completo (nessun isolamento) # --------------------------------------------------------------------------- def require_auth(x_api_key: str = Header(...)) -> str: if x_api_key not in API_KEYS: raise HTTPException(status_code=401, detail="API key non valida") # rate limit per chiave now = time.monotonic() window = _ratelimit.setdefault(x_api_key, []) window[:] = [t for t in window if now - t < 60] if len(window) >= 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: """Audit log in JSON lines (catturato da docker logs).""" entry = { "ts": datetime.now(timezone.utc).isoformat(), "key": key[:8] + "...", "action": action, "request_id": _request_id.get(), **extra, } log.info(json.dumps(entry, default=str)) def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() # --------------------------------------------------------------------------- # Guardrail di similarità pre-scrittura # --------------------------------------------------------------------------- def _normalize_text(text: str) -> str: """Normalizzazione canonica: lowercase, accenti rimossi, spazi normalizzati.""" import unicodedata 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: """SHA-256 del testo normalizzato (strato 1: duplicati esatti).""" return hashlib.sha256(_normalize_text(text).encode("utf-8")).hexdigest() def _find_similar(text: str, vector: list[float], top_k: int = 3) -> list[dict]: """Top-k record attivi più simili (esclude i superseded).""" qfilter = qm.Filter(must=[qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by"))]) hits = qdrant.query_points( collection_name=COLLECTION, query=vector, query_filter=qfilter, limit=top_k, with_payload=True, ).points return [ { "memory_id": h.id, "score": round(float(h.score), 4), "text": (h.payload or {}).get("text", ""), "kind": (h.payload or {}).get("kind", ""), "project_id": (h.payload or {}).get("project_id", ""), } for h in hits ] def _decide_guardrail(text: str, vector: list[float]) -> dict: """Applica il guardrail a 2 strati. Ritorna {decision, reason, matches}.""" # Strato 1 — hash esatto (duplicato identico) text_hash = _text_hash(text) qfilter = qm.Filter( must=[ qm.FieldCondition(key="text_hash", match=qm.MatchValue(value=text_hash)), qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")), ] ) exact = qdrant.query_points( collection_name=COLLECTION, query=vector, query_filter=qfilter, limit=1, with_payload=True, ).points if exact: return { "decision": "BLOCK", "reason": "EXACT_DUPLICATE", "matches": [{"memory_id": exact[0].id, "score": 1.0}], } # Strato 2 — similarità semantica top-3 matches = _find_similar(text, vector, top_k=3) if not matches: return {"decision": "ALLOW", "reason": "NO_CANDIDATE", "matches": []} top1 = matches[0]["score"] if top1 >= GUARDRAIL_BLOCK_THRESHOLD: return {"decision": "BLOCK", "reason": "KNOWN_SOLUTION", "matches": matches} if top1 >= GUARDRAIL_WARN_THRESHOLD: return {"decision": "WARN", "reason": "MODERATE_SIMILARITY", "matches": matches} return {"decision": "ALLOW", "reason": "NEW_SOLUTION", "matches": matches} def _parse_ts(value: Optional[str]) -> Optional[float]: """Converte ISO 8601 in timestamp Unix (per i range query Qdrant).""" if not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() except ValueError: return None # --------------------------------------------------------------------------- # Embedding via Ollama (BGE-M3) — client httpx riusato (creato lazy, chiuso a shutdown) # --------------------------------------------------------------------------- _http: Optional[httpx.AsyncClient] = None def _get_http() -> httpx.AsyncClient: global _http if _http is None: _http = httpx.AsyncClient(timeout=30) return _http async def embed(text: str) -> list[float]: if EMBED_API == "llamacpp": # llama.cpp: OpenAI-compatible /v1/embeddings headers = {"Content-Type": "application/json"} if EMBED_API_KEY: headers["Authorization"] = f"Bearer {EMBED_API_KEY}" r = await _get_http().post( f"{EMBED_URL}/v1/embeddings", json={"model": EMBED_MODEL, "input": text}, headers=headers, ) r.raise_for_status() return r.json()["data"][0]["embedding"] # Ollama (default) r = await _get_http().post( f"{EMBED_URL}/api/embed", json={"model": EMBED_MODEL, "input": text}, ) r.raise_for_status() return r.json()["embeddings"][0] # --------------------------------------------------------------------------- # Sparse encoding (BM25) per hybrid retrieval # --------------------------------------------------------------------------- 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]: """Vettore sparso BM25 per il testo (None se fastembed non disponibile).""" 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() -> None: """Migrazione: aggiunge il vettore sparso ai punti esistenti che ne sono privi.""" 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) # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @app.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: # Idempotency: replay della stessa richiesta (stessa key + stesso payload) → stessa risposta idem_key = f"{key}:{idempotency_key}" if idempotency_key else None if idem_key: _idempotency_cleanup() existing = _idempotency.get(idem_key) if existing: if existing["hash"] != _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()) # Supersede: il nuovo record corregge uno esistente, che resta in archivio marcato superseded_id: Optional[str] = None if body.supersedes_id: old = 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 vector = await embed(body.text) sparse = _sparse_encode(body.text) # Guardrail di similarità pre-scrittura (enforcement FUORI dall'LLM). # Il supersede esplicito è una correzione intenzionale: bypassa il guardrail. guardrail: Optional[dict] = None if GUARDRAIL_ENABLED and not body.supersedes_id: guardrail = _decide_guardrail(body.text, vector) if guardrail["decision"] == "BLOCK": _audit( key, "create_blocked", kind=body.kind, agent_id=body.agent_id or "shared", reason=guardrail["reason"], matches=[m["memory_id"] for m in guardrail["matches"]], ) raise HTTPException( status_code=409, detail={ "error": "duplicate_memory", "reason": guardrail["reason"], "matches": 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, "created_at": _now_iso(), "expires_at": _parse_ts(body.expires_at), "supersedes_id": superseded_id, "supersede_reason": body.supersede_reason, "embedding_model": EMBED_MODEL, "text_hash": _text_hash(body.text), } if guardrail: payload["guardrail"] = { "version": GUARDRAIL_VERSION, "decision": guardrail["decision"], "reason": guardrail["reason"], "matches": guardrail["matches"], } point_vector: dict[str, Any] = {"": vector} if sparse is not None: point_vector[SPARSE_VECTOR_NAME] = sparse qdrant.upsert( collection_name=COLLECTION, points=[qm.PointStruct(id=memory_id, vector=point_vector, payload=payload)], ) _invalidate_meta() if superseded_id: 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"]) 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} if idem_key: _idempotency[idem_key] = {"hash": _payload_hash(body), "response": response, "ts": time.time()} return response @app.post("/v1/memories:search") async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> dict: vector = await embed(body.query) must: list[Any] = [] if body.kind: must.append(qm.FieldCondition(key="kind", match=qm.MatchValue(value=body.kind))) if body.project_id: must.append(qm.FieldCondition(key="project_id", match=qm.MatchValue(value=body.project_id))) if body.scope: must.append(qm.FieldCondition(key="scope", match=qm.MatchValue(value=body.scope))) if not body.include_superseded: # default: esclude i record già corretti (superseded_by presente) must.append(qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by"))) qfilter = qm.Filter(must=must) if must else None if body.hybrid and SPARSE_AVAILABLE: # Hybrid retrieval: BM25 (sparso) + vettoriale, fusione RRF. # min_score applicato al prefetch denso (preserva la semantica anti-rumore); # i punteggi risultanti sono RRF, non cosine. sparse = _sparse_encode(body.query) if sparse is not None: hits = qdrant.query_points( collection_name=COLLECTION, prefetch=[ qm.Prefetch(query=vector, using="", limit=body.top_k * 4, score_threshold=body.min_score), qm.Prefetch(query=sparse, using=SPARSE_VECTOR_NAME, limit=body.top_k * 4), ], query=qm.FusionQuery(fusion=qm.Fusion.RRF), query_filter=qfilter, limit=body.top_k, with_payload=True, ).points else: hits = qdrant.query_points( collection_name=COLLECTION, query=vector, query_filter=qfilter, limit=body.top_k, score_threshold=body.min_score, with_payload=True, ).points else: hits = qdrant.query_points( collection_name=COLLECTION, query=vector, query_filter=qfilter, limit=body.top_k, score_threshold=body.min_score, # filtra a livello motore: sotto soglia = rumore with_payload=True, ).points results = [ { "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"), "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"), } for h in hits ] _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), ) _metrics["search_queries"] += 1 _metrics["search_hits"] += len(results) return {"results": results, "min_score": body.min_score, "total_hits": len(results)} @app.get("/v1/memories/{memory_id}") async def get_memory(memory_id: str, key: str = Depends(require_auth)) -> dict: point = qdrant.retrieve( collection_name=COLLECTION, ids=[memory_id], with_payload=True ) 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} @app.delete("/v1/memories/{memory_id}") async def delete_memory(memory_id: str, key: str = Depends(require_auth)) -> dict: point = qdrant.retrieve( collection_name=COLLECTION, ids=[memory_id], with_payload=True ) if not point: raise HTTPException(status_code=404, detail="Memoria non trovata") qdrant.delete(collection_name=COLLECTION, points_selector=[memory_id]) _invalidate_meta() _audit(key, "delete", memory_id=memory_id) return {"deleted": memory_id} @app.get("/v1/meta/overview") async def meta_overview(key: str = Depends(require_auth)) -> dict: """Panoramica della memoria: scope×kind con conteggi, progetti, agenti, superseduti. Usata dalla discovery per la ricerca settorializzata. Cache TTL 60s invalidata su write.""" now = time.time() cached = _meta_cache.get("overview") if cached and now - cached["ts"] < _META_TTL_SECONDS: _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 = qdrant.scroll( collection_name=COLLECTION, limit=1000, with_payload=["scope", "kind", "project_id", "agent_id", "superseded_by"], with_vectors=False, offset=offset, ) for p in points: pl = p.payload total += 1 s = pl.get("scope", "agent") k = pl.get("kind", "fact") scope_kinds.setdefault(s, Counter())[k] += 1 if pl.get("project_id"): projects[pl["project_id"]] += 1 agents[pl.get("agent_id", "shared")] += 1 if pl.get("superseded_by"): superseded += 1 if not next_offset: break offset = next_offset data = { "scopes": [ {"scope": s, "count": sum(c.values()), "kinds": [{"kind": k, "count": v} for k, v in sorted(c.items())]} for s, c in sorted(scope_kinds.items()) ], "projects": [{"project_id": pid, "count": c} for pid, c in projects.most_common()], "agents": [{"agent_id": aid, "count": c} for aid, c in agents.most_common()], "superseded": superseded, "total": total, } _meta_cache["overview"] = {"ts": now, "data": data} _audit(key, "meta", cached=False, total=total) return {**data, "cached": False} @app.get("/v1/status") async def status(request: Request) -> dict: # Endpoint pubblico (healthcheck): rate limit leggero per IP ip = request.client.host if request.client else "unknown" now = time.monotonic() window = _status_ratelimit.setdefault(ip, []) window[:] = [t for t in window if now - t < 60] if len(window) >= _STATUS_RATE_LIMIT_PER_MIN: raise HTTPException(status_code=429, detail="Rate limit superato") window.append(now) info = 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": GIT_COMMIT, "guardrail_version": GUARDRAIL_VERSION, } @app.get("/v1/version") async def version() -> dict: """Versione del codice in esecuzione: hash del commit Git da cui è stato creato il container Docker. Endpoint pubblico (nessun dato sensibile), utile per verificare programmaticamente l'allineamento del deploy.""" return { "version": GATEWAY_VERSION, "git_commit": 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, } @app.get("/v1/metrics") async def metrics(key: str = Depends(require_auth)) -> dict: """Riepilogo metriche in-memory (per verifica manuale; il push a VM è automatico).""" try: info = qdrant.get_collection(COLLECTION) points = info.points_count except Exception: # noqa: BLE001 points = None return { "requests": dict(_metrics["requests"]), "avg_duration_ms": { e: round(_metrics["duration_sum"][e] / _metrics["duration_count"][e] * 1000, 2) for e in _metrics["duration_count"] }, "errors": {f"{e}:{s}": c for (e, s), c in _metrics["errors"].items()}, "search_queries": _metrics["search_queries"], "search_hits": _metrics["search_hits"], "points": points, } def _prometheus_lines() -> list[str]: """Metriche in formato Prometheus text (senza timestamp, aggiunto dal push).""" lines: list[str] = [] for endpoint, count in _metrics["requests"].items(): lines.append(f'qmem_requests_total{{endpoint="{endpoint}"}} {count}') for endpoint, s in _metrics["duration_sum"].items(): c = _metrics["duration_count"][endpoint] lines.append(f'qmem_request_duration_seconds_sum{{endpoint="{endpoint}"}} {s:.6f}') lines.append(f'qmem_request_duration_seconds_count{{endpoint="{endpoint}"}} {c}') 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']}") try: info = qdrant.get_collection(COLLECTION) lines.append(f"qmem_points {info.points_count}") except Exception: # noqa: BLE001 pass return lines async def _metrics_push_loop() -> None: """Push periodico delle metriche a VictoriaMetrics (formato Prometheus + timestamp ms).""" while True: try: now_ms = int(time.time() * 1000) body = "\n".join(f"{l} {now_ms}" for l in _prometheus_lines()) + "\n" r = await _get_http().post( VM_PUSH_URL, content=body, headers={"Content-Type": "text/plain"}, ) if r.status_code >= 300: log.warning("metrics push: HTTP %s", r.status_code) except Exception as e: # noqa: BLE001 log.warning("metrics push error: %s", e) await asyncio.sleep(VM_PUSH_INTERVAL) return { "status": "ok", "collection": COLLECTION, "points": info.points_count, "embedding_model": EMBED_MODEL, "embedding_dim": EMBED_DIM, "access": "shared", "api_keys": len(API_KEYS), } # --------------------------------------------------------------------------- # Cleanup periodico: rimuove record scaduti (expires_at < now) # --------------------------------------------------------------------------- async def _cleanup_loop() -> None: while True: try: now = time.time() scroll = qdrant.scroll( collection_name=COLLECTION, scroll_filter=qm.Filter( must=[ qm.FieldCondition( key="expires_at", range=qm.Range(lt=now), ) ] ), limit=100, with_payload=False, ) ids = [p.id for p 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 e: # noqa: BLE001 log.warning("cleanup error: %s", e) await asyncio.sleep(3600) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8080)