""" Client di riferimento per AgileV (Python, solo libreria standard). Copiatelo nel vostro progetto. from agilev_client import AgileV with AgileV() as av: # chiave in AGILEV_KEY (o chiave=...) vettori = av.embed(testi) # una lista di vettori (1024 float), nell'ordine dei testi ordine = av.rerank(domanda, candidati, top_n=5) # [(indice, punteggio), ...] già ordinati Incorpora le regole imparate sul campo (API.md, sezione «Timeout del client e lotti»), così non dovete riscoprirle: 1. la connessione TLS si riusa (keep-alive): la stretta di mano (~250 ms) si paga una volta, non a ogni chiamata; 2. i lotti partono UNO ALLA VOLTA, mai in parallelo (il servizio calcola una richiesta per volta), 64 testi per lotto (massimo 128) e timeout 120 s; 3. i limiti (128 testi per richiesta, 32.000 caratteri per testo) si controllano PRIMA di spedire: un 413 non arriva mai per un errore di chi chiama; 4. si ritenta SOLO su 429 RATE_LIMIT, sui 5xx e sulla connessione caduta, con attesa crescente (1, 2, 4, 8 s, tetto 30) oppure ESATTAMENTE `Retry-After` se il servizio lo manda (sul 429 RATE_LIMIT dice quanti secondi mancano al minuto nuovo, fino a 60: si aspetta quello, non si indovina); su 400/401/403/413 e su 429 QUOTA_EXCEEDED no — e il `Retry-After` del QUOTA_EXCEEDED, che sono i secondi al MESE nuovo, non si legge: la richiesta va cambiata. Una connessione riusata che il server ha chiuso nel frattempo (keep-alive scaduto) NON è un ritentativo: si riconnette subito, senza attesa; 5. al timeout NON si ripete lo stesso lotto (era il caso peggiore di settembre: 47 lotti identici in coda): si DIMEZZA e si riprova con lotti più piccoli, fino a un testo solo; 6. il rerank legge di ogni candidato i primi 800 caratteri (misurato: stessa qualità, metà del tempo); 7. un documento lungo si SPEZZA prima di spedirlo: `spezza(testo)` fa frammenti da 1.500 caratteri (tagliati a paragrafi, poi a frasi, poi a parole) con 150 caratteri di frasi intere in comune fra un frammento e il successivo, e ripete in testa a ognuno l'intestazione che gli date (es. «Art. 25-septies Rubrica»): un articolo da 56.000 caratteri non è un 413 ma 40 frammenti che si ritrovano; 8. si FRENA prima del 429: ogni risposta autenticata dice quante richieste restano nel minuto (`X-RateLimit-Remaining`) e fra quanti secondi il minuto si rinnova (`X-RateLimit-Reset`, dal contratto 1.2); quando ne restano 0 la richiesta successiva aspetta quei secondi PRIMA di partire, invece di partire, prendere un 429 e aspettare comunque: nessuna richiesta a vuoto e nessun errore nei vostri log (`contatori["frenate"]`; le ultime intestazioni lette stanno in `av.rate`). Con l'1.1 le intestazioni non ci sono e il client non frena. 9. la DIMENSIONE si controlla all'avvio, una volta: `av.controlla_dim(1024)` legge `dim` da /health (senza chiave, senza consumo) e si ferma se non è quella del vostro indice, invece di farvelo scoprire a indice costruito; da quel momento un vettore della misura sbagliata diventa un `BAD_RESPONSE` invece di entrare nell'indice. Si confronta la dimensione e NON il nome: /health dichiara il modello sottostante (`bge-m3`), /v1/models e /v1/embeddings il nome di listino (`agilev-m3`), e confrontarli dà un falso allarme a ogni avvio (API.md, Regola 4: «il nome non è un segnale di salute, la dimensione sì»). Nessun testo viene registrato da questo client (niente log): solo i vostri dati vanno al servizio, sotto TLS. """ import http.client, ipaddress, json, os, re, socket, ssl, sys, time, urllib.parse from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union URL_DEFAULT = "https://api.agilev.agile.software" def dall_ambiente(nome: str) -> Tuple[Optional[str], str]: """ Una variabile d'ambiente, con una regola sola: un valore che c'è ma è VUOTO è un ERRORE, non un ripiego. Qualcuno l'ha impostata e non è arrivata; tirare dritto sul predefinito vuol dire mandare i testi all'indirizzo pubblico senza dirlo. In Python la sorgente è una sola (`os.environ`). In PHP sono tre e non portano lo stesso valore nella stessa stanza: se il vostro stack è PHP, guardate il gemello agilev_client.php. Ritorna (valore, sorgente) con sorgente `ambiente` o `assente`. """ 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 — " "non si tira dritto sul predefinito") return v.strip(), "ambiente" def in_chiaro_senza_internet(ospite: str) -> bool: """`http` (non cifrato) verso questo ospite resta dentro casa? Vero per il loopback, per le reti private (RFC1918, link-local, CGNAT) e per il NOME `localhost`; falso per tutto il resto, nome pubblico compreso. NESSUNA risoluzione di nomi, apposta: un nome non si sa dove guarda, risolverlo dentro un costruttore è una chiamata di rete che chi legge non si aspetta, e il risultato cambierebbe col DNS. Quindi un nome diverso da `localhost` non è ammesso in chiaro nemmeno se punta in casa: chi ha un fronte privato con un nome proprio lo dichiara con `in_chiaro_ammesso=True`. """ 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 # limiti del servizio (API.md, «Limiti») # I due tetti pubblicati, controllati PRIMA di spedire e in UN SOLO POSTO — come `norm_inputs` in # app.py sta in un posto solo dalla parte del servizio. Fino al 10/10 il controllo viveva dentro # `embed`, e `rerank` non ne aveva nessuno: la `query` e il numero dei `documents` 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 che sta 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 di sempre, parola per parola: le altre rotte hanno un campo solo. # # L'ORDINE È QUELLO DEL SERVIZIO e non è un dettaglio: prima il numero dei testi, poi la # lunghezza di ciascuno; e chi chiama passa la `query` prima dei `documents`. Con due difetti nella # stessa richiesta il servizio risponde TEXT_TOO_LONG della query, e un controllo locale in ordine # diverso darebbe al chiamante l'altro codice — cioè un ramo di gestione che il servizio non usa. 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}") LOTTO_DEFAULT, TIMEOUT_DEFAULT = 64, 120.0 RERANK_CARATTERI = 800 # per candidato: nDCG@10 0,814 contro 0,816 interi, in metà tempo ATTESA_MAX = 30.0 # tetto dell'attesa crescente (non di Retry-After) RETRY_AFTER_MAX = 60.0 # sul RATE_LIMIT il servizio non chiede mai più di un # minuto; sul QUOTA_EXCEEDED manda i secondi al mese nuovo # (fino a 2.678.400) e lì non si ritenta, quindi non si legge SPEZZA_CARATTERI, SPEZZA_SOVRAPPOSIZIONE = 1500, 150 # frammento e sovrapposizione consigliati (API.md, «Documenti lunghi») # Risposte che NON hanno la forma `{"error":{"code":…}}` perché non le scrive AgileV: le scrive il # framework (`{"detail":"Not Found"}`, `{"detail":"Method Not Allowed"}`) o il bordo (una pagina di # proxy). Sono reali — a settembre 2026 un cliente ha preso un `405` chiamando `/v1/embeddings` in # GET, e un altro un `502` durante un nostro riavvio — e prima di questa tabella arrivavano a chi # chiama come codice `?`, cioè come «non so»: un allarme che non si può smistare. # Il codice lo mette il CLIENT, e lo deduce dallo STATO, mai dal corpo: quei corpi non sono nostri e # possono cambiare senza preavviso. Il corpo non entra nel messaggio (vedi `_errore`). 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)"), } class AgileVErrore(Exception): """Risposta non 200 (o connessione impossibile). `stato` HTTP, `codice` del contratto, `messaggio`.""" def __init__(self, stato: int, codice: str, messaggio: str): super().__init__(f"{stato} {codice}: {messaggio}") self.stato, self.codice, self.messaggio = stato, codice, messaggio class AgileVTimeout(AgileVErrore): """Il servizio non ha risposto entro `timeout` secondi. embed() la gestisce dimezzando il lotto.""" def __init__(self, timeout: float, testi: int): super().__init__(0, "TIMEOUT", f"nessuna risposta in {timeout:g} s per un lotto di {testi} testi") class AgileV: def __init__(self, chiave: Optional[str] = None, url: Optional[str] = None, timeout: float = TIMEOUT_DEFAULT, lotto: int = LOTTO_DEFAULT, tentativi: int = 5, attesa: float = 1.0, dormi: Callable[[float], None] = time.sleep, indirizzo_obbligatorio: bool = False, in_chiaro_ammesso: bool = False): """ indirizzo_obbligatorio: se l'indirizzo finisce sul predefinito pubblico, si ferma invece di partire. È un argomento del COSTRUTTORE e non una variabile d'ambiente, apposta: una guardia che vive nello stesso ambiente sparisce insieme a ciò che deve proteggere, proprio nella stanza in cui serve. Mettetelo a True se puntate AgileV su un fronte privato (WireGuard, secondo box, smistatore) e i vostri testi non devono mai uscire su Internet. in_chiaro_ammesso: toglie la guardia che rifiuta `http://` verso un indirizzo che NON è loopback o rete privata. Mettetelo a True solo se sapete che quel filo non passa da Internet (un nome vostro dentro una VPN, per esempio): è l'uscita esplicita, dichiarata nel vostro codice, non un ripiego silenzioso. Per la stessa ragione di sopra è un argomento e non una variabile d'ambiente. """ k, sorgente_chiave = dall_ambiente("AGILEV_KEY") self.chiave = chiave or k or "" self.sorgente_chiave = "argomento" if chiave else sorgente_chiave 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") if indirizzo_obbligatorio and self.sorgente_indirizzo == "predefinito": raise ValueError("indirizzo AgileV obbligatorio ma AGILEV_URL non c'e': dentro un server web " f"l'ambiente del processo puo' essere vuoto. Senza questa guardia: {URL_DEFAULT}") self.indirizzo = (url or a or URL_DEFAULT).rstrip("/") u = urllib.parse.urlsplit(self.indirizzo) self._https, self._host, self._porta = u.scheme == "https", u.hostname or "", u.port # La guardia sul CHIARO. Non è «https e basta»: il chiaro verso casa (loopback, rete privata) # resta possibile, perché quel filo non passa da Internet — un fronte privato su 127.0.0.1 o # dentro WireGuard è una configurazione nostra e di diversi clienti. Quello che si rifiuta è # il chiaro verso un indirizzo che esce: lì il rimando del bordo arriva DOPO i byte. if not self._https and not in_chiaro_ammesso and not in_chiaro_senza_internet(self._host): raise ValueError( f"indirizzo AgileV in chiaro verso «{self._host or self.indirizzo}»: 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._base = u.path.rstrip("/") self.timeout, self.lotto = float(timeout), max(1, min(int(lotto), LOTTO_MAX)) self.tentativi, self.attesa, self._dormi = max(1, tentativi), attesa, dormi self._conn: Optional[http.client.HTTPConnection] = None self.contatori = {"richieste": 0, "ritentate": 0, "lotti_dimezzati": 0, "connessioni": 0, "riconnessioni": 0, "frenate": 0} self.rate: Dict[str, float] = {} # ultime X-RateLimit-* lette: limit, remaining, reset (secondi al minuto nuovo) self._freno: Optional[float] = None # istante (monotonic) prima del quale non si spedisce: minuto esaurito self._retry_after: Optional[float] = None self._usi = 0 # richieste servite dalla connessione corrente self._riusata = False # l'ultimo scambio è partito su una connessione già usata? self._subito = False # prossimo tentativo senza attesa (connessione riusata morta) self.dim: Optional[int] = None # dimensione attesa dei vettori, se controlla_dim() l'ha fissata # ---------------------------------------------------------------- trasporto: una connessione riusata def _connessione(self) -> http.client.HTTPConnection: if self._conn is None: if self._https: self._conn = http.client.HTTPSConnection(self._host, self._porta, timeout=self.timeout, context=ssl.create_default_context()) else: self._conn = http.client.HTTPConnection(self._host, self._porta, timeout=self.timeout) self.contatori["connessioni"] += 1; self._usi = 0 return self._conn def chiudi(self) -> None: if self._conn is not None: try: self._conn.close() finally: self._conn = None def __enter__(self): return self def __exit__(self, *_): self.chiudi() def _scambio(self, metodo: str, percorso: str, corpo: Optional[dict], con_chiave: bool) -> Tuple[int, Dict[str, str], str]: """Un solo giro HTTP sulla connessione viva. Solleva socket.timeout o un errore di connessione.""" intestazioni = {"Content-Type": "application/json", "Accept": "application/json"} if con_chiave: intestazioni["Authorization"] = f"Bearer {self.chiave}" dati = json.dumps(corpo).encode("utf-8") if corpo is not None else None c = self._connessione(); self._riusata = self._usi > 0 self.contatori["richieste"] += 1 # conta le richieste spedite, anche quelle senza risposta try: c.request(metodo, self._base + percorso, body=dati, headers=intestazioni) r = c.getresponse() testo = r.read().decode("utf-8", "replace"); self._usi += 1 if r.getheader("Connection", "").lower() == "close": self.chiudi() return r.status, {k.lower(): v for k, v in r.getheaders()}, testo except BaseException: self.chiudi() # dopo un timeout o un errore la connessione non è più affidabile raise def _nota_rate(self, intestazioni: Dict[str, str]) -> None: """Legge 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): return self.rate = {"limit": limite, "remaining": rimaste, "reset": reset} self._freno = time.monotonic() + min(RETRY_AFTER_MAX, 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. Se nel frattempo il minuto è già passato non si aspetta nulla.""" if self._freno is None: return resto = self._freno - time.monotonic(); self._freno = None if resto > 0: self.contatori["frenate"] += 1; self._dormi(resto) @staticmethod def _senza_codice(stato: int) -> Tuple[str, str]: """La risposta non porta un codice nostro. Lo mette il client, dallo STATO soltanto. Regola: il corpo non si legge e non entra nel messaggio. Quei corpi non sono di AgileV (FastAPI, il bordo, un proxy di mezzo) e nessuno ha promesso cosa ci metterà dentro: se finisce nel vostro log, un giorno ci finisce qualcosa che non volevate. Lo stato HTTP basta a smistare, ed è l'unica parte che il contratto garantisce.""" 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: # Il servizio non è arrivato a rispondere, o è caduto senza forma: ritentabile (e infatti # `embed` ritenta già su 5xx). Dalla 1.2 un guasto dell'app ha `INTERNAL` e non passa di qui. return "UPSTREAM_ERROR", f"guasto senza forma ({stato}): bordo o servizio, ritentabile" return "NO_CODE", f"risposta {stato} senza un codice del contratto" @staticmethod def _errore(stato: int, testo: str) -> Tuple[str, str]: """Codice e messaggio da mostrare a chi chiama. Il `422` non ha `code` (è il corpo di FastAPI) e RIMANDA INDIETRO I TESTI INVIATI in `detail[].input`: qui si legge solo QUALE campo è sbagliato, mai l'eco, così i vostri testi non finiscono nei vostri log passando per un nostro errore. Tutto quello che non ha la forma del contratto finisce in `_senza_codice`: **nessuna risposta esce di qui con codice `?`**, perché un `?` non si può smistare né contare.""" try: e = json.loads(testo) except ValueError: return AgileV._senza_codice(stato) if isinstance(e, dict) and isinstance(e.get("detail"), list): return "BAD_BODY", AgileV._campi(e["detail"]) # codice messo dal client: il servizio non ne manda try: e = e.get("detail", e).get("error", {}) codice = e.get("code") except AttributeError: return AgileV._senza_codice(stato) # `{"detail":"Not Found"}`: detail è una stringa if not isinstance(e, dict) or not codice: return AgileV._senza_codice(stato) return str(codice), str(e.get("message", "")) @staticmethod def _campi(dettagli: list, massimo: int = 5) -> str: """`body.input: Field required` — solo `loc`, `msg`, `type` del 422; `input` 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(": ")) resto = f" (+{len(dettagli) - massimo} altri)" if len(dettagli) > massimo else "" if not pezzi: return "corpo della richiesta non valido" return "corpo della richiesta non valido - " + "; ".join(pezzi) + resto @staticmethod 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 (lotto mutilato, proxy di mezzo, versione nuova) qui diventa un errore `BAD_RESPONSE` invece di una lista più corta: chi indicizza accoppierebbe i propri testi ai vettori sbagliati senza accorgersene. Segnalato da trpg il 28/9 sul proprio client. `dim` (la dimensione fissata da `controlla_dim()`) fa lo stesso sulla LUNGHEZZA dei vettori: un vettore della misura sbagliata ALZA LA VOCE invece di finire nell'indice o di essere scartato in silenzio — un `continue` su una dimensione che non combacia dà zero risultati su un indice pieno, ed è il guasto più silenzioso che possiate avere (Regola 4).""" 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 _chiama(self, metodo: str, percorso: str, corpo: Optional[dict] = None, con_chiave: bool = True, testi: int = 0) -> Any: """Politica di ritentativo: 429 RATE_LIMIT, 5xx e connessione caduta si ritentano con attesa crescente; il timeout risale come AgileVTimeout (embed lo gestisce); tutto il resto è definitivo.""" ultimo: Optional[AgileVErrore] = None for tentativo in range(self.tentativi): if tentativo and self._subito: # La connessione riusata era morta (keep-alive scaduto lato server o proxy): non è colpa del # servizio, si riconnette e si rispedisce subito. Sulla connessione nuova vale la politica normale. self._subito = False; self.contatori["riconnessioni"] += 1 elif tentativo: self.contatori["ritentate"] += 1 if self._retry_after is not None: attesa = min(RETRY_AFTER_MAX, self._retry_after) # il servizio sa quando else: attesa = min(ATTESA_MAX, self.attesa * 2 ** (tentativo - 1)) self._dormi(attesa); self._retry_after = None if con_chiave: self._frena() try: stato, intestazioni, testo = self._scambio(metodo, percorso, corpo, con_chiave) except socket.timeout: raise AgileVTimeout(self.timeout, testi) except (OSError, http.client.HTTPException) as e: ultimo = AgileVErrore(0, "CONNESSIONE", f"{type(e).__name__}: {e}"); self._subito = self._riusata; continue self._nota_rate(intestazioni) if stato == 200: return json.loads(testo) codice, messaggio = self._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 ra.replace(".", "", 1).isdigit() else None self._freno = None # l'attesa del ritentativo copre già il minuto: niente doppia attesa continue raise ultimo assert ultimo is not None raise ultimo # ---------------------------------------------------------------- API def health(self) -> dict: """GET /health, senza chiave: `ok`, `loaded`, `contract`.""" return self._chiama("GET", "/health", con_chiave=False) def controlla_dim(self, attesa: int) -> int: """Da chiamare UNA VOLTA all'avvio, prima di costruire o interrogare l'indice: è la regola 9. Confronta la DIMENSIONE dei vettori del servizio con quella del vostro indice e si ferma subito se non combacia, invece di farvelo scoprire a indice costruito. Non consuma niente e non vuole la chiave. Sul NOME non si confronta nulla, ed è voluto: /health dichiara il modello SOTTOSTANTE (`bge-m3`), mentre /v1/models e /v1/embeddings usano il nome di LISTINO (`agilev-m3`). Sono due vocabolari e i due valori NON sono confrontabili: chi li confronta ha un disallineamento a ogni avvio senza che ci sia un guasto (misurato: /v1/embeddings dice `agilev-m3`, /embed dice `bge-m3`, e i vettori sono identici bit per bit). Il nome si confronta solo DENTRO la stessa rotta, con quello che quella rotta dichiara. """ h = self.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, il vostro indice ne ha {int(attesa)}: " f"non indicizzate e non interrogate, sono spazi diversi") self.dim = dim return dim def usage(self) -> dict: """GET /v1/usage: i consumi della VOSTRA chiave, per mese e operazione.""" return self._chiama("GET", "/v1/usage") def embed(self, testi: Sequence[str], *, input_type: Optional[str] = None, return_sparse: bool = False, model: str = "agilev-m3") -> List[Any]: """Vettori nell'ordine dei testi: liste di 1024 float; con return_sparse=True coppie (denso, {token: peso}). I limiti si controllano qui, prima di spedire.""" testi = list(testi) # `quanti_al_massimo` resta None: `embed` i lotti li SPEZZA lui (il ciclo qui sotto), quindi # il tetto dei testi per richiesta non può sforare e un controllo sul numero rifiuterebbe # proprio le liste lunghe che questo client sa servire. _tetti(testi) fuori: List[Any] = [] for i in range(0, len(testi), self.lotto): fuori.extend(self._lotto(testi[i:i + self.lotto], input_type, return_sparse, model)) return fuori def _lotto(self, testi: List[str], input_type: Optional[str], return_sparse: bool, model: str) -> List[Any]: if not testi: return [] corpo: Dict[str, Any] = {"model": model, "input": testi} if input_type: corpo["input_type"] = input_type if return_sparse: corpo["return_sparse"] = True try: risposta = self._chiama("POST", "/v1/embeddings", corpo, testi=len(testi)) except AgileVTimeout: if len(testi) == 1: raise self.contatori["lotti_dimezzati"] += 1 meta = len(testi) // 2 # regola 5: mai lo stesso lotto di nuovo, uno più piccolo return self._lotto(testi[:meta], input_type, return_sparse, model) + \ self._lotto(testi[meta:], input_type, return_sparse, model) dati = self._dati(risposta, len(testi), return_sparse, self.dim) if not return_sparse: return [d["embedding"] for d in dati] return [(d["embedding"], dict(zip(d["sparse_embedding"]["indices"], d["sparse_embedding"]["values"]))) for d in dati] def rerank(self, query: str, documenti: Sequence[str], *, top_n: Optional[int] = None, caratteri: Optional[int] = RERANK_CARATTERI, model: str = "agilev-rerank") -> List[Tuple[int, float]]: """[(indice_nel_vostro_elenco, punteggio), ...] dal migliore. Ogni candidato è letto fino a `caratteri` (800, il valore misurato; None = intero). Pochi candidati (10) è la regola. I limiti si controllano qui, prima di spedire (regola 3), e sono TRE perché questa rotta ha due campi che portano testo: la `query` fino a CARATTERI_MAX, i documenti fino a LOTTO_MAX e ciascuno fino a CARATTERI_MAX. La `query` NON si taglia come i candidati: tagliarla cambierebbe la domanda, cioè la risposta — si rifiuta e si dice quanto è lunga.""" docs = [d[:caratteri] if caratteri else d 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(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 # `caratteri`: col predefinito a 800 un candidato lungo è legittimo, e il tetto per testo # morde solo quando chi chiama ha chiesto i documenti interi (`caratteri=None`). _tetti([query], campo="query") _tetti(docs, quanti_al_massimo=LOTTO_MAX, campo="documents") # top_n NEGATIVO si ferma qui, prima di spendere: non vuol dire niente, e i documenti si # pagano. Chi lo calcola (`top_n = me_ne_servono - ne_ho_gia`) sbaglia di segno, non di # intenzione, e un rifiuto locale costa zero richieste. `0` invece parte: per il servizio # e per questo client vuol dire «nessun limite», come `None`. if top_n is not None and top_n < 0: raise AgileVErrore(0, "BAD_TOP_N", f"top_n={top_n}: deve essere positivo (0 o assente = tutti)") corpo: Dict[str, Any] = {"model": model, "query": query, "documents": docs} if top_n is not None: corpo["top_n"] = top_n r = self._chiama("POST", "/v1/rerank", corpo, testi=len(docs)) ris = r.get("results") if isinstance(r, dict) else None # `if top_n` e non `if top_n is not None`: `top_n=0` non taglia, né qui né sul servizio, e # un `min(0, len(docs))` farebbe attendere zero risultati dove ne arrivano tutti. attesi = min(top_n, len(docs)) if top_n else len(docs) if not isinstance(ris, list) or len(ris) != attesi: quanti = len(ris) if isinstance(ris, list) else "nessun elenco" raise AgileVErrore(200, "BAD_RESPONSE", f"attesi {attesi} elementi in results, ricevuti {quanti}") for x in ris: if not (isinstance(x, dict) and isinstance(x.get("index"), int) and 0 <= x["index"] < len(docs) and isinstance(x.get("relevance_score"), (int, float))): raise AgileVErrore(200, "BAD_RESPONSE", "un elemento di results non ha index o relevance_score") return [(x["index"], float(x["relevance_score"])) for x in ris] # -------------------------------------------------------------------- documenti lunghi: spezza() prima di spedire _FINE_FRASE = re.compile(r"(?<=[.!?;])[ \t]+") # spazio dopo un segno di fine frase (stesso schema nel PHP) _BIANCHI = " \t\r\n\x0b\x00" # gli stessi caratteri di trim() in PHP def spezza(testo: str, massimo: int = SPEZZA_CARATTERI, sovrapposizione: int = SPEZZA_SOVRAPPOSIZIONE, intestazione: str = "") -> List[str]: """Spezza un documento in frammenti da al più `massimo` caratteri, pronti per embed(). Taglia dove il testo si taglia da solo: a fine riga, poi a fine frase (. ! ? ;), poi fra le parole, e solo per una parola più lunga del frammento a metà parola. Due frammenti consecutivi hanno in comune le ultime frasi intere del primo fino a `sovrapposizione` caratteri, così un concetto a cavallo del taglio resta intero in almeno uno dei due. `intestazione` (titolo, numero di articolo, rubrica) viene ripetuta in testa a ogni frammento, così ogni frammento sa di cosa parla anche da solo (è ciò che ha fatto la differenza sulle citazioni di articolo, vedi API.md). Il gemello PHP (AgileV::spezza) produce gli STESSI frammenti. Restituisce [] per un testo vuoto e [testo] per uno che sta già in un frammento.""" massimo = min(int(massimo), CARATTERI_MAX) intestazione = intestazione.strip(_BIANCHI) budget = massimo - (len(intestazione) + 1 if intestazione else 0) # spazio per il testo, tolta l'intestazione if budget < 50: raise ValueError(f"massimo={massimo} troppo piccolo per l'intestazione di {len(intestazione)} caratteri") sovrapposizione = max(0, min(int(sovrapposizione), budget // 3)) # mai più di un terzo del frammento # 1. unità: righe -> frasi -> (solo se una frase non entra) parole; ogni unità entra in un frammento unita: List[Tuple[str, str]] = [] # (testo, separatore da mettere PRIMA: "\n" a inizio riga, " " altrimenti) for riga in testo.replace("\r\n", "\n").split("\n"): riga = riga.strip(_BIANCHI); sep = "\n" if not riga: continue for frase in _FINE_FRASE.split(riga): frase = frase.strip(_BIANCHI) if not frase: continue if len(frase) <= budget: unita.append((frase, sep)); sep = " "; continue for parola in frase.split(" "): while parola: # una parola più lunga del budget si taglia unita.append((parola[:budget], sep)); sep = " "; parola = parola[budget:] if not unita: return [] # 2. frammenti: si riempie finché entra; il successivo riparte dalle ultime frasi intere del precedente frammenti: List[List[int]] = []; corrente: List[int] = []; lung = 0 for i, (u, _) in enumerate(unita): if corrente and lung + 1 + len(u) > budget: frammenti.append(corrente) coda: List[int] = []; lc = 0 for j in reversed(corrente): l = len(unita[j][0]) + (1 if coda else 0) if lc + l > sovrapposizione: break coda.insert(0, j); lc += l if lc + 1 + len(u) > budget: coda, lc = [], 0 # l'unità nuova deve entrare comunque corrente, lung = coda, lc lung += len(u) + (1 if corrente else 0); corrente.append(i) frammenti.append(corrente) fuori = [] for f in frammenti: t = "".join((unita[j][1] if k else "") + unita[j][0] for k, j in enumerate(f)) fuori.append(f"{intestazione}\n{t}" if intestazione else t) return fuori if __name__ == "__main__": # `python agilev_client.py --prova`: /health senza chiave; con AGILEV_KEY anche un embedding e un rerank veri. # `python agilev_client.py --spezza [massimo] [sovrapposizione] [intestazione] < documento.txt`: i frammenti in JSON # (è anche il modo in cui i test confrontano questo spezza() con il gemello PHP). if sys.argv[1:2] == ["--spezza"]: a = sys.argv[2:] + ["", "", ""] print(json.dumps(spezza(sys.stdin.read(), int(a[0] or SPEZZA_CARATTERI), int(a[1] or SPEZZA_SOVRAPPOSIZIONE), a[2]), ensure_ascii=False)); sys.exit(0) if sys.argv[1:] != ["--prova"]: print(__doc__); sys.exit(0) av = AgileV(chiave=os.environ.get("AGILEV_KEY") or "-") print(f"indirizzo: {av.indirizzo} (sorgente: {av.sorgente_indirizzo}; chiave: {av.sorgente_chiave})") h = av.health(); print(f"health: ok={h.get('ok')} loaded={h.get('loaded')} contract={h.get('contract')}") if os.environ.get("AGILEV_KEY"): v = av.embed(["Il contratto scade il 30 settembre", "Ricetta della carbonara"], input_type="document") print(f"embed: {len(v)} vettori da {len(v[0])} dimensioni") print(f"rerank: {av.rerank('scadenza del contratto', ['Il contratto scade il 30 settembre', 'Ricetta della carbonara'])}") print(f"contatori: {av.contatori}"); av.chiudi(); print("prova ok")