Spaces:

rahim237cm
/

jurix-api

Runtime error

App Files Files Community

jurix-api / tests /test_services /test_embedding_dimensions.py

rahim237cm

Claude Opus 5.5

feat(embeddings): voyage-4-large remplace EmbeddingGemma (API Voyage, halfvec 1024)

57c8117 25 days ago

Raw History Blame Contribute Delete

11.7 kB

"""
Tests de la dimension, de la normalisation et de la cle de cache des embeddings.
Chaque test correspond a un mecanisme qui n'existait pas :
- output_dimensionality n'etait pas demande a l'API ;
- la renormalisation des sorties tronquees (Matryoshka) n'existait pas, alors
  que l'operateur <=> de pgvector suppose des vecteurs unitaires ;
- le chemin batch n'avait ni garde de dimension ni controle du nombre de
  vecteurs rendus ;
- la cle de cache ignorait la dimension (un vecteur 3072 aurait ete resservi
  sous configuration 1536) et minusculait le texte.
"""

import json

import httpx
import numpy as np
import pytest

from app.services import embedding_service as module
from app.services.embedding_service import EmbeddingService, EmbeddingServiceError


@pytest.fixture
def recorder(monkeypatch):
    """Doublure HTTP de Voyage qui enregistre le corps de chaque requete."""
    state = {"payloads": [], "n_returned": None, "dim": None, "scale": 9.0}

    def _repondre(requete: httpx.Request) -> httpx.Response:
        payload = json.loads(requete.content)
        state["payloads"].append(payload)
        dim = state["dim"] or payload["output_dimension"]
        n = state["n_returned"] if state["n_returned"] is not None else len(payload["input"])
        rng = np.random.default_rng(7)
        # Vecteurs NON unitaires : c'est le service qui doit normaliser.
        return httpx.Response(200, json={"data": [
            {"index": i, "embedding": (rng.random(dim) * state["scale"]).tolist()}
            for i in range(n)
        ]})

    monkeypatch.setattr(module, "transport_http", httpx.MockTransport(_repondre))
    return state


@pytest.fixture
def service(recorder):
    return EmbeddingService(use_cache=False)


class TestDimension:

    def test_output_dimension_is_requested(self, service, recorder):
        service.generate_embedding("Article premier du code civil")

        assert recorder["payloads"][-1]["output_dimension"] == EmbeddingService.EMBEDDING_DIM
        # Valeur absolue et non derivee : le but est de remarquer un changement
        # accidentel de dimension, pas de le suivre.
        assert EmbeddingService.EMBEDDING_DIM == 1024

    def test_embedding_has_configured_dimension(self, service):
        embedding = service.generate_embedding("Article premier")

        assert embedding.shape == (EmbeddingService.EMBEDDING_DIM,)

    def test_batch_rejects_wrong_dimension(self, service, recorder):
        recorder["dim"] = 512  # l'API renvoie autre chose que ce qui est demande

        with pytest.raises(EmbeddingServiceError, match="Dimension"):
            service.generate_batch_embeddings(["a", "b"])

    def test_batch_rejects_truncated_response(self, service, recorder):
        """
        Deux vecteurs pour trois textes : zip() perdait silencieusement la
        queue du lot, puis un KeyError tombait bien plus loin.
        """
        recorder["n_returned"] = 2

        with pytest.raises(EmbeddingServiceError, match="incomplète"):
            service.generate_batch_embeddings(["a", "b", "c"])


class TestNormalisation:

    def test_embedding_is_unit_norm(self, service):
        embedding = service.generate_embedding("Responsabilité des dirigeants")

        assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5

    def test_batch_embeddings_are_unit_norm(self, service):
        embeddings = service.generate_batch_embeddings(["un", "deux", "trois"])

        for embedding in embeddings:
            assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5

    def test_normalisation_applies_even_when_not_requested(self, service):
        """
        La normalisation est INCONDITIONNELLE, a toute dimension y compris la
        native.
        Une version precedente la sautait quand normalize=False a la dimension
        native, en supposant que l'API renvoie deja un vecteur unitaire. C'est
        vrai de l'API, mais ca fait dependre d'un reglage une garantie dont tout
        l'aval depend : similarity() est un np.dot nu, et le score
        1 - distance borne a [0, 1] ne vaut que sur la sphere unite.
        """
        embedding = service.generate_embedding("Test", normalize=False)

        assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5


