From 33023eed7180697646e97166585d4f54f00d7a4e Mon Sep 17 00:00:00 2001 From: kaizencode Date: Mon, 10 Aug 2026 13:19:23 +0000 Subject: [PATCH] feat(semantic): search compatible OWIE snapshots --- docs/changes.md | 7 +++++-- docs/branch/semantic.md | 19 ++++++++++++++++++- owilix/cli/remote.py | 7 ++++--- docs/source/details/semantic-search.md | 28 +++++++++++++++++++--------- owilix/core/tasks/semantic.py | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------ tests/owilix/core/tasks/test_semantic.py | 68 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 6 file(s) changed, 278 insertion(s)(+), 45 deletion(s)(-) diff --git a/docs/changes.md b/docs/changes.md --- a/docs/changes.md +++ b/docs/changes.md @@ -56,9 +56,12 @@ 3.12 `.venv`; platform-specific Torch/ROCm installation remains explicit. - Automatic health check, detached startup and readiness wait for the local encoder; semantic-search users no longer start a separate command manually. +- Automatic fan-out across every local OWIE snapshot in the selected compatible + model/dimension space, with one-time query encoding, global ranking and URL + deduplication. Incompatible legacy snapshots are reported and skipped. - CLI controls for OWIE selection, encoder URL/timeout and cache rebuild, plus - synthetic Parquet/CLI coverage and end-to-end validation on the French - 2026-08-05 snapshot. + synthetic Parquet/CLI coverage and end-to-end validation across seven French + snapshots (363,808 vectors). **Breaking Changes**: None. Lexical search remains the default mode. diff --git a/docs/branch/semantic.md b/docs/branch/semantic.md --- a/docs/branch/semantic.md +++ b/docs/branch/semantic.md @@ -26,6 +26,8 @@ - [x] Add `owi remote search --mode semantic` - [x] Add the persistent Jina encoder process - [x] Automatically start and reuse the local encoder from semantic search +- [x] Search all local OWIE snapshots in the selected compatible vector space +- [x] Merge rankings and deduplicate pages across snapshots - [x] Document setup and usage ### Phase 1.3: Verification @@ -35,6 +37,12 @@ - [x] Real French OWIE auto-start smoke test passes (3 hits in 5.91s; 5.3ms scan) - [x] A second search reuses the encoder (1 hit in 0.40s; 5.5ms scan) - [x] Existing HTTP suite passes outside the network/thread sandbox (58 passed) +- [x] Multi-OWIE fan-out, one-time query encoding and cross-snapshot + deduplication covered by synthetic tests +- [x] Real French corpus selects 7 compatible snapshots out of 86 candidates + and scans 363,808 vectors (10 results in 24.17s while building all caches) +- [x] A subsequent multi-snapshot query reuses all 7 caches (5 results in + 2.01s; 54.1ms vector scan) --- @@ -52,11 +60,20 @@ - **Reason**: Vectors produced by another model are not comparable, even when their dimensions happen to match. - **Alternatives considered**: Allowing arbitrary encoders with a warning. +4. **Decision**: Search every local snapshot in one compatible vector space. + - **Reason**: A query should cover the available time slices without asking + the user to list dataset UUIDs, while vectors from different or unknown + models must never be compared. + - **Alternatives considered**: Searching only the newest snapshot and mixing + legacy vectors whose model identity is absent. + --- ## Learnings - OWIE schema 2.0 stores one vector per chunk and declares the embedding model in `schema_metadata`. +- Multi-snapshot search can encode once and fan out because compatible OWIEs + share the same model and dimension; URL deduplication prevents repeated pages. - Exact cosine search over the current French slice is fast enough once the vectors are memory-mapped locally. - Starlette's `TestClient` must run outside the restricted thread/network sandbox; all 58 HTTP tests pass there under Python 3.12. @@ -65,7 +82,7 @@ ## Merge Checklist -- [x] All tests pass (865 passed, 2 skipped across semantic and HTTP extras) +- [x] All tests pass (867 passed, 2 skipped across semantic and HTTP extras) - [x] Documentation updated - [x] `docs/changes.md` updated - [ ] `docs/epics.md` updated if epic complete diff --git a/owilix/cli/remote.py b/owilix/cli/remote.py --- a/owilix/cli/remote.py +++ b/owilix/cli/remote.py @@ -377,7 +377,7 @@ semantic_dataset: Optional[str] = typer.Option( None, "--semantic-dataset", - help="Local OWIE dataset UUID (semantic mode; newest compatible snapshot by default)", + help="Restrict semantic search to one local OWIE UUID (all compatible snapshots by default)", ), encoder_url: Optional[str] = typer.Option( None, @@ -401,8 +401,9 @@ Search the OWI lexical index or a locally pulled OWIE vector dataset. Lexical mode (the default) connects to the remote DuckLake/S3 index. - Semantic mode searches a local OWIE schema 2.0 snapshot and automatically - manages a persistent local encoder using the model declared by OWIE. + Semantic mode searches every compatible local OWIE schema 2.0 snapshot + and automatically manages a persistent local encoder using the model + declared by OWIE. In lexical mode, use --fetch to retrieve full document records from OWI. Semantic mode enriches hits from the locally pulled source OWI directly. diff --git a/docs/source/details/semantic-search.md b/docs/source/details/semantic-search.md --- a/docs/source/details/semantic-search.md +++ b/docs/source/details/semantic-search.md @@ -1,10 +1,10 @@ # Semantic search with OWIE -`owi remote search --mode semantic` searches the chunk vectors in a locally -pulled OWIE schema 2.0 dataset. The CLI calls a persistent Jina encoder for the -query, runs an exact cosine scan over a memory-mapped cache, keeps the best -chunk per document, then reads matching URLs, titles and passages from the -related source OWI. +`owi remote search --mode semantic` searches the chunk vectors in all locally +pulled, mutually compatible OWIE schema 2.0 datasets. The CLI encodes the query +once with a persistent Jina encoder, runs an exact cosine scan over each +memory-mapped cache, keeps the best chunk per document, merges the rankings, +then reads matching URLs, titles and passages from the related source OWIs. ## What runs where @@ -29,7 +29,8 @@ ``` The OWIE is sufficient for scores and document IDs. The source OWI is needed -for titles, URLs and relevant passages. +for titles, URLs and relevant passages. More snapshots can be pulled into the +same target: OWILIX discovers them automatically at query time. ## 2. Install the semantic extra @@ -62,10 +63,17 @@ ## 3. Search ```bash -owi remote search \ +.venv/bin/owi --target /media/winiga/owie-data remote search \ "Quels sont les risques et les enjeux de l'intelligence artificielle pour la société ?" \ --mode semantic --language fra --limit 10 ``` + +Without `--semantic-dataset`, OWILIX inspects every local OWIE for the selected +language and searches the largest compatible group sharing the exact same +embedding model and dimension. The query is encoded only once. Results from +all selected snapshots are globally ranked and deduplicated by URL (falling +back to the record ID). Incompatible or older formats are reported as skipped +instead of being mixed into the wrong vector space. No encoder command is required. OWILIX checks the local health endpoint, starts the persistent process when it is absent, waits for Jina to be ready, @@ -94,7 +102,7 @@ --explain ``` -The first query builds this cache: +The first query builds one cache per selected OWIE: ```text ~/.owi/.cache/semantic/// @@ -113,7 +121,9 @@ ## MVP limits - Exact scan only; no HNSW/FAISS index yet. -- One local OWIE snapshot is selected per query. +- Only OWIE schema 2.0 snapshots declaring the selected embedding model and + dimension can be combined. Legacy vectors with no trustworthy model identity + are skipped. - No hybrid lexical/vector score fusion yet. - No automatic remote pull. - `--fetch`, `--all`, `--repr`, `--select` and `--where` belong to lexical diff --git a/owilix/core/tasks/semantic.py b/owilix/core/tasks/semantic.py --- a/owilix/core/tasks/semantic.py +++ b/owilix/core/tasks/semantic.py @@ -70,6 +70,15 @@ @dataclass(frozen=True) +class OWIESelectionGroup: + """Compatible local OWIE snapshots selected for one semantic query.""" + + selections: tuple[OWIESelection, ...] + skipped: tuple[str, ...] + candidate_count: int + + +@dataclass(frozen=True) class SemanticCache: """Memory-mapped vectors and row metadata used during exact search.""" @@ -94,15 +103,15 @@ explain: bool = False, console: Optional[Console] = None, ) -> CommandResult: - """Search a local OWIE snapshot using exact cosine similarity. + """Search compatible local OWIE snapshots using exact cosine similarity. Args: manager: Active ``OWIlixManager``. terms: Natural-language query to encode. language: ISO 639-3 OWIE partition to search. limit: Maximum number of distinct documents to return. - dataset_id: Optional local OWIE dataset UUID. The newest compatible - local snapshot is selected when omitted. + dataset_id: Optional local OWIE dataset UUID. All compatible local + snapshots are searched when omitted. encoder_url: Base URL of the encoder. The default local encoder is started automatically when needed. encoder_timeout: HTTP timeout for model health and encoding requests. @@ -124,13 +133,8 @@ started = time.perf_counter() try: - selection = _select_owie(manager, language, dataset_id) - cache, cache_built = _load_or_build_cache( - manager, - selection, - rebuild=rebuild_cache, - console=console, - ) + selection_group = _select_owies(manager, language, dataset_id) + primary_selection = selection_group.selections[0] resolved_encoder_url = ( encoder_url or os.environ.get("OWI_SEMANTIC_ENCODER_URL") @@ -139,8 +143,8 @@ query_vector = _encode_query( query, resolved_encoder_url, - expected_model=selection.model, - expected_dimension=selection.dimension, + expected_model=primary_selection.model, + expected_dimension=primary_selection.dimension, timeout=encoder_timeout, state_directory=( Path(manager.owi_path).expanduser() / ".cache" / "semantic" / "encoder" @@ -148,31 +152,91 @@ console=console, ) - search_started = time.perf_counter() - hits = _rank_distinct_documents(cache, query_vector, limit) - search_duration = time.perf_counter() - search_started - source_dataset = _find_local_dataset(manager, selection.source_dataset_id) - source_rows = _read_source_rows(source_dataset, language, [hit["id"] for hit in hits]) - results = _enrich_hits(hits, source_rows, selection, source_dataset) + candidate_limit = max(50, limit * 5) + all_results: list[dict[str, Any]] = [] + searched: list[tuple[OWIESelection, SemanticCache, bool]] = [] + runtime_skips: list[str] = [] + search_duration = 0.0 + missing_sources = 0 + for selection in selection_group.selections: + try: + cache, cache_built = _load_or_build_cache( + manager, + selection, + rebuild=rebuild_cache, + console=console, + ) + except SemanticSearchError as error: + if dataset_id: + raise + runtime_skips.append(f"OWIE {selection.dataset_id}: {error}") + logger.warning("Skipping unusable OWIE %s: %s", selection.dataset_id, error) + continue + + search_started = time.perf_counter() + hits = _rank_distinct_documents(cache, query_vector, candidate_limit) + search_duration += time.perf_counter() - search_started + source_dataset = _find_local_dataset(manager, selection.source_dataset_id) + source_rows = _read_source_rows( + source_dataset, + language, + [hit["id"] for hit in hits], + ) + all_results.extend(_enrich_hits(hits, source_rows, selection, source_dataset)) + searched.append((selection, cache, cache_built)) + if source_dataset is None: + missing_sources += 1 + + if not searched: + detail = runtime_skips[0] if runtime_skips else "no usable semantic cache" + raise SemanticSearchError( + f"No compatible local OWIE could be searched ({detail}).", + ErrorType.DATA, + ExitCode.DATA_ERROR, + ) + + results = _merge_semantic_results(all_results, limit) + skipped_count = len(selection_group.skipped) + len(runtime_skips) + built_count = sum(1 for _selection, _cache, built in searched if built) + reused_count = len(searched) - built_count + vectors_scanned = sum(int(cache.manifest["count"]) for _selection, cache, _built in searched) if explain: - console.print(f"OWIE dataset: {selection.dataset_id}") - console.print(f"Source OWI dataset: {selection.source_dataset_id or 'not declared'}") - console.print(f"Embedding model: {selection.model} ({selection.dimension} dimensions)") - console.print(f"Vector cache: {cache.directory}") + console.print( + f"OWIE datasets: {len(searched)} searched, {skipped_count} skipped " + f"({selection_group.candidate_count} local candidates)" + ) + console.print( + f"Embedding model: {primary_selection.model} " + f"({primary_selection.dimension} dimensions)" + ) + console.print(f"Vectors scanned: {vectors_scanned}") + for selection, cache, cache_built in searched: + cache_state = "built" if cache_built else "reused" + console.print( + f"- {selection.dataset_id} -> " + f"{selection.source_dataset_id or 'source not declared'} " + f"({cache_state}: {cache.directory})" + ) console.print(f"Encoder: {resolved_encoder_url.rstrip('/')}") - if source_dataset is None: + if skipped_count: console.print( - "[yellow]The source OWI dataset is not available locally; " - "results contain vector IDs but no URL or passage.[/yellow]" + f"[yellow]Skipped {skipped_count} local OWIE dataset(s) whose format, " + "vector space, or cache is incompatible.[/yellow]" + ) + if missing_sources: + console.print( + f"[yellow]{missing_sources} searched OWIE dataset(s) have no local source OWI; " + "their results contain vector IDs but may lack URLs or passages.[/yellow]" ) total_duration = time.perf_counter() - started - cache_label = "built" if cache_built else "reused" message = ( f"{len(results)} results in {total_duration:.2f}s " - f"(vector scan {search_duration * 1000:.1f}ms, cache {cache_label})" + f"from {len(searched)} OWIE datasets " + f"(vector scan {search_duration * 1000:.1f}ms, " + f"cache reused {reused_count}/built {built_count})" ) console.print(f"Found {len(results)} semantic results in {total_duration:.2f}s") return CommandResult( @@ -203,8 +267,12 @@ ) -def _select_owie(manager: Any, language: str, dataset_id: Optional[str]) -> OWIESelection: - """Select the newest compatible local OWIE dataset for ``language``.""" +def _select_owies( + manager: Any, + language: str, + dataset_id: Optional[str], +) -> OWIESelectionGroup: + """Select every local OWIE in one compatible vector space.""" candidates: list[Any] = [] for access in ("public", "project", "user"): query = {"id": dataset_id} if dataset_id else {"resourceType": "owie"} @@ -244,7 +312,46 @@ ExitCode.DATA_ERROR, ) - return max(inspected, key=_selection_sort_key) + if dataset_id: + ordered = tuple(sorted(inspected, key=_selection_sort_key, reverse=True)) + return OWIESelectionGroup( + selections=ordered, + skipped=tuple(inspection_errors), + candidate_count=len(unique_candidates), + ) + + spaces: dict[tuple[str, int], list[OWIESelection]] = {} + for selection in inspected: + spaces.setdefault((selection.model, selection.dimension), []).append(selection) + + def space_sort_key( + item: tuple[tuple[str, int], list[OWIESelection]], + ) -> tuple[int, tuple[str, str]]: + selections = item[1] + return (len(selections), max(_selection_sort_key(selection) for selection in selections)) + + selected_space, selected = max(spaces.items(), key=space_sort_key) + for space, selections in spaces.items(): + if space == selected_space: + continue + for selection in selections: + inspection_errors.append( + f"OWIE {selection.dataset_id} uses vector space " + f"{selection.model}/{selection.dimension}, selected " + f"{selected_space[0]}/{selected_space[1]}" + ) + + ordered = tuple(sorted(selected, key=_selection_sort_key, reverse=True)) + return OWIESelectionGroup( + selections=ordered, + skipped=tuple(inspection_errors), + candidate_count=len(unique_candidates), + ) + + +def _select_owie(manager: Any, language: str, dataset_id: Optional[str]) -> OWIESelection: + """Return the newest selected OWIE for callers needing one snapshot.""" + return _select_owies(manager, language, dataset_id).selections[0] def _selection_sort_key(selection: OWIESelection) -> tuple[str, str]: @@ -725,6 +832,33 @@ if len(hits) >= limit: break return hits + + +def _merge_semantic_results( + results: Sequence[dict[str, Any]], + limit: int, +) -> list[dict[str, Any]]: + """Merge snapshot results, retaining the strongest hit per document.""" + best: dict[tuple[str, str], dict[str, Any]] = {} + for result in results: + url = str(result.get("url") or "").strip() + record_id = str(result.get("id") or "").strip() + identity = ("url", url) if url else ("id", record_id) + previous = best.get(identity) + if previous is None or _semantic_result_sort_key(result) > _semantic_result_sort_key(previous): + best[identity] = result + return sorted(best.values(), key=_semantic_result_sort_key, reverse=True)[:limit] + + +def _semantic_result_sort_key(result: dict[str, Any]) -> tuple[float, int, int, int, str]: + """Sort by score, preferring the newest snapshot for exact ties.""" + return ( + float(result.get("score") or 0.0), + int(result.get("year") or 0), + int(result.get("month") or 0), + int(result.get("day") or 0), + str(result.get("owie_dataset") or ""), + ) def _find_local_dataset(manager: Any, dataset_id: Optional[str]) -> Optional[Any]: diff --git a/tests/owilix/core/tasks/test_semantic.py b/tests/owilix/core/tasks/test_semantic.py --- a/tests/owilix/core/tasks/test_semantic.py +++ b/tests/owilix/core/tasks/test_semantic.py @@ -15,6 +15,7 @@ from owilix.core.tasks.semantic import ( SemanticSearchError, _select_owie, + _select_owies, _validate_encoder_identity, semantic_search, ) @@ -250,6 +251,73 @@ assert second.success is True build_cache.assert_not_called() assert "cache reused" in second.msg + + +def test_semantic_search_fans_out_and_deduplicates_snapshots( + semantic_manager: FakeManager, +): + """One query searches every compatible snapshot and merges duplicate pages.""" + first_owie = semantic_manager.local.datasets[0] + second_owie_id = "a9ae3346-91e0-11f1-a50d-3e22d9284a8a" + second_owie = FakeDataset( + Path(first_owie.path), + FakeMetadata( + second_owie_id, + "owie", + first_owie.metadata.source_dataset_id, + ), + ) + semantic_manager.local.datasets.append(second_owie) + + query_vector = np.array([1.0, 0.0], dtype=np.float32) + with patch( + "owilix.core.tasks.semantic._encode_query", + return_value=query_vector, + ) as encode_query: + result = semantic_search( + semantic_manager, + "risques de l'intelligence artificielle", + language="fra", + limit=10, + console=Console(file=None, quiet=True), + ) + + assert result.success is True + assert encode_query.call_count == 1 + assert "from 2 OWIE datasets" in result.msg + assert [row["id"] for row in result.json] == ["doc-one", "doc-two"] + assert len({row["url"] for row in result.json}) == 2 + for owie_id in (first_owie.metadata.id, second_owie_id): + cache = ( + Path(semantic_manager.owi_path) + / ".cache" + / "semantic" + / owie_id + / "fra" + ) + assert (cache / "manifest.json").is_file() + + +def test_semantic_selection_reports_incompatible_local_owie( + semantic_manager: FakeManager, + tmp_path: Path, +): + """An unusable legacy snapshot does not prevent compatible fan-out.""" + legacy_id = "legacy-owie" + legacy = FakeDataset( + tmp_path / "public" / "gpu" / legacy_id, + FakeMetadata(legacy_id, "owie"), + ) + semantic_manager.local.datasets.append(legacy) + + group = _select_owies(semantic_manager, "fra", None) + + assert [selection.dataset_id for selection in group.selections] == [ + "0c3b29fc-91db-11f1-882e-3e22d9284a8a" + ] + assert group.candidate_count == 2 + assert len(group.skipped) == 1 + assert "legacy-owie" in group.skipped[0] def test_encoder_model_mismatch_is_rejected(): -- tangled.sh