""" Ricerca ibrida in Qdrant con AgileV (contratto 1.2): denso + pesi lessicali nella stessa collection. Per chi ha già Qdrant (NIS2, LG231, taxai). Solo libreria standard: le chiamate sono le stesse REST che si fanno da PHP o da qualsiasi client; i corpi JSON qui sotto si copiano tali e quali. Chiave AgileV in AGILEV_KEY; Qdrant in QDRANT_URL (default http://127.0.0.1:6333), QDRANT_API_KEY se serve. python qdrant_ibrido.py indicizza doc1.txt doc2.txt ... # crea la collection e carica i punti python qdrant_ibrido.py cerca "art. 25-septies" # fusione RRF fatta da Qdrant python qdrant_ibrido.py cerca "art. 25-septies" --alfa 0.6 # fusione alfa fatta a mano python qdrant_ibrido.py cerca "art. 25-septies" --solo-denso # per confronto Cosa fa (provato su Qdrant 1.19 e sul corpus d.lgs. 231/2001, vedi NOVITA.md del 21/9/2026): 1. collection con DUE vettori per punto: "denso" (1024, Cosine) e "lessicale" (sparse, prodotto scalare); 2. ogni testo passa UNA volta da /v1/embeddings con return_sparse: il denso è identico a prima, i pesi lessicali vanno nel vettore sparso (indices/values: il formato di AgileV è già quello di Qdrant); 3. la domanda si trasforma allo stesso modo e si interroga con `prefetch` (un ramo denso, un ramo sparso) + `query: {"fusion": "rrf"}`: Qdrant fonde le due classifiche da solo. Oppure due query separate e alfa * coseno + (1 - alfa) * sparso / max_sparso a mano, come nell'esempio senza database. Chi ha già una collection con il solo denso: crea una collection nuova con i due vettori, reindicizza con return_sparse e sposta l'alias (come fatto l'11/9 per nis2_kb e lg231_docs). Il denso non cambia. """ import json, os, sys, urllib.request AGILEV_URL_DEFAULT = "https://api.agilev.agile.software/v1/embeddings" def dall_ambiente(nome): """Un valore che c'è ma è VUOTO è un errore, non un ripiego verso l'indirizzo pubblico.""" v = os.environ.get(nome) if v is None: return None, "assente" if not v.strip(): raise ValueError(f"{nome} c'e' ma e' vuota: toglila del tutto, oppure dalle un valore") return v.strip(), "ambiente" _u, _s = dall_ambiente("AGILEV_URL") AGILEV_URL = _u or AGILEV_URL_DEFAULT SORGENTE_URL = _s if _u else "predefinito" # `predefinito` = nessuno l'ha configurato: indirizzo PUBBLICO QDRANT_URL = os.environ.get("QDRANT_URL", "http://127.0.0.1:6333").rstrip("/") DENSO, LESSICALE = "denso", "lessicale" # nomi dei due vettori nella collection # --- corpi JSON delle chiamate Qdrant (copiabili in PHP) ----------------------------------------------- def corpo_collection(): """PUT /collections/: un vettore denso a 1024 con coseno e un vettore sparso.""" return {"vectors": {DENSO: {"size": 1024, "distance": "Cosine"}}, "sparse_vectors": {LESSICALE: {}}} def corpo_punti(ids, vettori, payloads): """PUT /collections//points: per ogni punto i due vettori; il payload è libero (id documento, titolo...).""" return {"points": [{"id": i, "vector": {DENSO: denso, LESSICALE: sparso}, "payload": p} for i, (denso, sparso), p in zip(ids, vettori, payloads)]} def corpo_query_rrf(denso, sparso, limite=10, candidati=50): """POST /collections//points/query: due rami di prefetch e fusione RRF fatta da Qdrant.""" return {"prefetch": [{"query": denso, "using": DENSO, "limit": candidati}, {"query": sparso, "using": LESSICALE, "limit": candidati}], "query": {"fusion": "rrf"}, "limit": limite, "with_payload": True} def corpo_query_uno(vettore, usando, limite=50): """POST /collections//points/query su un solo vettore (denso o lessicale).""" return {"query": vettore, "using": usando, "limit": limite, "with_payload": True} def fondi_alfa(densi, sparsi, alfa=0.6, limite=10): """ Fusione alfa a mano da due classifiche di Qdrant (liste di punti con id, score, payload). Lo sparso è normalizzato sul massimo della domanda. Un documento assente da una delle due liste prende il punteggio più basso visto in quella lista (denso) o 0 (sparso): è fuori dai candidati. """ d = {p["id"]: p["score"] for p in densi} s = {p["id"]: p["score"] for p in sparsi} pavimento = min(d.values()) if d else 0.0 massimo = max(s.values()) if s and max(s.values()) > 0 else 1.0 payload = {p["id"]: p.get("payload") for p in densi + sparsi} fusi = [{"id": i, "score": alfa * d.get(i, pavimento) + (1 - alfa) * s.get(i, 0.0) / massimo, "payload": payload[i]} for i in set(d) | set(s)] return sorted(fusi, key=lambda p: -p["score"])[:limite] # --- chiamate ------------------------------------------------------------------------------------------ def _http(metodo, url, corpo, intestazioni): req = urllib.request.Request(url, method=metodo, data=json.dumps(corpo).encode() if corpo is not None else None, headers={"Content-Type": "application/json", **intestazioni}) with urllib.request.urlopen(req, timeout=180) as r: return json.load(r) def qdrant(metodo, percorso, corpo=None): chiave = os.environ.get("QDRANT_API_KEY") return _http(metodo, QDRANT_URL + percorso, corpo, {"api-key": chiave} if chiave else {}) def embed(testi): """Chiama AgileV con return_sparse. Ritorna [(denso, {"indices": [...], "values": [...]}), ...].""" chiave = dall_ambiente("AGILEV_KEY")[0] or "" if not chiave: raise ValueError("chiave AgileV mancante: imposta AGILEV_KEY") dati = _http("POST", AGILEV_URL, {"model": "agilev-m3", "input": testi, "return_sparse": True}, {"Authorization": f"Bearer {chiave}"})["data"] return [(d["embedding"], d["sparse_embedding"]) for d in sorted(dati, key=lambda d: d["index"])] def indicizza(collection, percorsi, lotto=16): qdrant("PUT", f"/collections/{collection}", corpo_collection()) for inizio in range(0, len(percorsi), lotto): pezzo = percorsi[inizio:inizio + lotto] vettori = embed([open(p, encoding="utf-8").read() for p in pezzo]) qdrant("PUT", f"/collections/{collection}/points?wait=true", corpo_punti(range(inizio + 1, inizio + 1 + len(pezzo)), vettori, [{"file": p} for p in pezzo])) print(f"{len(percorsi)} punti in {collection}") def cerca(collection, domanda, modo="rrf", alfa=0.6, limite=10): denso, sparso = embed([domanda])[0] via = f"/collections/{collection}/points/query" if modo == "rrf": punti = qdrant("POST", via, corpo_query_rrf(denso, sparso, limite))["result"]["points"] elif modo == "alfa": densi = qdrant("POST", via, corpo_query_uno(denso, DENSO))["result"]["points"] sparsi = qdrant("POST", via, corpo_query_uno(sparso, LESSICALE))["result"]["points"] punti = fondi_alfa(densi, sparsi, alfa, limite) else: punti = qdrant("POST", via, corpo_query_uno(denso, DENSO, limite))["result"]["points"] return punti def main(argv): if len(argv) < 4: sys.exit(__doc__) comando, collection = argv[1], argv[2] if comando == "indicizza": indicizza(collection, argv[3:]) elif comando == "cerca": modo = "alfa" if "--alfa" in argv else "denso" if "--solo-denso" in argv else "rrf" alfa = float(argv[argv.index("--alfa") + 1]) if "--alfa" in argv else 0.6 for p in cerca(collection, argv[3], modo, alfa): print(f"{p['score']:.3f} {p['payload']}") else: sys.exit(__doc__) if __name__ == "__main__": main(sys.argv)