class TestCacheKey:

    def test_key_isolates_dimension(self, service):
        key_native = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT)
        original = EmbeddingService.EMBEDDING_DIM
        try:
            # Une dimension differente de celle configuree, sinon les deux cles
            # sont identiques et le test ne verifie rien.
            service.EMBEDDING_DIM = 512
            key_512 = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT)
        finally:
            service.EMBEDDING_DIM = original

        assert key_native != key_512

    def test_key_isolates_task_type(self, service):
        as_document = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT)
        as_query = service._cache_key("texte", EmbeddingService.TASK_QUERY)

        assert as_document != as_query

    def test_key_is_case_sensitive(self, service):
        """
        La cle minusculait le texte : "Article PREMIER" et "article premier"
        partageaient une entree alors que le modele les encode differemment.
        """
        upper = service._cache_key("Article PREMIER", EmbeddingService.TASK_DOCUMENT)
        lower = service._cache_key("article premier", EmbeddingService.TASK_DOCUMENT)

        assert upper != lower


class TestCacheRead:

    @pytest.mark.asyncio
    async def test_cache_read_rejects_wrong_dimension(self, recorder, db_session):
        """
        Une entree ecrite sous une autre dimension (ici 3072, la configuration
        precedente) doit etre ignoree, pas servie : elle serait rejetee par la
        colonne, ou pire, comparee de travers.
        """
        from sqlalchemy import text as sql_text

        service = EmbeddingService(use_cache=True)
        key = service._cache_key("texte périmé", EmbeddingService.TASK_DOCUMENT)

        await db_session.execute(
            sql_text(
                "INSERT INTO embedding_cache (text_hash, embedding_json, expires_at) "
                "VALUES (:key, :data, now() + interval '1 day')"
            ),
            {"key": key, "data": json.dumps([0.1] * 3072)},
        )
        await db_session.commit()

        assert service._get_from_pg_cache(key) is None


class TestOffLoop:

    @pytest.mark.asyncio
    async def test_generate_embedding_async_runs_off_the_event_loop(self, service):
        """
        La recherche semantique appelait generate_embedding depuis une
        coroutine : la boucle etait gelee pour l'aller-retour reseau, le
        time.sleep du backoff et les E/S psycopg2 du cache.
        """
        import threading

        loop_thread = threading.get_ident()
        seen = {}

        original = service._produire

        def _spy(*args, **kwargs):
            seen["thread"] = threading.get_ident()
            return original(*args, **kwargs)

        service._produire = _spy

        embedding = await service.generate_embedding_async("Une question")

        assert embedding.shape == (EmbeddingService.EMBEDDING_DIM,)
        assert seen["thread"] != loop_thread

    @pytest.mark.asyncio
    async def test_async_wrapper_defaults_to_query_task(self, service, recorder):
        """Son seul appelant encode une question, pas un document."""
        await service.generate_embedding_async("Une question")

        assert recorder["payloads"][-1]["input_type"] == "query"

    def test_batch_encodes_documents(self, service, recorder):
        """
        L'autre moitie de l'asymetrie de la recherche : une inversion encoderait
        les articles comme des questions, sans erreur.
        """
        service.generate_batch_embeddings(["Article premier", "Article 2"])

        assert recorder["payloads"][-1]["input_type"] == "document"


class TestRequestTimeout:
    """
    Le client Gemini doit imposer un delai maximal.
    Sans http_options, le client attend indefiniment. Observe en conditions
    reelles pendant l'ingestion du corpus : le processus est reste 40 minutes
    endormi sur une lecture de prise reseau, vivant, sans consommer d'UC.
    Aucune exception n'etant levee, la boucle de reprise du service ne se
    declenchait jamais — un blocage silencieux et sans fin.
    """

    def test_embedding_client_has_a_timeout(self):
        # Sans doublure : c'est le VRAI client Gemini qu'on inspecte. Sa
        # construction n'ouvre aucune connexion. Les delais de Voyage sont
        # verifies dans test_embedding_voyage.py.
        from app.core.config import settings
        from app.services.embedding_service import GeminiProvider

        service = EmbeddingService(use_cache=False, provider=GeminiProvider(api_key="x"))
        configured = service.client._api_client._http_options.timeout

        assert configured == settings.GEMINI_TIMEOUT_S * 1000  # millisecondes
        assert configured > 0

    def test_generation_client_has_a_timeout(self):
        from app.core.config import settings
        from app.services.gemini_service import GeminiService

        service = GeminiService()

        assert service.client._api_client._http_options.timeout == (
            settings.GEMINI_TIMEOUT_S * 1000
        )


