From e6db59ee8cb21c5a176556e8ae61e39a14c15b83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Louis-F=C3=A9lix=20Nothias?= Date: Fri, 10 Jul 2026 13:13:27 +0200 Subject: [PATCH 1/5] fix(embeddings): fail loudly instead of storing zero vectors when embedding fails --- CHANGELOG.md | 15 ++ src/perspicacite/llm/embeddings.py | 23 ++- src/perspicacite/rag/dynamic_kb.py | 7 + src/perspicacite/retrieval/chroma_store.py | 53 +++++- src/perspicacite/search/screening.py | 8 + src/perspicacite/web/app.py | 10 +- src/perspicacite/web/routers/kb.py | 30 ++++ tests/unit/test_embedding_degeneracy.py | 177 +++++++++++++++++++++ 8 files changed, 318 insertions(+), 5 deletions(-) create mode 100644 tests/unit/test_embedding_degeneracy.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d3071ad..4d7ca447 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed +- Ingest and retrieval no longer degrade silently when the embedding provider fails (for + example on an exhausted API quota). A zero vector is never stored or queried: `add_documents` + rejects a zero-norm vector produced for a non-empty chunk, `search` rejects a zero-norm query, + and `add_papers` propagates the failure instead of reporting a successful ingest of zero + chunks. Previously a poisoned collection answered every query with the same passages at a + constant score of 0.5 while the API reported `success: true`. +- `add_documents` rejects a provider that returns fewer vectors than texts, which otherwise + assigned each embedding to the wrong chunk. `screening.py` gained the same guard. + +### Added +- `GET /api/kb/{name}/stats` reports an `embedding_health` block (`probed_chunks`, + `zero_vector_chunks`, `degraded`) so a poisoned knowledge base is visible without reading + Chroma by hand. + ### Changed - **BREAKING:** the `indicia` and `adapters` optional extras are removed. They required the private, unpublished `indicium` stack, and `uv lock` resolves every extra whether or not it diff --git a/src/perspicacite/llm/embeddings.py b/src/perspicacite/llm/embeddings.py index 0a98a311..b026d487 100644 --- a/src/perspicacite/llm/embeddings.py +++ b/src/perspicacite/llm/embeddings.py @@ -1,5 +1,6 @@ """Embedding providers for vector search.""" +import math from pathlib import Path from typing import Any, Protocol @@ -8,8 +9,28 @@ logger = get_logger("perspicacite.llm.embeddings") +class EmbeddingFailedError(RuntimeError): + """An embedding could not be produced, so no vector may be stored or queried. + + Raised instead of substituting a zero vector. A zero vector is not a + neutral placeholder: in cosine space it sits at distance 1.0 from every + other vector, which silently turns retrieval into a constant score over an + arbitrary ordering rather than an error. + """ + + +def is_zero_vector(vector: list[float], tolerance: float = 1e-12) -> bool: + """Return True when the vector's L2 norm is indistinguishable from zero.""" + return math.sqrt(sum(component * component for component in vector)) <= tolerance + + class EmbeddingProvider(Protocol): - """Protocol for embedding providers.""" + """Protocol for embedding providers. + + Contract for implementers: raise on failure, never substitute a zero + vector for a text that has content. A zero vector is only ever a valid + answer for input that is itself empty or whitespace. + """ async def embed(self, texts: list[str]) -> list[list[float]]: """Embed a list of texts (document mode).""" diff --git a/src/perspicacite/rag/dynamic_kb.py b/src/perspicacite/rag/dynamic_kb.py index 796f3bc4..9da3b765 100644 --- a/src/perspicacite/rag/dynamic_kb.py +++ b/src/perspicacite/rag/dynamic_kb.py @@ -9,6 +9,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any +from perspicacite.llm.embeddings import EmbeddingFailedError from perspicacite.logging import get_logger from perspicacite.rag.paper_metadata_codec import decode_paper_metadata_json from perspicacite.rag.query_scope import PaperScopeResult, merge_scope_with_candidates @@ -116,6 +117,12 @@ async def add_papers( total_added += docs_added self._paper_ids.add(paper.id) + except EmbeddingFailedError: + # The embedder is down (bad key, exhausted quota). Every + # remaining paper would fail the same way, and swallowing it + # would report a successful ingest of zero usable chunks. + logger.error("add_paper_embedding_failed", paper_id=paper.id) + raise except Exception as e: logger.error("add_paper_error", paper_id=paper.id, error=str(e)) diff --git a/src/perspicacite/retrieval/chroma_store.py b/src/perspicacite/retrieval/chroma_store.py index 9fedacae..19082cd4 100644 --- a/src/perspicacite/retrieval/chroma_store.py +++ b/src/perspicacite/retrieval/chroma_store.py @@ -10,7 +10,7 @@ except ImportError: IncludeEnum = None # Will use literal values instead -from perspicacite.llm.embeddings import EmbeddingProvider +from perspicacite.llm.embeddings import EmbeddingFailedError, EmbeddingProvider, is_zero_vector from perspicacite.logging import get_logger from perspicacite.models.documents import ChunkMetadata, DocumentChunk from perspicacite.models.search import RetrievedChunk, SearchFilters @@ -18,6 +18,34 @@ logger = get_logger("perspicacite.retrieval.chroma") +def _reject_degenerate_embeddings( + texts: list[str], embeddings: list[list[float]], collection: str +) -> None: + """Refuse embeddings that would poison the collection. + + Chroma accepts a zero vector and then reports cosine distance 1.0 against + everything, so a poisoned collection answers every query with the same + passages at the same score. Catch it at the boundary instead. + """ + if len(embeddings) != len(texts): + raise EmbeddingFailedError( + f"embedding provider returned {len(embeddings)} vector(s) for " + f"{len(texts)} text(s) in collection {collection!r}; refusing to " + "store misaligned embeddings" + ) + degenerate = [ + index + for index, (text, vector) in enumerate(zip(texts, embeddings, strict=True)) + if text.strip() and is_zero_vector(vector) + ] + if degenerate: + raise EmbeddingFailedError( + f"embedding provider returned a zero vector for {len(degenerate)} " + f"non-empty chunk(s) in collection {collection!r} " + f"(first at index {degenerate[0]}); refusing to store them" + ) + + class ChromaVectorStore: """ ChromaDB-backed vector store. @@ -177,8 +205,15 @@ def _ctx_for(i: int) -> str | None: count=len(texts_to_embed), collection=collection, ) - embeddings = await self.embedding_provider.embed(texts_to_embed) - for idx, embedding in zip(indices_to_embed, embeddings): + try: + embeddings = await self.embedding_provider.embed(texts_to_embed) + except Exception as exc: + raise EmbeddingFailedError( + f"embedding {len(texts_to_embed)} chunk(s) for collection " + f"{collection!r} failed: {exc}" + ) from exc + _reject_degenerate_embeddings(texts_to_embed, embeddings, collection) + for idx, embedding in zip(indices_to_embed, embeddings, strict=True): chunks[idx].embedding = embedding # Prepare data for Chroma (lists must stay aligned — never filter embeddings) @@ -229,7 +264,19 @@ async def search( Returns: List of retrieved chunks with scores + + Raises: + EmbeddingFailedError: the query vector has zero norm, which Chroma + would answer with distance 1.0 for every document — a uniform + score over an arbitrary ordering rather than a ranking. """ + if is_zero_vector(query_embedding): + raise EmbeddingFailedError( + f"refusing to search collection {collection!r} with a zero-norm query " + "embedding: every document would score identically. The embedding " + "provider most likely failed (check the API key and its quota)." + ) + try: coll = self.client.get_collection(name=collection) except Exception as e: diff --git a/src/perspicacite/search/screening.py b/src/perspicacite/search/screening.py index 36aaad57..e82b3dbd 100644 --- a/src/perspicacite/search/screening.py +++ b/src/perspicacite/search/screening.py @@ -8,6 +8,7 @@ from rank_bm25 import BM25Plus +from perspicacite.llm.embeddings import EmbeddingFailedError from perspicacite.logging import get_logger if TYPE_CHECKING: @@ -487,6 +488,13 @@ async def screen_papers_embedding( try: ref_vecs = await embedding_provider.embed(list(flat)) cand_vecs = await embedding_provider.embed(to_embed) if to_embed else [] + # `bounds` indexes ref_vecs positionally, so a provider that drops + # empty texts would score candidates against the wrong paper. + if len(ref_vecs) != len(flat): + raise EmbeddingFailedError( + f"embedding provider returned {len(ref_vecs)} vector(s) for " + f"{len(flat)} reference text(s); refusing to score misaligned vectors" + ) except Exception as exc: return [ ScreenResult(item=c, score=0.0, kept=False, reason=f"embedding_error: {exc}") diff --git a/src/perspicacite/web/app.py b/src/perspicacite/web/app.py index 8475d933..c857dc40 100644 --- a/src/perspicacite/web/app.py +++ b/src/perspicacite/web/app.py @@ -30,9 +30,10 @@ # Now safe to import modules that use structlog. from fastapi import FastAPI -from fastapi.responses import HTMLResponse, Response +from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles +from perspicacite.llm.embeddings import EmbeddingFailedError from perspicacite.web.routers import ( chat as chat_router, ) @@ -105,6 +106,13 @@ async def _no_cache_for_assets(request, call_next): return response +@app.exception_handler(EmbeddingFailedError) +async def _embedding_failed_handler(request, exc: EmbeddingFailedError): + """Report a dead embedder as an upstream failure, never as a successful ingest.""" + logger.error("Embedding failed on %s: %s", request.url.path, exc) + return JSONResponse(status_code=502, content={"success": False, "error": str(exc)}) + + # Mount routers app.include_router(health_router.router) app.include_router(conversations_router.router) diff --git a/src/perspicacite/web/routers/kb.py b/src/perspicacite/web/routers/kb.py index 20bb8673..aef96f82 100644 --- a/src/perspicacite/web/routers/kb.py +++ b/src/perspicacite/web/routers/kb.py @@ -19,6 +19,7 @@ ingest_local_documents, validate_local_path, ) +from perspicacite.llm.embeddings import is_zero_vector from perspicacite.models.kb import ( ChunkConfig, KnowledgeBase, @@ -749,6 +750,32 @@ async def add_papers_to_kb(name: str, request: KBAddPapersRequest): } +EMBEDDING_PROBE_LIMIT = 128 + + +def _probe_embedding_health(coll, total_chunks: int) -> dict: + """Sample stored vectors and count the degenerate (zero-norm) ones. + + A zero vector sits at cosine distance 1.0 from every other vector, so a KB + full of them answers every query with the same passages at the same score. + Sampling a bounded prefix surfaces that; fetching every vector would not be + affordable on a large collection. + """ + empty = {"probed_chunks": 0, "zero_vector_chunks": 0, "degraded": False} + if not total_chunks: + return empty + probe = coll.get(limit=min(total_chunks, EMBEDDING_PROBE_LIMIT), include=["embeddings"]) + vectors = probe.get("embeddings") + if vectors is None or len(vectors) == 0: + return empty + zero_count = sum(1 for vector in vectors if is_zero_vector(list(vector))) + return { + "probed_chunks": len(vectors), + "zero_vector_chunks": zero_count, + "degraded": zero_count > 0, + } + + @router.get("/api/kb/{name}/stats") async def get_kb_stats(name: str): """Aggregate statistics for a KB, computed from ChromaDB metadata + the SQLite KB record.""" @@ -758,6 +785,7 @@ async def get_kb_stats(name: str): if not kb: return {"error": f"Knowledge base '{name}' not found"} scan_cap = 20000 + embedding_health = {"probed_chunks": 0, "zero_vector_chunks": 0, "degraded": False} try: coll = app_state.vector_store.client.get_collection(name=kb.collection_name) total_chunks = coll.count() @@ -765,6 +793,7 @@ async def get_kb_stats(name: str): limit=min(total_chunks, scan_cap) if total_chunks else 0, include=["metadatas"] ) metas = got.get("metadatas") or [] + embedding_health = _probe_embedding_health(coll, total_chunks) except Exception as e: logger.warning(f"kb stats: collection scan failed for {name}: {e}") metas, total_chunks = [], 0 @@ -801,6 +830,7 @@ async def get_kb_stats(name: str): "created_at": kb.created_at.isoformat() if kb.created_at else None, "scanned_chunks": len(metas), "scan_capped": total_chunks > scan_cap if total_chunks else False, + "embedding_health": embedding_health, } diff --git a/tests/unit/test_embedding_degeneracy.py b/tests/unit/test_embedding_degeneracy.py new file mode 100644 index 00000000..1c8cdefd --- /dev/null +++ b/tests/unit/test_embedding_degeneracy.py @@ -0,0 +1,177 @@ +"""Regression tests: a failing embedder must never produce a silent zero vector. + +A zero vector is not a neutral placeholder. Chroma stores it happily and then +reports cosine distance 1.0 against every document, so `score = 1/(1+distance)` +collapses to a constant 0.5 and retrieval returns arbitrary passages while the +API still reports success. These tests pin the loud behaviour at both ends: the +write path (ingest) and the read path (search). +""" + +import pytest + +from perspicacite.llm.embeddings import EmbeddingFailedError, is_zero_vector +from perspicacite.models.documents import ChunkMetadata, DocumentChunk +from perspicacite.models.papers import Paper, PaperSource +from perspicacite.retrieval.chroma_store import ChromaVectorStore + +DIMENSION = 8 + + +def _chunk(chunk_id: str, text: str) -> DocumentChunk: + """Build a minimal chunk carrying the given text.""" + return DocumentChunk( + id=chunk_id, + text=text, + metadata=ChunkMetadata( + paper_id="paper-1", + chunk_index=0, + title="Test Paper", + year=2024, + source=PaperSource.BIBTEX, + ), + ) + + +class _StubProvider: + """Embedding provider whose behaviour each test dictates.""" + + def __init__(self, behaviour): + self._behaviour = behaviour + + @property + def dimension(self) -> int: + return DIMENSION + + @property + def model_name(self) -> str: + return "stub-embeddings" + + async def embed(self, texts: list[str]) -> list[list[float]]: + return self._behaviour(texts) + + async def embed_query(self, texts: list[str]) -> list[list[float]]: + return self._behaviour(texts) + + +def _store(temp_dir, behaviour) -> ChromaVectorStore: + """Chroma store backed by a provider with the given behaviour.""" + return ChromaVectorStore(persist_dir=str(temp_dir), embedding_provider=_StubProvider(behaviour)) + + +def _raises_quota_error(_texts): + raise RuntimeError("OpenAIException - Error code: 429 ... insufficient_quota") + + +def _returns_zero_vectors(texts): + return [[0.0] * DIMENSION for _ in texts] + + +def _returns_unit_vectors(texts): + return [[1.0] + [0.0] * (DIMENSION - 1) for _ in texts] + + +def _returns_too_few_vectors(texts): + """Mimic a provider that silently drops an input, as LiteLLM does for empty text.""" + return [[1.0] + [0.0] * (DIMENSION - 1) for _ in texts[:-1]] + + +def test_is_zero_vector_distinguishes_degenerate_from_real(): + """The degeneracy check keys on norm, not on individual components.""" + assert is_zero_vector([0.0] * DIMENSION) + assert not is_zero_vector([0.0] * (DIMENSION - 1) + [1e-6]) + + +@pytest.mark.asyncio +async def test_provider_failure_surfaces_and_writes_nothing(temp_dir): + """A provider that raises must abort the ingest, not zero-fill it.""" + store = _store(temp_dir, _raises_quota_error) + with pytest.raises(EmbeddingFailedError, match="insufficient_quota"): + await store.add_documents("test-kb", [_chunk("chunk-1", "real content")]) + + collection = store.client.get_or_create_collection(name="test-kb") + assert collection.count() == 0, "a failed embed must not persist any chunk" + + +@pytest.mark.asyncio +async def test_zero_vector_for_nonempty_text_is_rejected(temp_dir): + """A zero vector for a text with content means the embedder failed.""" + store = _store(temp_dir, _returns_zero_vectors) + with pytest.raises(EmbeddingFailedError, match="zero vector"): + await store.add_documents("test-kb", [_chunk("chunk-1", "real content")]) + + collection = store.client.get_or_create_collection(name="test-kb") + assert collection.count() == 0 + + +@pytest.mark.asyncio +async def test_misaligned_embedding_count_is_rejected(temp_dir): + """A short vector list would misassign every embedding after the dropped text.""" + store = _store(temp_dir, _returns_too_few_vectors) + chunks = [_chunk("chunk-1", "first"), _chunk("chunk-2", "second")] + with pytest.raises(EmbeddingFailedError, match="misaligned"): + await store.add_documents("test-kb", chunks) + + collection = store.client.get_or_create_collection(name="test-kb") + assert collection.count() == 0 + + +@pytest.mark.asyncio +async def test_zero_norm_query_errors_instead_of_scoring_everything_equally(temp_dir): + """Chroma answers a zero query with distance 1.0 for every doc -> constant 0.5.""" + store = _store(temp_dir, _returns_unit_vectors) + await store.add_documents("test-kb", [_chunk("chunk-1", "real content")]) + + with pytest.raises(EmbeddingFailedError, match="zero-norm query"): + await store.search("test-kb", [0.0] * DIMENSION, top_k=5) + + +@pytest.mark.asyncio +async def test_real_query_still_searches(temp_dir): + """The guard must not disturb a healthy query.""" + store = _store(temp_dir, _returns_unit_vectors) + await store.add_documents("test-kb", [_chunk("chunk-1", "real content")]) + + results = await store.search("test-kb", [1.0] + [0.0] * (DIMENSION - 1), top_k=5) + assert len(results) == 1 + assert results[0].chunk.id == "chunk-1" + + +@pytest.mark.asyncio +async def test_empty_text_input_is_still_graceful(): + """An empty string is legitimately zero — only failures are errors.""" + from perspicacite.llm.embeddings import LiteLLMEmbeddingProvider + + provider = LiteLLMEmbeddingProvider() + result = await provider.embed(["", " "]) + assert len(result) == 2 + assert all(is_zero_vector(vector) for vector in result) + + +def _knowledge_base_raising(error: Exception): + """A DynamicKnowledgeBase whose per-paper ingest raises the given error.""" + from unittest.mock import AsyncMock, MagicMock + + from perspicacite.rag.dynamic_kb import DynamicKnowledgeBase + + kb = DynamicKnowledgeBase(MagicMock(), MagicMock()) + kb._initialized = True + kb._add_paper = AsyncMock(side_effect=error) + return kb + + +@pytest.mark.asyncio +async def test_add_papers_propagates_embedding_failure(): + """Ingest must not report a successful run of zero chunks when the embedder is down.""" + kb = _knowledge_base_raising(EmbeddingFailedError("quota exhausted")) + + with pytest.raises(EmbeddingFailedError): + await kb.add_papers([Paper(id="p1", title="T", source=PaperSource.BIBTEX)]) + + +@pytest.mark.asyncio +async def test_add_papers_still_tolerates_ordinary_paper_errors(): + """A single unparseable paper must not abort the batch, as before.""" + kb = _knowledge_base_raising(ValueError("bad pdf")) + + added = await kb.add_papers([Paper(id="p1", title="T", source=PaperSource.BIBTEX)]) + assert added == 0 From a5582477ad9a18b33e080171920967eeee3d0fb7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Louis-F=C3=A9lix=20Nothias?= Date: Fri, 10 Jul 2026 13:34:17 +0200 Subject: [PATCH 2/5] fix(retrieval): let a degenerate-query error escape the multi-KB fan-out --- src/perspicacite/rag/modes/advanced.py | 3 + src/perspicacite/rag/modes/deep_research.py | 5 ++ src/perspicacite/retrieval/multi_kb.py | 9 ++ tests/unit/test_multi_kb_embedding_failure.py | 86 +++++++++++++++++++ 4 files changed, 103 insertions(+) create mode 100644 tests/unit/test_multi_kb_embedding_failure.py diff --git a/src/perspicacite/rag/modes/advanced.py b/src/perspicacite/rag/modes/advanced.py index f90e8d61..c2e8bbe4 100644 --- a/src/perspicacite/rag/modes/advanced.py +++ b/src/perspicacite/rag/modes/advanced.py @@ -15,6 +15,7 @@ from typing import Any from perspicacite.config.schema import MultimodalMode +from perspicacite.llm.embeddings import EmbeddingFailedError from perspicacite.logging import get_logger from perspicacite.models.kb import chroma_collection_name_for_kb from perspicacite.models.rag import RAGMode, RAGRequest, RAGResponse, SourceReference, StreamEvent @@ -1050,6 +1051,8 @@ async def _wrrf_retrieval( query_embedding=query_embedding[0], top_k=self.initial_docs, ) + except EmbeddingFailedError: + raise # a degenerate query fails against every collection except Exception as e: logger.warning( "advanced_wrrf_fanout_search_failed", diff --git a/src/perspicacite/rag/modes/deep_research.py b/src/perspicacite/rag/modes/deep_research.py index d15d7be1..6742d418 100644 --- a/src/perspicacite/rag/modes/deep_research.py +++ b/src/perspicacite/rag/modes/deep_research.py @@ -21,6 +21,7 @@ from typing import Any from perspicacite.config.schema import MultimodalMode +from perspicacite.llm.embeddings import EmbeddingFailedError from perspicacite.logging import get_logger from perspicacite.models.kb import chroma_collection_name_for_kb from perspicacite.models.rag import RAGMode, RAGRequest, RAGResponse, SourceReference, StreamEvent @@ -1411,6 +1412,8 @@ async def _wrrf_retrieval( query_embedding=query_embedding[0], top_k=self.initial_docs, ) + except EmbeddingFailedError: + raise # a degenerate query fails against every collection except Exception as e: logger.warning( "profound_wrrf_fanout_search_failed", @@ -1488,6 +1491,8 @@ async def _basic_vector_retrieve( query_embedding=query_embedding[0], top_k=self.initial_docs, ) + except EmbeddingFailedError: + raise # a degenerate query fails against every collection except Exception as e: logger.warning( "profound_basic_fanout_search_failed", diff --git a/src/perspicacite/retrieval/multi_kb.py b/src/perspicacite/retrieval/multi_kb.py index feb81e2e..b74441e9 100644 --- a/src/perspicacite/retrieval/multi_kb.py +++ b/src/perspicacite/retrieval/multi_kb.py @@ -5,6 +5,7 @@ import asyncio from typing import TYPE_CHECKING, Any +from perspicacite.llm.embeddings import EmbeddingFailedError from perspicacite.logging import get_logger if TYPE_CHECKING: @@ -103,6 +104,10 @@ async def search( query_embedding=query_embedding, top_k=top_k * 2, ) + except EmbeddingFailedError: + # A degenerate query fails against every collection, so skipping + # them one by one would return an empty result reported as success. + raise except Exception as e: logger.warning("multi_kb_search_collection_failed", collection=coll, error=str(e)) continue @@ -252,6 +257,10 @@ async def _one(coll: str) -> list[Any]: collection=coll, query_embedding=query_embedding, top_k=top_k * 2 ) return results + except EmbeddingFailedError: + # Degenerate query: every collection fails identically, and an empty + # merged result would be indistinguishable from "nothing matched". + raise except Exception as e: logger.warning("fanout_search_failed", collection=coll, error=str(e)) return [] diff --git a/tests/unit/test_multi_kb_embedding_failure.py b/tests/unit/test_multi_kb_embedding_failure.py new file mode 100644 index 00000000..64b5e25d --- /dev/null +++ b/tests/unit/test_multi_kb_embedding_failure.py @@ -0,0 +1,86 @@ +"""A degenerate query must not be swallowed by the multi-KB fan-out. + +`MultiKBRetriever.search` and `query_chunks_across_collections` skip a collection +whose search fails, so the query can still be answered from the others. That is +right for a missing collection and wrong for a dead embedder: the query vector is +degenerate for *every* collection, so skipping them one by one turns an error into +an empty result reported as a success. +""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from perspicacite.llm.embeddings import EmbeddingFailedError +from perspicacite.retrieval.multi_kb import MultiKBRetriever, query_chunks_across_collections + + +def _embedding_service(vector): + service = MagicMock() + service.embed_query = AsyncMock(return_value=[vector]) + return service + + +def _kb_meta(collection_name: str): + return SimpleNamespace(collection_name=collection_name, name=collection_name) + + +@pytest.mark.asyncio +async def test_retriever_propagates_embedding_failure(): + """A zero-norm query must surface as an error, not an empty result set.""" + store = MagicMock() + store.search = AsyncMock(side_effect=EmbeddingFailedError("zero-norm query embedding")) + retriever = MultiKBRetriever( + vector_store=store, + embedding_service=_embedding_service([0.0, 0.0]), + kb_metas=[_kb_meta("kb-one"), _kb_meta("kb-two")], + ) + + with pytest.raises(EmbeddingFailedError): + await retriever.search("anything") + + +@pytest.mark.asyncio +async def test_retriever_still_skips_a_single_broken_collection(): + """An ordinary per-collection failure is still tolerated.""" + hit = SimpleNamespace( + chunk=SimpleNamespace( + id="c1", + text="text", + metadata=SimpleNamespace(paper_id="p1", title="T", year=2024), + ), + score=0.9, + ) + + async def _search(collection, **_kwargs): + if collection == "kb-broken": + raise RuntimeError("collection missing") + return [hit] + + store = MagicMock() + store.search = AsyncMock(side_effect=_search) + retriever = MultiKBRetriever( + vector_store=store, + embedding_service=_embedding_service([1.0, 0.0]), + kb_metas=[_kb_meta("kb-broken"), _kb_meta("kb-good")], + ) + + results = await retriever.search("anything") + assert len(results) == 1 + + +@pytest.mark.asyncio +async def test_fanout_helper_propagates_embedding_failure(): + """The standalone fan-out helper has the same contract.""" + store = MagicMock() + store.search = AsyncMock(side_effect=EmbeddingFailedError("zero-norm query embedding")) + + with pytest.raises(EmbeddingFailedError): + await query_chunks_across_collections( + vector_store=store, + embedding_service=_embedding_service([0.0, 0.0]), + collection_names=["kb-one", "kb-two"], + query="anything", + top_k=5, + ) From d00210703d768bf26f190fe9bc938b6f17185be9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Louis-F=C3=A9lix=20Nothias?= Date: Fri, 10 Jul 2026 13:42:31 +0200 Subject: [PATCH 3/5] fix(retrieval): refuse to store a zero vector for an empty chunk --- src/perspicacite/retrieval/chroma_store.py | 26 ++++++++++++------- tests/unit/test_embedding_degeneracy.py | 29 +++++++++++++++++++++- 2 files changed, 45 insertions(+), 10 deletions(-) diff --git a/src/perspicacite/retrieval/chroma_store.py b/src/perspicacite/retrieval/chroma_store.py index 19082cd4..8046ef16 100644 --- a/src/perspicacite/retrieval/chroma_store.py +++ b/src/perspicacite/retrieval/chroma_store.py @@ -25,7 +25,9 @@ def _reject_degenerate_embeddings( Chroma accepts a zero vector and then reports cosine distance 1.0 against everything, so a poisoned collection answers every query with the same - passages at the same score. Catch it at the boundary instead. + passages at the same score. No zero vector may be stored, whatever produced + it: for a text with content it means the embedder failed, and for an empty + text it means the chunk has nothing to retrieve. """ if len(embeddings) != len(texts): raise EmbeddingFailedError( @@ -35,15 +37,21 @@ def _reject_degenerate_embeddings( ) degenerate = [ index - for index, (text, vector) in enumerate(zip(texts, embeddings, strict=True)) - if text.strip() and is_zero_vector(vector) + for index, vector in enumerate(embeddings) + if is_zero_vector(vector) ] - if degenerate: - raise EmbeddingFailedError( - f"embedding provider returned a zero vector for {len(degenerate)} " - f"non-empty chunk(s) in collection {collection!r} " - f"(first at index {degenerate[0]}); refusing to store them" - ) + if not degenerate: + return + first = degenerate[0] + reason = ( + "the embedding provider returned a zero vector" + if texts[first].strip() + else "the chunk has no text to embed" + ) + raise EmbeddingFailedError( + f"refusing to store {len(degenerate)} zero vector(s) in collection " + f"{collection!r} (first at index {first}): {reason}" + ) class ChromaVectorStore: diff --git a/tests/unit/test_embedding_degeneracy.py b/tests/unit/test_embedding_degeneracy.py index 1c8cdefd..1f3b304e 100644 --- a/tests/unit/test_embedding_degeneracy.py +++ b/tests/unit/test_embedding_degeneracy.py @@ -96,13 +96,40 @@ async def test_provider_failure_surfaces_and_writes_nothing(temp_dir): async def test_zero_vector_for_nonempty_text_is_rejected(temp_dir): """A zero vector for a text with content means the embedder failed.""" store = _store(temp_dir, _returns_zero_vectors) - with pytest.raises(EmbeddingFailedError, match="zero vector"): + with pytest.raises(EmbeddingFailedError, match="provider returned a zero vector"): await store.add_documents("test-kb", [_chunk("chunk-1", "real content")]) collection = store.client.get_or_create_collection(name="test-kb") assert collection.count() == 0 +@pytest.mark.asyncio +async def test_zero_vector_for_an_empty_chunk_is_also_rejected(temp_dir): + """A cache-style provider zero-fills empty text; that must not reach Chroma. + + A chunk with no text and no title composes to an empty embedding text, so a + provider preserving positional alignment returns a zero vector for it. Stored, + it would answer every query at cosine distance 1.0. + """ + + def zero_for_empty(texts): + return [[0.0] * DIMENSION if not t.strip() else _returns_unit_vectors([t])[0] for t in texts] + + store = _store(temp_dir, zero_for_empty) + bare = DocumentChunk( + id="chunk-empty", + text="", + metadata=ChunkMetadata( + paper_id="paper-1", chunk_index=0, title=None, year=None, source=PaperSource.BIBTEX + ), + ) + with pytest.raises(EmbeddingFailedError, match="no text to embed"): + await store.add_documents("test-kb", [bare]) + + collection = store.client.get_or_create_collection(name="test-kb") + assert collection.count() == 0 + + @pytest.mark.asyncio async def test_misaligned_embedding_count_is_rejected(temp_dir): """A short vector list would misassign every embedding after the dropped text.""" From 9e1e9c5a4e3517bf18479703fd4e3cc4b2c7d7ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Louis-F=C3=A9lix=20Nothias?= Date: Fri, 10 Jul 2026 13:43:12 +0200 Subject: [PATCH 4/5] style(tests): wrap a long line in the degeneracy tests --- tests/unit/test_embedding_degeneracy.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_embedding_degeneracy.py b/tests/unit/test_embedding_degeneracy.py index 1f3b304e..739659fa 100644 --- a/tests/unit/test_embedding_degeneracy.py +++ b/tests/unit/test_embedding_degeneracy.py @@ -113,7 +113,8 @@ async def test_zero_vector_for_an_empty_chunk_is_also_rejected(temp_dir): """ def zero_for_empty(texts): - return [[0.0] * DIMENSION if not t.strip() else _returns_unit_vectors([t])[0] for t in texts] + unit = [1.0] + [0.0] * (DIMENSION - 1) + return [[0.0] * DIMENSION if not t.strip() else unit for t in texts] store = _store(temp_dir, zero_for_empty) bare = DocumentChunk( From cf6b1a1c168a79b049aff950a859cf773a4634a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Louis-F=C3=A9lix=20Nothias?= Date: Fri, 10 Jul 2026 13:56:25 +0200 Subject: [PATCH 5/5] fix(retrieval): screen caller-supplied embeddings and keep the DOI checkpoint recoverable --- src/perspicacite/pipeline/search_to_kb.py | 17 ++++++++- src/perspicacite/retrieval/chroma_store.py | 14 ++++++++ tests/unit/test_embedding_degeneracy.py | 29 ++++++++++++++++ tests/unit/test_ingest_dois_resume.py | 40 ++++++++++++++++++++++ 4 files changed, 99 insertions(+), 1 deletion(-) diff --git a/src/perspicacite/pipeline/search_to_kb.py b/src/perspicacite/pipeline/search_to_kb.py index 9bdb38eb..6fec2bf6 100644 --- a/src/perspicacite/pipeline/search_to_kb.py +++ b/src/perspicacite/pipeline/search_to_kb.py @@ -37,6 +37,7 @@ from dataclasses import dataclass, field from typing import Any +from perspicacite.llm.embeddings import EmbeddingFailedError from perspicacite.logging import get_logger logger = get_logger("perspicacite.pipeline.search_to_kb") @@ -547,6 +548,9 @@ async def ingest_dois_into_kb( cookies_path = pdf_config.cookies_path papers_to_add: list[Paper] = [] + # DOIs marked "added" on retrieval but not yet embedded. The checkpoint never + # re-offers an "added" id, so the mark must be undone if the batch never lands. + pending_dois: list[str] = [] skipped: list[dict] = [] failed: list[dict[str, str]] = [] dl: dict[str, int] = {"attempted": 0, "success": 0, "failed": 0} @@ -632,6 +636,7 @@ async def ingest_dois_into_kb( # Mark as added immediately on successful retrieval (Wave 3.3). ck_state.record(doi, "added") ckpt.save(ck_state) + pending_dois.append(doi) added_chunks = 0 if papers_to_add: @@ -641,7 +646,17 @@ async def ingest_dois_into_kb( ) dkb.collection_name = collection_name dkb._initialized = True - added_chunks = await dkb.add_papers(papers_to_add, include_full_text=True) + try: + added_chunks = await dkb.add_papers(papers_to_add, include_full_text=True) + except EmbeddingFailedError: + # These DOIs were marked "added" the moment they were retrieved, before + # the batch was embedded. remaining_ids() never re-offers an "added" id, + # so leaving the mark would make a resume skip them for ever. Chroma + # ignores duplicate ids, so re-ingesting any that did land is harmless. + for pending in pending_dois: + ck_state.processed.pop(pending, None) + ckpt.save(ck_state) + raise kb_meta.paper_count = (kb_meta.paper_count or 0) + len(papers_to_add) kb_meta.chunk_count = (kb_meta.chunk_count or 0) + added_chunks await app_state.session_store.save_kb_metadata(kb_meta) diff --git a/src/perspicacite/retrieval/chroma_store.py b/src/perspicacite/retrieval/chroma_store.py index 8046ef16..1f6906d1 100644 --- a/src/perspicacite/retrieval/chroma_store.py +++ b/src/perspicacite/retrieval/chroma_store.py @@ -230,6 +230,20 @@ def _ctx_for(i: int) -> str | None: embeddings = [chunk.embedding for chunk in chunks] if any(e is None for e in embeddings): raise ValueError("All chunks must have embeddings before add_documents") + # Callers may arrive with chunk.embedding already set (capsule_builder, + # capsule_reader, local_docs embed before calling). Those vectors skip the + # check above, so screen everything that is about to be written. + poisoned = [ + chunk.id + for chunk, e in zip(chunks, embeddings, strict=True) + if is_zero_vector(e) + ] + if poisoned: + raise EmbeddingFailedError( + f"refusing to store {len(poisoned)} zero vector(s) in collection " + f"{collection!r} (first chunk: {poisoned[0]!r}); a zero vector matches " + "every query at the same score" + ) metadatas = [_chunk_to_metadata(chunk.metadata) for chunk in chunks] # Add to Chroma diff --git a/tests/unit/test_embedding_degeneracy.py b/tests/unit/test_embedding_degeneracy.py index 739659fa..f584d2e4 100644 --- a/tests/unit/test_embedding_degeneracy.py +++ b/tests/unit/test_embedding_degeneracy.py @@ -131,6 +131,35 @@ def zero_for_empty(texts): assert collection.count() == 0 +@pytest.mark.asyncio +async def test_pre_embedded_zero_vector_is_rejected(temp_dir): + """capsule_builder, capsule_reader and local_docs embed before calling add_documents. + + Those chunks arrive with .embedding already set, so they never pass through the + provider branch. Screen them too, or the collection is poisoned by the back door. + """ + store = _store(temp_dir, _returns_unit_vectors) + chunk = _chunk("chunk-1", "real content") + chunk.embedding = [0.0] * DIMENSION + + with pytest.raises(EmbeddingFailedError, match="zero vector"): + await store.add_documents("test-kb", [chunk]) + + collection = store.client.get_or_create_collection(name="test-kb") + assert collection.count() == 0 + + +@pytest.mark.asyncio +async def test_pre_embedded_real_vector_still_stores(temp_dir): + """A caller-supplied healthy vector must still be accepted.""" + store = _store(temp_dir, _returns_unit_vectors) + chunk = _chunk("chunk-1", "real content") + chunk.embedding = [1.0] + [0.0] * (DIMENSION - 1) + + await store.add_documents("test-kb", [chunk]) + assert store.client.get_collection(name="test-kb").count() == 1 + + @pytest.mark.asyncio async def test_misaligned_embedding_count_is_rejected(temp_dir): """A short vector list would misassign every embedding after the dropped text.""" diff --git a/tests/unit/test_ingest_dois_resume.py b/tests/unit/test_ingest_dois_resume.py index 783e80a0..effe8ea1 100644 --- a/tests/unit/test_ingest_dois_resume.py +++ b/tests/unit/test_ingest_dois_resume.py @@ -81,3 +81,43 @@ async def fake_retrieve(doi, **kw): assert fetched_dois == ["10.3/c"] # The checkpoint should have been deleted on clean completion. assert not (ck_dir / "kb1__ingest_dois.json").exists() + + +@pytest.mark.asyncio +async def test_embedding_failure_unmarks_dois_so_a_resume_retries_them(tmp_path): + """DOIs are marked "added" on retrieval, before the batch is embedded. + + If the embedder dies, the batch never lands. remaining_ids() never re-offers an + "added" id, so leaving the mark would make a resume skip those DOIs for ever. + """ + from perspicacite.llm.embeddings import EmbeddingFailedError + + ck_dir = tmp_path / "ck" + ck_dir.mkdir() + checkpoint_path = ck_dir / "kb1__ingest_dois.json" + + async def fake_retrieve(doi, **kw): + return SimpleNamespace(success=True, full_text="x", abstract=None, metadata={}) + + app_state = _app_state(tmp_path) + with patch( + "perspicacite.pipeline.download.retrieve_paper_content", new=fake_retrieve + ), patch( + "perspicacite.pipeline.download.cookies.build_authenticated_client" + ) as mock_client_ctx, patch( + "perspicacite.rag.dynamic_kb.DynamicKnowledgeBase" + ) as mock_dkb: + mock_client_ctx.return_value.__aenter__ = AsyncMock(return_value=MagicMock()) + mock_client_ctx.return_value.__aexit__ = AsyncMock(return_value=False) + mock_dkb.return_value.add_papers = AsyncMock( + side_effect=EmbeddingFailedError("quota exhausted") + ) + + with pytest.raises(EmbeddingFailedError): + await ingest_dois_into_kb(app_state, "kb1", ["10.1/a", "10.2/b"]) + + # The checkpoint survives the abort, but must not claim the DOIs were added. + assert checkpoint_path.exists() + store = CheckpointStore(path=checkpoint_path, kb_name="kb1", operation="ingest_dois") + state = store.load_or_create(planned_ids=["10.1/a", "10.2/b"]) + assert sorted(state.remaining_ids()) == ["10.1/a", "10.2/b"]