Matteo Benedetto
eba9c19179
- gateway: MemoryIn + payload + risultati search; 422 su valore invalido - estensione: confidence in qmem_store e qmem_correct (eredita dal superseduto), mostrato nei risultati - backward compatible: record esistenti → confidence null - verificato sul server: high/medium/default, 422 su invalida (istanza di test)
621 lines
24 KiB
Python
621 lines
24 KiB
Python
"""
|
||
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", "")
|
||
OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
|
||
EMBED_MODEL = os.environ.get("EMBED_MODEL", "bge-m3")
|
||
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"))
|
||
|
||
# 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"):
|
||
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())
|
||
try:
|
||
yield
|
||
finally:
|
||
cleanup_task.cancel()
|
||
try:
|
||
await cleanup_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
global _http
|
||
if _http is not None:
|
||
await _http.aclose()
|
||
_http = None
|
||
|
||
|
||
app = FastAPI(title="Memory Gateway", version="2.5.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
|
||
|
||
# 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()
|
||
|
||
|
||
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]:
|
||
r = await _get_http().post(
|
||
f"{OLLAMA_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:
|
||
qdrant.upsert(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)
|
||
|
||
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,
|
||
}
|
||
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"])
|
||
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),
|
||
)
|
||
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),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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)
|