"""Reference solution — the knowledge foundation: authorized hybrid retrieval. Two things this file exists to demonstrate, and they are not the same thing: * retrieval QUALITY is engineering — chunking, BM25, dense recall, fusion, reranking; * retrieval SAFETY is architecture — namespaces, entitlement as a PRE-filter, citations. The failure this lab is built around has no runtime detection: a shared index with a post-hoc entitlement filter returns tenant B's chunk to tenant A, and the response is a 200 OK with a plausible answer. Deterministic: a hashing embedder, integer token estimates, no clock, no randomness. ``python solution.py`` runs the worked example. """ from __future__ import annotations import hashlib import math import re from collections import Counter from dataclasses import dataclass, field, replace from enum import Enum from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple # ====================================================================================== # 1. Documents, chunks and provenance # ====================================================================================== @dataclass(frozen=True) class Document: """A source document, with everything retrieval needs to be safe and defensible. ``tenant``, ``classification`` and ``barrier`` are not metadata to filter on later — they select the NAMESPACE a chunk is written to. That is the difference between isolation you can prove and isolation you hope for. """ doc_id: str tenant: str title: str text: str classification: str = "internal" barrier: Optional[str] = None # information barrier / desk, e.g. "advisory" version: int = 1 updated_tick: int = 0 def __post_init__(self) -> None: if not self.doc_id or not self.tenant: raise ValueError("doc_id and tenant are required") classification_rank(self.classification) # reject a typo at ingestion _CLASSIFICATION_ORDER = ("public", "internal", "confidential", "restricted") def classification_rank(name: str) -> int: try: return _CLASSIFICATION_ORDER.index(name) except ValueError: raise ValueError(f"unknown data classification: {name!r}") from None @dataclass(frozen=True) class Chunk: """A retrievable unit, carrying its provenance so an answer can cite it.""" chunk_id: str doc_id: str tenant: str text: str section: str start_char: int end_char: int classification: str barrier: Optional[str] doc_version: int updated_tick: int def citation(self) -> str: return f"{self.doc_id}@v{self.doc_version}#{self.section}[{self.start_char}:{self.end_char}]" # ====================================================================================== # 2. Chunking # ====================================================================================== _SECTION_RE = re.compile(r"^##\s+(.+)$", re.MULTILINE) def estimate_tokens(text: str) -> int: """Deterministic stand-in for a tokenizer: ceil(chars / 4).""" return (len(text) + 3) // 4 if text else 0 def chunk_document( document: Document, *, max_tokens: int = 96, overlap_tokens: int = 16, ) -> List[Chunk]: """Structure-aware chunking: split on ``## headings`` first, then on size. Splitting on structure before size is the single highest-leverage quality decision in retrieval, and it is the one most implementations skip. A fixed-size splitter cuts through the middle of a table, a definition, or a policy clause — and the resulting chunk is retrievable but not *usable*, which is worse than not retrieving it. Overlap buys boundary recall at the cost of index size. It must be smaller than the chunk size, or chunks stop advancing. """ if max_tokens <= 0: raise ValueError("max_tokens must be > 0") if not 0 <= overlap_tokens < max_tokens: raise ValueError("overlap_tokens must be in [0, max_tokens)") sections = _split_sections(document.text) chunks: List[Chunk] = [] index = 0 for section_name, section_text, section_offset in sections: for piece, start, end in _split_by_size(section_text, max_tokens, overlap_tokens): index += 1 chunks.append(Chunk( chunk_id=f"{document.doc_id}::{index}", doc_id=document.doc_id, tenant=document.tenant, text=piece, section=section_name, start_char=section_offset + start, end_char=section_offset + end, classification=document.classification, barrier=document.barrier, doc_version=document.version, updated_tick=document.updated_tick, )) return chunks def _split_sections(text: str) -> List[Tuple[str, str, int]]: matches = list(_SECTION_RE.finditer(text)) if not matches: return [("body", text.strip(), 0)] sections: List[Tuple[str, str, int]] = [] preamble = text[: matches[0].start()].strip() if preamble: sections.append(("body", preamble, 0)) for i, match in enumerate(matches): name = match.group(1).strip() start = match.end() end = matches[i + 1].start() if i + 1 < len(matches) else len(text) body = text[start:end].strip() if body: offset = start + (len(text[start:end]) - len(text[start:end].lstrip())) sections.append((name, body, offset)) return sections def _split_by_size(text: str, max_tokens: int, overlap_tokens: int) -> List[Tuple[str, int, int]]: """Split on word boundaries, never mid-word, with a token-measured window.""" words = text.split() if not words: return [] pieces: List[Tuple[str, int, int]] = [] start_word = 0 while start_word < len(words): window: List[str] = [] cursor = start_word while cursor < len(words): candidate = window + [words[cursor]] if window and estimate_tokens(" ".join(candidate)) > max_tokens: break window = candidate cursor += 1 piece = " ".join(window) offset = _char_offset(text, words, start_word) pieces.append((piece, offset, offset + len(piece))) if cursor >= len(words): break # Step back by roughly `overlap_tokens` worth of words for boundary recall. step_back = 0 overlap: List[str] = [] while step_back < len(window) - 1: trial = [window[len(window) - 1 - step_back]] + overlap if estimate_tokens(" ".join(trial)) > overlap_tokens: break overlap = trial step_back += 1 start_word = cursor - step_back return pieces def _char_offset(text: str, words: Sequence[str], word_index: int) -> int: offset = 0 for i in range(word_index): offset = text.index(words[i], offset) + len(words[i]) return text.index(words[word_index], offset) if word_index < len(words) else offset # ====================================================================================== # 3. Embeddings # ====================================================================================== def hash_embed(text: str, dimensions: int = 96) -> Tuple[float, ...]: """Signed feature hashing, L2-normalized. The random sign is not decoration: without it, hash collisions always add constructively and every pair of documents looks more similar than it is. With signs, collisions cancel in expectation. A real platform calls an embedding model. What matters here is that similarity is a pure function of the text, so retrieval behaviour is testable. """ vector = [0.0] * dimensions for token in _tokenize(text): digest = hashlib.blake2b(token.encode("utf-8"), digest_size=8).digest() index = int.from_bytes(digest[:4], "big") % dimensions vector[index] += 1.0 if digest[4] % 2 == 0 else -1.0 norm = math.sqrt(sum(v * v for v in vector)) if norm == 0.0: return tuple(vector) return tuple(v / norm for v in vector) def cosine(a: Sequence[float], b: Sequence[float]) -> float: return sum(x * y for x, y in zip(a, b)) _TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9\-]*") def _tokenize(text: str) -> List[str]: return _TOKEN_RE.findall(text.lower()) # ====================================================================================== # 4. The namespace — where isolation actually lives # ====================================================================================== @dataclass(frozen=True) class Principal: """Who is asking. Every retrieval is answered relative to this.""" user_id: str tenant: str max_classification: str = "internal" barriers: Tuple[str, ...] = () # barriers this principal is INSIDE def namespace_key(self) -> str: return self.tenant def namespace_of(chunk: Chunk) -> str: """The partition a chunk is written to. Tenant, and nothing else, decides it. Classification and barrier are enforced WITHIN a namespace; the tenant boundary is enforced BY the namespace. Two different mechanisms, deliberately: a bug in the classification check leaks within a tenant, which is bad; a bug in a tenant filter leaks across customers, which is a breach. """ return chunk.tenant # ====================================================================================== # 5. Lexical retrieval — BM25 # ====================================================================================== @dataclass(frozen=True) class Scored: chunk: Chunk score: float class BM25Index: """Okapi BM25, implemented rather than imported. ``k1`` controls term-frequency saturation: the tenth occurrence of a word adds much less than the second. ``b`` controls length normalization: at b=1 a long document is fully penalized for its length, at b=0 not at all. BM25 exists in a hybrid retriever because it is *exact*. A payment reference, an LEI, a product code — these are precisely what embeddings blur and lexical matching nails. """ def __init__(self, *, k1: float = 1.5, b: float = 0.75) -> None: if k1 < 0 or not 0.0 <= b <= 1.0: raise ValueError("k1 must be >= 0 and b in [0, 1]") self.k1 = k1 self.b = b self._by_namespace: Dict[str, List[Chunk]] = {} self._tf: Dict[str, List[Counter]] = {} self._df: Dict[str, Counter] = {} self._lengths: Dict[str, List[int]] = {} def add(self, chunk: Chunk) -> None: ns = namespace_of(chunk) tokens = _tokenize(chunk.text) self._by_namespace.setdefault(ns, []).append(chunk) self._tf.setdefault(ns, []).append(Counter(tokens)) self._lengths.setdefault(ns, []).append(len(tokens)) df = self._df.setdefault(ns, Counter()) for term in set(tokens): df[term] += 1 def _idf(self, ns: str, term: str) -> float: """The BM25 IDF, with the +0.5 smoothing that keeps it finite for a term in every document (and clamped at 0, because a negative contribution is nonsense).""" n = len(self._by_namespace.get(ns, ())) if n == 0: return 0.0 df = self._df[ns].get(term, 0) return max(0.0, math.log((n - df + 0.5) / (df + 0.5) + 1.0)) def search(self, query: str, *, namespace: str, limit: int = 10) -> List[Scored]: chunks = self._by_namespace.get(namespace, []) if not chunks: return [] lengths = self._lengths[namespace] avgdl = sum(lengths) / len(lengths) terms = _tokenize(query) results: List[Scored] = [] for i, chunk in enumerate(chunks): tf = self._tf[namespace][i] length = lengths[i] score = 0.0 for term in terms: frequency = tf.get(term, 0) if frequency == 0: continue denominator = frequency + self.k1 * (1 - self.b + self.b * length / avgdl) score += self._idf(namespace, term) * frequency * (self.k1 + 1) / denominator if score > 0: results.append(Scored(chunk, score)) results.sort(key=lambda s: (-s.score, s.chunk.chunk_id)) return results[:limit] # ====================================================================================== # 6. Dense retrieval # ====================================================================================== class VectorIndex: """A namespaced dense index. The topology decision — silo / pool / bridge — is made HERE, by keying the store on namespace. A pooled index with a post-hoc filter is the same class of defect as an un-namespaced cache: it works until a refactor moves the filter. """ def __init__(self, *, dimensions: int = 96) -> None: self.dimensions = dimensions self._by_namespace: Dict[str, List[Tuple[Chunk, Tuple[float, ...]]]] = {} def add(self, chunk: Chunk) -> None: ns = namespace_of(chunk) self._by_namespace.setdefault(ns, []).append( (chunk, hash_embed(chunk.text, self.dimensions))) def search(self, query: str, *, namespace: str, limit: int = 10) -> List[Scored]: entries = self._by_namespace.get(namespace, []) if not entries: return [] vector = hash_embed(query, self.dimensions) scored = [Scored(chunk, cosine(vector, embedding)) for chunk, embedding in entries] scored = [s for s in scored if s.score > 0.0] scored.sort(key=lambda s: (-s.score, s.chunk.chunk_id)) return scored[:limit] def namespaces(self) -> List[str]: return sorted(self._by_namespace) # ====================================================================================== # 7. Fusion and reranking # ====================================================================================== def reciprocal_rank_fusion( rankings: Sequence[Sequence[Scored]], *, k: int = 60, limit: int = 10, ) -> List[Scored]: """RRF: fused score is the sum of ``1 / (k + rank)`` over the rankings a chunk appears in. Score-FREE, which is the point. BM25 scores and cosine similarities live on incomparable scales, and any attempt to normalize them is a calibration that breaks the next time you change an embedding model. RRF only reads *positions*. Consequence worth knowing: consistent agreement beats a single strong signal. A chunk ranked 1st and 10th scores 1/61 + 1/70 = 0.0307; one ranked 3rd by both scores 2/63 = 0.0317. """ if k <= 0: raise ValueError("k must be > 0") fused: Dict[str, float] = {} chunks: Dict[str, Chunk] = {} for ranking in rankings: for rank, scored in enumerate(ranking, start=1): key = scored.chunk.chunk_id fused[key] = fused.get(key, 0.0) + 1.0 / (k + rank) chunks[key] = scored.chunk results = [Scored(chunks[key], score) for key, score in fused.items()] results.sort(key=lambda s: (-s.score, s.chunk.chunk_id)) return results[:limit] #: A reranker is a (query, chunk) -> score function. A real one is a cross-encoder that #: reads both together; ours is a deterministic stand-in with the same interface, so the #: *architecture* — rerank the top-k only, never the whole index — is what you build. Reranker = Callable[[str, Chunk], float] def lexical_overlap_reranker(query: str, chunk: Chunk) -> float: """Deterministic stand-in: coverage of the query's terms, weighted toward rare ones. Faithful in the property that matters: it scores the PAIR, so it can reward a chunk that answers the question over one that merely shares vocabulary with it. """ query_terms = set(_tokenize(query)) if not query_terms: return 0.0 chunk_terms = _tokenize(chunk.text) counts = Counter(chunk_terms) covered = sum(1 for term in query_terms if counts[term] > 0) density = sum(counts[term] for term in query_terms) / max(1, len(chunk_terms)) return covered / len(query_terms) + 0.25 * density def rerank(query: str, candidates: Sequence[Scored], *, reranker: Reranker, limit: int = 5, min_score: float = 0.0) -> List[Scored]: """Second stage over the top-k ONLY. A cross-encoder is orders of magnitude more expensive than a bi-encoder because it cannot precompute anything — it must run the model per (query, document) pair. That is why it never runs over the index, and why it is the first thing shed under latency pressure (Phase 00's degradation ladder). ``min_score`` is the relevance floor, and it matters more than it looks: first-stage retrieval ALWAYS returns something — an ANN index has a nearest neighbour even for a query about nothing in the corpus — so without a floor, "no relevant documents" is indistinguishable from "here are the least-irrelevant ones". A grounding check would then be asked to support claims against noise. """ scored = [Scored(c.chunk, reranker(query, c.chunk)) for c in candidates] scored = [s for s in scored if s.score > min_score] scored.sort(key=lambda s: (-s.score, s.chunk.chunk_id)) return scored[:limit] # ====================================================================================== # 8. The authorized retriever # ====================================================================================== @dataclass(frozen=True) class RetrievalPolicy: max_stale_ticks: Optional[int] = None # freshness contract; None = no contract enable_rerank: bool = True candidate_limit: int = 12 result_limit: int = 5 @dataclass(frozen=True) class RetrievalResult: chunks: Tuple[Scored, ...] degraded: Tuple[str, ...] = () # which stages were skipped or failed considered: int = 0 excluded_by_entitlement: int = 0 excluded_by_freshness: int = 0 @property def citations(self) -> Tuple[str, ...]: return tuple(s.chunk.citation() for s in self.chunks) class AuthorizedRetriever: """Hybrid retrieval that is authorized by construction. Three enforcement points, deliberately independent: 1. the NAMESPACE bounds the search to the caller's tenant — cross-tenant results are not filtered out, they are never retrieved; 2. classification and barrier are applied as a PRE-filter inside the namespace, before ranking, so an ineligible chunk cannot occupy a slot or leak by timing; 3. the freshness contract drops chunks that are too stale to rely on. Filtering after ranking is the defect this class exists to prevent. It leaks the *existence* of content through result-set size and ordering, and it is one refactor away from leaking the content itself. """ def __init__(self, *, bm25: BM25Index, vectors: VectorIndex, reranker: Reranker = lexical_overlap_reranker) -> None: self.bm25 = bm25 self.vectors = vectors self.reranker = reranker self._chunks: Dict[str, Chunk] = {} def ingest(self, document: Document, *, max_tokens: int = 96, overlap_tokens: int = 16) -> List[Chunk]: chunks = chunk_document(document, max_tokens=max_tokens, overlap_tokens=overlap_tokens) for chunk in chunks: self._chunks[chunk.chunk_id] = chunk self.bm25.add(chunk) self.vectors.add(chunk) return chunks def visible(self, principal: Principal, chunk: Chunk) -> bool: if chunk.tenant != principal.tenant: return False if classification_rank(chunk.classification) > classification_rank(principal.max_classification): return False if chunk.barrier is not None and chunk.barrier not in principal.barriers: return False return True def retrieve(self, query: str, principal: Principal, *, now_tick: int = 0, policy: RetrievalPolicy = RetrievalPolicy()) -> RetrievalResult: namespace = principal.namespace_key() degraded: List[str] = [] lexical = self.bm25.search(query, namespace=namespace, limit=policy.candidate_limit) try: dense = self.vectors.search(query, namespace=namespace, limit=policy.candidate_limit) except Exception: # pragma: no cover dense = [] degraded.append("dense") considered = len({s.chunk.chunk_id for s in lexical} | {s.chunk.chunk_id for s in dense}) # -- entitlement PRE-filter, before fusion and before ranking matters ---------- excluded_entitlement = 0 excluded_freshness = 0 def admit(scored: Sequence[Scored]) -> List[Scored]: nonlocal excluded_entitlement, excluded_freshness out: List[Scored] = [] for s in scored: if not self.visible(principal, s.chunk): excluded_entitlement += 1 continue if (policy.max_stale_ticks is not None and now_tick - s.chunk.updated_tick > policy.max_stale_ticks): excluded_freshness += 1 continue out.append(s) return out lexical = admit(lexical) dense = admit(dense) rankings = [r for r in (lexical, dense) if r] if not rankings: return RetrievalResult((), tuple(degraded), considered, excluded_entitlement, excluded_freshness) fused = reciprocal_rank_fusion(rankings, limit=policy.candidate_limit) if policy.enable_rerank: final = rerank(query, fused, reranker=self.reranker, limit=policy.result_limit) else: degraded.append("rerank") final = list(fused[: policy.result_limit]) return RetrievalResult(tuple(final), tuple(degraded), considered, excluded_entitlement, excluded_freshness) # ====================================================================================== # 9. Grounding # ====================================================================================== @dataclass(frozen=True) class Claim: text: str citation: Optional[str] = None @dataclass(frozen=True) class GroundingReport: supported: Tuple[Claim, ...] unsupported: Tuple[Claim, ...] @property def is_grounded(self) -> bool: return not self.unsupported @property def coverage(self) -> float: total = len(self.supported) + len(self.unsupported) return len(self.supported) / total if total else 1.0 def check_grounding(claims: Sequence[Claim], retrieved: Sequence[Scored], *, min_overlap: float = 0.6) -> GroundingReport: """Every claim must map to a retrieved span, or the answer is not defensible. A claim is supported when enough of its content words appear in some retrieved chunk — a deterministic stand-in for an entailment model. The architecture is the lesson: the check runs over the RETRIEVED SET, so a claim the model invented has nothing to match, and the report names it rather than the answer silently shipping. ``min_overlap`` is a threshold with a real trade-off: too low and paraphrase passes as evidence; too high and correct paraphrase is rejected. Tune it against a labelled set, not by feel. """ if not 0.0 < min_overlap <= 1.0: raise ValueError("min_overlap must be in (0, 1]") supported: List[Claim] = [] unsupported: List[Claim] = [] corpora = [(s.chunk, set(_tokenize(s.chunk.text))) for s in retrieved] for claim in claims: terms = set(_tokenize(claim.text)) - _STOPWORDS if not terms: supported.append(claim) continue best: Optional[Tuple[float, Chunk]] = None for chunk, chunk_terms in corpora: overlap = len(terms & chunk_terms) / len(terms) if best is None or overlap > best[0]: best = (overlap, chunk) if best is not None and best[0] >= min_overlap: supported.append(replace(claim, citation=best[1].citation())) else: unsupported.append(claim) return GroundingReport(tuple(supported), tuple(unsupported)) _STOPWORDS = { "the", "a", "an", "is", "are", "was", "were", "be", "been", "of", "to", "in", "on", "for", "and", "or", "it", "this", "that", "with", "as", "at", "by", "from", "has", "have", "had", "will", "would", "can", "could", "should", "may", "might", } # ====================================================================================== # 10. Context assembly # ====================================================================================== class Segment(str, Enum): """Prompt segments, in the order they must be emitted. Stable content first, volatile content last — because provider-side prefix caching is invalidated by the first changed byte (Phase 04). Getting this order wrong costs real money and nothing errors. """ INSTRUCTIONS = "instructions" # stable across every request TOOL_SCHEMAS = "tool_schemas" # stable per agent version POLICY = "policy" # stable per tenant MEMORY = "memory" # varies per session RETRIEVED = "retrieved" # varies per turn QUESTION = "question" # varies per turn — always last _SEGMENT_ORDER = tuple(Segment) @dataclass(frozen=True) class AssembledContext: text: str tokens: int included_chunks: Tuple[str, ...] dropped_chunks: Tuple[str, ...] stable_prefix_tokens: int @property def cacheable_fraction(self) -> float: return self.stable_prefix_tokens / self.tokens if self.tokens else 0.0 def assemble_context( *, budget_tokens: int, instructions: str, tool_schemas: str = "", policy: str = "", memory: str = "", question: str, retrieved: Sequence[Scored] = (), ) -> AssembledContext: """Fill a token budget in cache-friendly order, dropping retrieved chunks last-first. Two rules that are easy to get backwards: 1. the QUESTION is never dropped — an assembler that truncates the user's turn to fit more context has inverted its purpose; 2. when retrieval must be trimmed, drop the LOWEST-ranked chunks, and report which ones went, because "we answered without the third source" is a fact the grounding check and the audit record both need. """ if budget_tokens <= 0: raise ValueError("budget_tokens must be > 0") fixed = { Segment.INSTRUCTIONS: instructions, Segment.TOOL_SCHEMAS: tool_schemas, Segment.POLICY: policy, Segment.MEMORY: memory, Segment.QUESTION: question, } def render(included: Sequence[Scored]) -> Tuple[str, int]: parts: List[str] = [] stable = 0 for segment in _SEGMENT_ORDER: if segment is Segment.RETRIEVED: if included: body = "\n\n".join(_render_chunk(s.chunk) for s in included) parts.append(f"[{segment.value}]\n{body}") continue value = fixed.get(segment, "") if not value: continue block = f"[{segment.value}]\n{value}" parts.append(block) if segment in (Segment.INSTRUCTIONS, Segment.TOOL_SCHEMAS, Segment.POLICY): stable += estimate_tokens(block) return "\n\n".join(parts), stable # Measure the ASSEMBLED text, not the sum of its pieces. Segment headers and the # separators between blocks are real tokens, and a budget that ignores them # over-fills by exactly the amount nobody accounts for. included = list(retrieved) dropped: List[str] = [] while True: text, stable_tokens = render(included) if estimate_tokens(text) <= budget_tokens: break if not included: raise ValueError( f"fixed segments need {estimate_tokens(text)} tokens, budget is " f"{budget_tokens}; trim instructions or raise the budget — the " "question is never dropped") dropped.insert(0, included.pop().chunk.chunk_id) # lowest-ranked goes first return AssembledContext( text=text, tokens=estimate_tokens(text), included_chunks=tuple(s.chunk.chunk_id for s in included), dropped_chunks=tuple(dropped), stable_prefix_tokens=stable_tokens, ) def _render_chunk(chunk: Chunk) -> str: return f"<<{chunk.citation()}>>\n{chunk.text}" # ====================================================================================== # Worked example # ====================================================================================== WHOLESALE_POLICY = """## Hold release A payment on sanctions hold may be released only after a screening review is completed and recorded. Releases above AED 100,000 require dual control by two authorised officers. ## Escalation If a possible match scores above 0.80, escalate to Group Compliance before any release. """ WHOLESALE_PAYMENT = """## Payment PMT-771 Payment reference PMT-771 for AED 250,000 to beneficiary Acme Trading FZE is currently HELD. The hold was applied by the sanctions screening engine on value date 2026-03-12. ## Counterparty Acme Trading FZE is a UAE-registered entity, LEI 5493001KJTIIGC8Y1R12, ultimately controlled by Northgate Holdings Ltd. """ RETAIL_NOTE = """## Collections script When a retail customer disputes a late fee, apologise, check the statement date and offer a one-time goodwill reversal if the account has no prior reversals. """ MNPI_DEAL = """## Project Falcon Project Falcon is a confidential acquisition of Acme Trading FZE by a listed acquirer. Announcement is expected in Q3. This information is material and non-public. """ def main() -> None: # pragma: no cover - narrative output bm25 = BM25Index() vectors = VectorIndex() retriever = AuthorizedRetriever(bm25=bm25, vectors=vectors) documents = [ Document("pol-hold-release", "wholesale", "Sanctions hold release policy", WHOLESALE_POLICY, classification="internal", updated_tick=100), Document("pmt-771", "wholesale", "Payment PMT-771", WHOLESALE_PAYMENT, classification="confidential", updated_tick=140), Document("col-script", "retail", "Collections script", RETAIL_NOTE, classification="internal", updated_tick=90), Document("deal-falcon", "wholesale", "Project Falcon", MNPI_DEAL, classification="restricted", barrier="advisory", updated_tick=150), ] for document in documents: retriever.ingest(document) print("=" * 78) print("1. CHUNKING IS STRUCTURE-AWARE") print("=" * 78) chunks = chunk_document(documents[0]) for chunk in chunks: print(f" {chunk.chunk_id:<24} section={chunk.section!r:<16} " f"tokens={estimate_tokens(chunk.text):<3} cite={chunk.citation()}") print(" -> sections split first, size second. A fixed-size splitter cuts through") print(" the middle of a policy clause and produces a retrievable, unusable chunk.") print() print("=" * 78) print("2. BM25 AND DENSE FAIL ON OPPOSITE INPUTS") print("=" * 78) for query in ("PMT-771", "why would a transfer be stopped"): lex = bm25.search(query, namespace="wholesale", limit=3) den = vectors.search(query, namespace="wholesale", limit=3) print(f" query {query!r}") print(f" bm25 : {[s.chunk.chunk_id for s in lex]}") print(f" dense: {[s.chunk.chunk_id for s in den]}") print(" -> exact identifiers are BM25's strength and embeddings' weakness.") print(" That asymmetry is why hybrid wins — not because more is better.") print() print("=" * 78) print("3. RRF NEEDS NO CALIBRATION") print("=" * 78) lex = bm25.search("sanctions hold release dual control", namespace="wholesale", limit=5) den = vectors.search("sanctions hold release dual control", namespace="wholesale", limit=5) fused = reciprocal_rank_fusion([lex, den], limit=5) print(f" bm25 scores : {[round(s.score, 3) for s in lex]}") print(f" dense scores : {[round(s.score, 3) for s in den]}") print(f" fused : {[(s.chunk.chunk_id, round(s.score, 5)) for s in fused[:3]]}") print(" -> incomparable scales, never normalized. RRF reads positions only.") print(" 1st + 10th = 1/61 + 1/70 = 0.03068 ; 3rd + 3rd = 2/63 = 0.03175") print(" -> consistent agreement beats one strong signal.") print() print("=" * 78) print("4. ISOLATION IS THE NAMESPACE, NOT A FILTER") print("=" * 78) wholesale = Principal("u-42", "wholesale", max_classification="confidential") retail = Principal("u-7", "retail", max_classification="confidential") print(f" namespaces in the index: {vectors.namespaces()}") result = retriever.retrieve("collections goodwill reversal", wholesale) print(f" wholesale user asking a RETAIL question -> {len(result.chunks)} results, " f"considered={result.considered}") print(" the retail chunk was never a candidate: a different namespace was searched.") result = retriever.retrieve("collections goodwill reversal", retail) print(f" retail user, same question -> {[s.chunk.chunk_id for s in result.chunks]}") print() print("=" * 78) print("5. CLASSIFICATION AND BARRIERS, PRE-FILTERED") print("=" * 78) outside = Principal("u-42", "wholesale", max_classification="restricted") inside = Principal("u-99", "wholesale", max_classification="restricted", barriers=("advisory",)) for label, who in (("outside the barrier", outside), ("inside the barrier", inside)): got = retriever.retrieve("Acme Trading FZE acquisition", who) print(f" {label:<20}: {[s.chunk.chunk_id for s in got.chunks]}" f" excluded_by_entitlement={got.excluded_by_entitlement}") low = Principal("u-1", "wholesale", max_classification="internal") got = retriever.retrieve("PMT-771 held amount", low) print(f" cleared to 'internal' : {[s.chunk.chunk_id for s in got.chunks]}" f" (the confidential payment record is invisible)") print(" -> excluded BEFORE ranking. A post-hoc filter leaks existence through") print(" result-set size and is one refactor from leaking content.") print() print("=" * 78) print("6. FRESHNESS IS A CONTRACT") print("=" * 78) fresh = RetrievalPolicy(max_stale_ticks=30) got = retriever.retrieve("PMT-771 held", wholesale, now_tick=160, policy=fresh) print(f" now=160, max_stale=30 -> {[s.chunk.chunk_id for s in got.chunks]}" f" excluded_by_freshness={got.excluded_by_freshness}") got = retriever.retrieve("PMT-771 held", wholesale, now_tick=200, policy=fresh) print(f" now=200, max_stale=30 -> {[s.chunk.chunk_id for s in got.chunks]}" f" excluded_by_freshness={got.excluded_by_freshness}") print() print("=" * 78) print("7. RERANK IS THE FIRST THING YOU SHED") print("=" * 78) with_rerank = retriever.retrieve("dual control threshold for release", wholesale) without = retriever.retrieve("dual control threshold for release", wholesale, policy=RetrievalPolicy(enable_rerank=False)) print(f" with rerank : {[s.chunk.chunk_id for s in with_rerank.chunks]}") print(f" without : {[s.chunk.chunk_id for s in without.chunks]}" f" degraded={list(without.degraded)}") print(" -> the answer degrades in QUALITY and is still an answer. That is what makes") print(" retrieval a degradable dependency (Phase 00) rather than a serial one.") print() print("=" * 78) print("8. GROUNDING: EVERY CLAIM CITES, OR THE ANSWER FAILS") print("=" * 78) evidence = retriever.retrieve("PMT-771 hold reason and release rules", wholesale) claims = [ Claim("Payment PMT-771 for AED 250,000 to Acme Trading FZE is currently HELD"), Claim("Releases above AED 100,000 require dual control by two authorised officers"), Claim("The customer has been notified by email"), # invented ] report = check_grounding(claims, evidence.chunks) for claim in report.supported: print(f" OK {claim.text[:58]:<58} <- {claim.citation}") for claim in report.unsupported: print(f" FAIL {claim.text[:58]:<58} <- no supporting span") print(f" grounded={report.is_grounded} coverage={report.coverage:.0%}") print() print("=" * 78) print("9. CONTEXT ASSEMBLY, IN CACHE-FRIENDLY ORDER") print("=" * 78) context = assemble_context( budget_tokens=260, instructions="You are a wholesale payments investigator. Cite every claim.", tool_schemas="payments.lookup(reference) -> status", policy="Never disclose MNPI. Escalate matches above 0.80.", memory="The user is a relationship manager for Acme Trading FZE.", question="Why is PMT-771 held and can it be released?", retrieved=evidence.chunks, ) print(f" tokens={context.tokens}/260 included={len(context.included_chunks)}" f" dropped={len(context.dropped_chunks)}") print(f" stable prefix = {context.stable_prefix_tokens} tokens " f"({context.cacheable_fraction:.0%} of the prompt is cacheable)") print(f" segment order : {[line[1:-1] for line in context.text.split(chr(10)) if line.startswith('[') and line.endswith(']')]}") print(" -> stable first, question last. The first changed byte invalidates the") print(" provider-side prefix cache, so this ordering is a cost decision.") if __name__ == "__main__": # pragma: no cover main()