""" hv-split ======== Bundle of interpretations, not one answer. Every generative model picks one reading of an ambiguous query and answers it. hv-split refuses to pick. Given a query, it returns a *ranked bundle* of every interpretation it can detect — each with the ambiguity source, the ambiguous span, the reading, and a prior. This is a new output shape: not a label, not a completion, not a ranking of documents, but a distribution over *readings of the same query*. Ambiguity sources detected: referential pronouns with multiple candidate antecedents lexical polysemous terms with multiple senses scope negation scoping over/under a quantifier presupposition "why did X" presupposes X happened framing "in the language of Y" commits to a frame temporal vague temporal references ("recently", "soon") No dependencies. Pure stdlib. Runs anywhere Python 3.9+ runs. Author: zeechimp License: Apache-2.0 """ from __future__ import annotations import argparse import json import math import os import re import sys from collections import Counter from dataclasses import asdict, dataclass, field from typing import Callable, Dict, List, Optional, Tuple # --------------------------------------------------------------------------- # Lexicon # --------------------------------------------------------------------------- DEFAULT_POLYSEMOUS: Dict[str, List[str]] = { "bank": ["financial institution", "river edge", "memory bank", "blood bank"], "trust": ["confidence", "legal entity", "believe", "rely on"], "frame": ["physical border", "viewpoint", "incriminate falsely", "data unit"], "run": ["execute program", "physical movement", "operate", "flow"], "set": ["collection", "to place", "configuration", "sun decline"], "function": ["mathematical mapping", "role or purpose", "working state", "subroutine"], "model": ["mathematical object", "ML model", "fashion subject", "small replica"], "order": ["request or command", "sequence", "mathematical order", "purchase"], "memory": ["biological memory", "computer memory", "recollection", "memorial"], "state": ["political entity", "condition or mode", "verb: to say", "configuration"], "point": ["location", "argument", "score", "moment in time"], "line": ["geometric line", "queue", "text row", "connection"], "sense": ["perception", "meaning", "direction", "reason"], "light": ["illumination", "not heavy", "not dark", "to ignite"], "spring": ["season", "metal coil", "water source", "to jump"], "kind": ["type", "benevolent", "species"], "present": ["gift", "here now", "to show", "current time"], "mean": ["average", "unkind", "intend", "signify"], "note": ["musical note", "written message", "observe", "distinction"], "fine": ["good", "penalty", "thin", "precise"], } # --------------------------------------------------------------------------- # Pronoun forms # --------------------------------------------------------------------------- PRONOUN_FORMS: Dict[str, List[str]] = { "it": ["its"], "this": ["this"], "that": ["that"], "they": ["them", "their"], "these": ["these"], "those": ["those"], "he": ["him", "his"], "she": ["her"], } # --------------------------------------------------------------------------- # English pronouns — never valid noun phrase candidates # --------------------------------------------------------------------------- ENGLISH_PRONOUNS = { "i", "me", "my", "mine", "myself", "you", "your", "yours", "yourself", "he", "him", "his", "himself", "she", "her", "hers", "herself", "it", "its", "itself", "we", "us", "our", "ours", "ourselves", "they", "them", "their", "theirs", "themselves", "this", "that", "these", "those", "who", "whom", "whose", "which", "what", "anyone", "someone", "everyone", "nobody", } # --------------------------------------------------------------------------- # Second-word stop list for "the X Y" NP pattern # --------------------------------------------------------------------------- NP_SECOND_STOP = { # auxiliaries / verbs "is", "are", "was", "were", "be", "been", "being", "has", "have", "had", "do", "does", "did", "will", "would", "shall", "should", "can", "could", "may", "might", "must", "ought", # pronouns "i", "you", "he", "she", "it", "we", "they", "me", "him", "her", "us", "them", # conjunctions / prepositions "and", "or", "but", "nor", "so", "yet", "for", "of", "in", "on", "at", "to", "by", "with", "from", "into", "over", "under", "about", "through", "between", # wh-words "that", "which", "who", "whom", "whose", "where", "when", "why", "how", "what", "whether", "if", } # --------------------------------------------------------------------------- # Stop words that terminate a frame term # --------------------------------------------------------------------------- FRAME_STOP = { "had", "has", "have", "is", "was", "were", "will", "would", "should", "could", "can", "may", "might", "must", "do", "does", "did", "be", "been", "being", "and", "or", "but", "that", "which", "where", "when", "why", "how", "what", "who", "the", "a", "an", "to", "for", "of", "in", "on", "at", "by", } # --------------------------------------------------------------------------- # Presupposition patterns # --------------------------------------------------------------------------- PRESUP_PATTERNS: List[Tuple[str, str]] = [ (r"\bwhy (?:did|does|is|are|was|were)\b", "'why did X' presupposes X occurred or holds"), (r"\bwhen (?:did|does|will)\b", "'when did X' presupposes X occurs"), (r"\bhow (?:did|does)\b", "'how did X' presupposes X occurred"), (r"\bhave you stopped\b", "classical 'have you stopped X' presupposition trap"), (r"\bthe current\b", "'the current X' presupposes a unique current instance"), (r"\bthe only\b", "'the only X' presupposes uniqueness"), (r"\bthe last\b", "'the last X' presupposes a well-defined last occurrence"), ] # --------------------------------------------------------------------------- # Framing patterns # --------------------------------------------------------------------------- FRAME_PATTERNS: List[str] = [ r"\bin the (?:language|idiom|vocabulary|style|spirit) of\s+([^.,;!?]+)", r"\bfrom the (?:perspective|viewpoint|angle|standpoint) of\s+([^.,;!?]+)", r"\bthrough the (?:lens|frame|prism) of\s+([^.,;!?]+)", r"\bwithin the framework of\s+([^.,;!?]+)", r"\bunder the (?:assumption|interpretation|reading|hypothesis) that\s+([^.,;!?]+)", ] # --------------------------------------------------------------------------- # Temporal patterns # --------------------------------------------------------------------------- TEMPORAL_PATTERNS: Dict[str, str] = { r"\brecently\b": "recently = this week, this year, or this decade?", r"\bsoon\b": "soon = minutes, days, or years?", r"\blately\b": "lately = since when, exactly?", r"\bcurrently\b": "currently = as of now, as of this writing, or today?", r"\bpreviously\b": "previously = before what?", r"\bnow\b": "now = this instant, this era, or this moment in the argument?", r"\bbefore the (?:change|update|shift)\b": "which change/update/shift is meant?", r"\bafter the (?:change|update|shift)\b": "which change/update/shift is meant?", } # --------------------------------------------------------------------------- # Scope pattern # --------------------------------------------------------------------------- SCOPE_RE = re.compile( r"\b(all|every|each|any)\b.*?\b(not|n't|never|cannot|can't|" r"won't|doesn't|don't|isn't|aren't)\b[^.!?]*", re.IGNORECASE, ) # --------------------------------------------------------------------------- # Dataclasses # --------------------------------------------------------------------------- @dataclass class Interpretation: source: str span: str span_range: Tuple[int, int] reading: str prior: float rationale: str def to_dict(self) -> dict: d = asdict(self) d["span_range"] = list(self.span_range) return d @dataclass class SplitBundle: query: str interpretations: List[Interpretation] ambiguity_score: float confidence: float entropy: float dominant_source: str sources: List[str] n_interpretations: int def to_dict(self) -> dict: return { "query": self.query, "interpretations": [i.to_dict() for i in self.interpretations], "ambiguity_score": self.ambiguity_score, "confidence": self.confidence, "entropy": self.entropy, "dominant_source": self.dominant_source, "sources": self.sources, "n_interpretations": self.n_interpretations, } @dataclass class HVInterpretConfig: max_interpretations: int = 16 max_per_source: int = 6 window_chars: int = 220 source_weights: Dict[str, float] = field(default_factory=lambda: { "referential": 1.0, "lexical": 1.0, "scope": 1.0, "presuppositional": 1.0, "framing": 1.0, "temporal": 1.0, }) version: str = "0.1.1" # --------------------------------------------------------------------------- # Model # --------------------------------------------------------------------------- class HVInterpret: """Split an ambiguous query into a ranked bundle of interpretations.""" def __init__( self, lexicon: Optional[Dict[str, List[str]]] = None, config: Optional[HVInterpretConfig] = None, ): self.lexicon = lexicon if lexicon is not None else DEFAULT_POLYSEMOUS self.config = config or HVInterpretConfig() self._obs = 0 def __repr__(self) -> str: return ( f"HVInterpret(lexicon={len(self.lexicon)} words, " f"max_interps={self.config.max_interpretations})" ) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def split(self, query: str) -> SplitBundle: """Return a bundle of interpretations for the query.""" if not query or not query.strip(): return self._empty_bundle(query) per_source: Dict[str, List[Interpretation]] = { "referential": self._detect_referential(query), "lexical": self._detect_lexical(query), "scope": self._detect_scope(query), "presuppositional": self._detect_presuppositional(query), "framing": self._detect_framing(query), "temporal": self._detect_temporal(query), } # Within-source normalization, then apply source weights all_interps: List[Interpretation] = [] for source, interps in per_source.items(): if not interps: continue interps = interps[: self.config.max_per_source] s = sum(i.prior for i in interps) or 1.0 weight = self.config.source_weights.get(source, 1.0) for i in interps: i.prior = (i.prior / s) * weight all_interps.extend(interps) if not all_interps: return self._empty_bundle(query) # Global normalization to marginals total = sum(i.prior for i in all_interps) or 1.0 for i in all_interps: i.prior = i.prior / total all_interps.sort(key=lambda x: -x.prior) all_interps = all_interps[: self.config.max_interpretations] # Re-normalize after the cap total = sum(i.prior for i in all_interps) or 1.0 for i in all_interps: i.prior = i.prior / total priors = [i.prior for i in all_interps] max_prior = max(priors) entropy = -sum(p * math.log(p + 1e-12) for p in priors) max_entropy = math.log(len(priors)) if len(priors) > 1 else 1.0 ambiguity = entropy / max_entropy if max_entropy > 0 else 0.0 if len(all_interps) == 1: ambiguity = 0.0 confidence = 1.0 else: confidence = max_prior sources = sorted({i.source for i in all_interps}) dominant = all_interps[0].source return SplitBundle( query=query, interpretations=all_interps, ambiguity_score=float(ambiguity), confidence=float(confidence), entropy=float(entropy), dominant_source=dominant, sources=sources, n_interpretations=len(all_interps), ) def ambiguity(self, query: str) -> float: return self.split(query).ambiguity_score def most_likely(self, query: str) -> Optional[Interpretation]: b = self.split(query) return b.interpretations[0] if b.interpretations else None def answer_each( self, query: str, answer_fn: Callable[[str, Interpretation], str], ) -> List[Tuple[Interpretation, str]]: """Apply an answer callback to each interpretation. `answer_fn(query, interpretation)` -> str. Returns a list of (interpretation, answer) pairs. """ b = self.split(query) return [(i, answer_fn(b.query, i)) for i in b.interpretations] # ------------------------------------------------------------------ # Render # ------------------------------------------------------------------ def render(self, bundle: SplitBundle, mode: str = "text") -> str: if mode == "markdown": return self._render_markdown(bundle) if mode == "json": return json.dumps(bundle.to_dict(), indent=2) return self._render_text(bundle) # ------------------------------------------------------------------ # Detectors # ------------------------------------------------------------------ def _detect_referential(self, query: str) -> List[Interpretation]: nps = self._extract_noun_phrases(query) results: List[Interpretation] = [] seen_pronouns: set = set() for pronoun in PRONOUN_FORMS: if pronoun in seen_pronouns: continue m = re.search(rf"\b{pronoun}\b", query, re.IGNORECASE) if not m: continue prefix_start = max(0, m.start() - self.config.window_chars) candidates = [ (text, start, end) for text, start, end in nps if end <= m.start() and start >= prefix_start ] # Deduplicate by text seen_text = set() unique: List[Tuple[str, int, int]] = [] for text, start, end in candidates: key = text.lower() if key in seen_text: continue seen_text.add(key) unique.append((text, start, end)) # Cap at 3 most recent unique = sorted(unique, key=lambda c: -c[2])[:3] if len(unique) < 2: continue seen_pronouns.add(pronoun) for text, start, end in unique: results.append(Interpretation( source="referential", span=query[m.start():m.end()], span_range=(m.start(), m.end()), reading=f"'{pronoun}' refers to '{text}'", prior=1.0 / len(unique), rationale=( f"pronoun '{pronoun}' has {len(unique)} candidate " f"antecedents in the preceding window" ), )) return results def _detect_lexical(self, query: str) -> List[Interpretation]: results: List[Interpretation] = [] seen: set = set() for word, readings in self.lexicon.items(): if word in seen: continue m = re.search(rf"\b{re.escape(word)}\b", query, re.IGNORECASE) if not m: continue seen.add(word) for reading in readings: results.append(Interpretation( source="lexical", span=m.group(), span_range=(m.start(), m.end()), reading=f"'{word}' = {reading}", prior=1.0 / len(readings), rationale=f"polysemous term with {len(readings)} readings", )) return results def _detect_scope(self, query: str) -> List[Interpretation]: m = SCOPE_RE.search(query) if not m: return [] span_text = m.group() span_start = m.start() span_end = m.end() return [ Interpretation( source="scope", span=span_text, span_range=(span_start, span_end), reading="wide negation: NOT (all X Y)", prior=0.5, rationale="'not' can scope over the quantifier", ), Interpretation( source="scope", span=span_text, span_range=(span_start, span_end), reading="narrow negation: ALL X (NOT Y)", prior=0.5, rationale="'not' can scope under the quantifier", ), ] def _detect_presuppositional(self, query: str) -> List[Interpretation]: for pattern, rationale in PRESUP_PATTERNS: m = re.search(pattern, query, re.IGNORECASE) if not m: continue return [ Interpretation( source="presuppositional", span=m.group(), span_range=(m.start(), m.end()), reading="presupposition holds: the event or state is real", prior=0.7, rationale=rationale + " — default reading", ), Interpretation( source="presuppositional", span=m.group(), span_range=(m.start(), m.end()), reading="presupposition fails: the event or state may not hold", prior=0.3, rationale=rationale + " — can be challenged in a reply", ), ] return [] def _detect_framing(self, query: str) -> List[Interpretation]: for pattern in FRAME_PATTERNS: m = re.search(pattern, query, re.IGNORECASE) if not m: continue frame_term = (m.group(1).strip() if m.groups() else "the frame") # Truncate the frame term at the first function-word boundary tokens = frame_term.split() kept = [] for tok in tokens: if tok.lower() in FRAME_STOP: break kept.append(tok) frame_term = " ".join(kept).strip() if not frame_term: frame_term = "the frame" # Truncate the span to match the truncated frame term span_text = m.group() if m.groups(): original_term = m.group(1).strip() if frame_term != original_term: idx = span_text.find(frame_term) if idx >= 0: span_text = span_text[: idx + len(frame_term)] return [ Interpretation( source="framing", span=span_text, span_range=(m.start(), m.start() + len(span_text)), reading=f"answer strictly within the frame of '{frame_term}'", prior=0.7, rationale=f"frame '{frame_term}' requested explicitly", ), Interpretation( source="framing", span=span_text, span_range=(m.start(), m.start() + len(span_text)), reading=( f"answer outside the frame, noting that " f"'{frame_term}' is one choice among many" ), prior=0.3, rationale="frame may be a placeholder the user wants challenged", ), ] return [] def _detect_temporal(self, query: str) -> List[Interpretation]: results: List[Interpretation] = [] for pattern, note in TEMPORAL_PATTERNS.items(): m = re.search(pattern, query, re.IGNORECASE) if not m: continue results.append(Interpretation( source="temporal", span=m.group(), span_range=(m.start(), m.end()), reading=f"'{m.group()}' takes a narrow reading", prior=0.5, rationale=note + " — narrow vs broad", )) results.append(Interpretation( source="temporal", span=m.group(), span_range=(m.start(), m.end()), reading=f"'{m.group()}' takes a broad reading", prior=0.5, rationale=note + " — broad vs narrow", )) break return results # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _extract_noun_phrases( self, text: str ) -> List[Tuple[str, int, int]]: """Extract plausible noun phrases, filtering out pronouns and verb continuations.""" nps: List[Tuple[str, int, int]] = [] # 1. Capitalized sequences (proper nouns) — skip pronouns for m in re.finditer(r"\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b", text): captured = m.group() if captured.lower() in ENGLISH_PRONOUNS: continue nps.append((captured, m.start(), m.end())) # 2. Two-word NPs first: "the X Y" where Y is not a stop word two_re = re.compile( r"\b(?:the|a|an)\s+([a-z]+)\s+([a-z]+)\b", re.IGNORECASE ) used_starts = set() for m in two_re.finditer(text): second = m.group(2).lower() if second in NP_SECOND_STOP: continue captured = m.group(0) if captured.lower() in ENGLISH_PRONOUNS: continue used_starts.add(m.start()) nps.append((captured, m.start(), m.end())) # 3. Single-word NPs: "the X" / "a X" / "an X" single_re = re.compile( r"\b(?:the|a|an)\s+([a-z]+)\b", re.IGNORECASE ) for m in single_re.finditer(text): if m.start() in used_starts: continue captured = m.group(0) if captured.lower() in ENGLISH_PRONOUNS: continue nps.append((captured, m.start(), m.end())) # 4. Deduplicate by (text, start) seen = set() unique = [] for np_ in nps: key = (np_[0].lower(), np_[1]) if key in seen: continue seen.add(key) unique.append(np_) return unique def _empty_bundle(self, query: str) -> SplitBundle: return SplitBundle( query=query, interpretations=[], ambiguity_score=0.0, confidence=1.0, entropy=0.0, dominant_source="none", sources=[], n_interpretations=0, ) def _render_text(self, b: SplitBundle) -> str: lines = [] lines.append("=" * 78) lines.append(f"hv-split -- bundle for: {b.query!r}") lines.append("=" * 78) lines.append("") lines.append(f"ambiguity_score : {b.ambiguity_score:.3f}") lines.append(f"confidence : {b.confidence:.3f}") lines.append(f"sources : {', '.join(b.sources) or '(none)'}") lines.append(f"interpretations : {b.n_interpretations}") lines.append("") if not b.interpretations: lines.append("no ambiguity detected.") return "\n".join(lines) lines.append("-" * 78) for idx, i in enumerate(b.interpretations, 1): lines.append( f"#{idx:<3} prior {i.prior:.3f} [{i.source}] {i.reading}" ) lines.append(f" span : {i.span!r}") lines.append(f" rationale : {i.rationale}") lines.append("") return "\n".join(lines) def _render_markdown(self, b: SplitBundle) -> str: lines = [ f"# hv-split bundle", "", f"**Query:** `{b.query}`", "", f"- ambiguity_score: `{b.ambiguity_score:.3f}`", f"- confidence: `{b.confidence:.3f}`", f"- sources: `{', '.join(b.sources) or 'none'}`", f"- interpretations: `{b.n_interpretations}`", "", ] if not b.interpretations: lines.append("_No ambiguity detected._") return "\n".join(lines) lines.append("| # | prior | source | reading |") lines.append("|---:|---:|---|---|") for idx, i in enumerate(b.interpretations, 1): lines.append( f"| {idx} | {i.prior:.3f} | `{i.source}` | {i.reading} |" ) return "\n".join(lines) # ------------------------------------------------------------------ # Persistence # ------------------------------------------------------------------ def save_pretrained(self, save_dir: str) -> None: os.makedirs(save_dir, exist_ok=True) payload = { "config": asdict(self.config), "lexicon": self.lexicon, "calibrated_on": self._obs, } with open(os.path.join(save_dir, "config.json"), "w") as f: json.dump(payload, f, indent=2) @classmethod def from_pretrained(cls, save_dir: str) -> "HVInterpret": with open(os.path.join(save_dir, "config.json"), "r") as f: payload = json.load(f) cfg = HVInterpretConfig(**payload.get("config", {})) lexicon = payload.get("lexicon", DEFAULT_POLYSEMOUS) obj = cls(lexicon=lexicon, config=cfg) obj._obs = int(payload.get("calibrated_on", 0)) return obj # --------------------------------------------------------------------------- # Demo # --------------------------------------------------------------------------- SAMPLE_LEXICAL = "I need to access the bank" SAMPLE_REFERENTIAL = "She told her that the manager had changed it, and it broke" SAMPLE_SCOPE = "All that glitters is not gold" SAMPLE_PRESUP = "Why did the CEO resign last week?" SAMPLE_FRAMING = ( "in the language of category theory, what is an identity?" ) SAMPLE_TEMPORAL = "recently, has the function changed?" SAMPLE_COMPLEX = ( "why did the current bank say recently that the function she used " "in the language of category theory had changed?" ) SAMPLE_CLEAN = "compute 2 + 2" def _demo(output_dir: str = "./hv_split_output") -> None: os.makedirs(output_dir, exist_ok=True) m = HVInterpret() print("=" * 78) print("DEMO 1 -- lexical ambiguity") print("=" * 78) print(m.render(m.split(SAMPLE_LEXICAL))) print() print("=" * 78) print("DEMO 2 -- referential ambiguity") print("=" * 78) print(m.render(m.split(SAMPLE_REFERENTIAL))) print() print("=" * 78) print("DEMO 3 -- scope ambiguity") print("=" * 78) print(m.render(m.split(SAMPLE_SCOPE))) print() print("=" * 78) print("DEMO 4 -- presuppositional ambiguity") print("=" * 78) print(m.render(m.split(SAMPLE_PRESUP))) print() print("=" * 78) print("DEMO 5 -- framing ambiguity") print("=" * 78) print(m.render(m.split(SAMPLE_FRAMING))) print() print("=" * 78) print("DEMO 6 -- temporal ambiguity") print("=" * 78) print(m.render(m.split(SAMPLE_TEMPORAL))) print() print("=" * 78) print("DEMO 7 -- multi-source (the full bundle)") print("=" * 78) print(m.render(m.split(SAMPLE_COMPLEX))) print() print("=" * 78) print("DEMO 8 -- unambiguous query") print("=" * 78) print(m.render(m.split(SAMPLE_CLEAN))) print() print("=" * 78) print("DEMO 9 -- answer_each (call a callback per interpretation)") print("=" * 78) b = m.split(SAMPLE_LEXICAL) for interp, answer in m.answer_each( SAMPLE_LEXICAL, lambda q, i: f"[would answer as: {i.reading}]", ): print(f" prior {interp.prior:.3f} -> {answer}") print() print("=" * 78) print("DEMO 10 -- JSON output") print("=" * 78) print(m.render(m.split(SAMPLE_PRESUP), mode="json")) print() print("=" * 78) print("DEMO 11 -- save / load round trip") print("=" * 78) path = os.path.join(output_dir, "split_model") m.save_pretrained(path) m2 = HVInterpret.from_pretrained(path) a = m.split(SAMPLE_COMPLEX).ambiguity_score b = m2.split(SAMPLE_COMPLEX).ambiguity_score print(f" saved to : {path}") print(f" file written : config.json") print(f" reloaded : {m2!r}") print(f" ambiguity : {a:.6f} (reloaded: {b:.6f})") print(f" identical : {abs(a - b) < 1e-9}") print() print("all demos complete.") # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def _cli() -> None: p = argparse.ArgumentParser( description="hv-split: bundle interpretations of an ambiguous query." ) p.add_argument("--query", type=str, default="", help="query to split (or '-' to read from stdin)") p.add_argument("--mode", type=str, default="text", choices=["text", "markdown", "json"], help="output format") p.add_argument("--ambiguity", action="store_true", help="print only the ambiguity score") p.add_argument("--save-to", type=str, default="", help="save the model to this directory") p.add_argument("--outdir", type=str, default="./hv_split_output", help="directory for --save-to and demo output") args = p.parse_args() query = sys.stdin.read() if args.query == "-" else args.query if not query: _demo(args.outdir) return m = HVInterpret() if args.save_to: m.save_pretrained(args.save_to) print(f"saved to {args.save_to}", file=sys.stderr) if args.ambiguity: print(f"{m.ambiguity(query):.4f}") return b = m.split(query) print(m.render(b, mode=args.mode)) if __name__ == "__main__": _cli()