perf: cache PreparedQuery per stream, skip double-normalize in dedupe (#282)

Scoring hot path (_normalize_score_dedupe) re-tokenized the same
ranking_query ~240x per stream: once per item for local_relevance,
plus ~5x per item across snippet windows. Query tokens are immutable
within a stream, so compute them once as relevance.PreparedQuery and
thread through signals.annotate_stream and snippet.extract_best_snippet.

dedupe._PreparedText called normalize_text twice: once in __init__ and
again via get_ngrams. Factor out _ngrams_of_normalized so the prepared
path skips the redundant pass while get_ngrams keeps its public contract.

Behavior unchanged.

skills/last30days/scripts/lib/dedupe.py

@@ -39,11 +39,14 @@ def normalize_text(text: str) -> str:
     return re.sub(r"\s+", " ", text).strip()
 
 
+def _ngrams_of_normalized(norm: str, n: int = 3) -> set[str]:
+    if len(norm) < n:
+        return {norm} if norm else set()
+    return {norm[index:index + n] for index in range(len(norm) - n + 1)}
+
+
 def get_ngrams(text: str, n: int = 3) -> set[str]:
-    text = normalize_text(text)
-    if len(text) < n:
-        return {text} if text else set()
-    return {text[index:index + n] for index in range(len(text) - n + 1)}
+    return _ngrams_of_normalized(normalize_text(text), n)
 
 
 def jaccard_similarity(left: set[str], right: set[str]) -> float:
@@ -90,7 +93,7 @@ class _PreparedText:
 
     def __init__(self, raw: str) -> None:
         norm = normalize_text(raw)
-        self.ngrams = get_ngrams(norm) if norm else set()
+        self.ngrams = _ngrams_of_normalized(norm)
         self.tokens = _tokenize(norm)
 
 

skills/last30days/scripts/lib/pipeline.py

@@ -30,6 +30,7 @@ from . import (
     query,
     reddit,
     reddit_public,
+    relevance,
     rerank,
     schema,
     signals,
@@ -500,11 +501,12 @@ def _normalize_score_dedupe(
         source, raw_items, from_date, to_date,
         freshness_mode=freshness_mode,
     )
-    normalized = signals.annotate_stream(normalized, ranking_query, freshness_mode)
+    prepared_query = relevance.PreparedQuery(ranking_query)
+    normalized = signals.annotate_stream(normalized, prepared_query, freshness_mode)
     normalized = signals.prune_low_relevance(normalized)
     normalized = dedupe.dedupe_items(normalized)
     for item in normalized:
-        item.snippet = snippet.extract_best_snippet(item, ranking_query)
+        item.snippet = snippet.extract_best_snippet(item, prepared_query)
     return normalized
 
 

skills/last30days/scripts/lib/relevance.py

@@ -71,8 +71,29 @@ def _normalize_phrase(text: str) -> str:
     return ' '.join(re.sub(r'[^\w\s]', ' ', text.lower()).split())
 
 
+class PreparedQuery:
+    """Precomputed query shape reused across items in a stream.
+
+    Built once per ranking_query; reused by token_overlap_relevance so the
+    per-item normalize/score loops don't re-tokenize the same query N times.
+    """
+
+    __slots__ = ("raw", "q_tokens", "informative_q_tokens", "normalized_phrase")
+
+    def __init__(self, query: str) -> None:
+        self.raw = query
+        self.q_tokens = tokenize(query)
+        informative = {t for t in self.q_tokens if t not in LOW_SIGNAL_QUERY_TOKENS}
+        self.informative_q_tokens = informative or self.q_tokens
+        self.normalized_phrase = _normalize_phrase(query)
+
+
+def _as_prepared(query: "str | PreparedQuery") -> PreparedQuery:
+    return query if isinstance(query, PreparedQuery) else PreparedQuery(query)
+
+
 def token_overlap_relevance(
-    query: str,
+    query: "str | PreparedQuery",
     text: str,
     hashtags: Optional[List[str]] = None,
 ) -> float:
@@ -95,7 +116,8 @@ def token_overlap_relevance(
     Returns:
         Float between 0.0 and 1.0 (0.5 for empty queries)
     """
-    q_tokens = tokenize(query)
+    prepared = _as_prepared(query)
+    q_tokens = prepared.q_tokens
 
     # Combine text and hashtags for matching
     combined = text
@@ -119,9 +141,7 @@ def token_overlap_relevance(
     if overlap == 0:
         return 0.0
 
-    informative_q_tokens = {t for t in q_tokens if t not in LOW_SIGNAL_QUERY_TOKENS}
-    if not informative_q_tokens:
-        informative_q_tokens = q_tokens
+    informative_q_tokens = prepared.informative_q_tokens
 
     coverage = overlap / len(q_tokens)
     informative_overlap = len(informative_q_tokens & t_tokens) / len(informative_q_tokens)
@@ -129,7 +149,7 @@ def token_overlap_relevance(
     precision = overlap / precision_denominator
 
     phrase_bonus = 0.0
-    normalized_query = _normalize_phrase(query)
+    normalized_query = prepared.normalized_phrase
     normalized_text = _normalize_phrase(combined)
     if normalized_query and normalized_query in normalized_text:
         phrase_bonus = 0.12 if len(normalized_query.split()) > 1 else 0.16

skills/last30days/scripts/lib/signals.py

@@ -26,7 +26,10 @@ def source_quality(source: str) -> float:
     return SOURCE_QUALITY.get(source, 0.6)
 
 
-def local_relevance(item: schema.SourceItem, ranking_query: str) -> float:
+def local_relevance(
+    item: schema.SourceItem,
+    ranking_query: "str | relevance.PreparedQuery",
+) -> float:
     text = "\n".join(
         part
         for part in [item.title, item.body, item.snippet]
@@ -175,13 +178,14 @@ def normalize(values: list[float | None]) -> list[int | None]:
 
 def annotate_stream(
     items: list[schema.SourceItem],
-    ranking_query: str,
+    ranking_query: "str | relevance.PreparedQuery",
     freshness_mode: str,
 ) -> list[schema.SourceItem]:
     """Attach local scoring metadata and return items sorted by local_rank_score."""
+    prepared_query = ranking_query if isinstance(ranking_query, relevance.PreparedQuery) else relevance.PreparedQuery(ranking_query)
     engagement_scores = normalize([engagement_raw(item) for item in items])
     for item, eng_score in zip(items, engagement_scores, strict=True):
-        item.local_relevance = local_relevance(item, ranking_query)
+        item.local_relevance = local_relevance(item, prepared_query)
         item.freshness = freshness(item, freshness_mode)
         item.engagement_score = eng_score
         item.source_quality = source_quality(item.source)

skills/last30days/scripts/lib/snippet.py

@@ -26,7 +26,7 @@ def _windows(words: list[str], size: int, overlap: int) -> list[str]:
 
 def extract_best_snippet(
     item: schema.SourceItem,
-    ranking_query: str,
+    ranking_query: "str | relevance.PreparedQuery",
     max_words: int = 120,
 ) -> str:
     """Prefer existing snippets, else extract the best matching evidence window."""
@@ -43,8 +43,9 @@ def extract_best_snippet(
     if not candidates:
         return _truncate_words(body, max_words)
 
+    prepared_query = ranking_query if isinstance(ranking_query, relevance.PreparedQuery) else relevance.PreparedQuery(ranking_query)
     best = max(
         candidates,
-        key=lambda candidate: relevance.token_overlap_relevance(ranking_query, candidate),
+        key=lambda candidate: relevance.token_overlap_relevance(prepared_query, candidate),
     )
     return _truncate_words(best, max_words)