diff --git a/gateway/config.py b/gateway/config.py index d7f2d41..a66f817 100644 --- a/gateway/config.py +++ b/gateway/config.py @@ -13,6 +13,13 @@ EMBED_URL = os.environ.get("EMBED_URL", os.environ.get("OLLAMA_URL", "http://127 EMBED_MODEL = os.environ.get("EMBED_MODEL", "bge-m3") EMBED_API_KEY = os.environ.get("EMBED_API_KEY", "") EMBED_DIM = int(os.environ.get("EMBED_DIM", "1024")) +# Catena di fallback per gli embedding (JSON, formato RERANK_CHAIN + campo "api"). +# Vuota → comportamento legacy: endpoint singolo da EMBED_API/EMBED_URL/EMBED_API_KEY. +EMBED_CHAIN = os.environ.get("EMBED_CHAIN", "") +EMBED_TIMEOUT_MS = int(os.environ.get("EMBED_TIMEOUT_MS", "30000")) +EMBED_RETRY_COOLDOWN_S = int(os.environ.get("EMBED_RETRY_COOLDOWN_S", "60")) +# Retry transiente per le chiamate Qdrant (store/search inclusi) +QDRANT_RETRIES = int(os.environ.get("QDRANT_RETRIES", "3")) COLLECTION = os.environ.get("COLLECTION", "memories") 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")) @@ -23,7 +30,7 @@ GUARDRAIL_BLOCK_THRESHOLD = float(os.environ.get("GUARDRAIL_BLOCK_THRESHOLD", "0 GUARDRAIL_WARN_THRESHOLD = float(os.environ.get("GUARDRAIL_WARN_THRESHOLD", "0.70")) GUARDRAIL_VERSION = "similarity-v1" GIT_COMMIT = os.environ.get("GIT_COMMIT", "unknown").strip() -GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.9.0").strip() +GATEWAY_VERSION = os.environ.get("GATEWAY_VERSION", "2.10.0").strip() VM_PUSH_URL = os.environ.get("VM_PUSH_URL", "http://host.docker.internal:8428/api/v1/import/prometheus") VM_PUSH_INTERVAL = int(os.environ.get("VM_PUSH_INTERVAL", "30")) @@ -52,4 +59,7 @@ _metrics: dict[str, Any] = { "search_hits": 0, "rerank_calls": Counter(), "rerank_duration_sum": Counter(), + "embed_calls": Counter(), + "embed_duration_sum": Counter(), + "qdrant_retries": 0, } diff --git a/gateway/embed.py b/gateway/embed.py index 433086b..10b21b0 100644 --- a/gateway/embed.py +++ b/gateway/embed.py @@ -1,16 +1,38 @@ -"""Embedding denso (Ollama/llama.cpp) e sparse BM25.""" +"""Embedding denso (Ollama/llama.cpp) e sparse BM25, con catena di fallback resiliente. + +Catena da EMBED_CHAIN (JSON, stesso formato di RERANK_CHAIN + campo "api"): +il primo nodo raggiungibile vince, i nodi falliti entrano in cooldown. Se +EMBED_CHAIN è vuota si usa il comportamento legacy (endpoint singolo da +EMBED_API/EMBED_URL/EMBED_API_KEY). L'ultimo errore viene rilanciato al client +come gli endpoint precedenti: nessuna degradazione silenziosa della scrittura. +""" from __future__ import annotations +import json +import time +from dataclasses import dataclass from typing import Any, Optional import httpx from qdrant_client.http import models as qm -from config import EMBED_API, EMBED_API_KEY, EMBED_MODEL, EMBED_URL, SPARSE_VECTOR_NAME, log +import metrics +from config import ( + EMBED_API, + EMBED_API_KEY, + EMBED_CHAIN, + EMBED_DIM, + EMBED_MODEL, + EMBED_RETRY_COOLDOWN_S, + EMBED_TIMEOUT_MS, + EMBED_URL, + SPARSE_VECTOR_NAME, + log, +) try: from fastembed import SparseTextEmbedding - _sparse_model: Optional[SparseTextEmbedding] = None + _sparse_model: Optional[Any] = None SPARSE_AVAILABLE = True except Exception: # noqa: BLE001 _sparse_model = None @@ -20,6 +42,66 @@ except Exception: # noqa: BLE001 _http: Optional[httpx.AsyncClient] = None +@dataclass(frozen=True) +class EmbedNode: + """Un endpoint embedding nella catena di fallback.""" + + name: str + url: str + api: str # "llamacpp" (/v1/embeddings) | "ollama" (/api/embed) + key: str + timeout_ms: int + + +def parse_chain(raw: str, default_api: str, default_url: str, default_key: str) -> list[EmbedNode]: + """Parsa EMBED_CHAIN (JSON); vuota o invalida → endpoint legacy singolo.""" + nodes: list[EmbedNode] = [] + if raw: + try: + entries = json.loads(raw) + for entry in entries if isinstance(entries, list) else []: + if not isinstance(entry, dict) or not entry.get("url"): + continue + url = str(entry["url"]).rstrip("/") + api = str(entry.get("api") or "llamacpp") + if api not in ("llamacpp", "ollama") or not url.startswith(("http://", "https://")): + continue + nodes.append( + EmbedNode( + name=str(entry.get("name") or url), + url=url, + api=api, + key=str(entry.get("key") or ""), + timeout_ms=int(entry.get("timeout_ms", EMBED_TIMEOUT_MS)), + ) + ) + except (json.JSONDecodeError, TypeError, ValueError): + log.error("EMBED_CHAIN non è JSON valido: uso l'endpoint legacy") + if not nodes and default_url: + # Compatibilità legacy: endpoint singolo dagli env EMBED_* + nodes = [EmbedNode(name="embed", url=default_url.rstrip("/"), api=default_api, key=default_key, timeout_ms=EMBED_TIMEOUT_MS)] + return nodes + + +_chain: Optional[list[EmbedNode]] = None +_down_until: dict[str, float] = {} # url → monotonic deadline del cooldown +_http: Optional[httpx.AsyncClient] = None + + +def _get_chain() -> list[EmbedNode]: + global _chain + if _chain is None: + _chain = parse_chain(EMBED_CHAIN, EMBED_API, EMBED_URL, EMBED_API_KEY) + return _chain + + +def reset_chain_cache() -> None: + """Forza il re-parse della catena (usato dai test).""" + global _chain + _chain = None + _down_until.clear() + + def get_http() -> httpx.AsyncClient: global _http if _http is None: @@ -34,17 +116,58 @@ async def close_http() -> None: _http = None +def chain_nodes() -> list[EmbedNode]: + return _get_chain() + + async def embed(text: str) -> list[float]: - if EMBED_API == "llamacpp": + """Embedding con catena di fallback: ritorna il vettore o rilancia dopo l'ultimo fallimento.""" + chain = _get_chain() + if not chain: + raise RuntimeError("nessun endpoint embedding configurato") + now = time.monotonic() + live = [n for n in chain if _down_until.get(n.url, 0) <= now] + if not live: + # tutti in cooldown: ritenta comunque il primo (meglio di un fallimento immediato) + live = [chain[0]] + payload = {"model": EMBED_MODEL, "input": text} + started = time.monotonic() + last_exc: Optional[Exception] = None + for node in live: headers = {"Content-Type": "application/json"} - if EMBED_API_KEY: - headers["Authorization"] = f"Bearer {EMBED_API_KEY}" - response = await get_http().post(f"{EMBED_URL}/v1/embeddings", json={"model": EMBED_MODEL, "input": text}, headers=headers) - response.raise_for_status() - return response.json()["data"][0]["embedding"] - response = await get_http().post(f"{EMBED_URL}/api/embed", json={"model": EMBED_MODEL, "input": text}) - response.raise_for_status() - return response.json()["embeddings"][0] + if node.key: + headers["Authorization"] = f"Bearer {node.key}" + path = "/v1/embeddings" if node.api == "llamacpp" else "/api/embed" + try: + t0 = time.monotonic() + response = await get_http().post( + f"{node.url}{path}", + json=payload, + headers=headers, + timeout=httpx.Timeout(node.timeout_ms / 1000.0), + ) + response.raise_for_status() + data = response.json() + vector = data["data"][0]["embedding"] if node.api == "llamacpp" else data["embeddings"][0] + if len(vector) != EMBED_DIM: + raise ValueError(f"dimensione vettore {len(vector)} != EMBED_DIM {EMBED_DIM}") + took = int((time.monotonic() - started) * 1000) + metrics.record_embed(node.name, True, took) + return vector + except (httpx.HTTPError, ValueError, KeyError, IndexError, TypeError) as exc: + took = int((time.monotonic() - t0) * 1000) + _down_until[node.url] = time.monotonic() + EMBED_RETRY_COOLDOWN_S + last_exc = exc + metrics.record_embed(node.name, False, took) + log.warning( + "embed: nodo '%s' fallito dopo %dms (%s: %s) → cooldown %ds", + node.name, + took, + exc.__class__.__name__, + exc, + EMBED_RETRY_COOLDOWN_S, + ) + raise RuntimeError(f"tutti i nodi embedding falliti ({len(live)} tentativi)") from last_exc def get_sparse_model(): @@ -68,21 +191,32 @@ def backfill_sparse(qdrant: Any, collection: str) -> None: 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) + points, next_offset = qdrant.scroll( + collection_name=collection, + limit=100, + with_payload=["text"], + with_vectors=True, + offset=offset, + ) batch: list[qm.PointStruct] = [] - for point in points: - vectors = point.vector or {} - if SPARSE_VECTOR_NAME in vectors: + for p in points: + vecs = p.vector or {} + if SPARSE_VECTOR_NAME in vecs: continue - text = (point.payload or {}).get("text", "") - sparse = sparse_encode(text) if text else None - if sparse is not None: - batch.append(qm.PointStruct(id=point.id, vector={SPARSE_VECTOR_NAME: sparse})) + 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: + # update_vectors: aggiorna SOLO il vettore sparso, preservando payload e vettore denso + # (upsert parziale sostituirebbe l'intero punto — incidente 2026-08-16) qdrant.update_vectors(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) + log.info("backfill sparse: %d record aggiornati", updated) \ No newline at end of file diff --git a/gateway/metrics.py b/gateway/metrics.py index ee4bb47..e505ed9 100644 --- a/gateway/metrics.py +++ b/gateway/metrics.py @@ -25,6 +25,11 @@ def record_rerank(backend: str, ok: bool, took_ms: int) -> None: _metrics["rerank_duration_sum"][backend] += took_ms +def record_embed(backend: str, ok: bool, took_ms: int) -> None: + _metrics["embed_calls"][(backend, "ok" if ok else "fail")] += 1 + _metrics["embed_duration_sum"][backend] += took_ms + + def snapshot(qdrant: Any, collection: str) -> dict: try: points = qdrant.get_collection(collection).points_count @@ -41,6 +46,9 @@ def snapshot(qdrant: Any, collection: str) -> dict: "search_hits": _metrics["search_hits"], "rerank_calls": {f"{backend}:{outcome}": count for (backend, outcome), count in _metrics["rerank_calls"].items()}, "rerank_avg_ms": {backend: round(total / _metrics["rerank_calls"][(backend, "ok")], 2) for backend, total in _metrics["rerank_duration_sum"].items() if _metrics["rerank_calls"][(backend, "ok")]}, + "embed_calls": {f"{backend}:{outcome}": count for (backend, outcome), count in _metrics["embed_calls"].items()}, + "embed_avg_ms": {backend: round(total / _metrics["embed_calls"][(backend, "ok")], 2) for backend, total in _metrics["embed_duration_sum"].items() if _metrics["embed_calls"][(backend, "ok")]}, + "qdrant_retries": _metrics["qdrant_retries"], "points": points, } @@ -61,6 +69,11 @@ def prometheus_lines(qdrant: Any, collection: str) -> list[str]: lines.append(f'qmem_rerank_calls_total{{backend="{backend}",outcome="{outcome}"}} {count}') for backend, s in _metrics["rerank_duration_sum"].items(): lines.append(f'qmem_rerank_duration_seconds_sum{{backend="{backend}"}} {s / 1000:.6f}') + for (backend, outcome), count in _metrics["embed_calls"].items(): + lines.append(f'qmem_embed_calls_total{{backend="{backend}",outcome="{outcome}"}} {count}') + for backend, s in _metrics["embed_duration_sum"].items(): + lines.append(f'qmem_embed_duration_seconds_sum{{backend="{backend}"}} {s / 1000:.6f}') + lines.append(f"qmem_qdrant_retries_total {_metrics['qdrant_retries']}") try: lines.append(f"qmem_points {qdrant.get_collection(collection).points_count}") except Exception: # noqa: BLE001 diff --git a/gateway/routes.py b/gateway/routes.py index 48993db..7ecfa1e 100644 --- a/gateway/routes.py +++ b/gateway/routes.py @@ -11,6 +11,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Request from qdrant_client.http import models as qm import config +import embed import guardrail import metrics import rerank @@ -247,7 +248,7 @@ async def status(request: Request) -> dict: @router.get("/v1/version") async def version() -> dict: - return {"version": GATEWAY_VERSION, "git_commit": __import__("config").GIT_COMMIT, "guardrail_version": GUARDRAIL_VERSION, "guardrail_enabled": GUARDRAIL_ENABLED, "guardrail_block_threshold": GUARDRAIL_BLOCK_THRESHOLD, "guardrail_warn_threshold": GUARDRAIL_WARN_THRESHOLD, "embedding_model": EMBED_MODEL, "collection": COLLECTION, "rerank_enabled": rerank.enabled(), "rerank_model": config.RERANK_MODEL, "rerank_nodes": [n.name for n in rerank._get_chain()]} + return {"version": GATEWAY_VERSION, "git_commit": __import__("config").GIT_COMMIT, "guardrail_version": GUARDRAIL_VERSION, "guardrail_enabled": GUARDRAIL_ENABLED, "guardrail_block_threshold": GUARDRAIL_BLOCK_THRESHOLD, "guardrail_warn_threshold": GUARDRAIL_WARN_THRESHOLD, "embedding_model": EMBED_MODEL, "collection": COLLECTION, "embed_nodes": [n.name for n in embed.chain_nodes()], "rerank_enabled": rerank.enabled(), "rerank_model": config.RERANK_MODEL, "rerank_nodes": [n.name for n in rerank._get_chain()]} @router.get("/v1/metrics") diff --git a/gateway/state.py b/gateway/state.py index 15b78c0..187d73c 100644 --- a/gateway/state.py +++ b/gateway/state.py @@ -7,11 +7,57 @@ import json import time from typing import Any +import httpx from qdrant_client import QdrantClient -from config import QDRANT_API_KEY, QDRANT_URL +from config import QDRANT_API_KEY, QDRANT_URL, QDRANT_RETRIES, log, _metrics -qdrant = QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY) + +class ResilientQdrant: + """Proxy del client Qdrant che ritenta i metodi su errori di transport + (connessione/timeout transienti, es. riavvio del container Qdrant). + Gli errori applicativi (404, validazione) non vengono ritentati.""" + + def __init__(self, client: Any, attempts: int = QDRANT_RETRIES, backoff_s: float = 0.4): + self._client = client + self._attempts = max(1, attempts) + self._backoff = backoff_s + + @staticmethod + def _transient(exc: Exception) -> bool: + if isinstance(exc, httpx.TransportError): + return True + return type(exc).__name__ in ("ConnectionError", "TimeoutError") + + def __getattr__(self, name: str) -> Any: + attr = getattr(self._client, name) + if not callable(attr): + return attr + + def wrapped(*args: Any, **kwargs: Any) -> Any: + for attempt in range(self._attempts): + try: + return attr(*args, **kwargs) + except Exception as exc: # noqa: BLE001 + if attempt == self._attempts - 1 or not self._transient(exc): + raise + delay = self._backoff * (2**attempt) + _metrics["qdrant_retries"] += 1 + log.warning( + "qdrant.%s: errore transiente (%s: %s) → retry %d/%d tra %.1fs", + name, + exc.__class__.__name__, + exc, + attempt + 2, + self._attempts, + delay, + ) + time.sleep(delay) + + return wrapped + + +qdrant = ResilientQdrant(QdrantClient(url=QDRANT_URL, api_key=QDRANT_API_KEY)) request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-") ratelimit: dict[str, list[float]] = {} status_ratelimit: dict[str, list[float]] = {} diff --git a/gateway/tests/test_embed_chain.py b/gateway/tests/test_embed_chain.py new file mode 100644 index 0000000..76d22d6 --- /dev/null +++ b/gateway/tests/test_embed_chain.py @@ -0,0 +1,189 @@ +"""Test della catena di fallback per gli embedding e del wrapper Qdrant resilient.""" +from __future__ import annotations + +import asyncio +import json + +import httpx +import pytest + +import embed as embed_mod +import state +from test_api import auth_headers, make_record + +CHAIN = json.dumps( + [ + {"name": "primario", "url": "http://primario:9001", "api": "llamacpp", "key": "k1", "timeout_ms": 500}, + {"name": "fallback", "url": "http://fallback:9002", "api": "ollama", "key": "k2", "timeout_ms": 5000}, + ] +) + +VEC_1024 = [0.01] * 1024 + + +@pytest.fixture(autouse=True) +def _reset_chain(monkeypatch): + embed_mod.reset_chain_cache() + embed_mod._http = None + yield + embed_mod.reset_chain_cache() + embed_mod._http = None + + +def _mock_client(handler) -> list[str]: + calls: list[str] = [] + + def tracking_handler(request): + calls.append(f"{request.url.host}{request.url.path}") + return handler(request) + + embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking_handler)) + return calls + + +# --------------------------------------------------------------------------- +# parse della catena +# --------------------------------------------------------------------------- + + +def test_parse_chain_valida(monkeypatch): + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) + embed_mod.reset_chain_cache() + nodes = embed_mod.chain_nodes() + assert [n.name for n in nodes] == ["primario", "fallback"] + assert [n.api for n in nodes] == ["llamacpp", "ollama"] + + +def test_parse_legacy_quando_catena_vuota(monkeypatch): + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", "") + monkeypatch.setattr(embed_mod, "EMBED_API", "ollama") + monkeypatch.setattr(embed_mod, "EMBED_URL", "http://legacy:11434") + monkeypatch.setattr(embed_mod, "EMBED_API_KEY", "lk") + embed_mod.reset_chain_cache() + nodes = embed_mod.chain_nodes() + assert len(nodes) == 1 + assert nodes[0].name == "embed" + assert nodes[0].api == "ollama" + assert nodes[0].url == "http://legacy:11434" + assert nodes[0].key == "lk" + + +def test_parse_chain_json_invalido_cade_su_legacy(monkeypatch): + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", "non-json") + monkeypatch.setattr(embed_mod, "EMBED_URL", "http://legacy:11434") + embed_mod.reset_chain_cache() + assert [n.url for n in embed_mod.chain_nodes()] == ["http://legacy:11434"] + + +# --------------------------------------------------------------------------- +# fallback e cooldown +# --------------------------------------------------------------------------- + + +def test_fallback_primario_llamacpp_fallito(monkeypatch): + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) + embed_mod.reset_chain_cache() + + def handler(request): + if request.url.host == "primario": + return httpx.Response(500) + # nodo ollama-style: risposta con campo "embeddings" + return httpx.Response(200, json={"embeddings": [VEC_1024]}) + + embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + vector = asyncio.run(embed_mod.embed("test")) + assert vector == VEC_1024 + + +def test_cooldown_salta_primario(monkeypatch): + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) + embed_mod.reset_chain_cache() + calls: list[str] = [] + + def tracking(request): + calls.append(request.url.host) + if request.url.host == "primario": + return httpx.Response(500) + return httpx.Response(200, json={"embeddings": [VEC_1024]}) + + embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(tracking)) + asyncio.run(embed_mod.embed("test")) + asyncio.run(embed_mod.embed("test")) + # il primario fallito entra in cooldown: la seconda chiamata lo salta + assert calls == ["primario", "fallback", "fallback"] + + +def test_dimensione_errata_salta_nodo(monkeypatch): + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) + embed_mod.reset_chain_cache() + + def handler(request): + if request.url.host == "primario": + return httpx.Response(200, json={"data": [{"embedding": [0.0] * 512}]}) # dim sbagliata + return httpx.Response(200, json={"embeddings": [VEC_1024]}) + + embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + vector = asyncio.run(embed_mod.embed("test")) + assert len(vector) == 1024 + + +def test_tutti_nodi_falliti_rilancia(monkeypatch): + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) + embed_mod.reset_chain_cache() + embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(lambda r: httpx.Response(500))) + with pytest.raises(RuntimeError, match="tutti i nodi embedding falliti"): + asyncio.run(embed_mod.embed("test")) + + +def test_empty_text_comunque_chiamata(monkeypatch): + # il modello pydantic valida già la query; qui verifichiamo il passthrough + monkeypatch.setattr(embed_mod, "EMBED_CHAIN", CHAIN) + embed_mod.reset_chain_cache() + + def handler(request): + body = json.loads(request.content) + return httpx.Response(200, json={"embeddings": [VEC_1024] if body["input"] else []}) + + embed_mod._http = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + assert asyncio.run(embed_mod.embed("ok")) == VEC_1024 + + +# --------------------------------------------------------------------------- +# ResilientQdrant (retry transiente) +# --------------------------------------------------------------------------- + + +class FlakyQdrant: + def __init__(self, failures: int, exc: Exception): + self.calls = 0 + self.failures = failures + self.exc = exc + + def upsert(self, **kw): + self.calls += 1 + if self.calls <= self.failures: + raise self.exc + return "ok" + + +def test_retry_su_errore_transiente(monkeypatch): + fake = FlakyQdrant(2, httpx.ConnectError("conn")) + client = state.ResilientQdrant(fake, attempts=3, backoff_s=0.01) + assert client.upsert(x=1) == "ok" + assert fake.calls == 3 + + +def test_niente_retry_su_errore_applicativo(): + fake = FlakyQdrant(2, ValueError("404 logico")) + client = state.ResilientQdrant(fake, attempts=3, backoff_s=0.01) + with pytest.raises(ValueError): + client.upsert(x=1) + assert fake.calls == 1 + + +def test_retry_esaurito_rilancia(): + fake = FlakyQdrant(99, httpx.ReadTimeout("t")) + client = state.ResilientQdrant(fake, attempts=2, backoff_s=0.01) + with pytest.raises(httpx.ReadTimeout): + client.upsert(x=1) + assert fake.calls == 2 \ No newline at end of file