class TestDailyQuota:
    """
    Un quota JOURNALIER ne se rejoue pas.
    Google renvoie 429 aussi bien pour un quota par minute que par jour, et
    suggere un retryDelay d'une trentaine de secondes dans les deux cas. Pour un
    quota journalier ce delai est trompeur : reessayer trois fois brule une
    minute avant d'echouer de toute facon, et le lot suivant recommence. Seul le
    quotaId permet de trancher.
    """

    def test_recognises_a_daily_quota(self):
        from app.services.embedding_service import _is_daily_quota_error

        error = Exception(
            "429 RESOURCE_EXHAUSTED {'quotaId': "
            "'EmbedContentRequestsPerDayPerUserPerProjectPerModel-FreeTier'}"
        )

        assert _is_daily_quota_error(error) is True

    def test_a_per_minute_quota_stays_retryable(self):
        from app.services.embedding_service import _is_daily_quota_error

        error = Exception(
            "429 RESOURCE_EXHAUSTED {'quotaId': 'EmbedContentRequestsPerMinutePerProject'}"
        )

        assert _is_daily_quota_error(error) is False

    def test_other_errors_are_not_quota(self):
        from app.services.embedding_service import _is_daily_quota_error

        assert _is_daily_quota_error(Exception("500 INTERNAL")) is False
        assert _is_daily_quota_error(Exception("401 UNAUTHENTICATED")) is False

    def test_batch_raises_immediately_without_retrying(self, monkeypatch):
        """Aucune tentative supplementaire : on leve au premier 429 journalier."""
        from app.services.embedding_service import GeminiProvider, QuotaExhaustedError

        service = EmbeddingService(use_cache=False, provider=GeminiProvider(api_key="x"))

        calls = {"n": 0}

        def _always_quota(**kwargs):
            calls["n"] += 1
            raise RuntimeError(
                "429 RESOURCE_EXHAUSTED {'quotaId': "
                "'EmbedContentRequestsPerDayPerUserPerProjectPerModel-FreeTier'}"
            )

        monkeypatch.setattr(service.client.models, "embed_content", _always_quota)
        monkeypatch.setattr(module.time, "sleep", lambda *_: None)

        with pytest.raises(QuotaExhaustedError):
            service.generate_batch_embeddings(["a", "b"])

        assert calls["n"] == 1, "un quota journalier ne doit pas etre rejoue"

	"""
	Tests de la dimension, de la normalisation et de la cle de cache des embeddings.

	Chaque test correspond a un mecanisme qui n'existait pas :

	- output_dimensionality n'etait pas demande a l'API ;
	- la renormalisation des sorties tronquees (Matryoshka) n'existait pas, alors
	que l'operateur <=> de pgvector suppose des vecteurs unitaires ;
	- le chemin batch n'avait ni garde de dimension ni controle du nombre de
	vecteurs rendus ;
	- la cle de cache ignorait la dimension (un vecteur 3072 aurait ete resservi
	sous configuration 1536) et minusculait le texte.
	"""

	import json

	import httpx
	import numpy as np
	import pytest

	from app.services import embedding_service as module
	from app.services.embedding_service import EmbeddingService, EmbeddingServiceError


	@pytest.fixture
	def recorder(monkeypatch):
	"""Doublure HTTP de Voyage qui enregistre le corps de chaque requete."""
	state = {"payloads": [], "n_returned": None, "dim": None, "scale": 9.0}

	def _repondre(requete: httpx.Request) -> httpx.Response:
	payload = json.loads(requete.content)
	state["payloads"].append(payload)
	dim = state["dim"] or payload["output_dimension"]
	n = state["n_returned"] if state["n_returned"] is not None else len(payload["input"])
	rng = np.random.default_rng(7)
	# Vecteurs NON unitaires : c'est le service qui doit normaliser.
	return httpx.Response(200, json={"data": [
	{"index": i, "embedding": (rng.random(dim) * state["scale"]).tolist()}
	for i in range(n)
	]})

	monkeypatch.setattr(module, "transport_http", httpx.MockTransport(_repondre))
	return state


	@pytest.fixture
	def service(recorder):
	return EmbeddingService(use_cache=False)


	class TestDimension:

	def test_output_dimension_is_requested(self, service, recorder):
	service.generate_embedding("Article premier du code civil")

	assert recorder["payloads"][-1]["output_dimension"] == EmbeddingService.EMBEDDING_DIM
	# Valeur absolue et non derivee : le but est de remarquer un changement
	# accidentel de dimension, pas de le suivre.
	assert EmbeddingService.EMBEDDING_DIM == 1024

	def test_embedding_has_configured_dimension(self, service):
	embedding = service.generate_embedding("Article premier")

	assert embedding.shape == (EmbeddingService.EMBEDDING_DIM,)

	def test_batch_rejects_wrong_dimension(self, service, recorder):
	recorder["dim"] = 512 # l'API renvoie autre chose que ce qui est demande

	with pytest.raises(EmbeddingServiceError, match="Dimension"):
	service.generate_batch_embeddings(["a", "b"])

	def test_batch_rejects_truncated_response(self, service, recorder):
	"""
	Deux vecteurs pour trois textes : zip() perdait silencieusement la
	queue du lot, puis un KeyError tombait bien plus loin.
	"""
	recorder["n_returned"] = 2

	with pytest.raises(EmbeddingServiceError, match="incomplète"):
	service.generate_batch_embeddings(["a", "b", "c"])


	class TestNormalisation:

	def test_embedding_is_unit_norm(self, service):
	embedding = service.generate_embedding("Responsabilité des dirigeants")

	assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5

	def test_batch_embeddings_are_unit_norm(self, service):
	embeddings = service.generate_batch_embeddings(["un", "deux", "trois"])

	for embedding in embeddings:
	assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5

	def test_normalisation_applies_even_when_not_requested(self, service):
	"""
	La normalisation est INCONDITIONNELLE, a toute dimension y compris la
	native.

	Une version precedente la sautait quand normalize=False a la dimension
	native, en supposant que l'API renvoie deja un vecteur unitaire. C'est
	vrai de l'API, mais ca fait dependre d'un reglage une garantie dont tout
	l'aval depend : similarity() est un np.dot nu, et le score
	1 - distance borne a [0, 1] ne vaut que sur la sphere unite.
	"""
	embedding = service.generate_embedding("Test", normalize=False)

	assert abs(float(np.linalg.norm(embedding)) - 1.0) < 1e-5


	class TestCacheKey:

	def test_key_isolates_dimension(self, service):
	key_native = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT)
	original = EmbeddingService.EMBEDDING_DIM
	try:
	# Une dimension differente de celle configuree, sinon les deux cles
	# sont identiques et le test ne verifie rien.
	service.EMBEDDING_DIM = 512
	key_512 = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT)
	finally:
	service.EMBEDDING_DIM = original

	assert key_native != key_512

	def test_key_isolates_task_type(self, service):
	as_document = service._cache_key("texte", EmbeddingService.TASK_DOCUMENT)
	as_query = service._cache_key("texte", EmbeddingService.TASK_QUERY)

	assert as_document != as_query

	def test_key_is_case_sensitive(self, service):
	"""
	La cle minusculait le texte : "Article PREMIER" et "article premier"
	partageaient une entree alors que le modele les encode differemment.
	"""
	upper = service._cache_key("Article PREMIER", EmbeddingService.TASK_DOCUMENT)
	lower = service._cache_key("article premier", EmbeddingService.TASK_DOCUMENT)

	assert upper != lower


	class TestCacheRead:

	@pytest.mark.asyncio
	async def test_cache_read_rejects_wrong_dimension(self, recorder, db_session):
	"""
	Une entree ecrite sous une autre dimension (ici 3072, la configuration
	precedente) doit etre ignoree, pas servie : elle serait rejetee par la
	colonne, ou pire, comparee de travers.
	"""
	from sqlalchemy import text as sql_text

	service = EmbeddingService(use_cache=True)
	key = service._cache_key("texte périmé", EmbeddingService.TASK_DOCUMENT)

	await db_session.execute(
	sql_text(
	"INSERT INTO embedding_cache (text_hash, embedding_json, expires_at) "
	"VALUES (:key, :data, now() + interval '1 day')"
	),
	{"key": key, "data": json.dumps([0.1] * 3072)},
	)
	await db_session.commit()

	assert service._get_from_pg_cache(key) is None


	class TestOffLoop:

	@pytest.mark.asyncio
	async def test_generate_embedding_async_runs_off_the_event_loop(self, service):
	"""
	La recherche semantique appelait generate_embedding depuis une
	coroutine : la boucle etait gelee pour l'aller-retour reseau, le
	time.sleep du backoff et les E/S psycopg2 du cache.
	"""
	import threading

	loop_thread = threading.get_ident()
	seen = {}

	original = service._produire

	def _spy(args, *kwargs):
	seen["thread"] = threading.get_ident()
	return original(args, *kwargs)

	service._produire = _spy

	embedding = await service.generate_embedding_async("Une question")

	assert embedding.shape == (EmbeddingService.EMBEDDING_DIM,)
	assert seen["thread"] != loop_thread

	@pytest.mark.asyncio
	async def test_async_wrapper_defaults_to_query_task(self, service, recorder):
	"""Son seul appelant encode une question, pas un document."""
	await service.generate_embedding_async("Une question")

	assert recorder["payloads"][-1]["input_type"] == "query"

	def test_batch_encodes_documents(self, service, recorder):
	"""
	L'autre moitie de l'asymetrie de la recherche : une inversion encoderait
	les articles comme des questions, sans erreur.
	"""
	service.generate_batch_embeddings(["Article premier", "Article 2"])

	assert recorder["payloads"][-1]["input_type"] == "document"


	class TestRequestTimeout:
	"""
	Le client Gemini doit imposer un delai maximal.

	Sans http_options, le client attend indefiniment. Observe en conditions
	reelles pendant l'ingestion du corpus : le processus est reste 40 minutes
	endormi sur une lecture de prise reseau, vivant, sans consommer d'UC.
	Aucune exception n'etant levee, la boucle de reprise du service ne se
	declenchait jamais — un blocage silencieux et sans fin.
	"""

	def test_embedding_client_has_a_timeout(self):
	# Sans doublure : c'est le VRAI client Gemini qu'on inspecte. Sa
	# construction n'ouvre aucune connexion. Les delais de Voyage sont
	# verifies dans test_embedding_voyage.py.
	from app.core.config import settings
	from app.services.embedding_service import GeminiProvider

	service = EmbeddingService(use_cache=False, provider=GeminiProvider(api_key="x"))
	configured = service.client._api_client._http_options.timeout

	assert configured == settings.GEMINI_TIMEOUT_S * 1000 # millisecondes
	assert configured > 0

	def test_generation_client_has_a_timeout(self):
	from app.core.config import settings
	from app.services.gemini_service import GeminiService

	service = GeminiService()

	assert service.client._api_client._http_options.timeout == (
	settings.GEMINI_TIMEOUT_S * 1000
	)


	class TestDailyQuota:
	"""
	Un quota JOURNALIER ne se rejoue pas.

	Google renvoie 429 aussi bien pour un quota par minute que par jour, et
	suggere un retryDelay d'une trentaine de secondes dans les deux cas. Pour un
	quota journalier ce delai est trompeur : reessayer trois fois brule une
	minute avant d'echouer de toute facon, et le lot suivant recommence. Seul le
	quotaId permet de trancher.
	"""

	def test_recognises_a_daily_quota(self):
	from app.services.embedding_service import _is_daily_quota_error

	error = Exception(
	"429 RESOURCE_EXHAUSTED {'quotaId': "
	"'EmbedContentRequestsPerDayPerUserPerProjectPerModel-FreeTier'}"
	)

	assert _is_daily_quota_error(error) is True

	def test_a_per_minute_quota_stays_retryable(self):
	from app.services.embedding_service import _is_daily_quota_error

	error = Exception(
	"429 RESOURCE_EXHAUSTED {'quotaId': 'EmbedContentRequestsPerMinutePerProject'}"
	)

	assert _is_daily_quota_error(error) is False

	def test_other_errors_are_not_quota(self):
	from app.services.embedding_service import _is_daily_quota_error

	assert _is_daily_quota_error(Exception("500 INTERNAL")) is False
	assert _is_daily_quota_error(Exception("401 UNAUTHENTICATED")) is False

	def test_batch_raises_immediately_without_retrying(self, monkeypatch):
	"""Aucune tentative supplementaire : on leve au premier 429 journalier."""
	from app.services.embedding_service import GeminiProvider, QuotaExhaustedError

	service = EmbeddingService(use_cache=False, provider=GeminiProvider(api_key="x"))

	calls = {"n": 0}

	def _always_quota(**kwargs):
	calls["n"] += 1
	raise RuntimeError(
	"429 RESOURCE_EXHAUSTED {'quotaId': "
	"'EmbedContentRequestsPerDayPerUserPerProjectPerModel-FreeTier'}"
	)

	monkeypatch.setattr(service.client.models, "embed_content", _always_quota)
	monkeypatch.setattr(module.time, "sleep", lambda *_: None)

	with pytest.raises(QuotaExhaustedError):
	service.generate_batch_embeddings(["a", "b"])

	assert calls["n"] == 1, "un quota journalier ne doit pas etre rejoue"