cache.normalization

Overview

Converts a raw user query into a canonical form used both as the exact-match cache key input and as the text fed to the embedding model. The module docstring enumerates the pipeline stages in order:

1. Unicode NFC normalization
2. Lower-case
3. Strip control characters / zero-width characters
4. Collapse whitespace
5. Remove punctuation (configurable)
6. Stop-word removal (configurable, English + Arabic minimal set)
7. Token deduplication (optional)
8. Semantic normalization hints (synonym map, optional)

The actual method applies these in a slightly different order than the docstring lists them (synonym expansion happens before stop-word removal) — see Internal Workflow below for the verified, as-implemented order.

NormalizationConfig

class NormalizationConfig:
    def __init__(
        self,
        remove_punctuation: bool = True,
        remove_stopwords: bool = False,
        deduplicate_tokens: bool = False,
        apply_synonyms: bool = True,
        extra_synonyms: Optional[Dict[str, str]] = None,
        max_length: int = 2048,
    ) -> None

Plain __init__-based configuration (not a @dataclass, unlike most other config types in this codebase). synonyms is built by copying a built-in default map (llm → "large language model", rag → "retrieval augmented generation", ai → "artificial intelligence", spelling normalizations like summarise→summarize, etc.) and then updating it with extra_synonyms if provided.

QueryNormalizer

class QueryNormalizer:
    def __init__(self, config: Optional[NormalizationConfig] = None) -> None

Stateless (holds only its config) and explicitly documented as safe to share across threads.

normalize(self, query: str) -> str

Overview

Why it exists: two differently-worded but semantically identical queries ("What's RAG?" vs. "what is rag") should hit the same exact-match cache key and produce comparable embeddings; raw text alone won't do that.

When to use it: always, as the first step before hashing or embedding a query — this is exactly what CacheManager.get()/put() do internally.

When not to use it: if you need the original, human-readable text for display — normalize() is lossy (it lower-cases, may strip punctuation and stop-words, and truncates long input); keep the original alongside the normalized form, as IntelligentCacheEntry does with its separate raw_query/normalized_query fields.

Signature

def normalize(self, query: str) -> str

Returns

str — the canonical form. Returns "" immediately if query is falsy (empty string, None, etc. — note the parameter is annotated str but the if not query: guard also protects against None being passed despite the type hint).

Raises

No exceptions are raised by this method under normal operation; all steps operate on built-in string operations and precompiled regexes.

Internal Workflow (as implemented, in source order)

1. unicodedata.normalize("NFC", query)
2. Strip control/zero-width characters via _CONTROL_CHAR_RE (replaced with a space)
3. .lower()
4. If remove_punctuation: strip via _PUNCT_RE (replaced with a space)
5. Collapse repeated whitespace via _WHITESPACE_RE and .strip()
6. If apply_synonyms: split on whitespace, map each token through self._cfg.synonyms (unmapped tokens pass through unchanged), rejoin with single spaces
7. If remove_stopwords: filter out tokens present in the combined English+Arabic stop-word set (_ALL_STOPWORDS)
8. If deduplicate_tokens: order-preserving de-duplication using a set to track seen tokens
9. Truncate to self._cfg.max_length characters if longer

Algorithm

Simple, sequential text-transformation pipeline — no parsing, no external calls. All regexes are module-level, precompiled once at import time.

Complexity

• Time: O(n) in the length of query for each regex/string pass; O(1) synonym lookup per token via dict access. Overall O(n) with a modest constant factor for ~4-5 sequential passes over the text.
• Space: O(n) for intermediate string copies.

Thread Safety

Thread-safe and async-safe — the method is a pure function of its input and the (immutable-after-construction) config; there is no shared mutable state accessed. Multiprocessing-safe as well, since nothing here depends on process-local state.

Side Effects

None — no logging, no I/O, no mutation of the QueryNormalizer or its config.

Examples

from zerda.cache.normalization import QueryNormalizer, NormalizationConfig

norm = QueryNormalizer()
print(norm.normalize("What is RAG???"))
# -> "what is retrieval augmented generation"   (RAG expanded via synonym map)

norm2 = QueryNormalizer(NormalizationConfig(remove_stopwords=True))
print(norm2.normalize("What is the capital of France?"))
# stop-words like "what", "is", "the", "of" are removed

# Custom synonyms
norm3 = QueryNormalizer(NormalizationConfig(
    extra_synonyms={"k8s": "kubernetes"}
))
print(norm3.normalize("How do I scale a k8s deployment?"))

# Deduplication
norm4 = QueryNormalizer(NormalizationConfig(deduplicate_tokens=True))
print(norm4.normalize("test test test one two"))
# -> "test one two"

# Very long input gets truncated
norm5 = QueryNormalizer(NormalizationConfig(max_length=10))
print(norm5.normalize("this is a very long query indeed"))
# -> exactly 10 characters

# Error-handling: no exceptions to catch, but empty/None-ish input is safe
print(repr(norm.normalize("")))    # ''

Best Practices

• Reuse a single QueryNormalizer instance — it's stateless and cheap to share.
• Keep remove_stopwords=False (the default) unless you've verified it doesn't hurt semantic-match recall for your domain — the source comment explicitly flags this risk.
• If you extend synonyms, prefer extra_synonyms over mutating .synonyms on a shared instance from multiple threads (dict mutation is not otherwise protected).

Common Mistakes

• Assuming normalize() preserves enough information to display back to the user — store the original text separately if you need it.
• Expecting stop-word removal to apply to compound synonym expansions (e.g. "large language model") — synonym expansion happens before stop-word removal in this implementation, so none of "large", "language", or "model" will be stripped even though "the"/"a" would be had they appeared literally in the input.

exact_cache_key(self, tenant_id: str, normalized_query: str) -> str

Overview

Produces a deterministic, tenant-scoped key for exact-match lookups, so identical queries from different tenants never collide.

Signature

def exact_cache_key(self, tenant_id: str, normalized_query: str) -> str

Algorithm

hashlib.sha256(f"{tenant_id}:{normalized_query}".encode()).hexdigest().

Complexity

O(n) in the combined length of the two strings (hashing cost); O(1) space beyond the fixed 64-character hex output.

Thread Safety

Thread-safe (pure function, stateless).

Example

key_a = norm.exact_cache_key("tenant-a", "what is retrieval augmented generation")
key_b = norm.exact_cache_key("tenant-b", "what is retrieval augmented generation")
assert key_a != key_b   # tenant isolation, even for identical normalized text

Related APIs

• CacheManager.get() / put() — the sole caller of both methods
• IntelligentCacheEntry — stores both raw_query and normalized_query