diff --git a/python/hachimoji_citation.py b/python/hachimoji_citation.py new file mode 100644 index 00000000..fa7df266 --- /dev/null +++ b/python/hachimoji_citation.py @@ -0,0 +1,406 @@ +#!/usr/bin/env python3 +""" +hachimoji_citation.py — Equation → Hachimoji State → arxiv Citations + +Pipeline: + equation string + → parse shape (n_vars, n_ops, max_depth, n_quantifiers, n_relations) + → classify to one of 8 Hachimoji states (mirrors HachimojiCodec.lean) + → build semantic query from state meaning + → hybrid_search(query_text, embedding, top_k) via arxiv-pg podman container + → return top-k arxiv citations for that equation's regime + +DB: podman container 'arxiv-pg', database 'arxiv'. +Embedding: sentence-transformers/static-retrieval-mrl-en-v1 (1024-dim). +Search: hybrid_search() — RRF merger of pg_trgm + pgvector HNSW (22ms). + +Mirrors formal definitions in: + formal/CoreFormalism/HachimojiCodec.lean — classifyEquation, admission + formal/CoreFormalism/HachimojiLUT.lean — equationPosition +""" + +from __future__ import annotations + +import json +import re +import subprocess +from dataclasses import dataclass, field +from typing import Optional + +MODEL_NAME = "sentence-transformers/static-retrieval-mrl-en-v1" + +# ── State semantic queries ─────────────────────────────────────────────────── +# Each Hachimoji state maps to a natural-language query that retrieves +# papers relevant to that regime from the arxiv corpus. +# Source: HachimojiBase.lean §4 semantic cross-reference. + +STATE_QUERIES: dict[str, str] = { + "Φ": ( + "density boundary ordered lattice phase transition Erdős-Rényi " + "trivial regime fundamental equation constants few variables" + ), + "Λ": ( + "Barnes-Wall lattice room regime bounded quantifiers linear algebra " + "shallow depth ordered exploration" + ), + "Ρ": ( + "spectral radius stability boundary tight regime operator norm " + "high operator count convergence fixed point" + ), + "Κ": ( + "complementarity threshold marginal regime softplus many variables " + "near threshold complementarity duality" + ), + "Ω": ( + "Goormaghtigh collision eigensolid terminal fixed point Baker bound " + "contradiction repunit equation zero lambda" + ), + "Σ": ( + "Goormaghtigh symmetric partner solution swap algebraic geometry " + "symmetric balanced equation equal structure" + ), + "Π": ( + "potential violation density probe Baker threshold calculus " + "high complexity integral limit series path integral" + ), + "Ζ": ( + "Riemann zeta near cancellation zero region near zero no integer point " + "fallback empty undefined bare expression" + ), +} + +# ── Equation shape (mirrors HachimojiCodec.lean EquationShape) ────────────── + +@dataclass +class EquationShape: + n_vars: int = 0 + n_ops: int = 0 + max_depth: int = 0 + n_quantifiers: int = 0 + n_relations: int = 0 + + +# ── Classifier (mirrors HachimojiCodec.lean classifyEquation) ─────────────── +# MUST stay in sync with the Lean definition. If HachimojiCodec.lean changes, +# update this function and add a note here. + +def classify_equation(s: EquationShape) -> str: + """Return the Hachimoji state label for an EquationShape. + + Exact mirror of HachimojiCodec.lean §8 classifyEquation, order preserved: + 1. Ω contradiction (n_vars=0, n_ops=0, n_relations≥1) + 2. Λ bounded quantifiers, shallow depth + 3. Ζ empty/bare + 4. Φ fundamental (few vars, few ops, has relation) + 5. Π high complexity + 6. Σ symmetric structure + 7. Ρ high ops, no quantifiers + 8. Κ many variables, shallow + 9. Ζ default fallback + """ + if s.n_vars == 0 and s.n_ops == 0 and s.n_relations >= 1: + return "Ω" + if s.n_quantifiers > 0 and s.max_depth <= 2: + return "Λ" + if s.n_vars <= 1 and s.n_ops == 0 and s.n_relations == 0: + return "Ζ" + if s.n_vars <= 3 and s.n_quantifiers == 0 and s.n_ops <= 5 and s.n_relations >= 1: + return "Φ" + if s.n_ops + s.n_vars * s.max_depth + s.n_quantifiers * 2 >= 8 or s.n_ops > 8: + return "Π" + is_symmetric = ( + s.n_relations >= 1 and s.n_vars >= 2 + and 1 <= s.n_ops <= 10 and s.n_quantifiers == 0 + ) + if is_symmetric: + return "Σ" + if s.n_ops > 5 and s.n_quantifiers == 0: + return "Ρ" + if s.n_vars > 5 and s.max_depth <= 1: + return "Κ" + return "Ζ" + + +def admission(state: str) -> str: + """Mirror of HachimojiCodec.lean §4 admission. + + ADMIT: Φ, Λ, Ρ, Κ, Σ + QUARANTINE: Ω, Π, Ζ + """ + return "ADMIT" if state in {"Φ", "Λ", "Ρ", "Κ", "Σ"} else "QUARANTINE" + + +# ── Simple equation shape parser ──────────────────────────────────────────── + +def parse_shape(equation: str) -> EquationShape: + """Heuristic shape extraction from a raw equation string. + + Good enough for classification; not a full parse. Replace with a proper + AST parser when available. + """ + ops = len(re.findall(r'[+\-*/^√∫∑∏∂∇]', equation)) + vars_ = len(set(re.findall(r'\b[a-zA-Z]\b', equation))) + rels = len(re.findall(r'[=<>≤≥≠≈]', equation)) + quants = len(re.findall(r'[∀∃]', equation)) + # depth: count nested parens / brackets + depth = 0 + max_d = 0 + for ch in equation: + if ch in '([{': + depth += 1 + max_d = max(max_d, depth) + elif ch in ')]}': + depth = max(0, depth - 1) + return EquationShape( + n_vars=vars_, + n_ops=ops, + max_depth=max_d, + n_quantifiers=quants, + n_relations=rels, + ) + + +# ── DB helpers ─────────────────────────────────────────────────────────────── + +def psql(sql: str, timeout: int = 30) -> tuple[str, str]: + """Run SQL in the arxiv-pg podman container.""" + r = subprocess.run( + ['podman', 'exec', 'arxiv-pg', 'psql', + '-U', 'postgres', '-d', 'arxiv', '-t', '-A', '-c', sql], + capture_output=True, text=True, timeout=timeout, + ) + return r.stdout.strip(), r.stderr.strip() + + +def embed(text: str) -> list[float]: + """Embed a query string with the canonical model (1024-dim, normalised).""" + from sentence_transformers import SentenceTransformer + model = SentenceTransformer(MODEL_NAME) + vec = model.encode(text, normalize_embeddings=True) + return vec.tolist() + + +# ── Citation dataclass ─────────────────────────────────────────────────────── + +@dataclass +class Citation: + paper_id: str + title: str + rrf_score: float + trigram_rank: int + vector_rank: int + + +# ── Main entry point ───────────────────────────────────────────────────────── + +@dataclass +class EquationCitation: + equation: str + shape: EquationShape + state: str + admission: str + phase_deg: int + query: str + citations: list[Citation] = field(default_factory=list) + + +STATE_PHASE: dict[str, int] = { + "Φ": 0, "Λ": 45, "Ρ": 90, "Κ": 135, + "Ω": 180, "Σ": 225, "Π": 270, "Ζ": 315, +} + + +def cite_equation( + equation: str, + shape: Optional[EquationShape] = None, + top_k: int = 5, + use_embedding: bool = True, +) -> EquationCitation: + """Classify an equation and retrieve its top-k arxiv citations. + + Args: + equation: Raw equation string (LaTeX or ASCII math). + shape: Pre-computed shape; parsed from equation if None. + top_k: Number of citations to retrieve. + use_embedding: If True, use semantic embedding (requires model); + if False, trigram-only (faster, no model needed). + + Returns: + EquationCitation with state, phase, admission, and top-k papers. + """ + if shape is None: + shape = parse_shape(equation) + + state = classify_equation(shape) + adm = admission(state) + phase = STATE_PHASE[state] + query = STATE_QUERIES[state] + + # Augment query with the equation itself (first 200 chars) + full_query = f"{query} {equation[:200]}".strip() + full_query_esc = full_query.replace("'", "''") + + if use_embedding: + emb = embed(full_query) + emb_s = '[' + ','.join(f'{x:.6f}' for x in emb) + ']' + sql = ( + f"SELECT paper_id, title, trigram_rank, vector_rank, rrf_score " + f"FROM hybrid_search('{full_query_esc}', '{emb_s}'::vector(1024), {top_k})" + ) + else: + # trigram-only fallback via ts_rank + sql = ( + f"SELECT paper_id, title, " + f"row_number() OVER () AS trigram_rank, " + f"0 AS vector_rank, " + f"ts_rank(to_tsvector('english', title || ' ' || COALESCE(abstract, '')), " + f" plainto_tsquery('english', '{full_query_esc}')) AS rrf_score " + f"FROM arxiv_papers " + f"WHERE to_tsvector('english', title || ' ' || COALESCE(abstract, '')) " + f" @@ plainto_tsquery('english', '{full_query_esc}') " + f"ORDER BY rrf_score DESC LIMIT {top_k}" + ) + + out, err = psql(sql) + citations: list[Citation] = [] + if err and "ERROR" in err: + print(f"DB error: {err[:120]}") + else: + for line in out.split("\n"): + if not line.strip(): + continue + parts = line.split("|") + if len(parts) >= 5: + try: + citations.append(Citation( + paper_id = parts[0].strip(), + title = parts[1].strip()[:200], + trigram_rank = int(parts[2].strip()), + vector_rank = int(parts[3].strip()), + rrf_score = float(parts[4].strip()), + )) + except (ValueError, IndexError): + continue + + return EquationCitation( + equation = equation, + shape = shape, + state = state, + admission = adm, + phase_deg = phase, + query = full_query[:120], + citations = citations, + ) + + +def cite_batch( + equations: list[str], + top_k: int = 5, + use_embedding: bool = True, +) -> list[EquationCitation]: + """Classify and cite a batch of equations. + + Loads the embedding model once for efficiency. + """ + if use_embedding: + from sentence_transformers import SentenceTransformer + model = SentenceTransformer(MODEL_NAME) + else: + model = None + + results = [] + for eq in equations: + shape = parse_shape(eq) + state = classify_equation(shape) + query = f"{STATE_QUERIES[state]} {eq[:200]}".strip() + query_esc = query.replace("'", "''") + + if model is not None: + emb = model.encode(query, normalize_embeddings=True).tolist() + emb_s = '[' + ','.join(f'{x:.6f}' for x in emb) + ']' + sql = ( + f"SELECT paper_id, title, trigram_rank, vector_rank, rrf_score " + f"FROM hybrid_search('{query_esc}', '{emb_s}'::vector(1024), {top_k})" + ) + else: + sql = ( + f"SELECT paper_id, title, 0, 0, " + f"ts_rank(to_tsvector('english', title), " + f"plainto_tsquery('english', '{query_esc}')) " + f"FROM arxiv_papers " + f"WHERE to_tsvector('english', title) @@ plainto_tsquery('english', '{query_esc}') " + f"ORDER BY 5 DESC LIMIT {top_k}" + ) + + out, err = psql(sql) + citations: list[Citation] = [] + for line in out.split("\n"): + parts = line.split("|") + if len(parts) >= 5: + try: + citations.append(Citation( + paper_id=parts[0].strip(), title=parts[1].strip()[:200], + trigram_rank=int(parts[2].strip()), + vector_rank=int(parts[3].strip()), + rrf_score=float(parts[4].strip()), + )) + except (ValueError, IndexError): + continue + + results.append(EquationCitation( + equation=eq, shape=shape, state=state, + admission=admission(state), + phase_deg=STATE_PHASE[state], + query=query[:120], citations=citations, + )) + return results + + +# ── CLI ────────────────────────────────────────────────────────────────────── + +if __name__ == "__main__": + import argparse, sys + + parser = argparse.ArgumentParser( + description="Classify an equation and retrieve arxiv citations.") + parser.add_argument("equation", nargs="?", + help="Equation string (quote it)") + parser.add_argument("--top-k", type=int, default=5) + parser.add_argument("--no-embed", action="store_true", + help="Trigram-only; skip embedding model") + parser.add_argument("--json", action="store_true", + help="Output JSON") + args = parser.parse_args() + + if not args.equation: + # Demo batch + demo = [ + "E = mc^2", + "a^2 + b^2 = c^2", + "∀x. P(x) → Q(x)", + "0 = 1", + "∫ f(x) dx = F(x) + C", + ] + results = cite_batch(demo, top_k=args.top_k, + use_embedding=not args.no_embed) + else: + results = [cite_equation(args.equation, top_k=args.top_k, + use_embedding=not args.no_embed)] + + if args.json: + def _serial(obj): + if hasattr(obj, '__dict__'): + return obj.__dict__ + return str(obj) + print(json.dumps([r.__dict__ for r in results], default=_serial, indent=2)) + else: + for r in results: + print(f"\n{'='*60}") + print(f"Equation : {r.equation}") + print(f"State : {r.state} phase={r.phase_deg}° admission={r.admission}") + print(f"Query : {r.query[:80]}…") + print(f"Citations:") + if not r.citations: + print(" (none)") + for i, c in enumerate(r.citations, 1): + print(f" {i}. [{c.rrf_score:.4f}] {c.title} ({c.paper_id})")