""" AgileV per LangChain (contratto 1.2): tre pezzi, ognuno usabile da solo. AgileVEmbeddings langchain_core.embeddings.Embeddings sopra /v1/embeddings (denso, 1024 dim). Con return_sparse=True chiede in una sola chiamata anche i pesi lessicali e li tiene da parte per AgileVSparseEmbeddings: così la ricerca ibrida costa UNA passata del modello per testo, non due. AgileVSparseEmbeddings i pesi lessicali (sparse_embedding) nel formato che langchain-qdrant si aspetta da un SparseEmbeddings (SparseVector con indices/values): serve per QdrantVectorStore(..., retrieval_mode=RetrievalMode.HYBRID). AgileVRerank BaseDocumentCompressor sopra /v1/rerank: riordina i candidati di un retriever (ContextualCompressionRetriever). Legge di ogni documento solo i primi max_caratteri (800 di default): misurato sul corpus vero, a 800 caratteri 10 candidati costano la metà (6 s invece di 13 su 4 vCPU) senza perdere nulla (nDCG@10 0,814 contro 0,816); vedi NOVITA.md del 21/9/2026. AgileVSpezza lo splitter: split_text / create_documents / split_documents come un TextSplitter di LangChain, ma con spezza() del client di riferimento (agilev_client.py, da tenere nella stessa cartella): frammenti da 1.500 caratteri a righe/frasi/parole, 150 di sovrapposizione a frasi intere, e l'intestazione presa da un campo dei metadati (es. "titolo") ripetuta in testa a ogni frammento: è ciò che fa ritrovare un articolo citato per numero. RecursiveCharacterTextSplitter non lo fa. Dipendenze: langchain-core (facoltativa per gli embedding, necessaria per il compressore); requests (facoltativa: se c'è, la connessione TLS viene riusata, ~250 ms in meno a chiamata). Chiave in AGILEV_KEY, indirizzo in AGILEV_URL (default https://api.agilev.agile.software). Esempio (ricerca ibrida in Qdrant, come nei docs di langchain-qdrant): from langchain_qdrant import QdrantVectorStore, RetrievalMode emb = AgileVEmbeddings(return_sparse=True) docs = AgileVSpezza(intestazione_da="titolo").split_documents(documenti_interi) # metadata["titolo"] in testa a ogni frammento store = QdrantVectorStore.from_documents(docs, embedding=emb, sparse_embedding=emb.sparse(), retrieval_mode=RetrievalMode.HYBRID, url="http://localhost:6333", collection_name="nis2_kb", vector_name="denso", sparse_vector_name="lessicale") retriever = ContextualCompressionRetriever(base_compressor=AgileVRerank(top_n=5), base_retriever=store.as_retriever(search_kwargs={"k": 10})) """ from __future__ import annotations import hashlib, ipaddress, json, os, time, urllib.parse, urllib.request, urllib.error from collections import OrderedDict from typing import Any, List, Optional, Sequence try: from langchain_core.embeddings import Embeddings except ImportError: # senza LangChain le classi funzionano lo stesso class Embeddings: # type: ignore[no-redef] pass try: # il tipo che langchain-qdrant usa per i vettori sparsi from langchain_qdrant.sparse_embeddings import SparseEmbeddings, SparseVector except ImportError: class SparseVector: # type: ignore[no-redef] """Stessa forma di langchain_qdrant.SparseVector: indices e values, già nel formato di Qdrant.""" def __init__(self, indices: List[int], values: List[float]): self.indices, self.values = indices, values def __eq__(self, altro): return isinstance(altro, SparseVector) and (self.indices, self.values) == (altro.indices, altro.values) def __repr__(self): return f"SparseVector(indices={self.indices!r}, values={self.values!r})" class SparseEmbeddings: # type: ignore[no-redef] pass URL_DEFAULT = "https://api.agilev.agile.software" def dall_ambiente(nome: str) -> "tuple[Optional[str], str]": """ Un valore che c'è ma è VUOTO è un ERRORE, non un ripiego verso l'indirizzo pubblico: qualcuno l'ha impostato e non è arrivato. Stessa regola del client di riferimento (agilev_client.py). """ 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" def in_chiaro_senza_internet(ospite: str) -> bool: """`http` (non cifrato) verso questo ospite resta dentro casa? Gemello di agilev_client.py. Vero per il loopback, per le reti private e per il NOME `localhost`; falso per tutto il resto, nome pubblico compreso. NESSUNA risoluzione di nomi: un nome non si sa dove guarda, e risolverlo dentro un costruttore è una chiamata di rete che chi legge non si aspetta. """ if ospite.lower() in ("localhost", "localhost.localdomain"): return True try: indirizzo = ipaddress.ip_address(ospite) except ValueError: # non è un indirizzo: è un nome, e non lo risolviamo return False return bool(indirizzo.is_private or indirizzo.is_loopback) LOTTO_MAX, CARATTERI_MAX = 128, 32000 # i due tetti del servizio (API.md, «Limiti») class AgileVErrore(RuntimeError): """Errore restituito dal servizio: stato HTTP, codice (KEY_INVALID, RATE_LIMIT, ...) e messaggio.""" def __init__(self, stato: int, codice: str, messaggio: str): super().__init__(f"AgileV {stato} {codice}: {messaggio}") self.stato, self.codice, self.messaggio = stato, codice, messaggio # I DUE TETTI CONTROLLATI PRIMA DI SPEDIRE, in un posto solo — come `_tetti()` nel client di # riferimento e come `norm_inputs` in app.py dalla parte del servizio. È la regola 3 dell'elenco # che agilev_client.py porta in intestazione, la sola delle nove che questo modulo non applicava: # fino al 10/10 il numero dei `documents` di un rerank e la lunghezza di ogni testo partivano # sempre, e il cliente pagava una richiesta — più un posto del minuto, che `auth()` consuma prima # di qualunque controllo sull'input — per farsi dire un numero scritto qui sopra. # # `campo` NOMINA il campo come fa il servizio, perché /v1/rerank ne ha due che portano testo e un # 413 che non dice quale ha sforato lascia solo la strada di indovinare. Senza `campo` il messaggio # resta quello del client di riferimento, parola per parola: chi tiene i due moduli nello stesso # progetto deve leggere la stessa frase per lo stesso fatto. def _tetti(testi: Sequence[str], *, quanti_al_massimo: Optional[int] = None, campo: str = "") -> None: dove = f" (campo `{campo}`)" if campo else "" if quanti_al_massimo is not None and len(testi) > quanti_al_massimo: raise AgileVErrore(413, "BATCH_TOO_LARGE", f"massimo {quanti_al_massimo} testi per richiesta{dove}") for i, t in enumerate(testi): if not isinstance(t, str): raise AgileVErrore(400, "BAD_INPUT", f"il testo {i} non e' una stringa{dove}") if len(t) > CARATTERI_MAX: raise AgileVErrore(413, "TEXT_TOO_LONG", f"il testo {i} ha {len(t)} caratteri, massimo {CARATTERI_MAX}: spezzatelo{dove}") class _Client: """POST JSON verso AgileV. Con `requests` riusa la connessione (keep-alive), altrimenti urllib.""" def __init__(self, chiave: Optional[str] = None, url: Optional[str] = None, timeout: float = 180.0, in_chiaro_ammesso: bool = False, tentativi: int = 5, attesa: float = 1.0, dormi=None): k, _ = dall_ambiente("AGILEV_KEY") self.chiave = chiave or k or "" if not self.chiave: raise ValueError("chiave AgileV mancante: passa chiave=... o imposta AGILEV_KEY") if url is not None and not url.strip(): raise ValueError("indirizzo AgileV vuoto: passa None per il predefinito, non una stringa vuota") a, sorgente = dall_ambiente("AGILEV_URL") self.sorgente_indirizzo = "argomento" if url else (sorgente if a else "predefinito") self.url = (url or a or URL_DEFAULT).rstrip("/") # La guardia sul CHIARO, come nel client di riferimento: il chiaro verso casa resta possibile, # il chiaro verso un indirizzo che esce no — lì il rimando del bordo arriva DOPO i byte. Qui # conta il doppio: con `requests` quel rimando lo seguiva la libreria, e il chiamante vedeva 200. u = urllib.parse.urlsplit(self.url) ospite = u.hostname or "" if u.scheme != "https" and not in_chiaro_ammesso and not in_chiaro_senza_internet(ospite): raise ValueError( f"indirizzo AgileV in chiaro verso «{ospite or self.url}»: la chiave e i vostri testi " "partirebbero NON cifrati. Il bordo risponde a http con un rimando a https, ma quel " "rimando arriva DOPO: quando lo leggete, chiave e testi hanno gia' attraversato Internet. " f"Usate {URL_DEFAULT} (https), oppure in_chiaro_ammesso=True se quel filo non passa da " "Internet — loopback, reti private e localhost sono gia' ammessi senza dire niente") self.timeout = timeout # Il freno e i ritentativi (regole 4 e 8 del client di riferimento, contratto 1.2). Lo stato sta # sull'ISTANZA e non in un `threading.local()`: un `_Client` è condiviso fra AgileVEmbeddings e # il suo compagno sparso, e un thread-local in una piscina di fili dà un filo nuovo a ogni # richiesta quando sono poche e lo riusa sotto carico — cioè si comporta bene nel banco e male # in produzione. self.tentativi = max(1, int(tentativi)) self.attesa = float(attesa) self._dormi = dormi or time.sleep self.contatori = {"richieste": 0, "ritentate": 0, "frenate": 0} self.rate: dict = {} # ultime X-RateLimit-* lette: limit, remaining, reset self._freno: Optional[float] = None # istante (monotonic) prima del quale non si spedisce self._retry_after: Optional[float] = None try: import requests self._sessione = requests.Session() except ImportError: self._sessione = None # ---------------------------------------------------------------- un giro HTTP, i due rami def _scambio(self, metodo: str, percorso: str, corpo: Optional[dict], con_chiave: bool): """(stato, intestazioni in minuscolo, testo). I due rami vanno tenuti ENTRAMBI: il ramo urllib è quello che gira quando `requests` non è installata, cioè proprio quando non c'è niente che freni per noi.""" intestazioni = {"Content-Type": "application/json"} if corpo is not None else {} if con_chiave: intestazioni["Authorization"] = f"Bearer {self.chiave}" self.contatori["richieste"] += 1 if self._sessione is not None: # allow_redirects=False: un rimando su una POST rispedisce il corpo, e il corpo è il vostro # testo. `requests` i rimandi li segue da sé, e li seguiva: il ramo REDIRECT di `_errore` # era codice che su questa strada non si raggiungeva mai, e una fuga in chiaro verso la # porta 80 arrivava al chiamante come un 200. Gli altri tre client non li seguono dal primo # giorno; qui lo si dichiara anche sulla GET, perché la regola è la stessa. if corpo is None: r = self._sessione.get(self.url + percorso, headers=intestazioni, timeout=self.timeout, allow_redirects=False) else: r = self._sessione.post(self.url + percorso, json=corpo, headers=intestazioni, timeout=self.timeout, allow_redirects=False) return r.status_code, {k.lower(): v for k, v in r.headers.items()}, r.text dati = json.dumps(corpo).encode() if corpo is not None else None req = urllib.request.Request(self.url + percorso, data=dati, headers=intestazioni, method=metodo) try: with urllib.request.urlopen(req, timeout=self.timeout) as r: return r.status, {k.lower(): v for k, v in r.headers.items()}, r.read().decode() except urllib.error.HTTPError as e: # `HTTPError` È una risposta: porta `.headers`, e sono le intestazioni su cui si frena e si # legge Retry-After. Leggerle solo sul 200 vorrebbe dire non frenare mai, perché le # intestazioni che contano arrivano proprio col rifiuto. return e.code, {k.lower(): v for k, v in (e.headers or {}).items()}, e.read().decode() def _leggi_rate(self, intestazioni: dict) -> None: """X-RateLimit-* (contratto 1.2): se il minuto è esaurito arma il freno fino al minuto nuovo.""" try: limite, rimaste, reset = (float(intestazioni[k]) for k in ("x-ratelimit-limit", "x-ratelimit-remaining", "x-ratelimit-reset")) except (KeyError, ValueError, TypeError): return self.rate = {"limit": limite, "remaining": rimaste, "reset": reset} self._freno = time.monotonic() + min(60.0, reset) if rimaste <= 0 else None def _frena(self) -> None: """Regola 8: se l'ultima risposta diceva 0 richieste rimaste, si aspetta il minuto nuovo PRIMA di spedire, invece di spedire, prendere un 429 e aspettare comunque.""" if self._freno is None: return resto = self._freno - time.monotonic() self._freno = None if resto > 0: self.contatori["frenate"] += 1 self._dormi(resto) def _chiama(self, metodo: str, percorso: str, corpo: Optional[dict], con_chiave: bool) -> dict: """Regola 4: si ritenta SOLO su 429 RATE_LIMIT e sui 5xx, con attesa crescente (tetto 30 s) o ESATTAMENTE `Retry-After` quando c'è (tetto 60 s). Tutto il resto — 400, 401, 403, 413, il 429 QUOTA_EXCEEDED, un 200 di forma sbagliata — è definitivo e risale subito a chi chiama. Il timeout NON si ritenta qui, ed è una scelta: il rimedio del timeout è dimezzare il LOTTO (regola 5), e il lotto lo decide chi chiama (AgileVEmbeddings), non questo trasporto; ripetere la stessa richiesta pesante sarebbe il caso peggiore di settembre, 47 lotti identici in coda.""" ultimo: Optional[AgileVErrore] = None for tentativo in range(self.tentativi): if tentativo: self.contatori["ritentate"] += 1 if self._retry_after is not None: attesa = min(60.0, self._retry_after) # il servizio sa quando tornare else: attesa = min(30.0, self.attesa * 2 ** (tentativo - 1)) self._dormi(attesa) self._retry_after = None if con_chiave: self._frena() stato, intestazioni, testo = self._scambio(metodo, percorso, corpo, con_chiave) self._leggi_rate(intestazioni) if stato == 200: return json.loads(testo) codice, messaggio = _errore(stato, testo) ultimo = AgileVErrore(stato, codice, messaggio) if (stato == 429 and codice == "RATE_LIMIT") or stato >= 500: ra = intestazioni.get("retry-after") self._retry_after = (float(ra) if ra and str(ra).strip().replace(".", "", 1).isdigit() else None) self._freno = None # l'attesa del ritentativo copre già il minuto: non due volte continue raise ultimo assert ultimo is not None raise ultimo def get(self, percorso: str) -> dict: """GET senza chiave: serve solo per /health, che non vuole la chiave e non consuma nulla.""" return self._chiama("GET", percorso, None, False) def post(self, percorso: str, corpo: dict) -> dict: return self._chiama("POST", percorso, corpo, True) SENZA_FORMA = { 404: ("NOT_FOUND", "percorso inesistente: controllate l'URL di base e la rotta"), 405: ("BAD_METHOD", "metodo HTTP sbagliato per questa rotta (gli embedding vogliono POST)"), } def _senza_codice(stato: int) -> tuple: """La risposta non porta un codice nostro: lo mette il client, dallo STATO soltanto. Il corpo non si legge e non entra nel messaggio — non è di AgileV (FastAPI, il bordo, un proxy) e nessuno ha promesso cosa ci metterà dentro.""" if stato in SENZA_FORMA: return SENZA_FORMA[stato] if 300 <= stato < 400: # Regola 5, ramo `3xx` (1/10). NON è un errore: è la stessa risposta all'indirizzo giusto. # Lo scrive il framework quando il percorso ha una BARRA IN FONDO (`/v1/embeddings/`), e il # `307` conserva metodo e corpo. Questo client non segue i rimandi apposta: un rimando su una # POST rispedisce il corpo, e il corpo è il vostro testo. Seguirlo sarebbe sicuro — il # `Location` è assoluto e in `https`, misurato — ma non scrivere la barra lo è di più. # Non consuma nulla: il rimando non arriva al nostro codice (misurato: 0 richieste contate). return "REDIRECT", f"rimando ({stato}): togliete la barra finale dall'URL, oppure seguite Location" if stato >= 500: return "UPSTREAM_ERROR", f"guasto senza forma ({stato}): bordo o servizio, ritentabile" return "NO_CODE", f"risposta {stato} senza un codice del contratto" def _errore(stato: int, testo: str) -> tuple: """Codice e messaggio da mostrare. Il 422 non ha `code` (è il corpo di FastAPI) e RIMANDA INDIETRO I TESTI INVIATI in detail[].input: si legge solo QUALE campo è sbagliato, mai l'eco. Tutto il resto che non ha la forma del contratto passa da `_senza_codice`: niente esce di qui col codice `?`.""" try: e = json.loads(testo) except ValueError: return _senza_codice(stato) if isinstance(e, dict) and isinstance(e.get("detail"), list): return "BAD_BODY", _campi(e["detail"]) # codice messo dal client: il 422 non ne manda try: e = e.get("detail", e).get("error", {}) codice = e.get("code") except AttributeError: return _senza_codice(stato) if not isinstance(e, dict) or not codice: return _senza_codice(stato) return str(codice), str(e.get("message", "")) def _campi(dettagli: list, massimo: int = 5) -> str: """`body.input: Field required` — solo loc/msg/type del 422; detail[].input (l'eco) non si legge.""" pezzi = [] for d in dettagli[:massimo]: if not isinstance(d, dict): continue dove = ".".join(str(x) for x in d.get("loc") or ()) pezzi.append(f"{dove}: {d.get('msg') or d.get('type') or '?'}".lstrip(": ")) if not pezzi: return "corpo della richiesta non valido" resto = f" (+{len(dettagli) - massimo} altri)" if len(dettagli) > massimo else "" return "corpo della richiesta non valido - " + "; ".join(pezzi) + resto def _dati(risposta: Any, attesi: int, sparso: bool, dim: Optional[int] = None) -> List[dict]: """`data` in ordine di `index`, con TANTI elementi quanti i testi inviati. Un 200 di forma inattesa diventa un errore BAD_RESPONSE invece di una lista più corta: il vector store accoppierebbe i testi ai vettori sbagliati senza accorgersene (e con zip() li perderebbe in silenzio). `dim` (fissata da AgileVEmbeddings.controlla_dim()) fa lo stesso sulla LUNGHEZZA: un vettore della misura sbagliata alza la voce invece di entrare nella collezione o di essere scartato in silenzio.""" dati = risposta.get("data") if isinstance(risposta, dict) else None if not isinstance(dati, list) or len(dati) != attesi: quanti = len(dati) if isinstance(dati, list) else "nessun elenco" raise AgileVErrore(200, "BAD_RESPONSE", f"attesi {attesi} elementi in data, ricevuti {quanti}") for d in dati: if not (isinstance(d, dict) and isinstance(d.get("embedding"), list)): raise AgileVErrore(200, "BAD_RESPONSE", "un elemento di data non ha embedding") if sparso and not isinstance(d.get("sparse_embedding"), dict): raise AgileVErrore(200, "BAD_RESPONSE", "chiesto return_sparse, ma un elemento non ha sparse_embedding") lunghezze = sorted({len(d["embedding"]) for d in dati}) if len(lunghezze) > 1: raise AgileVErrore(200, "BAD_RESPONSE", f"i vettori del lotto hanno lunghezze diverse: {lunghezze}") if dim is not None and lunghezze and lunghezze != [dim]: raise AgileVErrore(200, "BAD_RESPONSE", f"attesi vettori da {dim} valori, ricevuti da {lunghezze[0]}") indici = sorted(i for i in (d.get("index") for d in dati) if isinstance(i, int) and not isinstance(i, bool)) if indici != list(range(attesi)): raise AgileVErrore(200, "BAD_RESPONSE", f"gli index di data non sono 0..{attesi - 1}") return sorted(dati, key=lambda d: d["index"]) def _impronta(testo: str) -> str: return hashlib.sha256(testo.encode("utf-8")).hexdigest() class AgileVEmbeddings(Embeddings): """Embedding denso di AgileV (agilev-m3, 1024 dimensioni, già normalizzato) per LangChain. return_sparse=True: ogni chiamata chiede anche i pesi lessicali e li mette da parte (per impronta sha256 del testo, al massimo `memoria` voci) così che AgileVSparseEmbeddings non rifaccia la chiamata. """ def __init__(self, chiave: Optional[str] = None, url: Optional[str] = None, model: str = "agilev-m3", return_sparse: bool = False, lotto: int = 64, timeout: float = 180.0, memoria: int = 10000): self._client = _Client(chiave, url, timeout) self.model, self.return_sparse, self.lotto = model, return_sparse, max(1, min(lotto, LOTTO_MAX)) self._sparsi: "OrderedDict[str, SparseVector]" = OrderedDict() self._memoria = memoria self.dim: Optional[int] = None # fissata da controlla_dim(): la guardia sulla lunghezza def controlla_dim(self, attesa: int) -> int: """Da chiamare all'avvio, PRIMA di costruire o interrogare la collezione: GET /health (senza chiave, nessun consumo) e si confronta la DIMENSIONE con quella della collezione, fermandosi se non combacia. Sul nome non si confronta nulla, ed è voluto: /health dichiara il modello SOTTOSTANTE (`bge-m3`), /v1/models e /v1/embeddings il nome di LISTINO (`agilev-m3`). Sono vocabolari diversi e confrontarli dà un disallineamento a ogni avvio senza che ci sia un guasto: quello che si confronta è la dimensione, che si MISURA. Fissa anche la guardia sulla lunghezza dei vettori che arrivano.""" h = self._client.get("/health") dim = h.get("dim") if not isinstance(dim, int) or isinstance(dim, bool): raise AgileVErrore(200, "BAD_RESPONSE", "/health non dichiara `dim`: non si puo' controllare la dimensione") if dim != int(attesa): raise AgileVErrore(0, "DIM_MISMATCH", f"il servizio produce vettori da {dim} valori, la vostra collezione ne ha {int(attesa)}: " f"non indicizzate e non interrogate, sono spazi diversi") self.dim = dim return dim def _chiama(self, testi: Sequence[str], input_type: str, sparso: bool) -> List[dict]: """Un lotto per volta, nell'ordine di ingresso. Ritorna gli elementi `data` del servizio.""" # `quanti_al_massimo` resta None: i lotti li SPEZZA il ciclo qui sotto (a `self.lotto`, che il # costruttore tiene <= LOTTO_MAX), quindi il tetto dei testi per richiesta non può sforare e # un controllo sul numero rifiuterebbe proprio le liste lunghe che questo modulo sa servire. _tetti(testi) fuori: List[dict] = [] for i in range(0, len(testi), self.lotto): lotto = list(testi[i:i + self.lotto]) corpo: dict[str, Any] = {"model": self.model, "input": lotto, "input_type": input_type} if sparso: corpo["return_sparse"] = True fuori.extend(_dati(self._client.post("/v1/embeddings", corpo), len(lotto), sparso, self.dim)) if self.return_sparse and sparso: for testo, d in zip(testi, fuori): se = d["sparse_embedding"] self._sparsi[_impronta(testo)] = SparseVector(indices=list(se["indices"]), values=list(se["values"])) while len(self._sparsi) > self._memoria: self._sparsi.popitem(last=False) return fuori def embed_documents(self, texts: List[str]) -> List[List[float]]: return [d["embedding"] for d in self._chiama(texts, "document", self.return_sparse)] def embed_query(self, text: str) -> List[float]: return self._chiama([text], "query", self.return_sparse)[0]["embedding"] def embed_ibrido(self, texts: List[str], input_type: str = "document") -> List[tuple]: """Fuori da LangChain: [(denso, SparseVector), ...] in una chiamata sola, qualunque sia return_sparse.""" dati = self._chiama(texts, input_type, True) return [(d["embedding"], SparseVector(indices=list(d["sparse_embedding"]["indices"]), values=list(d["sparse_embedding"]["values"]))) for d in dati] def sparse(self) -> "AgileVSparseEmbeddings": """Il compagno lessicale, che legge i pesi già ricevuti da questa istanza.""" return AgileVSparseEmbeddings(denso=self) class AgileVSparseEmbeddings(SparseEmbeddings): """I pesi lessicali di agilev-m3 come SparseEmbeddings di langchain-qdrant (ricerca ibrida). Costruito da AgileVEmbeddings.sparse() riusa i pesi già scaricati (zero chiamate in più quando QdrantVectorStore chiede prima il denso e poi lo sparso degli stessi testi); da solo chiama il servizio. """ def __init__(self, denso: Optional[AgileVEmbeddings] = None, chiave: Optional[str] = None, url: Optional[str] = None, lotto: int = 64, timeout: float = 180.0): self._denso = denso or AgileVEmbeddings(chiave, url, return_sparse=True, lotto=lotto, timeout=timeout) def _prendi(self, testi: List[str], input_type: str) -> List[SparseVector]: fuori: List[Optional[SparseVector]] = [self._denso._sparsi.pop(_impronta(t), None) for t in testi] mancanti = [i for i, v in enumerate(fuori) if v is None] if mancanti: nuovi = self._denso.embed_ibrido([testi[i] for i in mancanti], input_type) for i, (_, sv) in zip(mancanti, nuovi): fuori[i] = sv return fuori # type: ignore[return-value] def embed_documents(self, texts: List[str]) -> List[SparseVector]: return self._prendi(texts, "document") def embed_query(self, text: str) -> SparseVector: return self._prendi([text], "query")[0] class AgileVSpezza: """Splitter per LangChain sopra spezza() del client di riferimento. Ha la stessa forma di un TextSplitter (split_text, create_documents, split_documents) senza dipendere da langchain-text-splitters. intestazione_da: nome del campo dei metadati da ripetere in testa a ogni frammento ("titolo", "articolo"...); None = nessuna intestazione. Ogni frammento eredita i metadati del documento più "frammento" (indice, da 0) e "frammenti" (quanti sono). Richiede agilev_client.py nella stessa cartella di questo file. """ def __init__(self, massimo: int = 1500, sovrapposizione: int = 150, intestazione_da: Optional[str] = None): self.massimo, self.sovrapposizione, self.intestazione_da = massimo, sovrapposizione, intestazione_da @staticmethod def _spezza(): try: from agilev_client import spezza except ImportError: import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) try: from agilev_client import spezza except ImportError as e: raise ImportError("AgileVSpezza usa spezza() di agilev_client.py: copiatelo nella stessa cartella") from e return spezza def split_text(self, text: str, intestazione: str = "") -> List[str]: return self._spezza()(text, self.massimo, self.sovrapposizione, intestazione) def create_documents(self, texts: List[str], metadatas: Optional[List[dict]] = None) -> List[Any]: from langchain_core.documents import Document fuori = [] for i, testo in enumerate(texts): meta = dict((metadatas or [{}] * len(texts))[i]) intest = str(meta.get(self.intestazione_da) or "") if self.intestazione_da else "" pezzi = self.split_text(testo, intest) fuori.extend(Document(page_content=p, metadata={**meta, "frammento": j, "frammenti": len(pezzi)}) for j, p in enumerate(pezzi)) return fuori def split_documents(self, documents: Sequence[Any]) -> List[Any]: docs = list(documents) return self.create_documents([d.page_content for d in docs], [d.metadata for d in docs]) def transform_documents(self, documents: Sequence[Any], **kwargs: Any) -> Sequence[Any]: return self.split_documents(documents) # forma di BaseDocumentTransformer try: from langchain_core.callbacks import Callbacks from langchain_core.documents import Document from langchain_core.documents.compressor import BaseDocumentCompressor from pydantic import ConfigDict, PrivateAttr class AgileVRerank(BaseDocumentCompressor): """Riordina i documenti di un retriever con /v1/rerank (bge-reranker-v2-m3) e tiene i primi top_n. max_caratteri: quanto testo di ogni documento vede il reranker. 800 è il valore misurato sul corpus di normativa (stessa qualità del testo intero, metà del tempo); 0 = tutto (fino a 1.024 token). Il punteggio finisce in metadata["relevance_score"]. """ model_config = ConfigDict(arbitrary_types_allowed=True) chiave: Optional[str] = None url: Optional[str] = None model: str = "agilev-rerank" top_n: int = 5 max_caratteri: int = 800 timeout: float = 180.0 _client: Any = PrivateAttr(default=None) def _cli(self) -> _Client: if self._client is None: self._client = _Client(self.chiave, self.url, self.timeout) return self._client def compress_documents(self, documents: Sequence[Document], query: str, callbacks: Optional[Callbacks] = None) -> Sequence[Document]: documenti = list(documents) if not documenti: return [] testi = [d.page_content[: self.max_caratteri] if self.max_caratteri else d.page_content for d in documenti] # I TETTI PRIMA DI SPEDIRE, nell'ordine del servizio: la `query` è guardata per prima # perché /v1/rerank fa `norm_inputs([req.query], ...)` e solo dopo # `norm_inputs(req.documents, ...)`, e con due difetti nella stessa richiesta il codice # che il chiamante riceve deve essere quello che avrebbe ricevuto dal servizio. I # documenti si misurano DOPO il taglio a `max_caratteri`: col predefinito a 800 un # candidato lungo è legittimo, e il tetto per testo morde solo quando chi chiama ha # chiesto i documenti interi (`max_caratteri=0`). La `query` non si taglia: tagliarla # cambierebbe la domanda, cioè la risposta. _tetti([query], campo="query") _tetti(testi, quanti_al_massimo=LOTTO_MAX, campo="documents") # top_n NEGATIVO si ferma qui, prima di spendere: i documenti si pagano a uno a uno e un # numero negativo non vuol dire niente. `0` parte: per il servizio e per noi vuol dire # «nessun limite», cioè tieni tutti i documenti riordinati. Viene DOPO i tetti perché # il servizio il top_n non lo rifiuta (`_come_si_taglia`: torna tutti i documenti e lo # dice in `ignored_values`): con un top_n negativo E 129 documenti il servizio direbbe # BATCH_TOO_LARGE, e questo rifiuto locale in più non deve coprirglielo. if self.top_n < 0: raise AgileVErrore(0, "BAD_TOP_N", f"top_n={self.top_n}: deve essere positivo (0 = tutti)") corpo = {"model": self.model, "query": query, "documents": testi, "top_n": self.top_n} risposta = self._cli().post("/v1/rerank", corpo) # già ordinati, index = posizione risultati = risposta.get("results") if isinstance(risposta, dict) else None attesi = min(self.top_n, len(testi)) if self.top_n else len(testi) if not isinstance(risultati, list) or len(risultati) != attesi: quanti = len(risultati) if isinstance(risultati, list) else "nessun elenco" raise AgileVErrore(200, "BAD_RESPONSE", f"attesi {attesi} elementi in results, ricevuti {quanti}") for r in risultati: if not (isinstance(r, dict) and isinstance(r.get("index"), int) and 0 <= r["index"] < len(testi) and isinstance(r.get("relevance_score"), (int, float))): raise AgileVErrore(200, "BAD_RESPONSE", "un elemento di results non ha index o relevance_score") fuori = [] for r in risultati: d = documenti[r["index"]] fuori.append(Document(page_content=d.page_content, metadata={**d.metadata, "relevance_score": float(r["relevance_score"])})) return fuori except ImportError: # LangChain assente: gli embedding restano disponibili AgileVRerank = None # type: ignore[assignment,misc]