From 53737bf8ab10432c405eba6e822d930db4aeaba0 Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:00:42 +0100 Subject: [PATCH 1/2] feat(find): index column names + semantic_type in dataset search MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The topic_index already carries full column defs (~6.2k across the Lab) with name, role, semantic_type, description — but find() never scored them, so a query matching only a column name/type returned nothing. Columns are now a scored field on datasets (col name+desc as text, semantic_type as tag), unlocking queries like: find("cig") → anac_aggiudicatari (cig_code column) find("codice_fiscale") → anac_aggiudicatari (fiscal_code column) No new pipeline, no artifact growth — pure query-time change on data already present. Level A only; distinct-value extraction (level C) is explicitly out of scope, that's toolkit_query's job. --- src/agent_context_builder/navigation.py | 37 +++++++++++++++++++++++++ tests/test_navigation.py | 30 ++++++++++++++++++++ 2 files changed, 67 insertions(+) diff --git a/src/agent_context_builder/navigation.py b/src/agent_context_builder/navigation.py index 8849359..ff0b4d9 100644 --- a/src/agent_context_builder/navigation.py +++ b/src/agent_context_builder/navigation.py @@ -412,6 +412,36 @@ def _score_entity(query: str, fields: list[tuple[str | None, str]]) -> int: return score +def _columns_index_text(columns: list[dict[str, Any]] | None) -> tuple[str, str]: + """Join column names+descriptions and semantic types for search indexing. + + The topic_index already carries full column defs (~6k across the Lab); + indexing them lets find() answer "which dataset has a CIG column?" + without a separate pipeline. + """ + if not columns: + return "", "" + names = " ".join( + filter( + None, + (str(c.get("name") or "") for c in columns), + ) + ) + descs = " ".join( + filter( + None, + (str(c.get("description") or "") for c in columns), + ) + ) + sem = " ".join( + filter( + None, + (str(c.get("semantic_type") or "") for c in columns), + ) + ) + return f"{names} {descs}".strip(), sem + + def search_index( topic_index: dict[str, Any], query: str, @@ -453,11 +483,13 @@ def search_index( "category": ds.get("category") or meta.get("category"), "tags": ds.get("tags") or meta.get("tags"), "description": ds.get("description") or meta.get("description"), + "columns": ds.get("columns") or meta.get("columns"), } ) if type_filter in (None, "dataset", "datasets"): for ds in dataset_entries: + col_text, col_sem = _columns_index_text(ds.get("columns")) score = _score_entity( query, [ @@ -468,6 +500,11 @@ def search_index( (ds.get("registry_source"), "tag"), (ds.get("description"), "text"), (" ".join(ds.get("tags") or []), "tag"), + # Column schema already lives in the artifact — names + + # descriptions as discovery text, semantic_type as a + # structured tag ("cig_code" → query "cig"). + (col_text, "text"), + (col_sem, "tag"), ], ) if score <= 0: diff --git a/tests/test_navigation.py b/tests/test_navigation.py index 6a59e8b..3011f99 100644 --- a/tests/test_navigation.py +++ b/tests/test_navigation.py @@ -71,6 +71,22 @@ def __init__(self, description="", url=""): "description": ( "Bandi di gara pubblici italiani (CIG): oggetto, importi, stazione appaltante" ), + "columns": [ + { + "name": "cig", + "type": "VARCHAR", + "role": "dimension", + "semantic_type": "cig_code", + "description": "Codice identificativo gara", + }, + { + "name": "importo", + "type": "DOUBLE", + "role": "metric", + "semantic_type": "", + "description": "Importo aggiudicazione", + }, + ], }, ] @@ -296,6 +312,20 @@ def test_search_index_morphology_plural(): assert results and results[0]["slug"] == "anac_bandi_gara" +def test_search_index_column_name(): + """Regression: a query matching a column name (not in slug/name/tags/desc) + must find the dataset — columns are now an indexed field.""" + # 'importo' lives only in the column name, not in any other field + results = search_index(_index(), "importo aggiudicazione", entity_type="dataset") + assert results and results[0]["slug"] == "anac_bandi_gara" + + +def test_search_index_column_semantic_type(): + """semantic_type 'cig_code' is indexed as a tag: query 'cig' finds it.""" + results = search_index(_index(), "cig", entity_type="dataset") + assert results and results[0]["slug"] == "anac_bandi_gara" + + def test_search_index_single_word_still_works(): results = search_index(_index(), "rifiuti", entity_type="dataset") assert results and results[0]["slug"] == "ispra_ru_base" From ff349a966e7743984eae23f640dcffcce08e598e Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Thu, 15 Oct 2026 20:15:46 +0100 Subject: [PATCH 2/2] perf(search): cache field tokenization + ascii fast-path deaccent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Profiling showed 95% of search_index time in matching, dominated by _deaccent (unicodedata.normalize + combining per char) re-running on every field of every entity for every query — even pure-ASCII fields. - _deaccent: skip normalize when text.isascii() (~99% of Lab fields) - _field_words: optional per-search word_cache keyed by field value; search_index threads one shared dict so identical values (source_id, category, registry_source repeated across datasets) tokenize once - _score_entity: precompute token forms once, tokenize each field once per entity (was once per token×field); drop the double _field_words call in the prepared listcomp Wall-clock on real CI artifact (436 datasets, 10-entity limit): ~150ms → ~70ms per query. _field_words calls 173k → 97k (cache hits are dict lookups), _deaccent 173k → 63k (rest are ascii no-ops). Behavior unchanged — same scores, same hits on all regression queries. --- src/agent_context_builder/navigation.py | 100 +++++++++++++++++------- 1 file changed, 72 insertions(+), 28 deletions(-) diff --git a/src/agent_context_builder/navigation.py b/src/agent_context_builder/navigation.py index ff0b4d9..4929ac4 100644 --- a/src/agent_context_builder/navigation.py +++ b/src/agent_context_builder/navigation.py @@ -305,17 +305,35 @@ def _norm(text: str | None) -> str: def _deaccent(text: str) -> str: + # Fast path: most Lab fields (slugs, tags, semantic_types, ASCII names) + # have no accented characters, so skip normalize entirely. + if text.isascii(): + return text return "".join(c for c in unicodedata.normalize("NFKD", text) if not unicodedata.combining(c)) -def _field_words(value: str | None) -> list[str]: +# Per-search cache: field value → word list. Passed down through +# search_index → _score_entity → _token_in_words so each distinct field +# string is tokenized once per query, not once per (token × entity). +_WordCache = dict[str, list[str]] + + +def _field_words(value: str | None, cache: _WordCache | None = None) -> list[str]: """Lowercase alphanumeric words; underscores/punctuation are separators. Makes snake_case slugs word-matchable: "anac_bandi_gara" → - ["anac", "bandi", "gara"]. + ["anac", "bandi", "gara"]. When a cache dict is supplied, repeated + values reuse the precomputed word list. """ + if value is None: + return [] + if cache is not None and value in cache: + return cache[value] v = _deaccent(_norm(value)) - return [w for w in re.split(r"[^a-z0-9]+", v) if w] + words = [w for w in re.split(r"[^a-z0-9]+", v) if w] + if cache is not None: + cache[value] = words + return words def _tokenize_query(query: str) -> list[str]: @@ -341,15 +359,13 @@ def _word_forms(word: str) -> set[str]: return forms -def _token_in_field(token: str, value: str | None) -> bool: - """Match a query token against any word in a field value. +def _token_in_words(token: str, words: list[str], token_forms: set[str]) -> bool: + """Match a query token against a pre-tokenized field (word list). - Whole word (with light morphology) or typed prefix ("pension" → - "pensioni"). Snake_case slugs are word-split, so "bandi" matches - "anac_bandi_gara". + Uses a token_forms set computed once per query — the common case, since + the same token is tested against every field of every entity. """ - token_forms = _word_forms(token) - for field_word in _field_words(value): + for field_word in words: if _word_forms(field_word) & token_forms: return True if len(token) >= 3 and field_word.startswith(token): @@ -366,10 +382,19 @@ def query_matches(query: str, *texts: str | None) -> bool: tokens = _tokenize_query(query) if not tokens: return False - return all(any(_token_in_field(tok, text) for text in texts) for tok in tokens) + token_forms_list = [_word_forms(t) for t in tokens] + field_words_list = [_field_words(text) for text in texts] + return all( + any(_token_in_words(tok, words, tf) for words in field_words_list) + for tok, tf in zip(tokens, token_forms_list) + ) -def _score_entity(query: str, fields: list[tuple[str | None, str]]) -> int: +def _score_entity( + query: str, + fields: list[tuple[str | None, str]], + word_cache: _WordCache | None = None, +) -> int: """Score an entity against a query. fields: (value, weight_name). Tiers: @@ -394,21 +419,33 @@ def _score_entity(query: str, fields: list[tuple[str | None, str]]) -> int: score = max(score, _SUBSTR_WEIGHTS.get(kind, 15)) tokens = _tokenize_query(query) - if tokens: - matched_kinds: set[str] = set() - all_matched = True - for token in tokens: - token_hit = False - for value, kind in fields: - if _token_in_field(token, value): - matched_kinds.add(kind) - token_hit = True - if not token_hit: - all_matched = False - break - if all_matched and matched_kinds: - best = max(_FIELD_TOKEN_WEIGHTS.get(kind, 18) for kind in matched_kinds) - score = max(score, best) + if not tokens: + return score + token_forms_list = [_word_forms(t) for t in tokens] + # Tokenize each field once per entity (not once per token×field). Empty + # fields are dropped so they are never re-scanned. The shared word_cache + # dedupes identical field values across entities in one search. + prepared = [] + for value, kind in fields: + if not value: + continue + words = _field_words(value, word_cache) + if words: + prepared.append((kind, words)) + matched_kinds: set[str] = set() + all_matched = True + for tok, tok_forms in zip(tokens, token_forms_list): + token_hit = False + for kind, words in prepared: + if _token_in_words(tok, words, tok_forms): + matched_kinds.add(kind) + token_hit = True + if not token_hit: + all_matched = False + break + if all_matched and matched_kinds: + best = max(_FIELD_TOKEN_WEIGHTS.get(kind, 18) for kind in matched_kinds) + score = max(score, best) return score @@ -456,6 +493,9 @@ def search_index( """ results: list[dict[str, Any]] = [] type_filter = entity_type.lower() if entity_type else None + # Shared word cache for the whole query: identical field values (e.g. + # source_id "anac" repeated across 9 datasets) tokenize once. + word_cache: _WordCache = {} by_slug = topic_index.get("by_slug") or {} datasets_by_source = topic_index.get("datasets") or {} @@ -506,6 +546,7 @@ def search_index( (col_text, "text"), (col_sem, "tag"), ], + word_cache, ) if score <= 0: continue @@ -536,6 +577,7 @@ def search_index( (card.get("role"), "tag"), (card.get("domain"), "tag"), ], + word_cache, ) if score <= 0: continue @@ -566,6 +608,7 @@ def search_index( (dom.get("name"), "name"), (dom.get("description"), "text"), ], + word_cache, ) if score <= 0: continue @@ -593,6 +636,7 @@ def search_index( (a.get("description"), "text"), (a.get("source"), "tag"), ], + word_cache, ) if score <= 0: continue @@ -615,7 +659,7 @@ def search_index( if sid: source_counts[sid] = source_counts.get(sid, 0) + 1 for sid, count in source_counts.items(): - score = _score_entity(query, [(sid, "slug")]) + score = _score_entity(query, [(sid, "slug")], word_cache) if score <= 0: continue results.append(