diff --git a/generate/__init__.py b/generate/__init__.py index 0f426560..869ab987 100644 --- a/generate/__init__.py +++ b/generate/__init__.py @@ -1 +1,18 @@ +from .proposition import ( + FrameRegistry, + FrameSlot, + Proposition, + PropositionFrame, + propose, +) from .stream import generate, agenerate + +__all__ = [ + "FrameRegistry", + "FrameSlot", + "Proposition", + "PropositionFrame", + "agenerate", + "generate", + "propose", +] diff --git a/generate/proposition.py b/generate/proposition.py new file mode 100644 index 00000000..3e5c585d --- /dev/null +++ b/generate/proposition.py @@ -0,0 +1,333 @@ +""" +Structured proposition generation. + +A proposition is the first structured assertion above the surface walk: +prompt and field form a grade-2 relation blade; a frame is selected by exact +CGA inner product against that relation; vocabulary points then instantiate +the frame slots. + +No normalization happens here. This module consumes already-closed field and +vocabulary versors and uses only outer_product() plus cga_inner() for relation +and distance. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import json +from pathlib import Path +from typing import Iterable + +import numpy as np + +from algebra.cga import cga_inner, outer_product +from field.state import FieldState +from generate.stream import _articulate + +_PROJECT_ROOT = Path(__file__).resolve().parents[1] +_STOP_SURFACES = frozenset({"what", "who", "how", "why", "when", "which", "it", "to"}) + + +@dataclass(frozen=True, slots=True) +class FrameSlot: + name: str + required: bool + semantic_role: str | None = None + agreement_target: str | None = None + + +@dataclass(frozen=True, slots=True) +class PropositionFrame: + frame_id: str + language: str + predicate_type: str + slots: tuple[FrameSlot, ...] + constraints: tuple[str, ...] = field(default_factory=tuple) + relation: np.ndarray = field(default_factory=lambda: np.zeros(32, dtype=np.float32)) + + def __post_init__(self) -> None: + object.__setattr__(self, "slots", tuple(self.slots)) + object.__setattr__(self, "constraints", tuple(self.constraints)) + object.__setattr__(self, "relation", np.asarray(self.relation, dtype=np.float32).copy()) + + +@dataclass(frozen=True, slots=True) +class Proposition: + subject: str + predicate: str + object_: str | None + surface: str + frame_id: str + subject_versor: np.ndarray + predicate_versor: np.ndarray + object_versor: np.ndarray | None = None + relation: np.ndarray = field(default_factory=lambda: np.zeros(32, dtype=np.float32)) + + def __post_init__(self) -> None: + subject_versor = np.asarray(self.subject_versor, dtype=np.float32).copy() + predicate_versor = np.asarray(self.predicate_versor, dtype=np.float32).copy() + object.__setattr__(self, "subject_versor", subject_versor) + object.__setattr__(self, "predicate_versor", predicate_versor) + if self.object_versor is not None: + object_versor = np.asarray(self.object_versor, dtype=np.float32).copy() + object.__setattr__(self, "object_versor", object_versor) + object.__setattr__(self, "relation", np.asarray(self.relation, dtype=np.float32).copy()) + + +class FrameRegistry: + """Exact frame selection over precompiled frame relation blades.""" + + def __init__(self, frames: Iterable[PropositionFrame]) -> None: + self._frames = tuple(frames) + if not self._frames: + raise ValueError("FrameRegistry requires at least one frame.") + + @classmethod + def from_pack(cls, pack: str, vocab) -> "FrameRegistry": + """ + Load frames from packs//frames.jsonl. + + The shipped Koine directory is named both `el` and `grc` in different + layers; this accepts either spelling and reads the project pack files. + """ + pack_dir = _PROJECT_ROOT / "packs" / pack + if not pack_dir.exists() and pack == "el": + pack_dir = _PROJECT_ROOT / "packs" / "grc" + if not pack_dir.exists() and pack == "grc": + pack_dir = _PROJECT_ROOT / "packs" / "el" + return cls.from_jsonl(pack_dir / "frames.jsonl", vocab) + + @classmethod + def from_jsonl(cls, path: str | Path, vocab) -> "FrameRegistry": + path = Path(path) + frames: list[PropositionFrame] = [] + for line in path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + payload = json.loads(line) + frame = _parse_frame(payload, vocab) + frames.append(frame) + return cls(frames) + + def select(self, relation: np.ndarray) -> PropositionFrame: + relation = np.asarray(relation, dtype=np.float32) + return max(self._frames, key=lambda frame: cga_inner(relation, frame.relation)) + + def __iter__(self): + return iter(self._frames) + + def __len__(self) -> int: + return len(self._frames) + + +def propose(field_state: FieldState, vault, vocab, frame_registry: FrameRegistry) -> Proposition: + """ + Generate one structured proposition from the live field. + + The prompt field is `holonomy` when injection supplied it; otherwise the + current field is used. The selected subject is nearest the prompt. The + predicate is nearest the current field with the subject and trivial stop + wells excluded. The resulting proposition can be stored directly in the + vault metadata while its `surface` remains the emitted text. + """ + prompt = _prompt_versor(field_state) + relation = outer_product(prompt, field_state.F) + frame = frame_registry.select(relation) + + subject_word, subject_idx = _nearest_content_word( + vocab, + prompt, + exclude_indices=frozenset(), + preferred_pos=frozenset({"noun", "pronoun"}), + ) + predicate_word, predicate_idx = _nearest_content_word( + vocab, + field_state.F, + exclude_indices=frozenset({subject_idx}), + ) + + object_word: str | None = None + object_versor: np.ndarray | None = None + if _frame_wants_object(frame): + object_word, object_idx = _nearest_content_word( + vocab, + relation, + exclude_indices=frozenset({subject_idx, predicate_idx}), + preferred_pos=frozenset({"noun", "pronoun"}), + ) + object_versor = vocab.get_versor_at(object_idx) + + subject = _articulate(vocab, subject_word) + predicate = _articulate(vocab, predicate_word) + object_surface = _articulate(vocab, object_word) if object_word is not None else None + surface = _render_surface(frame, subject, predicate, object_surface) + + proposition = Proposition( + subject=subject, + predicate=predicate, + object_=object_surface, + surface=surface, + frame_id=frame.frame_id, + subject_versor=vocab.get_versor_at(subject_idx), + predicate_versor=vocab.get_versor_at(predicate_idx), + object_versor=object_versor, + relation=relation, + ) + if vault is not None: + vault.store(field_state.F, {"kind": "proposition", "proposition": proposition}) + return proposition + + +def _parse_frame(payload: dict, vocab) -> PropositionFrame: + slots = tuple( + FrameSlot( + name=slot["name"], + required=bool(slot["required"]), + semantic_role=slot.get("semantic_role"), + agreement_target=slot.get("agreement_target"), + ) + for slot in payload.get("slots", ()) + ) + relation = _frame_relation(payload, vocab) + return PropositionFrame( + frame_id=payload["frame_id"], + language=payload["language"], + predicate_type=payload["predicate_type"], + slots=slots, + constraints=tuple(payload.get("constraints", ())), + relation=relation, + ) + + +def _frame_relation(payload: dict, vocab) -> np.ndarray: + left = _first_existing(vocab, _role_anchor_candidates(payload)) + right = _first_existing(vocab, _predicate_anchor_candidates(payload)) + if left is None or right is None: + left = vocab.get_word_at(0) + right = vocab.get_word_at(1 if len(vocab) > 1 else 0) + return outer_product(vocab.get_versor(left), vocab.get_versor(right)) + + +def _role_anchor_candidates(payload: dict) -> tuple[str, ...]: + text = " ".join( + [payload.get("predicate_type", "")] + + [ + " ".join(str(slot.get(k, "")) for k in ("name", "semantic_role")) + for slot in payload.get("slots", ()) + ] + ).lower() + if "creation" in text or "agent" in text: + return ("create", "κτίζω", "ברא", "λόγος", "דבר", "word") + if "pros" in text or "direction" in text or "accompan" in text: + return ("with", "λόγος", "דבר", "word") + return ( + "light", + "φῶς", + "אוֹר", + "אור", + "truth", + "ἀλήθεια", + "אמת", + "word", + "λόγος", + "דבר", + ) + + +def _predicate_anchor_candidates(payload: dict) -> tuple[str, ...]: + predicate_type = payload.get("predicate_type", "").lower() + if "existential" in predicate_type: + return ("exist", "is", "was", "ζωή", "חיים") + if "relational" in predicate_type or "prepositional" in predicate_type: + return ("with", "to", "λόγος", "דבר") + if "creation" in predicate_type or "verbal" in predicate_type: + return ("create", "κτίζω", "ברא") + return ("is", "was", "true", "real", "ἀλήθεια", "אמת", "truth") + + +def _first_existing(vocab, candidates: tuple[str, ...]) -> str | None: + for candidate in candidates: + try: + vocab.get_versor(candidate) + except KeyError: + continue + return candidate + return None + + +def _prompt_versor(field_state: FieldState) -> np.ndarray: + return field_state.holonomy if field_state.holonomy is not None else field_state.F + + +def _nearest_content_word( + vocab, + query: np.ndarray, + exclude_indices: frozenset[int], + preferred_pos: frozenset[str] = frozenset(), +) -> tuple[str, int]: + stop_indices = { + vocab.index_of(surface) + for surface in _STOP_SURFACES + if _has_word(vocab, surface) + } + blocked = set(exclude_indices) | stop_indices + if preferred_pos: + selected = _nearest_by_pos(vocab, query, blocked, preferred_pos) + if selected is not None: + return selected + try: + return vocab.nearest(query, exclude_indices=blocked) + except ValueError: + return vocab.nearest(query, exclude_indices=set(exclude_indices)) + + +def _nearest_by_pos( + vocab, + query: np.ndarray, + blocked: set[int], + preferred_pos: frozenset[str], +) -> tuple[str, int] | None: + best_score = -np.inf + best: tuple[str, int] | None = None + for idx in range(len(vocab)): + if idx in blocked: + continue + word = vocab.get_word_at(idx) + morphology_for_word = getattr(vocab, "morphology_for_word", None) + morphology = morphology_for_word(word) if morphology_for_word is not None else None + pos = None if morphology is None else dict(morphology.inflection).get("pos") + if pos not in preferred_pos: + continue + score = cga_inner(query, vocab.get_versor_at(idx)) + if score > best_score: + best_score = score + best = (word, idx) + return best + + +def _has_word(vocab, word: str) -> bool: + try: + vocab.index_of(word) + except KeyError: + return False + return True + + +def _frame_wants_object(frame: PropositionFrame) -> bool: + object_names = {"object", "ground", "locus", "location", "genitive", "created"} + return any(slot.required and slot.name in object_names for slot in frame.slots) + + +def _render_surface( + frame: PropositionFrame, + subject: str, + predicate: str, + object_surface: str | None, +) -> str: + if frame.language == "he" and frame.predicate_type == "copular": + return f"{subject} {predicate}" + if object_surface is not None: + return f"{subject} {predicate} {object_surface}" + if frame.predicate_type.startswith("copular"): + return f"{subject} {predicate}" + return f"{subject} {predicate}" diff --git a/pyproject.toml b/pyproject.toml index 0b22f0db..cb1beec6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,6 +9,7 @@ dependencies = [ "mlx>=0.18; sys_platform == 'darwin'", "pytest>=9.0.3", "pytest-asyncio>=1.3.0", + "ruff>=0.15.12", ] [project.optional-dependencies] diff --git a/tests/test_proposition.py b/tests/test_proposition.py new file mode 100644 index 00000000..5dc4282f --- /dev/null +++ b/tests/test_proposition.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from algebra.cga import cga_inner +from generate.proposition import FrameRegistry, Proposition, propose +from ingest.gate import inject +from language_packs.compiler import load_mounted_packs +from vault.store import VaultStore + + +def test_light_prompt_generates_structured_proposition_near_prompt(): + vocab = load_mounted_packs( + ("en_minimal_v1", "he_logos_micro_v1", "grc_logos_micro_v1") + ) + state = inject(["light", "אוֹר", "φῶς"], vocab) + vault = VaultStore() + random_idx = vault.store(vocab.get_versor("λόγος"), {"kind": "random"}) + registry = FrameRegistry.from_pack("grc", vocab) + + proposition = propose(state, vault, vocab, registry) + + assert isinstance(proposition, Proposition) + assert proposition.subject + assert proposition.predicate + assert proposition.surface + random_entry = vault.recall(vocab.get_versor("λόγος"), top_k=1)[0]["versor"] + prompt = state.F + + assert cga_inner(proposition.subject_versor, prompt) > cga_inner( + proposition.subject_versor, + random_entry, + ) + assert cga_inner(proposition.predicate_versor, prompt) > cga_inner( + proposition.predicate_versor, + random_entry, + ) + stored = vault.recall(state.F, top_k=2) + assert any(hit["metadata"].get("kind") == "proposition" for hit in stored) + assert random_idx == 0