303 lines
11 KiB
Python
303 lines
11 KiB
Python
"""
|
|
Vertical slice: one cognitive pulse from injection to token recall.
|
|
|
|
V4 — coupled forward-correction loop (Threshold 2: Dual-Correction).
|
|
|
|
Two operators run in lockstep each iteration:
|
|
|
|
GraphDiffusionOperator — spreads context pressure across token edges
|
|
ConstraintCorrectionOperator — pulls the output node toward the intent target
|
|
|
|
Both must converge (delta < threshold) before the pulse ends.
|
|
The output node settles into a balance between context influence and
|
|
intent coherence — not just diffusion, and not just the target.
|
|
|
|
Usage:
|
|
python -m scripts.run_pulse "What is truth?"
|
|
python -m scripts.run_pulse --top-k 10 "Compare knowledge and wisdom"
|
|
python -m scripts.run_pulse --no-glove "light"
|
|
python -m scripts.run_pulse --no-correction "grace" # V3 pure-diffusion mode
|
|
python -m scripts.run_pulse --correction-rate 0.1 "the beginning" # soft correction
|
|
|
|
Flags:
|
|
--top-k N Return N nearest vault words (default 5)
|
|
--max-words N Load at most N words from GloVe (default 50000)
|
|
--no-glove Use compiled en_core_cognition_v1 pack (no download)
|
|
--no-correction Disable ConstraintCorrectionOperator (V3 mode)
|
|
--correction-rate R Blend weight toward target per step (default 0.3)
|
|
-v Verbose logging
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
|
|
import numpy as np
|
|
|
|
from algebra.backend import cga_inner
|
|
from algebra.versor import construction_seed_versor
|
|
from field.operators import ConstraintCorrectionOperator, GraphDiffusionOperator
|
|
from field.state import ManifoldState
|
|
from generate.graph_planner import graph_from_intent, ground_graph, plan_articulation
|
|
from generate.intent import classify_intent
|
|
from generate.realizer import realize_semantic
|
|
from sensorium.adapters.text import deterministic_hash_versor
|
|
from vocab.manifold import VocabManifold
|
|
|
|
from dataclasses import dataclass
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
CONVERGENCE_THRESHOLD = 1e-6
|
|
MAX_STEPS = 2000
|
|
TOP_K = 5
|
|
COMPILED_PACK_ID = "en_core_cognition_v1"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PulseResult:
|
|
recalled_words: tuple[str, ...]
|
|
surface: str
|
|
steps: int
|
|
converged: bool
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Manifold loading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _load_manifold(use_glove: bool, max_words: int) -> VocabManifold:
|
|
if use_glove:
|
|
from packs.en_seeder import seed_english_manifold
|
|
log.info("[pulse] Seeding English manifold (max_words=%d) …", max_words)
|
|
manifold = seed_english_manifold(max_words=max_words)
|
|
log.info("[pulse] Manifold ready: %d words", len(manifold))
|
|
return manifold
|
|
|
|
from packs.compiler import load_pack
|
|
_, manifold = load_pack(COMPILED_PACK_ID)
|
|
return manifold
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Token injection and graph construction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _inject_token(token: str, manifold: VocabManifold) -> np.ndarray:
|
|
"""Project one token into Cl(4,1). Manifold lookup first, hash fallback."""
|
|
try:
|
|
return manifold.get_versor(token.lower()).astype(np.float64)
|
|
except KeyError:
|
|
return deterministic_hash_versor(token).astype(np.float64)
|
|
|
|
|
|
def _build_manifold(
|
|
text: str,
|
|
manifold: VocabManifold,
|
|
) -> tuple[ManifoldState, list[str], np.ndarray]:
|
|
"""Build a per-token graph with an input-driven output node.
|
|
|
|
Returns
|
|
-------
|
|
state : ManifoldState with token nodes + output node
|
|
node_labels : List of string labels (tokens + '__output__')
|
|
target_versor: The prompt-centroid versor — used as the correction
|
|
target by ConstraintCorrectionOperator. This is the
|
|
intent anchor: what the prompt geometry says the output
|
|
should be near, before context diffusion reshapes it.
|
|
|
|
Topology:
|
|
- Each input token → one node (versor from manifold or hash fallback)
|
|
- One output node → initialised from centroid of input versors
|
|
- Star edges: every input node → output node
|
|
- Chain edges: sequential input nodes for adjacency pressure
|
|
"""
|
|
tokens = text.strip().lower().split()
|
|
if not tokens:
|
|
tokens = ["__empty__"]
|
|
|
|
token_versors = [_inject_token(t, manifold) for t in tokens]
|
|
|
|
centroid = np.mean(token_versors, axis=0)
|
|
max_abs = float(np.max(np.abs(centroid)))
|
|
if max_abs > 1e-9:
|
|
centroid = centroid * (0.9 / max_abs)
|
|
target_versor = construction_seed_versor(centroid).astype(np.float32)
|
|
|
|
node_labels = list(tokens) + ["__output__"]
|
|
fields = np.stack(
|
|
[np.asarray(v, dtype=np.float32) for v in token_versors]
|
|
+ [target_versor],
|
|
axis=0,
|
|
)
|
|
|
|
output_idx = len(tokens)
|
|
edges: list[list[int]] = []
|
|
for i in range(len(tokens)):
|
|
edges.append([i, output_idx])
|
|
for i in range(len(tokens) - 1):
|
|
edges.append([i, i + 1])
|
|
|
|
edge_array = (
|
|
np.array(edges, dtype=np.int32)
|
|
if edges
|
|
else np.empty((0, 2), dtype=np.int32)
|
|
)
|
|
return ManifoldState(fields=fields, edges=edge_array), node_labels, target_versor
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Recall
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _recall_from_manifold(
|
|
output_versor: np.ndarray,
|
|
manifold: VocabManifold,
|
|
top_k: int,
|
|
) -> list[tuple[str, float]]:
|
|
"""Top-k words from VocabManifold by CGA inner product."""
|
|
exclude: set[int] = set()
|
|
results: list[tuple[str, float]] = []
|
|
for _ in range(top_k):
|
|
try:
|
|
word, idx = manifold.nearest(
|
|
output_versor, exclude_indices=frozenset(exclude),
|
|
)
|
|
except ValueError:
|
|
break
|
|
score = float(cga_inner(output_versor, manifold.get_versor_at(idx)))
|
|
exclude.add(idx)
|
|
results.append((word, score))
|
|
return results
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pulse loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def run_pulse(
|
|
text: str,
|
|
*,
|
|
top_k: int = TOP_K,
|
|
max_words: int = 50_000,
|
|
use_glove: bool = True,
|
|
use_correction: bool = True,
|
|
correction_rate: float = 0.3,
|
|
) -> PulseResult:
|
|
"""Execute one cognitive pulse and return recalled words + realized surface.
|
|
|
|
Parameters
|
|
----------
|
|
use_correction : Enable ConstraintCorrectionOperator (default True).
|
|
Set False to reproduce V3 pure-diffusion behaviour.
|
|
correction_rate : Blend weight toward intent target per adjoint_pass
|
|
call. Lower = softer correction, more steps.
|
|
"""
|
|
manifold = _load_manifold(use_glove, max_words)
|
|
state, node_labels, target_versor = _build_manifold(text, manifold)
|
|
|
|
diffusion_op = GraphDiffusionOperator(damping=0.5)
|
|
correction_op = ConstraintCorrectionOperator(
|
|
target_versor=target_versor,
|
|
correction_rate=correction_rate,
|
|
node_index=-1,
|
|
) if use_correction else None
|
|
|
|
n_input = len(node_labels) - 1
|
|
print(f"[pulse] input : {text!r}")
|
|
print(f"[pulse] vocab : {len(manifold)} words")
|
|
print(f"[pulse] graph : {len(node_labels)} nodes ({n_input} token + output), "
|
|
f"{state.edges.shape[0]} edges")
|
|
print(f"[pulse] correction : {'enabled (rate=%.2f)' % correction_rate if use_correction else 'disabled (V3 mode)'}")
|
|
|
|
step = 0
|
|
delta_fwd = float("inf")
|
|
delta_corr = float("inf") if use_correction else 0.0
|
|
converged = False
|
|
|
|
while step < MAX_STEPS:
|
|
# --- Forward pass (diffusion) ---
|
|
state, delta_fwd = diffusion_op.forward(state)
|
|
step = state.step
|
|
|
|
# --- Adjoint pass (correction) ---
|
|
if correction_op is not None:
|
|
state, delta_corr = correction_op.adjoint_pass(state)
|
|
|
|
if step <= 5 or step % 50 == 0:
|
|
if use_correction:
|
|
print(f"[pulse] step {step:4d} Δ_fwd={delta_fwd:.2e} Δ_corr={delta_corr:.2e}")
|
|
else:
|
|
print(f"[pulse] step {step:4d} delta={delta_fwd:.2e}")
|
|
|
|
if delta_fwd < CONVERGENCE_THRESHOLD and delta_corr < CONVERGENCE_THRESHOLD:
|
|
converged = True
|
|
print(f"[pulse] converged at step {step} "
|
|
f"(Δ_fwd={delta_fwd:.2e}, Δ_corr={delta_corr:.2e})")
|
|
break
|
|
else:
|
|
print(f"[pulse] WARNING: max_steps ({MAX_STEPS}) reached — "
|
|
f"Δ_fwd={delta_fwd:.2e} Δ_corr={delta_corr:.2e}")
|
|
|
|
output_idx = len(node_labels) - 1
|
|
output_versor = state.fields[output_idx]
|
|
results = _recall_from_manifold(output_versor, manifold, top_k)
|
|
recalled_words = tuple(w for w, _ in results)
|
|
|
|
print(f"[pulse] output -> top-{top_k} recall:")
|
|
for rank, (word, score) in enumerate(results, 1):
|
|
marker = " <-" if word in [t.lower() for t in node_labels[:-1]] else ""
|
|
print(f"[pulse] {rank}. {word!r:20s} score={score:+.6f}{marker}")
|
|
|
|
# --- Surface realizer join ---
|
|
intent = classify_intent(text)
|
|
graph = graph_from_intent(intent)
|
|
grounded = ground_graph(graph, recalled_words)
|
|
target = plan_articulation(grounded)
|
|
plan = realize_semantic(target, grounded)
|
|
surface = plan.surface
|
|
|
|
print(f"[pulse] surface : {surface}")
|
|
|
|
return PulseResult(
|
|
recalled_words=recalled_words,
|
|
surface=surface,
|
|
steps=step,
|
|
converged=converged,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _parse_args() -> argparse.Namespace:
|
|
p = argparse.ArgumentParser(description="CORE cognitive pulse (V4 — dual correction)")
|
|
p.add_argument("text", nargs="*", default=["What is truth?"])
|
|
p.add_argument("--top-k", type=int, default=5, metavar="N")
|
|
p.add_argument("--max-words", type=int, default=50_000, metavar="N")
|
|
p.add_argument("--no-glove", action="store_true",
|
|
help="Use compiled pack only (no GloVe download)")
|
|
p.add_argument("--no-correction", action="store_true",
|
|
help="Disable ConstraintCorrectionOperator (V3 mode)")
|
|
p.add_argument("--correction-rate", type=float, default=0.3, metavar="R")
|
|
p.add_argument("-v", "--verbose", action="store_true")
|
|
return p.parse_args()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = _parse_args()
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s",
|
|
)
|
|
input_text = " ".join(args.text)
|
|
run_pulse(
|
|
input_text,
|
|
top_k=args.top_k,
|
|
max_words=args.max_words,
|
|
use_glove=not args.no_glove,
|
|
use_correction=not args.no_correction,
|
|
correction_rate=args.correction_rate,
|
|
)
|