enne2/pi-qmem
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: hybrid retrieval (BM25 + vettoriale, RRF) opt-in via hybrid=true

Browse Source 
- sparse vector bm25 (modifier IDF) + indice TEXT su text
- migrazione automatica: create_vector_name su collection esistente + backfill dei punti privi di sparse
- write: sparse vector a ogni upsert (fastembed Qdrant/bm25)
- search: hybrid=true → prefetch denso (min_score anti-rumore) + sparso, fusione RRF; default invariato (punteggi cosine)
- qdrant-client 1.13.0 → 1.19.0 (create_vector_name, query_points; search rimosso in 1.19)
- fastembed 0.5.1 + pre-download modello BM25 nel Dockerfile (appuser)
- estensione: parametro hybrid in qmem_search
- verificato sul server: migrazione + backfill OK, termine esatto trovato con RRF 1.0 (istanza di test)
This commit is contained in:
Matteo Benedetto
2026-08-16 19:41:42 +02:00
parent 2d354bfde6
commit 7abced1566
4 changed files with 144 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/main.py
+130 -10
View File
@@ -52,6 +52,19 @@ API_KEYS: set[str] = {k.strip() for k in os.environ.get("API_KEYS", "").split(",
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")
@@ -63,6 +76,9 @@ async def lifespan(_app: FastAPI):
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(
@@ -70,9 +86,25 @@ async def lifespan(_app: FastAPI):
field_name=field,
field_schema=qm.PayloadSchemaType.KEYWORD,
)
log.info("collection %s creata con indici", COLLECTION)
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:
@@ -173,6 +205,7 @@ class SearchIn(BaseModel):
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.")
# ---------------------------------------------------------------------------
@@ -239,6 +272,61 @@ async def embed(text: str) -> list[float]:
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
# ---------------------------------------------------------------------------
@@ -272,6 +360,7 @@ async def add_memory(
superseded_id = body.supersedes_id
vector = await embed(body.text)
sparse = _sparse_encode(body.text)
payload: dict[str, Any] = {
"text": body.text,
@@ -286,9 +375,12 @@ async def add_memory(
"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=vector, payload=payload)],
points=[qm.PointStruct(id=memory_id, vector=point_vector, payload=payload)],
)
_invalidate_meta()
@@ -326,14 +418,42 @@ async def search_memories(body: SearchIn, key: str = Depends(require_auth)) -> d
# default: esclude i record già corretti (superseded_by presente)
must.append(qm.IsEmptyCondition(is_empty=qm.PayloadField(key="superseded_by")))
hits = qdrant.search(
collection_name=COLLECTION,
query_vector=vector,
query_filter=qm.Filter(must=must) if must else None,
limit=body.top_k,
score_threshold=body.min_score, # filtra a livello motore: sotto soglia = rumore
with_payload=True,
)
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,