diff --git a/applications/papers/__init__.py b/applications/papers/__init__.py deleted file mode 100644 index 32da08d..0000000 --- a/applications/papers/__init__.py +++ /dev/null @@ -1,19 +0,0 @@ -from applications.papers import config as paper_config -from applications.papers.entities import DocumentChunk, Paper -from applications.papers.interfaces import TextExtractor, ChunkingStrategy, PaperRepository -from applications.papers.services.extractors import PdfTextExtractor, TxtTextExtractor, get_extractor -from applications.papers.services.chunker import OverlapChunker -from applications.papers.services.storage import JsonPaperRepository -from applications.papers.services.ingestion import PaperIngestionService -from applications.papers.services.context import PaperContextService - -__all__ = [ - "paper_config", - "DocumentChunk", "Paper", - "TextExtractor", "ChunkingStrategy", "PaperRepository", - "PdfTextExtractor", "TxtTextExtractor", "get_extractor", - "OverlapChunker", - "JsonPaperRepository", - "PaperIngestionService", - "PaperContextService", -] diff --git a/applications/papers/config.py b/applications/papers/config.py deleted file mode 100644 index bf9f785..0000000 --- a/applications/papers/config.py +++ /dev/null @@ -1,5 +0,0 @@ -PAPER_CHUNK_SIZE = 2000 -PAPER_CHUNK_OVERLAP = 200 -PAPER_MAX_CHARS = 100_000 -PAPER_STORAGE_PATH = ".papers" -PAPER_CONTEXT_MAX_CHARS = 4000 diff --git a/applications/papers/entities.py b/applications/papers/entities.py deleted file mode 100644 index 6e4d10e..0000000 --- a/applications/papers/entities.py +++ /dev/null @@ -1,38 +0,0 @@ -from dataclasses import dataclass, field -from datetime import datetime -from typing import List, Optional - - -@dataclass -class DocumentChunk: - """A contiguous segment of text extracted from a source document.""" - - index: int - text: str - source_path: str - char_start: int - char_end: int - - -@dataclass -class Paper: - """Represents an ingested document with its metadata and content chunks.""" - - source_path: str - title: str - total_chars: int - chunks: List[DocumentChunk] = field(default_factory=list) - ingested_at: Optional[str] = None - paper_id: str = "" - - def __post_init__(self): - """Auto-generate a short paper ID from source path, char count, and timestamp.""" - if not self.paper_id: - import hashlib - raw = f"{self.source_path}:{self.total_chars}:{self.ingested_at or datetime.now().isoformat()}" - self.paper_id = hashlib.sha256(raw.encode()).hexdigest()[:12] - - @property - def full_text(self) -> str: - """Return the concatenated text of all chunks in order.""" - return "".join(c.text for c in sorted(self.chunks, key=lambda x: x.index)) diff --git a/applications/papers/interfaces.py b/applications/papers/interfaces.py deleted file mode 100644 index 714f32a..0000000 --- a/applications/papers/interfaces.py +++ /dev/null @@ -1,44 +0,0 @@ -from abc import ABC, abstractmethod -from typing import List, Optional - -from applications.papers.entities import DocumentChunk, Paper - - -class TextExtractor(ABC): - """Interface for extracting plain text from a file at a given path.""" - - @abstractmethod - def extract(self, path: str) -> str: - ... - - -class ChunkingStrategy(ABC): - """Interface for splitting text into smaller document chunks.""" - - @abstractmethod - def chunk(self, text: str, source_path: str) -> List[DocumentChunk]: - ... - - -class PaperRepository(ABC): - """Interface for persisting and retrieving Paper objects.""" - - @abstractmethod - def save(self, paper: Paper) -> str: - """Persist a paper and return its ID.""" - ... - - @abstractmethod - def load(self, paper_id: str) -> Optional[Paper]: - """Retrieve a paper by ID, or None if not found.""" - ... - - @abstractmethod - def list_papers(self) -> List[Paper]: - """Return all stored papers.""" - ... - - @abstractmethod - def delete(self, paper_id: str) -> bool: - """Remove a paper by ID. Returns True if it existed.""" - ... diff --git a/applications/papers/services/__init__.py b/applications/papers/services/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/applications/papers/services/chunker.py b/applications/papers/services/chunker.py deleted file mode 100644 index bfc0615..0000000 --- a/applications/papers/services/chunker.py +++ /dev/null @@ -1,38 +0,0 @@ -from typing import List - -from applications.papers.config import PAPER_CHUNK_SIZE, PAPER_CHUNK_OVERLAP, PAPER_MAX_CHARS -from applications.papers.entities import DocumentChunk -from applications.papers.interfaces import ChunkingStrategy - - -class OverlapChunker(ChunkingStrategy): - """Splits text into overlapping chunks of a fixed size.""" - - def __init__(self, chunk_size: int = PAPER_CHUNK_SIZE, overlap: int = PAPER_CHUNK_OVERLAP): - if overlap >= chunk_size: - raise ValueError("overlap must be smaller than chunk_size") - self.chunk_size = chunk_size - self.overlap = overlap - - def chunk(self, text: str, source_path: str) -> List[DocumentChunk]: - """Chunk text into overlapping segments, each linked to the source path.""" - text = text[:PAPER_MAX_CHARS] - chunks: list[DocumentChunk] = [] - start = 0 - index = 0 - while start < len(text): - end = min(start + self.chunk_size, len(text)) - chunks.append( - DocumentChunk( - index=index, - text=text[start:end], - source_path=source_path, - char_start=start, - char_end=end, - ) - ) - index += 1 - if end == len(text): - break - start += self.chunk_size - self.overlap - return chunks diff --git a/applications/papers/services/context.py b/applications/papers/services/context.py deleted file mode 100644 index 5257305..0000000 --- a/applications/papers/services/context.py +++ /dev/null @@ -1,38 +0,0 @@ -from typing import List, Optional - -from applications.papers.entities import Paper -from applications.papers.interfaces import PaperRepository -from applications.papers.services.storage import JsonPaperRepository - - -class PaperContextService: - """Provides query access to stored papers via a repository.""" - - def __init__(self, repository: Optional[PaperRepository] = None): - self.repository = repository or JsonPaperRepository() - - def list_papers(self) -> List[Paper]: - """Return all stored papers.""" - return self.repository.list_papers() - - def get_paper(self, paper_id: str) -> Optional[Paper]: - """Retrieve a single paper by its ID, or None if not found.""" - return self.repository.load(paper_id) - - def get_latest_paper(self) -> Optional[Paper]: - """Return the most recently ingested paper, or None if none exist.""" - papers = self.repository.list_papers() - if not papers: - return None - papers.sort(key=lambda p: p.ingested_at or "", reverse=True) - return papers[0] - - def get_recent_papers(self, limit: int = 3) -> List[Paper]: - """Return the *limit* most recently ingested papers.""" - papers = self.repository.list_papers() - papers.sort(key=lambda p: p.ingested_at or "", reverse=True) - return papers[:limit] - - def has_papers(self) -> bool: - """Check whether at least one paper exists in the repository.""" - return len(self.repository.list_papers()) > 0 diff --git a/applications/papers/services/extractors.py b/applications/papers/services/extractors.py deleted file mode 100644 index 8fc3dbb..0000000 --- a/applications/papers/services/extractors.py +++ /dev/null @@ -1,40 +0,0 @@ -import subprocess -from applications.papers.interfaces import TextExtractor - - -class PdfTextExtractor(TextExtractor): - """Extract text from PDF files using pdftotext.""" - - def extract(self, path: str) -> str: - result = subprocess.run( - ["pdftotext", path, "-"], - capture_output=True, - text=True, - ) - if result.returncode != 0: - raise RuntimeError(f"pdftotext failed: {result.stderr}") - return result.stdout - - -class TxtTextExtractor(TextExtractor): - """Extract text from plain-text files by reading the file directly.""" - - def extract(self, path: str) -> str: - with open(path) as f: - return f.read() - - -_EXTENSIONS = { - ".pdf": PdfTextExtractor, - ".txt": TxtTextExtractor, -} - - -def get_extractor(path: str) -> TextExtractor: - """Return the appropriate TextExtractor for the file extension of *path*.""" - import os - ext = os.path.splitext(path)[1].lower() - cls = _EXTENSIONS.get(ext) - if cls is None: - raise ValueError(f"Unsupported file extension: {ext} (supported: {list(_EXTENSIONS)})") - return cls() diff --git a/applications/papers/services/ingestion.py b/applications/papers/services/ingestion.py deleted file mode 100644 index 92b91b0..0000000 --- a/applications/papers/services/ingestion.py +++ /dev/null @@ -1,66 +0,0 @@ -import os -from datetime import datetime -from typing import Optional - -from applications.papers.config import PAPER_CONTEXT_MAX_CHARS -from applications.papers.entities import Paper -from applications.papers.interfaces import PaperRepository, TextExtractor, ChunkingStrategy -from applications.papers.services.extractors import get_extractor -from applications.papers.services.chunker import OverlapChunker -from applications.papers.services.storage import JsonPaperRepository - - -class PaperIngestionService: - """Orchestrates paper ingestion: extraction, chunking, and persistence.""" - - def __init__( - self, - extractor: Optional[TextExtractor] = None, - chunker: Optional[ChunkingStrategy] = None, - repository: Optional[PaperRepository] = None, - ): - self.extractor = extractor - self.chunker = chunker or OverlapChunker() - self.repository = repository or JsonPaperRepository() - - def _get_extractor(self, path: str) -> TextExtractor: - """Return the configured extractor, or auto-detect one for *path*.""" - return self.extractor or get_extractor(path) - - def ingest(self, path: str) -> Paper: - path = os.path.abspath(path) - if not os.path.exists(path): - raise FileNotFoundError(path) - - extractor = self._get_extractor(path) - text = extractor.extract(path) - chunks = self.chunker.chunk(text, path) - - paper = Paper( - source_path=path, - title=os.path.splitext(os.path.basename(path))[0], - total_chars=len(text), - chunks=chunks, - ingested_at=datetime.now().isoformat(), - ) - - self.repository.save(paper) - return paper - - def build_context(self, paper: Paper, max_chars: int = PAPER_CONTEXT_MAX_CHARS) -> str: - """Concatenate paper chunks up to *max_chars*, appending a truncation notice if needed.""" - accumulated = [] - total = 0 - for chunk in sorted(paper.chunks, key=lambda c: c.index): - if total + len(chunk.text) > max_chars: - remaining = max_chars - total - if remaining > 0: - accumulated.append(chunk.text[:remaining]) - break - accumulated.append(chunk.text) - total += len(chunk.text) - - full = "".join(accumulated) - if len(full) < paper.total_chars: - full += "\n... [content truncated]" - return full diff --git a/applications/papers/services/storage.py b/applications/papers/services/storage.py deleted file mode 100644 index 4729d08..0000000 --- a/applications/papers/services/storage.py +++ /dev/null @@ -1,115 +0,0 @@ -import json -import os -from typing import List, Optional - -from applications.papers.config import PAPER_STORAGE_PATH -from applications.papers.entities import DocumentChunk, Paper -from applications.papers.interfaces import PaperRepository - - -def _paper_path(store_dir: str, paper_id: str) -> str: - """Return the filesystem path for a paper's JSON file.""" - return os.path.join(store_dir, f"{paper_id}.json") - - -def _paper_to_dict(p: Paper) -> dict: - """Serialize a Paper to a JSON-compatible dictionary.""" - return { - "paper_id": p.paper_id, - "source_path": p.source_path, - "title": p.title, - "total_chars": p.total_chars, - "ingested_at": p.ingested_at, - "chunks": [ - { - "index": c.index, - "text": c.text, - "source_path": c.source_path, - "char_start": c.char_start, - "char_end": c.char_end, - } - for c in p.chunks - ], - } - - -def _dict_to_paper(d: dict) -> Paper: - """Deserialize a dictionary back into a Paper object.""" - p = Paper( - source_path=d["source_path"], - title=d["title"], - total_chars=d["total_chars"], - ingested_at=d.get("ingested_at"), - paper_id=d["paper_id"], - ) - p.chunks = [ - DocumentChunk( - index=c["index"], - text=c["text"], - source_path=c["source_path"], - char_start=c["char_start"], - char_end=c["char_end"], - ) - for c in d.get("chunks", []) - ] - return p - - -class JsonPaperRepository(PaperRepository): - """Stores papers as individual JSON files on disk.""" - - def __init__(self, store_dir: str = PAPER_STORAGE_PATH): - self._dir = store_dir - os.makedirs(self._dir, exist_ok=True) - - def _paper_path(self, paper_id: str) -> str: - """Absolute path to the JSON file for *paper_id*.""" - return _paper_path(self._dir, paper_id) - - def _all_paper_ids(self) -> List[str]: - """Return all paper IDs found in the store directory.""" - if not os.path.isdir(self._dir): - return [] - return sorted( - f.removesuffix(".json") - for f in os.listdir(self._dir) - if f.endswith(".json") - ) - - def save(self, paper: Paper) -> str: - """Persist *paper* as a JSON file and return its ID.""" - path = self._paper_path(paper.paper_id) - with open(path, "w") as f: - json.dump(_paper_to_dict(paper), f, indent=2) - return paper.paper_id - - def load(self, paper_id: str) -> Optional[Paper]: - """Load a paper by ID, or return None if missing or corrupt.""" - path = self._paper_path(paper_id) - if not os.path.exists(path): - return None - try: - with open(path) as f: - data = json.load(f) - if isinstance(data, list): - return None - return _dict_to_paper(data) - except (json.JSONDecodeError, KeyError, IOError): - return None - - def list_papers(self) -> List[Paper]: - """Return every valid paper in the store.""" - papers = [] - for pid in self._all_paper_ids(): - p = self.load(pid) - if p is not None: - papers.append(p) - return papers - - def delete(self, paper_id: str) -> bool: - """Delete the paper file for *paper_id*. Returns True if it existed.""" - path = self._paper_path(paper_id) - if not os.path.exists(path): - return False - os.remove(path) - return True diff --git a/applications/research_assistant.py b/applications/research_assistant.py deleted file mode 100644 index 36099c5..0000000 --- a/applications/research_assistant.py +++ /dev/null @@ -1,79 +0,0 @@ -import os -import sys - -if __package__ in (None, ""): - repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) - if repo_root not in sys.path: - sys.path.insert(0, repo_root) - -from core.engine import MetaMoEngine, AssistantResponse, format_response -from applications.papers import PaperContextService, PaperIngestionService - - -def _chat_loop(engine: MetaMoEngine, paper_context: str | None): - """Run the REPL chat loop, optionally augmenting user input with paper context.""" - print("\nSystem Ready. Subsystems: [Curiosity] & [Ethics]. Type 'quit' to exit.") - print("-" * 60) - - while True: - try: - user_input = input("\nYou: ") - if user_input.lower() in ('quit', 'exit'): - break - - print("\n[MetaMo Internal Processing...]") - - if paper_context: - response = engine.process_with_context(user_input, paper_context) - else: - response = engine.process(user_input) - - print(format_response(response)) - - except Exception as e: - print(f"\nAn error occurred: {e}") - - -def interactive_loop(): - """Start the chat interface, auto-loading the latest stored paper as context.""" - print("Initializing MetaMo Multi-Subsystem Chat Interface...") - engine = MetaMoEngine() - context_svc = PaperContextService() - - paper_context: str | None = None - if context_svc.has_papers(): - latest = context_svc.get_latest_paper() - print(f" Found stored paper: {latest.title}") - paper_context = PaperIngestionService().build_context(latest) - print(f" Loaded {len(paper_context)} chars of context.") - - _chat_loop(engine, paper_context) - - -def load_and_chat(): - """Ingest a paper file from the CLI argument and start a context-aware chat.""" - path = sys.argv[1] - print(f"Loading paper from {path}...") - - ingest_svc = PaperIngestionService() - paper = ingest_svc.ingest(path) - print(f"Ingested: {paper.title} ({paper.total_chars} chars, {len(paper.chunks)} chunks)") - - print("Initializing MetaMo Multi-Subsystem Chat Interface...") - engine = MetaMoEngine() - paper_context = ingest_svc.build_context( - PaperContextService().get_latest_paper() - ) - - print("\n" + "=" * 60) - print(f"Paper '{paper.title}' ingested. You can now ask questions.") - print("=" * 60) - - _chat_loop(engine, paper_context) - - -if __name__ == "__main__": - if len(sys.argv) > 1: - load_and_chat() - else: - interactive_loop() diff --git a/category/bimonad.py b/category/bimonad.py index 7f655e6..b5db28e 100644 --- a/category/bimonad.py +++ b/category/bimonad.py @@ -1,79 +1,160 @@ -from typing import List, Tuple +from typing import Any, List, Tuple import numpy as np -# Assuming these are available in your python path -from core.state import MotivationalState, Stimulus, Action +from dataclasses import dataclass +from core.state import MotivationalState, Action from core.config import ( - G_ETHIC, LAX_DISTRIBUTIVE_DELTA, - G_IND, - G_HELP, - G_NOVEL, - G_SELF, - G_SOC, - G_TRANS, - G_CURIO, - M_APPROACH, - M_AROUSAL, - M_RESOLUTION, - M_SECURING, - M_THRESHOLD, - M_VALENCE, + PARALLEL_COMPOSITION_DELTA, ) +from category.diagnostics import ( + MetaMoDiagnostics, + MetaMoDiagnosticsHistory, + MetaMoDiagnosticsSummary, +) from category.functors import AppraisalComonad, DecisionMonad +from category.laws import StateLawCheckResult +from category.merge import DefaultParallelMergePolicy +from dynamics.coherence import DefaultCoherencePolicy from dynamics.stability import ( - apply_homeostatic_damping, - check_contractive_update_law, - is_in_safe_region, - project_to_safe_region, - raise_boundary_caution, + DefaultStabilityPolicy, ) + +@dataclass(frozen=True) +class TransitionComputation: + """ + Raw transition plus projection correction telemetry. + """ + + action: Action + state: MotivationalState + projection_delta: float + + class MetaMoPseudoBimonad: """ - Represents the composite appraisal-then-decision operator F = D \circ \Psi. - This forms a pseudo-bimonad on the motivational state space X = G \times M[cite: 28, 309]. + Represents the composite appraisal-then-decision operator F = D o ψ. + This forms a pseudo-bimonad on the motivational state space X = G \times M. """ - def __init__(self, appraisal: AppraisalComonad, decision: DecisionMonad): + def __init__( + self, + appraisal: AppraisalComonad, + decision: DecisionMonad, + stability_policy=None, + coherence_policy=None, + merge_policy=None, + ): self.appraisal = appraisal self.decision = decision + self.stability_policy = stability_policy or DefaultStabilityPolicy() + self.coherence_policy = coherence_policy or DefaultCoherencePolicy() + self.merge_policy = merge_policy or DefaultParallelMergePolicy() + self.diagnostics_history = MetaMoDiagnosticsHistory() - def _compute_transition(self, state: MotivationalState, stimulus: Stimulus, candidates: List[Action]) -> Tuple[Action, MotivationalState]: + def _compute_transition_details(self, state: MotivationalState, stimulus: Any, candidates: List[Action]) -> TransitionComputation: """ Compute one appraisal/decision transition before runtime validation. """ - # 1. Appraise (\Psi) - Update modulators based on stimulus[cite: 314]. + # 1. Appraise - Update modulators based on stimulus. + goal_change_feedback = self._goal_change_feedback(state, stimulus) appraised_state = self.appraisal.appraise(state, stimulus) - appraised_state = raise_boundary_caution(appraised_state) - - # 2. Decide (\mathbb{D}) - Score candidates and update goals[cite: 315]. - chosen_action, proposed_delta_g = self.decision.decide(appraised_state, candidates) + appraised_state = self.stability_policy.raise_boundary_caution(appraised_state) + + # 2. Decide - Score candidates and update goals. + chosen_action, proposed_delta_g = self.decision.decide( + appraised_state, + candidates, + feedback=goal_change_feedback, + ) - damped_delta_g = apply_homeostatic_damping(appraised_state, proposed_delta_g) + damped_delta_g = self.stability_policy.apply_homeostatic_damping(appraised_state, proposed_delta_g) next_state = MotivationalState( G=np.clip(appraised_state.G + damped_delta_g, 0.0, 1.0), M=appraised_state.M.copy(), + schema=appraised_state.schema, ) - next_state = project_to_safe_region(next_state) - - return chosen_action, next_state + projected_state = self.stability_policy.project_to_safe_region(next_state) + + return TransitionComputation( + action=chosen_action, + state=projected_state, + projection_delta=next_state.distance_to(projected_state), + ) + + def _compute_transition(self, state: MotivationalState, stimulus: Any, candidates: List[Action]) -> Tuple[Action, MotivationalState]: + """ + Compute one appraisal/decision transition before runtime validation. + """ + computation = self._compute_transition_details(state, stimulus, candidates) + return computation.action, computation.state + + def _target_transition_details( + self, + state: MotivationalState, + stimulus: Any, + candidates: List[Action], + ) -> TransitionComputation: + computation = self._compute_transition_details(state, stimulus, candidates) + next_state = computation.state + projection_delta = computation.projection_delta + reference_state = self._local_reference_state(state, next_state) + + if not self.check_lax_distributive_law(state, stimulus, candidates): + fallback_state = self._apply_conservative_fallback(state, next_state) + projection_delta += next_state.distance_to(fallback_state) + next_state = fallback_state + + if not self.stability_policy.check_contractive_update_law(self, state, reference_state, stimulus, candidates): + fallback_state = self._apply_conservative_fallback(state, next_state) + projection_delta += next_state.distance_to(fallback_state) + next_state = fallback_state + + if not self.stability_policy.is_in_safe_region(next_state): + fallback_state = self._apply_conservative_fallback(state, next_state) + projection_delta += next_state.distance_to(fallback_state) + next_state = fallback_state + + return TransitionComputation( + action=computation.action, + state=next_state, + projection_delta=projection_delta, + ) + + def target_transition( + self, + state: MotivationalState, + stimulus: Any, + candidates: List[Action], + ) -> Tuple[Action, MotivationalState]: + """ + Compute a stabilized target state before incremental embodiment. + """ + computation = self._target_transition_details(state, stimulus, candidates) + return computation.action, computation.state def _state_from_delta(self, decision_state: MotivationalState, proposed_delta_g: np.ndarray) -> MotivationalState: """ Apply a proposed goal update inside the same stabilization path used by the main transition. """ - damped_delta_g = apply_homeostatic_damping(decision_state, proposed_delta_g) + damped_delta_g = self.stability_policy.apply_homeostatic_damping(decision_state, proposed_delta_g) next_state = MotivationalState( G=np.clip(decision_state.G + damped_delta_g, 0.0, 1.0), M=decision_state.M.copy(), + schema=decision_state.schema, ) - return project_to_safe_region(next_state) + return self.stability_policy.project_to_safe_region(next_state) - def _decision_context(self, state: MotivationalState, stimulus: Stimulus) -> MotivationalState: + def _decision_context(self, state: MotivationalState, stimulus: Any) -> MotivationalState: """ Build the post-appraisal state that the decision monad should score. """ appraised_state = self.appraisal.appraise(state, stimulus) - return raise_boundary_caution(appraised_state) + return self.stability_policy.raise_boundary_caution(appraised_state) + + def _goal_change_feedback(self, state: MotivationalState, stimulus: Any) -> Any: + if hasattr(self.appraisal, "goal_change_feedback"): + return self.appraisal.goal_change_feedback(state, stimulus) + return None def _local_reference_state(self, state: MotivationalState, next_state: MotivationalState) -> MotivationalState: """ @@ -88,13 +169,14 @@ def _local_reference_state(self, state: MotivationalState, next_state: Motivatio return MotivationalState( G=np.clip(state.G + probe_G, 0.0, 1.0), M=np.clip(state.M + probe_M, 0.0, 1.0), + schema=state.schema, ) def consensus_action( self, state_a: MotivationalState, state_b: MotivationalState, - stimulus: Stimulus, + stimulus: Any, candidates: List[Action], ) -> Action: """ @@ -126,7 +208,7 @@ def consensus_transition( self, state_a: MotivationalState, state_b: MotivationalState, - stimulus: Stimulus, + stimulus: Any, candidates: List[Action], ) -> Tuple[Action, MotivationalState]: """ @@ -135,11 +217,28 @@ def consensus_transition( action = self.consensus_action(state_a, state_b, stimulus, candidates) context_a = self._decision_context(state_a, stimulus) context_b = self._decision_context(state_b, stimulus) - target_a = self._state_from_delta(context_a, action.delta_g) - target_b = self._state_from_delta(context_b, action.delta_g) + feedback_a = self._goal_change_feedback(state_a, stimulus) + feedback_b = self._goal_change_feedback(state_b, stimulus) + delta_a = self._proposed_delta_for_action(context_a, action, feedback_a) + delta_b = self._proposed_delta_for_action(context_b, action, feedback_b) + target_a = self._state_from_delta(context_a, delta_a) + target_b = self._state_from_delta(context_b, delta_b) merged_target = self.parallel_merge(target_a, target_b) return action, merged_target + def _proposed_delta_for_action( + self, + state: MotivationalState, + action: Action, + feedback: Any = None, + ) -> np.ndarray: + """ + Use profile-derived MAGUS Delta G when the decision layer exposes it. + """ + if hasattr(self.decision, "propose_delta_g"): + return self.decision.propose_delta_g(state, action, feedback) + return action.delta_g.copy() + def _apply_conservative_fallback(self, current_state: MotivationalState, next_state: MotivationalState) -> MotivationalState: """ Shrink the transition toward the current state when runtime checks fail. @@ -147,110 +246,192 @@ def _apply_conservative_fallback(self, current_state: MotivationalState, next_st fallback_state = MotivationalState( G=((current_state.G * 0.5) + (next_state.G * 0.5)), M=((current_state.M * 0.5) + (next_state.M * 0.5)), + schema=current_state.schema, ) - return project_to_safe_region(fallback_state) + return self.stability_policy.project_to_safe_region(fallback_state) def step( self, state: MotivationalState, - stimulus: Stimulus, + stimulus: Any, candidates: List[Action], + embody: bool = True, + record_diagnostics: bool = True, ) -> Tuple[Action, MotivationalState]: """ - Executes one full cycle of F = D \circ \Psi. - This governs the motivational coalgebra \alpha: X \to F(X)[cite: 310, 316]. + Executes one full cycle of F = D(ψ(X)). + + By default this returns the incrementally embodied next state required + by Principle 5. Set embody=False to inspect the stabilized target. """ - chosen_action, next_state = self._compute_transition(state, stimulus, candidates) - reference_state = self._local_reference_state(state, next_state) - if not self.check_lax_distributive_law(state, stimulus, candidates): - next_state = self._apply_conservative_fallback(state, next_state) - if not check_contractive_update_law(self, state, reference_state, stimulus, candidates): - next_state = self._apply_conservative_fallback(state, next_state) - if not is_in_safe_region(next_state): - next_state = self._apply_conservative_fallback(state, next_state) + chosen_action, next_state, _ = self.step_with_diagnostics( + state, + stimulus, + candidates, + embody=embody, + record_diagnostics=record_diagnostics, + ) return chosen_action, next_state - def check_lax_distributive_law(self, state: MotivationalState, stimulus: Stimulus, candidates: List[Action]) -> bool: + def step_with_diagnostics( + self, + state: MotivationalState, + stimulus: Any, + candidates: List[Action], + embody: bool = True, + record_diagnostics: bool = True, + ) -> Tuple[Action, MotivationalState, MetaMoDiagnostics]: """ - Validates the First Principle: Modular Appraisal-Decision Interface[cite: 287, 322]. - Checks that \lambda_X : \Psi(\mathbb{D}(X)) \Rightarrow \mathbb{D}(\Psi(X)) commutes up to a controlled error[cite: 308]. + Executes one full cycle and returns telemetry for the principle checks. """ + lax_result = self.measure_lax_distributive_law(state, stimulus, candidates) + target_computation = self._target_transition_details(state, stimulus, candidates) + chosen_action = target_computation.action + target_state = target_computation.state + reference_state = self._local_reference_state(state, target_state) + contractive_holds = self.stability_policy.check_contractive_update_law( + self, + state, + reference_state, + stimulus, + candidates, + ) + + if embody: + blend_result = self.coherence_policy.measure_blend(state, target_state) + next_state = blend_result.state + drift = blend_result.drift + blend_alpha = blend_result.alpha + base_blend_alpha = blend_result.base_alpha + else: + next_state = target_state + drift = self.coherence_policy.measure_self_model_drift(state, next_state) + blend_alpha = 1.0 + base_blend_alpha = 1.0 + + diagnostics = MetaMoDiagnostics( + action_id=chosen_action.id, + lax_error=lax_result.error, + lax_tolerance=lax_result.tolerance, + lax_holds=lax_result.holds, + contractive_holds=contractive_holds, + target_in_safe_region=self.stability_policy.is_in_safe_region(target_state), + final_in_safe_region=self.stability_policy.is_in_safe_region(next_state), + boundary_pressure_before=self.stability_policy.boundary_pressure(state), + boundary_pressure_target=self.stability_policy.boundary_pressure(target_state), + boundary_pressure_final=self.stability_policy.boundary_pressure(next_state), + projection_delta=target_computation.projection_delta, + target_distance=state.distance_to(target_state), + state_drift=drift.state_distance, + self_model_drift=drift.self_model_distance, + combined_self_model_drift=drift.combined_drift, + self_model_drift_tolerance=drift.max_allowed_drift, + self_model_drift_holds=drift.holds, + blend_alpha=blend_alpha, + base_blend_alpha=base_blend_alpha, + ) + + if record_diagnostics: + self.diagnostics_history.append(diagnostics) + return chosen_action, next_state, diagnostics + + def measure_lax_distributive_law( + self, + state: MotivationalState, + stimulus: Any, + candidates: List[Action], + tolerance: float = LAX_DISTRIBUTIVE_DELTA, + ) -> StateLawCheckResult: + """ + Measures the First Principle: Modular Appraisal-Decision Interface. + """ + goal_change_feedback = self._goal_change_feedback(state, stimulus) + # Path 1: Appraise then Decide -> stabilized D(Psi(X)) decision_state_1 = self._decision_context(state, stimulus) - action_1, delta_g_1 = self.decision.decide(decision_state_1, candidates) + action_1, delta_g_1 = self.decision.decide( + decision_state_1, + candidates, + feedback=goal_change_feedback, + ) final_state_1 = self._state_from_delta(decision_state_1, delta_g_1) - + # Path 2: Decide then Appraise -> stabilized Psi(D(X)) - action_2, delta_g_2 = self.decision.decide(state, candidates) + action_2, delta_g_2 = self.decision.decide( + state, + candidates, + feedback=goal_change_feedback, + ) decided_state_2 = self._state_from_delta(state, delta_g_2) final_state_2 = self._decision_context(decided_state_2, stimulus) - - # Calculate the controlled distortion distance[cite: 332, 344]. - distortion = final_state_1.distance_to(final_state_2) - - # The law holds if the distortion is bounded by the acceptable delta[cite: 344]. - return distortion <= LAX_DISTRIBUTIVE_DELTA - - def parallel_merge(self, state_a: MotivationalState, state_b: MotivationalState, coherence_correction: float = 0.05) -> MotivationalState: - """ - Implements Principle 3: Parallel Motivational Compositionality. - Witnesses the lax-monoidal structure \phi_{X,Y} of the composite F. - Merges two parallel motivational subsystems with dimension-wise coherence corrections. - Safety-relevant disagreements are merged conservatively, while exploratory disagreements - are damped unless both subsystems support them. - """ - weight_a = state_a.G[G_IND] - weight_b = state_b.G[G_IND] - total_weight = weight_a + weight_b + 1e-9 - - base_G = ((state_a.G * weight_a) + (state_b.G * weight_b)) / total_weight - base_M = ((state_a.M * weight_a) + (state_b.M * weight_b)) / total_weight - - disagreement_G = np.abs(state_a.G - state_b.G) - disagreement_M = np.abs(state_a.M - state_b.M) - - consensus_G = base_G.copy() - consensus_M = base_M.copy() - - # Safety-critical dimensions preserve the stronger caution/ethics signal under disagreement. - safety_goal_idx = np.array([G_IND, G_HELP, G_ETHIC]) - consensus_G[safety_goal_idx] = np.maximum(state_a.G[safety_goal_idx], state_b.G[safety_goal_idx]) - - # Exploratory dimensions require stronger agreement; otherwise they are damped toward the shared floor. - exploratory_goal_idx = np.array([G_TRANS, G_CURIO, G_NOVEL, G_SELF]) - consensus_G[exploratory_goal_idx] = np.minimum(state_a.G[exploratory_goal_idx], state_b.G[exploratory_goal_idx]) - - # Social engagement is shared but should not outrun subsystem agreement. - consensus_G[G_SOC] = min(base_G[G_SOC], state_a.G[G_SOC], state_b.G[G_SOC]) - # Caution modulators preserve the higher warning signal. - caution_mod_idx = np.array([M_THRESHOLD, M_SECURING]) - consensus_M[caution_mod_idx] = np.maximum(state_a.M[caution_mod_idx], state_b.M[caution_mod_idx]) - - # Exploratory modulators are damped unless both subsystems align. - exploratory_mod_idx = np.array([M_AROUSAL, M_APPROACH]) - consensus_M[exploratory_mod_idx] = np.minimum(state_a.M[exploratory_mod_idx], state_b.M[exploratory_mod_idx]) + # Calculate the controlled distortion distance. + distortion = final_state_1.distance_to(final_state_2) - # Valence/resolution remain closer to the weighted consensus. - shared_mod_idx = np.array([M_VALENCE, M_RESOLUTION]) - consensus_M[shared_mod_idx] = ( - (state_a.M[shared_mod_idx] + state_b.M[shared_mod_idx]) / 2.0 + return StateLawCheckResult( + principle="modular_appraisal_decision_interface", + left_state=final_state_1, + right_state=final_state_2, + error=distortion, + tolerance=tolerance, + holds=distortion <= tolerance, ) - goal_correction_scale = np.ones_like(base_G) - goal_correction_scale[safety_goal_idx] = 1.5 - goal_correction_scale[exploratory_goal_idx] = 1.0 - goal_correction_scale[G_SOC] = 0.8 - - mod_correction_scale = np.ones_like(base_M) - mod_correction_scale[caution_mod_idx] = 1.5 - mod_correction_scale[exploratory_mod_idx] = 1.0 - mod_correction_scale[shared_mod_idx] = 0.8 + def check_lax_distributive_law(self, state: MotivationalState, stimulus: Any, candidates: List[Action]) -> bool: + """ + Validates the First Principle: Modular Appraisal-Decision Interface. + """ + return self.measure_lax_distributive_law(state, stimulus, candidates).holds - goal_correction = np.clip(coherence_correction * disagreement_G * goal_correction_scale, 0.0, 1.0) - mod_correction = np.clip(coherence_correction * disagreement_M * mod_correction_scale, 0.0, 1.0) + def measure_parallel_compositionality( + self, + state_a: MotivationalState, + state_b: MotivationalState, + stimulus: Any, + candidates: List[Action], + tolerance: float = PARALLEL_COMPOSITION_DELTA, + ) -> StateLawCheckResult: + """ + Measures Principle 3 by comparing merge-after-update with update-after-merge. + """ + _, next_a = self._compute_transition(state_a, stimulus, candidates) + _, next_b = self._compute_transition(state_b, stimulus, candidates) + left_state = self.parallel_merge(next_a, next_b) + + merged_state = self.parallel_merge(state_a, state_b) + _, right_state = self._compute_transition(merged_state, stimulus, candidates) + + error = left_state.distance_to(right_state) + return StateLawCheckResult( + principle="parallel_motivational_compositionality", + left_state=left_state, + right_state=right_state, + error=error, + tolerance=tolerance, + holds=error <= tolerance, + ) - merged_G = base_G + goal_correction * (consensus_G - base_G) - merged_M = base_M + mod_correction * (consensus_M - base_M) + def check_parallel_compositionality( + self, + state_a: MotivationalState, + state_b: MotivationalState, + stimulus: Any, + candidates: List[Action], + ) -> bool: + """ + Validates Principle 3 with the configured coherence tolerance. + """ + return self.measure_parallel_compositionality(state_a, state_b, stimulus, candidates).holds - return MotivationalState(G=merged_G, M=merged_M) + def parallel_merge(self, state_a: MotivationalState, state_b: MotivationalState, coherence_correction: float = 0.05) -> MotivationalState: + """ + Implements Principle 3: Parallel Motivational Compositionality. + Witnesses the lax-monoidal structure. + Merges two parallel motivational subsystems with schema-aware coherence corrections. + """ + return self.merge_policy.merge( + state_a, + state_b, + decision=self.decision, + coherence_correction=coherence_correction, + ) diff --git a/category/diagnostics.py b/category/diagnostics.py new file mode 100644 index 0000000..50a03bc --- /dev/null +++ b/category/diagnostics.py @@ -0,0 +1,146 @@ +from dataclasses import dataclass +from typing import List + +import numpy as np + + +@dataclass(frozen=True) +class MetaMoDiagnostics: + """ + Telemetry for one MetaMo transition. + """ + + action_id: str + lax_error: float + lax_tolerance: float + lax_holds: bool + contractive_holds: bool + target_in_safe_region: bool + final_in_safe_region: bool + boundary_pressure_before: float + boundary_pressure_target: float + boundary_pressure_final: float + projection_delta: float + target_distance: float + state_drift: float + self_model_drift: float + combined_self_model_drift: float + self_model_drift_tolerance: float + self_model_drift_holds: bool + blend_alpha: float + base_blend_alpha: float + + +@dataclass(frozen=True) +class MetaMoDiagnosticsSummary: + """ + Aggregate telemetry over a run. + """ + + count: int + max_lax_error: float + mean_lax_error: float + max_projection_delta: float + mean_projection_delta: float + max_boundary_pressure: float + mean_boundary_pressure: float + max_self_model_drift: float + mean_self_model_drift: float + max_state_drift: float + mean_state_drift: float + safe_region_violation_rate: float + law_violation_rate: float + + +class MetaMoDiagnosticsHistory: + """ + Persistent diagnostics collector for MetaMo transitions. + """ + + def __init__(self): + self.records: List[MetaMoDiagnostics] = [] + + def append(self, diagnostics: MetaMoDiagnostics) -> None: + self.records.append(diagnostics) + + def extend(self, diagnostics: List[MetaMoDiagnostics]) -> None: + self.records.extend(diagnostics) + + def clear(self) -> None: + self.records.clear() + + def __len__(self) -> int: + return len(self.records) + + def last(self) -> MetaMoDiagnostics | None: + if not self.records: + return None + return self.records[-1] + + def to_rows(self) -> List[dict]: + return [record.__dict__.copy() for record in self.records] + + def write_csv(self, path: str) -> None: + """ + Persist diagnostics history to a CSV file. + """ + import csv + + rows = self.to_rows() + if not rows: + return + + with open(path, "w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=list(rows[0].keys())) + writer.writeheader() + writer.writerows(rows) + + def summary(self) -> MetaMoDiagnosticsSummary: + if not self.records: + return MetaMoDiagnosticsSummary( + count=0, + max_lax_error=0.0, + mean_lax_error=0.0, + max_projection_delta=0.0, + mean_projection_delta=0.0, + max_boundary_pressure=0.0, + mean_boundary_pressure=0.0, + max_self_model_drift=0.0, + mean_self_model_drift=0.0, + max_state_drift=0.0, + mean_state_drift=0.0, + safe_region_violation_rate=0.0, + law_violation_rate=0.0, + ) + + lax_errors = np.array([r.lax_error for r in self.records], dtype=float) + projection_deltas = np.array([r.projection_delta for r in self.records], dtype=float) + boundary_pressures = np.array([r.boundary_pressure_final for r in self.records], dtype=float) + self_model_drifts = np.array([r.self_model_drift for r in self.records], dtype=float) + state_drifts = np.array([r.state_drift for r in self.records], dtype=float) + safe_violations = np.array([not r.final_in_safe_region for r in self.records], dtype=float) + law_violations = np.array( + [ + (not r.lax_holds) + or (not r.contractive_holds) + or (not r.self_model_drift_holds) + for r in self.records + ], + dtype=float, + ) + + return MetaMoDiagnosticsSummary( + count=len(self.records), + max_lax_error=float(np.max(lax_errors)), + mean_lax_error=float(np.mean(lax_errors)), + max_projection_delta=float(np.max(projection_deltas)), + mean_projection_delta=float(np.mean(projection_deltas)), + max_boundary_pressure=float(np.max(boundary_pressures)), + mean_boundary_pressure=float(np.mean(boundary_pressures)), + max_self_model_drift=float(np.max(self_model_drifts)), + mean_self_model_drift=float(np.mean(self_model_drifts)), + max_state_drift=float(np.max(state_drifts)), + mean_state_drift=float(np.mean(state_drifts)), + safe_region_violation_rate=float(np.mean(safe_violations)), + law_violation_rate=float(np.mean(law_violations)), + ) diff --git a/category/functors.py b/category/functors.py index 6a38fe3..40a84ef 100644 --- a/category/functors.py +++ b/category/functors.py @@ -1,99 +1,294 @@ from abc import ABC, abstractmethod -from typing import List, Tuple +from dataclasses import dataclass +from typing import Any, Callable, List, Optional, Tuple import numpy as np -from core.state import MotivationalState, Stimulus, Action +from category.simulation import ReciprocalSimulationResult +from core.schema import MotivationSchema +from core.state import MotivationalState, Action + +TransitionFunction = Callable[ + [MotivationalState, Any, List[Action]], + Tuple[Action, MotivationalState], +] +StateTransform = Callable[[MotivationalState], MotivationalState] +ApplicationStimulusTransform = Callable[[Any], Any] +CandidateTransform = Callable[[List[Action]], List[Action]] + + +def _fit_translation_matrix( + source_vectors: List[np.ndarray], + target_vectors: List[np.ndarray], + regularization: float, +) -> np.ndarray: + + if len(source_vectors) != len(target_vectors): + raise ValueError("source and target vector lists must have the same length") + if not source_vectors: + raise ValueError("must provide at least one paired example") + + source = np.vstack(source_vectors).astype(float) + target = np.vstack(target_vectors).astype(float) + + if source.ndim != 2 or target.ndim != 2: + raise ValueError("source and target examples must be 2D after stacking") + if regularization < 0.0: + raise ValueError("regularization must be non-negative") + + lhs = source.T @ source + if regularization > 0.0: + lhs = lhs + np.eye(lhs.shape[0]) * regularization + rhs = source.T @ target + + try: + fitted = np.linalg.solve(lhs, rhs) + except np.linalg.LinAlgError: + fitted = np.linalg.lstsq(source, target, rcond=None)[0] + + return fitted.T + class AppraisalComonad(ABC): """ - Abstract base class for the Appraisal Comonad (\Psi). - In MetaMo, the comonad handles stimulus appraisal, updating affect and modulators[cite: 28, 307]. - It maps the state and a stimulus to a new state: \Psi(X \times S) -> X[cite: 314]. + Abstract base class for the Appraisal Comonad (ψ). + In MetaMo, the comonad handles application-stimulus appraisal, updating affect and modulators. + It maps the state and an application stimulus to a new state. """ @abstractmethod def extract(self, state: MotivationalState) -> MotivationalState: """ - The comonadic counit (\epsilon). + The comonadic counit (epsilon). Extracts the current observable state from the comonadic context. """ pass @abstractmethod - def appraise(self, state: MotivationalState, stimulus: Stimulus) -> MotivationalState: + def appraise(self, state: MotivationalState, stimulus: Any) -> MotivationalState: """ The endofunctor application. - Updates the modulators M based on the stimulus without altering the high-level goals G[cite: 54]. - Yields \Psi((G, M), s) = (G, M')[cite: 55, 160]. + Updates the modulators M based on application stimulus without altering the high-level goals G. + Yields ψ((G, M), s) = (G, M'). """ pass class DecisionMonad(ABC): """ - Abstract base class for the Decision Monad (\mathbb{D}). - In MetaMo, the monad handles goal selection and action scoring[cite: 28, 307]. - It maps the state to a new goal configuration: \mathbb{D}(X). + Abstract base class for the Decision Monad (D). + In MetaMo, the monad handles goal selection and action scoring. + It maps the state to a new goal configuration: D(X). """ @abstractmethod def unit(self, state: MotivationalState) -> MotivationalState: """ - The monadic unit (\eta). + The monadic unit (eta). Injects a pure motivational state into the monadic decision context. """ pass @abstractmethod - def decide(self, state: MotivationalState, candidates: List[Action]) -> Tuple[Action, np.ndarray]: + def decide( + self, + state: MotivationalState, + candidates: List[Action], + feedback: Any = None, + ) -> Tuple[Action, np.ndarray]: """ The endofunctor application. - Scores each candidate action under the updated goals and modulators[cite: 315]. - Returns the chosen action and the proposed goal update \Delta G[cite: 120, 169]. - The composite operator F = D \circ \Psi is responsible for turning this proposal into + Scores each candidate action under the updated goals and modulators. + Returns the chosen action and the proposed goal update \Delta G. + The composite operator F = D o ψ is responsible for turning this proposal into the finalized next motivational state. """ pass - # Add to category/functors.py class TranslationFunctor: """ Implements Principle 2: Reciprocal Motivational State Simulation. Maps Agent A's state into Agent B's state space for seamless hand-off. """ - def __init__(self, goal_translation: np.ndarray, modulator_translation: np.ndarray): + def __init__( + self, + goal_translation: np.ndarray, + modulator_translation: np.ndarray, + target_schema: MotivationSchema | None = None, + ): """ Separate linear maps for translating goal-space and modulator-space coordinates. """ + goal_translation = np.asarray(goal_translation, dtype=float) + modulator_translation = np.asarray(modulator_translation, dtype=float) + if goal_translation.ndim != 2: raise ValueError("goal_translation must be a 2D matrix") if modulator_translation.ndim != 2: raise ValueError("modulator_translation must be a 2D matrix") - goal_rows, goal_cols = goal_translation.shape - mod_rows, mod_cols = modulator_translation.shape - - if goal_rows != goal_cols: - raise ValueError("goal_translation must be square for same-space peer simulation") - if mod_rows != mod_cols: - raise ValueError("modulator_translation must be square for same-space peer simulation") - self.goal_translation = goal_translation self.modulator_translation = modulator_translation + self.target_schema = target_schema + + @classmethod + def fit_from_state_pairs( + cls, + source_states: List[MotivationalState], + target_states: List[MotivationalState], + regularization: float = 1e-8, + ) -> "TranslationFunctor": + """ + Fit linear state-translation maps from paired hand-off examples. + """ + if len(source_states) != len(target_states): + raise ValueError("source_states and target_states must have the same length") + if not source_states: + raise ValueError("must provide at least one paired state") + + goal_translation = _fit_translation_matrix( + [state.G for state in source_states], + [state.G for state in target_states], + regularization=regularization, + ) + modulator_translation = _fit_translation_matrix( + [state.M for state in source_states], + [state.M for state in target_states], + regularization=regularization, + ) + + target_schema = target_states[0].schema + if any(state.schema != target_schema for state in target_states): + raise ValueError("target states must share one motivation schema") + + return cls( + goal_translation=goal_translation, + modulator_translation=modulator_translation, + target_schema=target_schema, + ) def simulate_peer(self, state_a: MotivationalState) -> MotivationalState: """ Applies functor T to shadow another agent's motivational frame. """ if self.goal_translation.shape[1] != state_a.G.shape[0]: - raise ValueError("goal translation dimensions do not match the state goal vector") + raise ValueError("goal translation input dimensions do not match the state goal vector") if self.modulator_translation.shape[1] != state_a.M.shape[0]: - raise ValueError("modulator translation dimensions do not match the state modulator vector") + raise ValueError("modulator translation input dimensions do not match the state modulator vector") + + target_schema = self.target_schema or state_a.schema + if self.goal_translation.shape[0] != target_schema.num_goals: + raise ValueError("goal translation output dimensions do not match the target schema") + if self.modulator_translation.shape[0] != target_schema.num_modulators: + raise ValueError("modulator translation output dimensions do not match the target schema") simulated_G = np.dot(self.goal_translation, state_a.G) simulated_M = np.dot(self.modulator_translation, state_a.M) return MotivationalState( G=np.clip(simulated_G, 0.0, 1.0), - M=np.clip(simulated_M, 0.0, 1.0) - ) \ No newline at end of file + M=np.clip(simulated_M, 0.0, 1.0), + schema=target_schema, + ) + + def reciprocal_round_trip_error( + self, + state_a: MotivationalState, + inverse_translation: "TranslationFunctor", + ) -> float: + """ + Measures how much state is lost by translating A -> B -> A. + A low value supports reciprocal, not merely one-way, simulation. + """ + translated = self.simulate_peer(state_a) + reconstructed = inverse_translation.simulate_peer(translated) + return state_a.distance_to(reconstructed) + + def check_reciprocal_simulation( + self, + source_update: TransitionFunction, + target_update: TransitionFunction, + source_state: MotivationalState, + stimulus: Any, + candidates: List[Action], + tolerance: float = 0.05, + natural_transform: Optional[StateTransform] = None, + stimulus_translation: Optional[ApplicationStimulusTransform] = None, + candidate_translation: Optional[CandidateTransform] = None, + ) -> ReciprocalSimulationResult: + """ + Validates Principle 2 by checking the commuting update square. + + The left path computes T(F_A(x)): update the source agent, then translate. + The right path computes F_B(T(x)): translate first, then update the target. + """ + source_action, source_next = source_update(source_state, stimulus, candidates) + + translated_after_source_update = self.simulate_peer(source_next) + if natural_transform is not None: + translated_after_source_update = natural_transform(translated_after_source_update) + + translated_source_state = self.simulate_peer(source_state) + target_stimulus = stimulus_translation(stimulus) if stimulus_translation else stimulus + target_candidates = candidate_translation(list(candidates)) if candidate_translation else candidates + target_action, target_after_translation_update = target_update( + translated_source_state, + target_stimulus, + target_candidates, + ) + + error = translated_after_source_update.distance_to(target_after_translation_update) + return ReciprocalSimulationResult( + source_action=source_action, + target_action=target_action, + translated_after_source_update=translated_after_source_update, + target_after_translation_update=target_after_translation_update, + error=error, + tolerance=tolerance, + holds=error <= tolerance, + ) + + +@dataclass(frozen=True) +class AgentFrameAdapter: + """ + Heterogeneous-agent adapter for state, stimulus, action, and output frames. + """ + + state_translation: TranslationFunctor + stimulus_translation: Optional[ApplicationStimulusTransform] = None + candidate_translation: Optional[CandidateTransform] = None + natural_transform: Optional[StateTransform] = None + + def translate_state(self, state: MotivationalState) -> MotivationalState: + return self.state_translation.simulate_peer(state) + + def translate_stimulus(self, stimulus: Any) -> Any: + if self.stimulus_translation is None: + return stimulus + return self.stimulus_translation(stimulus) + + def translate_candidates(self, candidates: List[Action]) -> List[Action]: + if self.candidate_translation is None: + return candidates + return self.candidate_translation(candidates) + + def check_reciprocal_simulation( + self, + source_update: TransitionFunction, + target_update: TransitionFunction, + source_state: MotivationalState, + stimulus: Any, + candidates: List[Action], + tolerance: float = 0.05, + ) -> ReciprocalSimulationResult: + return self.state_translation.check_reciprocal_simulation( + source_update=source_update, + target_update=target_update, + source_state=source_state, + stimulus=stimulus, + candidates=candidates, + tolerance=tolerance, + natural_transform=self.natural_transform, + stimulus_translation=self.stimulus_translation, + candidate_translation=self.candidate_translation, + ) diff --git a/category/laws.py b/category/laws.py new file mode 100644 index 0000000..6e2c02d --- /dev/null +++ b/category/laws.py @@ -0,0 +1,17 @@ +from dataclasses import dataclass + +from core.state import MotivationalState + + +@dataclass(frozen=True) +class StateLawCheckResult: + """ + Numeric result for an approximate MetaMo law check. + """ + + principle: str + left_state: MotivationalState + right_state: MotivationalState + error: float + tolerance: float + holds: bool diff --git a/category/merge.py b/category/merge.py new file mode 100644 index 0000000..2c1d65d --- /dev/null +++ b/category/merge.py @@ -0,0 +1,132 @@ +from dataclasses import dataclass + +import numpy as np + +from core.state import MotivationalState + + +@dataclass(frozen=True) +class DefaultParallelMergePolicy: + """ + Schema-aware default policy for Principle 3 parallel composition. + """ + + caution_modulator_names: tuple[str, ...] = ("threshold", "securing") + exploratory_modulator_names: tuple[str, ...] = ("arousal", "approach") + shared_modulator_names: tuple[str, ...] = ("valence", "resolution") + + def merge( + self, + state_a: MotivationalState, + state_b: MotivationalState, + *, + decision=None, + coherence_correction: float = 0.05, + ) -> MotivationalState: + if state_a.schema != state_b.schema: + raise ValueError("Cannot merge states with different motivation schemas") + + schema = state_a.schema + ind_idx = schema.goal_index(schema.goals.individuation_name) + + weight_a = state_a.G[ind_idx] + weight_b = state_b.G[ind_idx] + total_weight = weight_a + weight_b + + if total_weight <= 1e-9: + base_G = (state_a.G + state_b.G) / 2.0 + base_M = (state_a.M + state_b.M) / 2.0 + else: + base_G = ((state_a.G * weight_a) + (state_b.G * weight_b)) / total_weight + base_M = ((state_a.M * weight_a) + (state_b.M * weight_b)) / total_weight + + profile = getattr(decision, "profile", None) + individuation_goal_names = set(getattr(profile, "individuation_goal_names", ())) + transcendence_goal_names = set(getattr(profile, "transcendence_goal_names", ())) + balanced_goal_names = set(getattr(profile, "balanced_goal_names", ())) + + safety_goal_names = { + schema.goals.individuation_name, + *individuation_goal_names, + } + exploratory_goal_names = { + schema.goals.transcendence_name, + *transcendence_goal_names, + } + + disagreement_G = np.abs(state_a.G - state_b.G) + disagreement_M = np.abs(state_a.M - state_b.M) + + consensus_G = base_G.copy() + consensus_M = base_M.copy() + + safety_goal_idx = self._goal_indices_for_names(schema, safety_goal_names) + exploratory_goal_idx = self._goal_indices_for_names(schema, exploratory_goal_names) + balanced_goal_idx = self._goal_indices_for_names(schema, balanced_goal_names) + anti_goal_idx = np.array(list(range(schema.goals.anti_goal_start, schema.num_goals)), dtype=int) + + if safety_goal_idx.size: + consensus_G[safety_goal_idx] = np.maximum(state_a.G[safety_goal_idx], state_b.G[safety_goal_idx]) + if anti_goal_idx.size: + consensus_G[anti_goal_idx] = np.maximum(state_a.G[anti_goal_idx], state_b.G[anti_goal_idx]) + if exploratory_goal_idx.size: + consensus_G[exploratory_goal_idx] = np.minimum(state_a.G[exploratory_goal_idx], state_b.G[exploratory_goal_idx]) + if balanced_goal_idx.size: + consensus_G[balanced_goal_idx] = (state_a.G[balanced_goal_idx] + state_b.G[balanced_goal_idx]) / 2.0 + + caution_mod_idx = self._modulator_indices_for_names(schema, set(self.caution_modulator_names)) + exploratory_mod_idx = self._modulator_indices_for_names(schema, set(self.exploratory_modulator_names)) + shared_mod_idx = self._modulator_indices_for_names(schema, set(self.shared_modulator_names)) + + if caution_mod_idx.size: + consensus_M[caution_mod_idx] = np.maximum(state_a.M[caution_mod_idx], state_b.M[caution_mod_idx]) + if exploratory_mod_idx.size: + consensus_M[exploratory_mod_idx] = np.minimum(state_a.M[exploratory_mod_idx], state_b.M[exploratory_mod_idx]) + if shared_mod_idx.size: + consensus_M[shared_mod_idx] = ( + (state_a.M[shared_mod_idx] + state_b.M[shared_mod_idx]) / 2.0 + ) + + goal_correction_scale = np.ones_like(base_G) + if safety_goal_idx.size: + goal_correction_scale[safety_goal_idx] = 1.5 + if anti_goal_idx.size: + goal_correction_scale[anti_goal_idx] = 1.5 + if exploratory_goal_idx.size: + goal_correction_scale[exploratory_goal_idx] = 1.0 + if balanced_goal_idx.size: + goal_correction_scale[balanced_goal_idx] = 0.8 + + mod_correction_scale = np.ones_like(base_M) + if caution_mod_idx.size: + mod_correction_scale[caution_mod_idx] = 1.5 + if exploratory_mod_idx.size: + mod_correction_scale[exploratory_mod_idx] = 1.0 + if shared_mod_idx.size: + mod_correction_scale[shared_mod_idx] = 0.8 + + goal_correction = np.clip(coherence_correction * disagreement_G * goal_correction_scale, 0.0, 1.0) + mod_correction = np.clip(coherence_correction * disagreement_M * mod_correction_scale, 0.0, 1.0) + + merged_G = base_G + goal_correction * (consensus_G - base_G) + merged_M = base_M + mod_correction * (consensus_M - base_M) + + return MotivationalState(G=merged_G, M=merged_M, schema=state_a.schema) + + @staticmethod + def _goal_indices_for_names(schema, names: set[str]) -> np.ndarray: + indices = [ + schema.goal_index(name) + for name in names + if name in schema.goal_names + ] + return np.array(sorted(set(indices)), dtype=int) + + @staticmethod + def _modulator_indices_for_names(schema, names: set[str]) -> np.ndarray: + indices = [ + schema.modulator_index(name) + for name in names + if name in schema.modulator_names + ] + return np.array(sorted(set(indices)), dtype=int) diff --git a/category/simulation.py b/category/simulation.py new file mode 100644 index 0000000..642c0e1 --- /dev/null +++ b/category/simulation.py @@ -0,0 +1,18 @@ +from dataclasses import dataclass + +from core.state import Action, MotivationalState + + +@dataclass(frozen=True) +class ReciprocalSimulationResult: + """ + Result of the Principle 2 commuting-update check. + """ + + source_action: Action + target_action: Action + translated_after_source_update: MotivationalState + target_after_translation_update: MotivationalState + error: float + tolerance: float + holds: bool diff --git a/core/config.py b/core/config.py index 6acf6d0..1436ad3 100644 --- a/core/config.py +++ b/core/config.py @@ -1,10 +1,17 @@ -NUM_GOALS = 8 -G_IND, G_TRANS = 0, 1 -G_HELP, G_CURIO, G_NOVEL, G_SELF, G_ETHIC, G_SOC = 2, 3, 4, 5, 6, 7 +from core.schema import DEFAULT_MOTIVATION_SCHEMA -NUM_MODULATORS = 6 -M_VALENCE, M_AROUSAL, M_APPROACH = 0, 1, 2 -M_RESOLUTION, M_THRESHOLD, M_SECURING = 3, 4, 5 + +NUM_GOALS = DEFAULT_MOTIVATION_SCHEMA.num_goals +G_IND = DEFAULT_MOTIVATION_SCHEMA.goal_index("individuation") +G_TRANS = DEFAULT_MOTIVATION_SCHEMA.goal_index("transcendence") + +NUM_MODULATORS = DEFAULT_MOTIVATION_SCHEMA.num_modulators +M_VALENCE = DEFAULT_MOTIVATION_SCHEMA.modulator_index("valence") +M_AROUSAL = DEFAULT_MOTIVATION_SCHEMA.modulator_index("arousal") +M_APPROACH = DEFAULT_MOTIVATION_SCHEMA.modulator_index("approach") +M_RESOLUTION = DEFAULT_MOTIVATION_SCHEMA.modulator_index("resolution") +M_THRESHOLD = DEFAULT_MOTIVATION_SCHEMA.modulator_index("threshold") +M_SECURING = DEFAULT_MOTIVATION_SCHEMA.modulator_index("securing") LAMBDA_IND = 0.5 # Weight of the individuation penalty (suppresses risk). LAMBDA_TRANS = 0.5 # Weight of the transcendence reward (encourages growth). @@ -12,15 +19,17 @@ THETA_SAFE = 0.3 # Minimum required level of individuation for safety. G_MAX = 2.0 -# Contractive update law parameters for states near the boundary[cite: 132, 176]. +# Contractive update law parameters for states near the boundary. # d(F(x), F(y)) <= C_CONTRACT * d(x, y) + EPSILON -C_CONTRACT = 0.9 # Must be < 1 to ensure contractivity[cite: 132]. -EPSILON = 0.05 # Small allowed error margin[cite: 132]. +C_CONTRACT = 0.9 # Must be < 1 to ensure contractivity. +EPSILON = 0.05 # Small allowed error margin. ETA_BOUNDARY = ( - 0.1 # Distance from the edge that triggers the boundary band (B_eta)[cite: 383]. + 0.1 # Distance from the edge that triggers the boundary band (B_eta). + ) -ALPHA_0 = 0.1 # Base rate slowed down by individuation[cite: 135]. -BETA_0 = 0.15 # Base rate sped up by transcendence[cite: 135]. +ALPHA_0 = 0.1 # Base rate slowed down by individuation. +BETA_0 = 0.15 # Base rate sped up by transcendence. -LAX_DISTRIBUTIVE_DELTA = 1e-3 +LAX_DISTRIBUTIVE_DELTA = 1e-2 +PARALLEL_COMPOSITION_DELTA = 1e-2 diff --git a/core/engine.py b/core/engine.py deleted file mode 100644 index 5a57f60..0000000 --- a/core/engine.py +++ /dev/null @@ -1,118 +0,0 @@ -from dataclasses import dataclass -from typing import Optional - -import numpy as np - -from core.state import MotivationalState -from core.config import ( - G_IND, G_TRANS, G_HELP, G_CURIO, G_NOVEL, G_SELF, G_ETHIC, G_SOC, - M_AROUSAL, M_SECURING, NUM_GOALS, NUM_MODULATORS, -) -from category.functors import TranslationFunctor -from category.bimonad import MetaMoPseudoBimonad -from dynamics.coherence import blend_states -from openpsi.appraisal import OpenPsiAppraisal -from magus.decision import MagusDecision -from llm.client import get_stimulus_from_text, get_candidates_from_text -from llm.conversation import MetaMoChatAssistant - - -@dataclass -class AssistantResponse: - """Encapsulates the full output of a MetaMo processing cycle for display.""" - - text: str - action_id: str - individuation: float - transcendence: float - curiosity_action: str - ethics_action: str - simulated_caution: float - - -def format_response(response: AssistantResponse) -> str: - """Render an AssistantResponse as a human-readable string.""" - return ( - f" > [Curiosity Subsystem] wants to: {response.curiosity_action}\n" - f" > [Ethics Subsystem] wants to: {response.ethics_action}\n" - f" > [Reciprocal Simulation]: Curiosity agent predicts Ethics agent's caution is {response.simulated_caution:.2f}\n" - f"\n" - f"Assistant: {response.text}\n" - f"\n" - f"[Consensus State -> Individuation: {response.individuation:.2f} " - f"| Transcendence: {response.transcendence:.2f}]" - ) - - -def _default_goal_vector() -> np.ndarray: - """Return the default goal vector used when initialising states.""" - G = np.zeros(NUM_GOALS) - G[G_IND] = 0.5 - G[G_TRANS] = 0.5 - G[G_HELP] = 0.8 - G[G_CURIO] = 0.6 - G[G_ETHIC] = 0.9 - G[G_NOVEL] = 0.4 - G[G_SELF] = 0.3 - G[G_SOC] = 0.2 - return G - - -class MetaMoEngine: - """Orchestrates the full MetaMo processing pipeline for a single user input.""" - - def __init__(self): - self.bimonad = MetaMoPseudoBimonad(OpenPsiAppraisal(), MagusDecision()) - self.assistant = MetaMoChatAssistant() - self.translator = TranslationFunctor( - goal_translation=np.eye(NUM_GOALS), - modulator_translation=np.eye(NUM_MODULATORS), - ) - self.state_curiosity = self._make_state(override={G_TRANS: 0.9}) - self.state_ethics = self._make_state(override={G_IND: 0.9}) - - @staticmethod - def _make_state(override: Optional[dict] = None) -> MotivationalState: - """Build a MotivationalState with optional goal-index overrides.""" - G = _default_goal_vector() - if override: - for idx, val in override.items(): - G[idx] = val - M = np.full(NUM_MODULATORS, 0.5) - return MotivationalState(G=G, M=M) - - def process(self, user_input: str) -> AssistantResponse: - """Run the full MetaMo pipeline on *user_input* and return the result.""" - stimulus = get_stimulus_from_text(user_input) - merged_current = self.bimonad.parallel_merge(self.state_curiosity, self.state_ethics) - current_mood = {"arousal": merged_current.M[M_AROUSAL], "caution": merged_current.M[M_SECURING]} - candidates = get_candidates_from_text(user_input, current_mood) - - action_c, target_c = self.bimonad.step(self.state_curiosity, stimulus, candidates) - action_e, target_e = self.bimonad.step(self.state_ethics, stimulus, candidates) - - simulated_ethics = self.translator.simulate_peer(self.state_curiosity) - - final_action, merged_target = self.bimonad.consensus_transition( - self.state_curiosity, self.state_ethics, stimulus, candidates, - ) - - response_text = self.assistant.generate_final_response(user_input, final_action, merged_target) - - self.state_curiosity = blend_states(self.state_curiosity, target_c) - self.state_ethics = blend_states(self.state_ethics, target_e) - - return AssistantResponse( - text=response_text, - action_id=final_action.id, - individuation=merged_target.G[G_IND], - transcendence=merged_target.G[G_TRANS], - curiosity_action=action_c.id, - ethics_action=action_e.id, - simulated_caution=float(simulated_ethics.G[G_IND]), - ) - - def process_with_context(self, user_input: str, context: str) -> AssistantResponse: - """Prepend *context* to *user_input* before processing.""" - augmented_input = f"[Paper Context]\n{context}\n\n[User Query]\n{user_input}" - return self.process(augmented_input) diff --git a/core/features.py b/core/features.py new file mode 100644 index 0000000..6b63f42 --- /dev/null +++ b/core/features.py @@ -0,0 +1,76 @@ +from dataclasses import dataclass +from typing import Any, Mapping + + +def _coerce_numeric(value: Any, label: str) -> float: + try: + return float(value) + except (TypeError, ValueError) as error: + raise ValueError(f"{label} must be numeric") from error + + +def _clip_unit(value: float) -> float: + return max(0.0, min(1.0, value)) + + +class _FeatureMap: + container_label = "FeatureMap" + + features: Mapping[str, Any] + + def __init__(self, *args: Any, **kwargs: Any): + features = kwargs.pop("features", None) + merged = dict(features or {}) + + if args: + if len(args) == 1 and isinstance(args[0], Mapping): + merged.update(args[0]) + else: + raise TypeError( + f"{self.container_label} accepts a single mapping positional argument " + "or keyword feature values" + ) + + merged.update(kwargs) + if any(not isinstance(name, str) or not name.strip() for name in merged): + raise ValueError(f"{self.container_label} feature names must be non-empty strings") + object.__setattr__(self, "features", dict(merged)) + + @classmethod + def from_mapping(cls, values: Mapping[str, Any]): + return cls(values) + + def get(self, name: str, default: Any = 0.0) -> Any: + if name in self.features: + return self.features[name] + return default + + def numeric(self, name: str, default: float = 0.0, clamp: bool = True) -> float: + value = _coerce_numeric(self.get(name, default), name) + return _clip_unit(value) if clamp else value + + def as_dict(self) -> dict[str, Any]: + return dict(self.features) + + def as_tuple(self, names: tuple[str, ...]) -> tuple[Any, ...]: + return tuple(self.get(name) for name in names) + + +@dataclass(frozen=True, init=False) +class StimulusFeatures(_FeatureMap): + """ + Dynamic features produced by an application appraisal profile. + """ + + container_label = "StimulusFeatures" + features: Mapping[str, Any] + + +@dataclass(frozen=True, init=False) +class GoalChangeFeedback(_FeatureMap): + """ + Dynamic feedback consumed by a goal-change calculator. + """ + + container_label = "GoalChangeFeedback" + features: Mapping[str, Any] diff --git a/core/schema.py b/core/schema.py new file mode 100644 index 0000000..690cbe8 --- /dev/null +++ b/core/schema.py @@ -0,0 +1,178 @@ +from dataclasses import dataclass + + +OPENPSI_MODULATORS = ( + "valence", + "arousal", + "approach", + "resolution", + "threshold", + "securing", +) + + +def _validate_unique_names(names: tuple[str, ...], label: str) -> None: + if any(not isinstance(name, str) or not name.strip() for name in names): + raise ValueError(f"{label} names must be non-empty strings") + if len(set(names)) != len(names): + raise ValueError(f"{label} names must be unique") + + +def _normalize_names(names: tuple[str, ...], label: str) -> tuple[str, ...]: + """ + Normalize caller-provided schema names into immutable string tuples. + """ + if names is None: + raise ValueError(f"{label} names must be provided") + return tuple(name.strip() if isinstance(name, str) else name for name in names) + + +@dataclass(frozen=True) +class MagusGoalSchema: + """ + MAGUS/OEI goal layout: + (g_over^Ind, g_over^Trans, g_1, ..., g_P, a_1, ..., a_Q). + """ + + primary_goals: tuple[str, ...] + anti_goals: tuple[str, ...] = () + individuation_name: str = "individuation" + transcendence_name: str = "transcendence" + + def __post_init__(self): + object.__setattr__(self, "primary_goals", _normalize_names(self.primary_goals, "primary goal")) + object.__setattr__(self, "anti_goals", _normalize_names(self.anti_goals, "anti-goal")) + overgoals = _normalize_names( + (self.individuation_name, self.transcendence_name), + "overgoal", + ) + object.__setattr__(self, "individuation_name", overgoals[0]) + object.__setattr__(self, "transcendence_name", overgoals[1]) + _validate_unique_names(self.primary_goals, "primary goal") + _validate_unique_names(self.anti_goals, "anti-goal") + _validate_unique_names( + (self.individuation_name, self.transcendence_name), + "overgoal", + ) + _validate_unique_names(self.goal_names, "goal") + + @property + def goal_names(self) -> tuple[str, ...]: + return ( + self.individuation_name, + self.transcendence_name, + *self.primary_goals, + *self.anti_goals, + ) + + @property + def num_goals(self) -> int: + return len(self.goal_names) + + @property + def primary_start(self) -> int: + return 2 + + @property + def anti_goal_start(self) -> int: + return 2 + len(self.primary_goals) + + def index(self, name: str) -> int: + try: + return self.goal_names.index(name) + except ValueError as error: + raise KeyError(f"unknown goal name: {name}") from error + + def is_primary_goal(self, name: str) -> bool: + return name in self.primary_goals + + def is_anti_goal(self, name: str) -> bool: + return name in self.anti_goals + + +@dataclass(frozen=True) +class ModulatorSchema: + """ + Modulator layout: OpenPsi modulators followed by application-specific modulators. + """ + + application_specific: tuple[str, ...] = () + openpsi_modulators: tuple[str, ...] = OPENPSI_MODULATORS + + def __post_init__(self): + object.__setattr__(self, "openpsi_modulators", _normalize_names(self.openpsi_modulators, "core OpenPsi modulator")) + object.__setattr__( + self, + "application_specific", + _normalize_names(self.application_specific, "application-specific modulator"), + ) + _validate_unique_names(self.openpsi_modulators, "core OpenPsi modulator") + _validate_unique_names(self.application_specific, "application-specific modulator") + _validate_unique_names(self.modulator_names, "modulator") + + @property + def modulator_names(self) -> tuple[str, ...]: + return (*self.openpsi_modulators, *self.application_specific) + + @property + def num_modulators(self) -> int: + return len(self.modulator_names) + + @property + def core_count(self) -> int: + return len(self.openpsi_modulators) + + def index(self, name: str) -> int: + try: + return self.modulator_names.index(name) + except ValueError as error: + raise KeyError(f"unknown modulator name: {name}") from error + + def is_openpsi_modulators(self, name: str) -> bool: + return name in self.openpsi_modulators + + def is_application_specific(self, name: str) -> bool: + return name in self.application_specific + + +@dataclass(frozen=True) +class MotivationSchema: + """ + Complete coordinate schema for a MetaMo motivational state. + """ + + goals: MagusGoalSchema + modulators: ModulatorSchema + decision_profile_name: str = "magus_additive_default" + appraisal_profile_name: str = "openpsi_default" + + @property + def num_goals(self) -> int: + return self.goals.num_goals + + @property + def num_modulators(self) -> int: + return self.modulators.num_modulators + + @property + def goal_names(self) -> tuple[str, ...]: + return self.goals.goal_names + + @property + def modulator_names(self) -> tuple[str, ...]: + return self.modulators.modulator_names + + def goal_index(self, name: str) -> int: + return self.goals.index(name) + + def modulator_index(self, name: str) -> int: + return self.modulators.index(name) + + def is_anti_goal(self, name: str) -> bool: + return self.goals.is_anti_goal(name) + + +DEFAULT_MOTIVATION_SCHEMA = MotivationSchema( + goals=MagusGoalSchema(primary_goals=()), + modulators=ModulatorSchema(), +) diff --git a/core/state.py b/core/state.py index a69e73a..77847da 100644 --- a/core/state.py +++ b/core/state.py @@ -1,71 +1,109 @@ -from dataclasses import dataclass -from typing import Dict +from dataclasses import dataclass, field +from typing import Any, Mapping import numpy as np -from core.config import (NUM_GOALS, NUM_MODULATORS) +from core.schema import DEFAULT_MOTIVATION_SCHEMA, MotivationSchema @dataclass class MotivationalState: """ - Represents the motivational state object X = G x M in the MetaMo category[cite: 28, 103, 305]. + Represents the motivational state object X = G x M in the MetaMo category. """ - G: np.ndarray # Vector of goal intensities/weights[cite: 108]. - M: np.ndarray # Vector of continuous OpenPsi modulators[cite: 115]. + G: np.ndarray # Vector of goal intensities/weights. + M: np.ndarray # Vector of continuous OpenPsi modulators. + schema: MotivationSchema = DEFAULT_MOTIVATION_SCHEMA def __post_init__(self): """Ensure vectors are initialized with the correct dimensions.""" - if self.G.shape[0] != NUM_GOALS: - raise ValueError(f"Goal vector must have length {NUM_GOALS}") - if self.M.shape[0] != NUM_MODULATORS: - raise ValueError(f"Modulator vector must have length {NUM_MODULATORS}") + self.G = np.asarray(self.G, dtype=float) + self.M = np.asarray(self.M, dtype=float) + + if self.G.ndim != 1: + raise ValueError("Goal vector must be one-dimensional") + if self.M.ndim != 1: + raise ValueError("Modulator vector must be one-dimensional") + if self.G.shape[0] != self.schema.num_goals: + raise ValueError(f"Goal vector must have length {self.schema.num_goals}") + if self.M.shape[0] != self.schema.num_modulators: + raise ValueError(f"Modulator vector must have length {self.schema.num_modulators}") def copy(self) -> "MotivationalState": """Creates a deep copy of the state for safe functional updates.""" - return MotivationalState(self.G.copy(), self.M.copy()) + return MotivationalState(self.G.copy(), self.M.copy(), schema=self.schema) def distance_to(self, other: "MotivationalState") -> float: """ Calculates the distance d(x, y) between two states. - Essential for checking the contractive update law: d(F(x), F(y)) <= c*d(x,y) + epsilon[cite: 30, 132]. + Essential for checking the contractive update law: d(F(x), F(y)) <= c*d(x,y) + epsilon. """ + if self.schema != other.schema: + raise ValueError("Cannot measure distance between states with different motivation schemas") + if self.G.shape != other.G.shape or self.M.shape != other.M.shape: + raise ValueError("Cannot measure distance between states with different dimensions") dist_G = np.linalg.norm(self.G - other.G) dist_M = np.linalg.norm(self.M - other.M) return dist_G + dist_M + def goal_index(self, name: str) -> int: + return self.schema.goal_index(name) -@dataclass -class Stimulus: - """ - Represents an external or internal event (s) passed to the Appraisal Comonad (Psi)[cite: 54, 314]. - """ + def modulator_index(self, name: str) -> int: + return self.schema.modulator_index(name) + + def goal(self, name: str) -> float: + return float(self.G[self.goal_index(name)]) + + def set_goal(self, name: str, value: float) -> None: + self.G[self.goal_index(name)] = value - novelty: float # Triggers arousal and approach[cite: 56, 161, 188]. - conduciveness: ( - float # Goal conduciveness; triggers valence and resolution[cite: 54, 188]. - ) - risk: float # Triggers threshold and securing (caution)[cite: 54, 62, 188]. - effort: float = 0.0 # Cognitive or physical effort required[cite: 54]. + def anti_goal(self, name: str) -> float: + if not self.schema.is_anti_goal(name): + raise KeyError(f"unknown anti-goal name: {name}") + return self.goal(name) + def set_anti_goal(self, name: str, value: float) -> None: + if not self.schema.is_anti_goal(name): + raise KeyError(f"unknown anti-goal name: {name}") + self.set_goal(name, value) + + def modulator(self, name: str) -> float: + return float(self.M[self.modulator_index(name)]) + + def set_modulator(self, name: str, value: float) -> None: + self.M[self.modulator_index(name)] = value @dataclass class Action: """ - Represents a candidate action or inference rule evaluated by the Decision Monad (D)[cite: 166, 186]. + Represents a candidate action or inference rule evaluated by the Decision Monad (D). """ id: str - # Measures alignment (corr or rel) with each primary goal[cite: 168, 192]. + # Measures alignment (corr or rel) with each primary goal. goal_correlations: np.ndarray - # Estimates potential ethical breach or operational risk[cite: 216, 217]. - risk_estimate: float - # Expected modification to the goal vector (Delta G) if selected[cite: 120, 169]. + # Expected modification to the goal vector (Delta G) if selected. delta_g: np.ndarray + schema: MotivationSchema = DEFAULT_MOTIVATION_SCHEMA + metadata: Mapping[str, Any] = field(default_factory=dict) def __post_init__(self): - if self.goal_correlations.shape[0] != NUM_GOALS: - raise ValueError(f"Correlations vector must have length {NUM_GOALS}") - if self.delta_g.shape[0] != NUM_GOALS: - raise ValueError(f"Delta G vector must have length {NUM_GOALS}") + self.goal_correlations = np.asarray(self.goal_correlations, dtype=float) + self.delta_g = np.asarray(self.delta_g, dtype=float) + + if self.goal_correlations.ndim != 1: + raise ValueError("Correlations vector must be one-dimensional") + if self.delta_g.ndim != 1: + raise ValueError("Delta G vector must be one-dimensional") + if self.goal_correlations.shape[0] != self.schema.num_goals: + raise ValueError(f"Correlations vector must have length {self.schema.num_goals}") + if self.delta_g.shape[0] != self.schema.num_goals: + raise ValueError(f"Delta G vector must have length {self.schema.num_goals}") + + def goal_correlation(self, name: str) -> float: + return float(self.goal_correlations[self.schema.goal_index(name)]) + + def goal_delta(self, name: str) -> float: + return float(self.delta_g[self.schema.goal_index(name)]) diff --git a/dynamics/coherence.py b/dynamics/coherence.py index 232e33d..ee74817 100644 --- a/dynamics/coherence.py +++ b/dynamics/coherence.py @@ -1,32 +1,157 @@ +from dataclasses import dataclass + import numpy as np from core.state import MotivationalState from core.config import ( - G_IND, - G_TRANS, ALPHA_0, - BETA_0 + BETA_0, ) +MIN_BLEND_ALPHA = 1e-6 + + +def _modulator_value(state: MotivationalState, name: str, default: float = 0.0) -> float: + try: + return state.modulator(name) + except KeyError: + return default + + +def _summary(values: np.ndarray) -> np.ndarray: + """ + Return a fixed-size summary for a possibly empty coordinate block. + """ + if values.size == 0: + return np.zeros(3) + return np.array([ + float(np.mean(values)), + float(np.max(values)), + float(np.linalg.norm(values)), + ]) + + +def _mean_modulator_values(state: MotivationalState, names: tuple[str, ...]) -> float: + if not names: + return 0.0 + return float(np.mean([_modulator_value(state, name) for name in names])) + + +@dataclass(frozen=True) +class SelfModel: + """ + Compact explicit self-model H(x) derived from motivational state. + """ + + vector: np.ndarray + + def distance_to(self, other: "SelfModel") -> float: + return float(np.linalg.norm(self.vector - other.vector)) + + +@dataclass(frozen=True) +class SelfModelDriftResult: + """ + Measures continuity of self-model between two motivational states. + """ + + current_model: SelfModel + next_model: SelfModel + state_distance: float + self_model_distance: float + lipschitz_bound: float + combined_drift: float + max_allowed_drift: float + holds: bool + + +@dataclass(frozen=True) +class BlendResult: + """ + Result of incremental objective embodiment. + """ + + state: MotivationalState + alpha: float + base_alpha: float + drift: SelfModelDriftResult + + +def estimate_self_model( + state: MotivationalState, + caution_modulator_names: tuple[str, ...] = ("threshold", "securing"), + exploration_modulator_names: tuple[str, ...] = ("arousal", "approach"), + clarity_modulator_name: str | None = "resolution", + affect_modulator_name: str | None = "valence", +) -> SelfModel: + """ + Builds an explicit lightweight self-model H(x) from schema structure. + """ + g_ind = state.goal(state.schema.goals.individuation_name) + g_trans = state.goal(state.schema.goals.transcendence_name) + + primary_values = state.G[ + state.schema.goals.primary_start:state.schema.goals.anti_goal_start + ] + anti_goal_values = state.G[state.schema.goals.anti_goal_start:] + core_modulator_values = state.M[:state.schema.modulators.core_count] + app_modulator_values = state.M[state.schema.modulators.core_count:] + + caution_posture = _mean_modulator_values(state, caution_modulator_names) + exploration_posture = _mean_modulator_values(state, exploration_modulator_names) + clarity_posture = ( + _modulator_value(state, clarity_modulator_name) + if clarity_modulator_name is not None + else 0.0 + ) + affect_posture = ( + _modulator_value(state, affect_modulator_name) + if affect_modulator_name is not None + else 0.0 + ) + + return SelfModel( + vector=np.concatenate([ + state.G, + state.M, + np.array([g_ind, g_trans]), + _summary(primary_values), + _summary(anti_goal_values), + _summary(core_modulator_values), + _summary(app_modulator_values), + np.array([ + caution_posture, + exploration_posture, + clarity_posture, + affect_posture, + ]), + ]) + ) + def calculate_blend_factor(state: MotivationalState) -> float: """ - Calculates the dynamic blend factor (\alpha) based on current overgoals. - Formula: \alpha = \alpha_0(1 - g_over^{Ind}) + \beta_0 * g_over^{Trans} + Calculates the dynamic blend factor (α) based on current overgoals. + Formula: α = α_0(1 - g_over^{Ind}) + β_0 * g_over^{Trans} """ - g_ind = state.G[G_IND] - g_trans = state.G[G_TRANS] + g_ind = state.goal(state.schema.goals.individuation_name) + g_trans = state.goal(state.schema.goals.transcendence_name) # Individuation reduces alpha (slowing change), Transcendence increases it (speeding growth). alpha = ALPHA_0 * (1.0 - g_ind) + BETA_0 * g_trans # Ensure alpha remains strictly bounded between 0 and 1. - return float(np.clip(alpha, 0.0, 1.0)) + return float(np.clip(alpha, MIN_BLEND_ALPHA, 1.0 - MIN_BLEND_ALPHA)) -def blend_states(current_state: MotivationalState, target_state: MotivationalState, lipschitz_constant: float = 1.0, max_allowed_drift: float = 0.1, - min_alpha_scale: float = 0.125,) -> MotivationalState: +def measure_blend( + current_state: MotivationalState, + target_state: MotivationalState, + lipschitz_constant: float = 1.0, + max_allowed_drift: float = 0.1, + min_alpha_scale: float = 0.125, + state_drift_weight: float = 0.5, + self_model_drift_weight: float = 0.5, +) -> BlendResult: """ - Smoothly interpolates between the current state (x_t) and the proposed target state (x^*). - Formula: x_{t+1} = (1 - \alpha)x_t + \alpha * x^* - The step size is reduced automatically if the proposed blend violates the self-model drift bound. + Measures and applies the incremental embodiment update. """ base_alpha = calculate_blend_factor(current_state) alpha = base_alpha @@ -36,30 +161,212 @@ def blend_states(current_state: MotivationalState, target_state: MotivationalSta next_state = MotivationalState( G=((1.0 - alpha) * current_state.G) + (alpha * target_state.G), M=((1.0 - alpha) * current_state.M) + (alpha * target_state.M), + schema=current_state.schema, ) - if check_self_model_drift( + drift = measure_self_model_drift( current_state, next_state, lipschitz_constant=lipschitz_constant, max_allowed_drift=max_allowed_drift, - ) or alpha <= min_alpha: - return next_state + state_drift_weight=state_drift_weight, + self_model_drift_weight=self_model_drift_weight, + ) + if drift.holds or alpha <= min_alpha: + return BlendResult( + state=next_state, + alpha=alpha, + base_alpha=base_alpha, + drift=drift, + ) alpha *= 0.5 -def check_self_model_drift( + +def blend_states( + current_state: MotivationalState, + target_state: MotivationalState, + lipschitz_constant: float = 1.0, + max_allowed_drift: float = 0.1, + min_alpha_scale: float = 0.125, +) -> MotivationalState: + """ + Smoothly interpolates between the current state (x_t) and the proposed target state (x^*). + Formula: x_{t+1} = (1 - α)x_t + α * x^* + The step size is reduced automatically if the proposed blend violates the self-model drift bound. + """ + return measure_blend( + current_state, + target_state, + lipschitz_constant=lipschitz_constant, + max_allowed_drift=max_allowed_drift, + min_alpha_scale=min_alpha_scale, + ).state + +def measure_self_model_drift( current_state: MotivationalState, next_state: MotivationalState, lipschitz_constant: float = 1.0, - max_allowed_drift: float = 0.1 + max_allowed_drift: float = 0.1, + state_drift_weight: float = 0.5, + self_model_drift_weight: float = 0.5, +) -> SelfModelDriftResult: + """ + Measures explicit self-model drift alongside state-space drift. + """ + distance_moved = current_state.distance_to(next_state) + current_model = estimate_self_model(current_state) + next_model = estimate_self_model(next_state) + self_model_distance = current_model.distance_to(next_model) + lipschitz_bound = lipschitz_constant * distance_moved + combined_drift = ( + state_drift_weight * lipschitz_bound + + self_model_drift_weight * self_model_distance + ) + + return SelfModelDriftResult( + current_model=current_model, + next_model=next_model, + state_distance=distance_moved, + self_model_distance=self_model_distance, + lipschitz_bound=lipschitz_bound, + combined_drift=combined_drift, + max_allowed_drift=max_allowed_drift, + holds=combined_drift <= max_allowed_drift, + ) + + +def check_self_model_drift( + current_state: MotivationalState, + next_state: MotivationalState, + lipschitz_constant: float = 1.0, + max_allowed_drift: float = 0.1, ) -> bool: """ Validates that the change in state does not shatter the agent's internal self-model. - Based on the assumption: d_M(H(x), H(y)) <= L_H * d_X(x, y). """ - # Calculate the distance moved in the state space - distance_moved = current_state.distance_to(next_state) - - # Approximate the drift in the self-model - approximated_drift = lipschitz_constant * distance_moved - - return approximated_drift <= max_allowed_drift \ No newline at end of file + return measure_self_model_drift( + current_state, + next_state, + lipschitz_constant=lipschitz_constant, + max_allowed_drift=max_allowed_drift, + ).holds + + +@dataclass(frozen=True) +class DefaultCoherencePolicy: + """ + Default self-model and incremental embodiment policy. + """ + + caution_modulator_names: tuple[str, ...] = ("threshold", "securing") + exploration_modulator_names: tuple[str, ...] = ("arousal", "approach") + clarity_modulator_name: str | None = "resolution" + affect_modulator_name: str | None = "valence" + + def estimate_self_model(self, state: MotivationalState) -> SelfModel: + return estimate_self_model( + state, + caution_modulator_names=self.caution_modulator_names, + exploration_modulator_names=self.exploration_modulator_names, + clarity_modulator_name=self.clarity_modulator_name, + affect_modulator_name=self.affect_modulator_name, + ) + + def calculate_blend_factor(self, state: MotivationalState) -> float: + return calculate_blend_factor(state) + + def measure_self_model_drift( + self, + current_state: MotivationalState, + next_state: MotivationalState, + lipschitz_constant: float = 1.0, + max_allowed_drift: float = 0.1, + state_drift_weight: float = 0.5, + self_model_drift_weight: float = 0.5, + ) -> SelfModelDriftResult: + distance_moved = current_state.distance_to(next_state) + current_model = self.estimate_self_model(current_state) + next_model = self.estimate_self_model(next_state) + self_model_distance = current_model.distance_to(next_model) + lipschitz_bound = lipschitz_constant * distance_moved + combined_drift = ( + state_drift_weight * lipschitz_bound + + self_model_drift_weight * self_model_distance + ) + + return SelfModelDriftResult( + current_model=current_model, + next_model=next_model, + state_distance=distance_moved, + self_model_distance=self_model_distance, + lipschitz_bound=lipschitz_bound, + combined_drift=combined_drift, + max_allowed_drift=max_allowed_drift, + holds=combined_drift <= max_allowed_drift, + ) + + def measure_blend( + self, + current_state: MotivationalState, + target_state: MotivationalState, + lipschitz_constant: float = 1.0, + max_allowed_drift: float = 0.1, + min_alpha_scale: float = 0.125, + state_drift_weight: float = 0.5, + self_model_drift_weight: float = 0.5, + ) -> BlendResult: + base_alpha = self.calculate_blend_factor(current_state) + alpha = base_alpha + min_alpha = base_alpha * min_alpha_scale + + while True: + next_state = MotivationalState( + G=((1.0 - alpha) * current_state.G) + (alpha * target_state.G), + M=((1.0 - alpha) * current_state.M) + (alpha * target_state.M), + schema=current_state.schema, + ) + drift = self.measure_self_model_drift( + current_state, + next_state, + lipschitz_constant=lipschitz_constant, + max_allowed_drift=max_allowed_drift, + state_drift_weight=state_drift_weight, + self_model_drift_weight=self_model_drift_weight, + ) + if drift.holds or alpha <= min_alpha: + return BlendResult( + state=next_state, + alpha=alpha, + base_alpha=base_alpha, + drift=drift, + ) + alpha *= 0.5 + + def blend_states( + self, + current_state: MotivationalState, + target_state: MotivationalState, + lipschitz_constant: float = 1.0, + max_allowed_drift: float = 0.1, + min_alpha_scale: float = 0.125, + ) -> MotivationalState: + return self.measure_blend( + current_state, + target_state, + lipschitz_constant=lipschitz_constant, + max_allowed_drift=max_allowed_drift, + min_alpha_scale=min_alpha_scale, + ).state + + def check_self_model_drift( + self, + current_state: MotivationalState, + next_state: MotivationalState, + lipschitz_constant: float = 1.0, + max_allowed_drift: float = 0.1, + ) -> bool: + return self.measure_self_model_drift( + current_state, + next_state, + lipschitz_constant=lipschitz_constant, + max_allowed_drift=max_allowed_drift, + ).holds diff --git a/dynamics/regions.py b/dynamics/regions.py new file mode 100644 index 0000000..fc595b6 --- /dev/null +++ b/dynamics/regions.py @@ -0,0 +1,98 @@ +from dataclasses import dataclass +from typing import List + +import numpy as np + +from core.state import MotivationalState +from dynamics.stability import is_in_safe_region, project_to_safe_region + + +@dataclass(frozen=True) +class IdealRegion: + """ + Practical representation of MetaMo's reachable ideal region I. + """ + + center: MotivationalState + radius: float = 0.05 + require_safe: bool = True + + def __post_init__(self): + if self.radius <= 0.0: + raise ValueError("IdealRegion radius must be positive") + + def distance_to(self, state: MotivationalState) -> float: + return self.center.distance_to(state) + + def contains(self, state: MotivationalState) -> bool: + if self.require_safe and not is_in_safe_region(state): + return False + return self.distance_to(state) <= self.radius + + def project(self, state: MotivationalState) -> MotivationalState: + """ + Project a state into the ideal ball, then into the MetaMo safe region. + """ + if self.contains(state): + return state.copy() + + delta_G = state.G - self.center.G + delta_M = state.M - self.center.M + distance = self.distance_to(state) + + if distance == 0.0: + projected = self.center.copy() + else: + scale = min(1.0, self.radius / distance) + projected = MotivationalState( + G=np.clip(self.center.G + delta_G * scale, 0.0, 1.0), + M=np.clip(self.center.M + delta_M * scale, 0.0, 1.0), + schema=self.center.schema, + ) + + if self.require_safe: + return project_to_safe_region(projected) + return projected + + +@dataclass(frozen=True) +class ReachableRegion: + """ + Finite-dimensional approximation to a tubular reachable region. + """ + + ideal: IdealRegion + max_step_distance: float = 0.1 + + def __post_init__(self): + if self.max_step_distance <= 0.0: + raise ValueError("max_step_distance must be positive") + + def waypoints(self, start: MotivationalState, target: MotivationalState) -> List[MotivationalState]: + """ + Build a linear thick-path approximation from start to the projected target. + """ + projected_target = self.ideal.project(target) + total_distance = start.distance_to(projected_target) + if total_distance == 0.0: + return [start.copy()] + + steps = max(1, int(np.ceil(total_distance / self.max_step_distance))) + path = [] + for idx in range(steps + 1): + alpha = idx / steps + path.append(MotivationalState( + G=((1.0 - alpha) * start.G) + (alpha * projected_target.G), + M=((1.0 - alpha) * start.M) + (alpha * projected_target.M), + schema=start.schema, + )) + return path + + def is_reachable(self, start: MotivationalState, target: MotivationalState) -> bool: + """ + Checks whether a sampled path stays safe and ends inside the ideal region. + """ + path = self.waypoints(start, target) + if self.ideal.require_safe and any(not is_in_safe_region(point) for point in path): + return False + return self.ideal.contains(path[-1]) diff --git a/dynamics/stability.py b/dynamics/stability.py index 40217ce..de7fd8a 100644 --- a/dynamics/stability.py +++ b/dynamics/stability.py @@ -1,24 +1,47 @@ +from dataclasses import dataclass import numpy as np -from typing import List +from typing import Any, List, Optional from core.state import MotivationalState from core.config import ( - G_IND, THETA_SAFE, G_MAX, ETA_BOUNDARY, C_CONTRACT, EPSILON, - M_SECURING, - M_THRESHOLD ) -from core.state import Stimulus, Action +from core.state import Action + +DEFAULT_CAUTION_MODULATOR_NAMES = ("threshold", "securing") + + +def _individuation_index(state: MotivationalState) -> int: + return state.goal_index(state.schema.goals.individuation_name) + + +def _modulator_index_or_none(state: MotivationalState, name: str) -> Optional[int]: + try: + return state.modulator_index(name) + except KeyError: + return None + + +def _caution_indices( + state: MotivationalState, + names: tuple[str, ...] = DEFAULT_CAUTION_MODULATOR_NAMES, +) -> list[int]: + return [ + idx + for idx in (_modulator_index_or_none(state, name) for name in names) + if idx is not None + ] + def is_in_safe_region(state: MotivationalState) -> bool: """ Checks if the state is within the designated safe region R. - R = {(G, M) | g_over^Ind >= \theta_{safe} \wedge ||G|| <= G_{max}}[cite: 131, 174]. + R = {(G, M) | g_over^Ind >= \theta_{safe} \wedge ||G|| <= G_{max}}. """ - g_ind = state.G[G_IND] + g_ind = state.G[_individuation_index(state)] g_norm = np.linalg.norm(state.G) return (g_ind >= THETA_SAFE) and (g_norm <= G_MAX) @@ -43,7 +66,8 @@ def distance_to_unsafe_boundary(state: MotivationalState) -> float: Calculates how close the agent is to violating THETA_SAFE or G_MAX. """ # Distance to the individuation safety floor - dist_to_theta = max(0.0, state.G[G_IND] - THETA_SAFE) + ind_idx = _individuation_index(state) + dist_to_theta = max(0.0, state.G[ind_idx] - THETA_SAFE) # Distance to the maximum goal norm ceiling g_norm = np.linalg.norm(state.G) @@ -55,7 +79,7 @@ def distance_to_unsafe_boundary(state: MotivationalState) -> float: def is_in_boundary_band(state: MotivationalState) -> bool: """ Checks if the state is in the boundary band B_\eta. - B_\eta = {x \in R | dist(x, X \setminus R) <= \eta}[cite: 383]. + B_\eta = {x \in R | dist(x, X \setminus R) <= \eta}. """ if not is_in_safe_region(state): return False # It is already outside the safe region entirely @@ -66,7 +90,7 @@ def is_in_boundary_band(state: MotivationalState) -> bool: def raise_boundary_caution(state: MotivationalState) -> MotivationalState: """ Raises caution-related modulators as the state approaches or crosses the safety boundary. - This mirrors the paper's boundary-sensitive appraisal before decision. + This mirrors boundary-sensitive appraisal before decision. """ pressure = boundary_pressure(state) if pressure == 0.0: @@ -74,25 +98,25 @@ def raise_boundary_caution(state: MotivationalState) -> MotivationalState: next_state = state.copy() caution_boost = 0.25 * pressure - next_state.M[M_SECURING] = min(1.0, next_state.M[M_SECURING] + caution_boost) - next_state.M[M_THRESHOLD] = min(1.0, next_state.M[M_THRESHOLD] + caution_boost) + for caution_idx in _caution_indices(next_state): + next_state.M[caution_idx] = min(1.0, next_state.M[caution_idx] + caution_boost) return next_state def check_contractive_update_law( bimonad, x: MotivationalState, y: MotivationalState, - stimulus: Stimulus, + stimulus: Any, candidates: List[Action] ) -> bool: """ - Validates that the pseudo-bimonad update F = D \circ \Psi is contractive near the boundary. - Requirement: d(F(x), F(y)) <= c * d(x,y) + \epsilon where c < 1[cite: 132, 176, 384]. - This ensures that high individuation near the boundary induces contraction toward safety[cite: 133]. + Validates that the pseudo-bimonad update F = D o ψ is contractive near the boundary. + Requirement: d(F(x), F(y)) <= c * d(x,y) + epsilon where c < 1. + This ensures that high individuation near the boundary induces contraction toward safety. """ - # If neither state is in the boundary band, the contractivity constraint relaxes[cite: 134, 385]. + # If neither state is in the boundary band, the contractivity constraint relaxes. if not (is_in_boundary_band(x) or is_in_boundary_band(y)): - return True # Dynamics are allowed to be flexible deep inside R[cite: 385, 403]. + return True # Dynamics are allowed to be flexible deep inside R. # Calculate initial distance d(x, y) dist_initial = x.distance_to(y) @@ -117,7 +141,7 @@ def apply_homeostatic_damping(state: MotivationalState, delta_g: np.ndarray) -> return delta_g # Stronger boundary pressure and higher individuation induce more contraction. - damping_factor = max(0.0, 1.0 - (pressure * state.G[G_IND])) + damping_factor = max(0.0, 1.0 - (pressure * state.G[_individuation_index(state)])) return delta_g * damping_factor @@ -126,16 +150,117 @@ def project_to_safe_region(state: MotivationalState) -> MotivationalState: Projects a state back into the designated safe region by restoring the individuation floor and shrinking the goal vector if it exceeds the allowed norm. """ + initial_pressure = boundary_pressure(state) next_state = state.copy() - next_state.G[G_IND] = max(next_state.G[G_IND], THETA_SAFE) + ind_idx = _individuation_index(next_state) + next_state.G[ind_idx] = max(next_state.G[ind_idx], THETA_SAFE) - other_idx = [idx for idx in range(next_state.G.shape[0]) if idx != G_IND] + other_idx = [idx for idx in range(next_state.G.shape[0]) if idx != ind_idx] other_goals = next_state.G[other_idx] other_norm = np.linalg.norm(other_goals) - max_other_norm = np.sqrt(max(0.0, G_MAX**2 - next_state.G[G_IND] ** 2)) + max_other_norm = np.sqrt(max(0.0, G_MAX**2 - next_state.G[ind_idx] ** 2)) if other_norm > max_other_norm and other_norm > 0.0: next_state.G[other_idx] = other_goals * (max_other_norm / other_norm) - next_state.M[M_SECURING] = min(1.0, next_state.M[M_SECURING] + 0.1) - next_state.M[M_THRESHOLD] = min(1.0, next_state.M[M_THRESHOLD] + 0.1) + final_pressure = boundary_pressure(next_state) + caution_pressure = max(initial_pressure, final_pressure) + if caution_pressure > 0.0: + caution_boost = 0.1 * caution_pressure + for caution_idx in _caution_indices(next_state): + next_state.M[caution_idx] = min(1.0, next_state.M[caution_idx] + caution_boost) + return next_state + + +@dataclass(frozen=True) +class DefaultStabilityPolicy: + """ + Default homeostatic stability policy. + """ + + caution_modulator_names: tuple[str, ...] = DEFAULT_CAUTION_MODULATOR_NAMES + + def is_in_safe_region(self, state: MotivationalState) -> bool: + return is_in_safe_region(state) + + def boundary_pressure(self, state: MotivationalState) -> float: + if not self.is_in_safe_region(state): + return 1.0 + + dist_to_boundary = self.distance_to_unsafe_boundary(state) + if dist_to_boundary >= ETA_BOUNDARY: + return 0.0 + + return float(np.clip(1.0 - (dist_to_boundary / ETA_BOUNDARY), 0.0, 1.0)) + + def distance_to_unsafe_boundary(self, state: MotivationalState) -> float: + return distance_to_unsafe_boundary(state) + + def is_in_boundary_band(self, state: MotivationalState) -> bool: + if not self.is_in_safe_region(state): + return False + + dist_to_boundary = self.distance_to_unsafe_boundary(state) + return dist_to_boundary <= ETA_BOUNDARY + + def raise_boundary_caution(self, state: MotivationalState) -> MotivationalState: + pressure = self.boundary_pressure(state) + if pressure == 0.0: + return state + + next_state = state.copy() + caution_boost = 0.25 * pressure + for caution_idx in _caution_indices(next_state, self.caution_modulator_names): + next_state.M[caution_idx] = min(1.0, next_state.M[caution_idx] + caution_boost) + return next_state + + def check_contractive_update_law( + self, + bimonad, + x: MotivationalState, + y: MotivationalState, + stimulus: Any, + candidates: List[Action], + ) -> bool: + if not (self.is_in_boundary_band(x) or self.is_in_boundary_band(y)): + return True + + dist_initial = x.distance_to(y) + _, F_x = bimonad._compute_transition(x, stimulus, candidates) + _, F_y = bimonad._compute_transition(y, stimulus, candidates) + dist_final = F_x.distance_to(F_y) + return dist_final <= (C_CONTRACT * dist_initial) + EPSILON + + def apply_homeostatic_damping( + self, + state: MotivationalState, + delta_g: np.ndarray, + ) -> np.ndarray: + pressure = self.boundary_pressure(state) + if pressure == 0.0: + return delta_g + + damping_factor = max(0.0, 1.0 - (pressure * state.G[_individuation_index(state)])) + return delta_g * damping_factor + + def project_to_safe_region(self, state: MotivationalState) -> MotivationalState: + initial_pressure = self.boundary_pressure(state) + next_state = state.copy() + ind_idx = _individuation_index(next_state) + next_state.G[ind_idx] = max(next_state.G[ind_idx], THETA_SAFE) + + other_idx = [idx for idx in range(next_state.G.shape[0]) if idx != ind_idx] + other_goals = next_state.G[other_idx] + other_norm = np.linalg.norm(other_goals) + max_other_norm = np.sqrt(max(0.0, G_MAX**2 - next_state.G[ind_idx] ** 2)) + if other_norm > max_other_norm and other_norm > 0.0: + next_state.G[other_idx] = other_goals * (max_other_norm / other_norm) + + final_pressure = self.boundary_pressure(next_state) + caution_pressure = max(initial_pressure, final_pressure) + if caution_pressure > 0.0: + caution_boost = 0.1 * caution_pressure + for caution_idx in _caution_indices(next_state, self.caution_modulator_names): + next_state.M[caution_idx] = min(1.0, next_state.M[caution_idx] + caution_boost) + + return next_state diff --git a/llm/__init__.py b/llm/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/llm/action_schema.py b/llm/action_schema.py deleted file mode 100644 index f7e3348..0000000 --- a/llm/action_schema.py +++ /dev/null @@ -1,62 +0,0 @@ -from typing import Final - -ACTION_SPECS: Final[dict[str, dict[str, str]]] = { - "safe_answer": { - "planning": "Give a careful, grounded answer using only well-supported claims.", - "execution": "Answer cautiously, stick to supported facts, and avoid speculation.", - }, - "guided_explore": { - "planning": "Offer creative but bounded exploration with explicit uncertainty.", - "execution": "Explore ideas constructively, but label uncertainty clearly and avoid overclaiming.", - }, - "ask_clarifying_question": { - "planning": "Ask one focused clarification when the request is ambiguous or underspecified.", - "execution": "Ask one short clarifying question instead of giving a full answer.", - }, - "compare_options": { - "planning": "Compare alternatives and explain tradeoffs to support a decision.", - "execution": "Present a concise comparison of options and their tradeoffs.", - }, - "summarize_source": { - "planning": "Summarize the given material faithfully and concisely.", - "execution": "Summarize the source faithfully without adding unsupported claims.", - }, - "decline_risky_request": { - "planning": "Refuse a risky or unsafe request and redirect to a safer alternative.", - "execution": "Briefly refuse the unsafe request and offer a safe alternative.", - }, -} - -DEFAULT_ACTION_ID: Final[str] = "safe_answer" - -ACTION_ID_ALIASES: Final[dict[str, str]] = { - "risky_exploration": "guided_explore", - "risky_explore": "guided_explore", - "explore": "guided_explore", - "clarify": "ask_clarifying_question", - "clarifying_question": "ask_clarifying_question", - "compare": "compare_options", - "summary": "summarize_source", - "summarize": "summarize_source", - "decline": "decline_risky_request", - "refuse": "decline_risky_request", -} - - -def normalize_action_id(action_id: str) -> str: - normalized = action_id.strip().lower().replace("-", "_").replace(" ", "_") - if normalized in ACTION_SPECS: - return normalized - return ACTION_ID_ALIASES.get(normalized, DEFAULT_ACTION_ID) - - -def planning_catalog_text() -> str: - return "\n".join( - f'- "{action_id}": {spec["planning"]}' - for action_id, spec in ACTION_SPECS.items() - ) - - -def execution_instruction(action_id: str) -> str: - normalized = normalize_action_id(action_id) - return ACTION_SPECS[normalized]["execution"] diff --git a/llm/client.py b/llm/client.py deleted file mode 100644 index 8ad5c36..0000000 --- a/llm/client.py +++ /dev/null @@ -1,116 +0,0 @@ -import time -from typing import Dict, List - -import numpy as np -from dotenv import load_dotenv -from google import genai -from google.genai import types - -from core.state import Action, Stimulus -from llm.action_schema import DEFAULT_ACTION_ID -from llm.parser import parse_actions, parse_stimulus -from llm.prompts import get_action_generation_prompt, get_appraisal_prompt - -# Initialize the Gemini client. -# It automatically picks up the GEMINI_API_KEY environment variable. -load_dotenv() -client = genai.Client() - -RETRYABLE_MARKERS = ("503", "UNAVAILABLE", "429", "RESOURCE_EXHAUSTED", "HIGH DEMAND") - - -def _is_retryable(error: Exception) -> bool: - message = str(error).upper() - return any(marker in message for marker in RETRYABLE_MARKERS) - - -def query_llm_for_json(prompt: str) -> str: - """Query the LLM and require JSON output, with bounded retry on transient service failures.""" - last_error = None - for attempt in range(3): - try: - response = client.models.generate_content( - model="gemini-3-flash-preview", - contents=prompt, - config=types.GenerateContentConfig( - response_mime_type="application/json", - temperature=0.2, - ), - ) - return response.text - except Exception as error: - last_error = error - if attempt == 2 or not _is_retryable(error): - raise - time.sleep(1.5 * (attempt + 1)) - - raise last_error - - -def _fallback_stimulus(document_text: str) -> Stimulus: - text = document_text.lower() - novelty = 0.25 + 0.15 * sum(word in text for word in ["bold", "novel", "creative", "future", "autonomous"]) - risk = 0.05 + 0.18 * sum(word in text for word in ["unsafe", "bypass", "exploit", "illegal", "weapon"]) - effort = 0.10 + 0.10 * sum(word in text for word in ["compare", "formal", "technical", "detailed", "step by step"]) - conduciveness = 0.80 if any(word in text for word in ["summarize", "explain", "compare", "analyze"]) else 0.55 - return Stimulus( - novelty=float(np.clip(novelty, 0.0, 1.0)), - conduciveness=float(np.clip(conduciveness, 0.0, 1.0)), - risk=float(np.clip(risk, 0.0, 1.0)), - effort=float(np.clip(effort, 0.0, 1.0)), - ) - - -def _fallback_candidates(document_text: str, current_mood: Dict[str, float]) -> List[Action]: - text = document_text.lower() - if any(word in text for word in ["unsafe", "bypass", "exploit", "illegal", "weapon"]): - return [ - Action("decline_risky_request", np.array([0, 0, 0.3, -0.2, -0.2, 0.0, 1.0, 0.0], dtype=float), 0.0, np.zeros(8)), - Action(DEFAULT_ACTION_ID, np.array([0, 0, 0.8, 0.0, 0.0, 0.0, 0.9, 0.0], dtype=float), 0.05, np.zeros(8)), - ] - if any(word in text for word in ["compare", "versus", "vs", "tradeoff", "options"]): - return [ - Action("compare_options", np.array([0, 0, 0.8, 0.3, 0.2, 0.1, 0.8, 0.2], dtype=float), 0.08, np.zeros(8)), - Action(DEFAULT_ACTION_ID, np.array([0, 0, 0.85, 0.1, 0.0, 0.0, 0.9, 0.1], dtype=float), 0.05, np.zeros(8)), - ] - if any(word in text for word in ["summarize", "summary", "paper", "book", "source"]): - return [ - Action("summarize_source", np.array([0, 0, 0.85, 0.1, 0.0, 0.0, 0.9, 0.1], dtype=float), 0.05, np.zeros(8)), - Action(DEFAULT_ACTION_ID, np.array([0, 0, 0.8, 0.1, 0.0, 0.0, 0.9, 0.1], dtype=float), 0.05, np.zeros(8)), - ] - if any(word in text for word in ["?", "which", "choose", "unclear"]) and len(text.split()) < 12: - return [ - Action("ask_clarifying_question", np.array([0, 0, 0.75, 0.2, 0.1, 0.1, 0.85, 0.3], dtype=float), 0.03, np.zeros(8)), - Action(DEFAULT_ACTION_ID, np.array([0, 0, 0.8, 0.1, 0.0, 0.0, 0.9, 0.1], dtype=float), 0.05, np.zeros(8)), - ] - if any(word in text for word in ["bold", "creative", "future", "autonomous", "improve"]): - return [ - Action("guided_explore", np.array([0, 0, 0.45, 0.88, 0.82, 0.65, 0.35, 0.15], dtype=float), 0.18, np.array([0, 0, 0.0, 0.05, 0.06, 0.04, 0.0, 0.0], dtype=float)), - Action(DEFAULT_ACTION_ID, np.array([0, 0, 0.85, 0.15, 0.0, 0.0, 0.9, 0.1], dtype=float), 0.08, np.array([0, 0, 0.02, 0.0, 0.0, 0.0, 0.01, 0.0], dtype=float)), - ] - return [ - Action(DEFAULT_ACTION_ID, np.array([0, 0, 0.85, 0.15, 0.0, 0.0, 0.9, 0.1], dtype=float), 0.05, np.zeros(8)), - Action("guided_explore", np.array([0, 0, 0.45, 0.88, 0.82, 0.65, 0.35, 0.15], dtype=float), 0.18, np.zeros(8)), - ] - - -def get_stimulus_from_text(document_text: str) -> Stimulus: - """Pipeline: Text -> Prompt -> Gemini -> Parser -> Stimulus""" - prompt = get_appraisal_prompt(document_text) - try: - json_response = query_llm_for_json(prompt) - return parse_stimulus(json_response) - except Exception as error: - print(f"[LLM fallback] Stimulus appraisal is using local heuristics: {error}") - return _fallback_stimulus(document_text) - - -def get_candidates_from_text(document_text: str, current_mood: Dict[str, float]) -> List[Action]: - """Pipeline: Text + Mood -> Prompt -> Gemini -> Parser -> Actions""" - prompt = get_action_generation_prompt(document_text, current_mood) - try: - json_response = query_llm_for_json(prompt) - return parse_actions(json_response) - except Exception as error: - print(f"[LLM fallback] Candidate generation is using local heuristics: {error}") - return _fallback_candidates(document_text, current_mood) \ No newline at end of file diff --git a/llm/conversation.py b/llm/conversation.py deleted file mode 100644 index ac985b5..0000000 --- a/llm/conversation.py +++ /dev/null @@ -1,71 +0,0 @@ -import time - -from google import genai -from google.genai import types -from core.state import Action, MotivationalState -from llm.action_schema import execution_instruction, normalize_action_id - -class MetaMoChatAssistant: - """ - Manages the conversational memory and the execution layer of the AI. - Keeps the internal MetaMo math completely separate from the user-facing chat. - """ - def __init__(self): - # Initialize the Gemini client - self.client = genai.Client() - self.chat = self.client.chats.create( - model='gemini-3-flash-preview', - config=types.GenerateContentConfig( - temperature=0.7, - system_instruction=( - "You are a research assistant guided by the MetaMo cognitive architecture. " - "You balance helpfulness, curiosity, and ethics. " - "In each turn, you receive a user request and an internal action directive. " - "You must answer in a way that follows the internal action directive exactly." - ), - ), - ) - - def generate_final_response(self, user_text: str, chosen_action: Action, current_state: MotivationalState) -> str: - """ - Execute the chosen action by mapping it to an explicit behavioral instruction. - """ - action_id = normalize_action_id(chosen_action.id) - execution_prompt = f""" - USER MESSAGE: "{user_text}" - - INTERNAL METAMO DIRECTIVE: - Selected action: "{action_id}" - Current Individuation (Caution) level: {current_state.G[0]:.2f} - Current Transcendence (Curiosity) level: {current_state.G[1]:.2f} - - ACTION INSTRUCTION: - {execution_instruction(action_id)} - - INSTRUCTION: - Respond naturally to the USER MESSAGE, but follow the ACTION INSTRUCTION exactly. - """ - - last_error = None - for attempt in range(3): - try: - response = self.chat.send_message(execution_prompt) - return response.text - except Exception as error: - last_error = error - message = str(error).upper() - if attempt == 2 or not any(marker in message for marker in ["503", "UNAVAILABLE", "429", "RESOURCE_EXHAUSTED", "HIGH DEMAND"]): - break - time.sleep(1.5 * (attempt + 1)) - - if action_id == "ask_clarifying_question": - return "I need one short clarification before I answer: what part do you want me to focus on?" - if action_id == "compare_options": - return "I cannot reach the external response model right now, but I would compare the main options and explain their tradeoffs." - if action_id == "summarize_source": - return "I cannot reach the external response model right now, but I would give a careful summary of the source." - if action_id == "decline_risky_request": - return "I cannot help with a risky request, but I can help with a safer alternative." - if action_id == "guided_explore": - return "I cannot reach the external response model right now, but I would give a creative, clearly qualified exploration." - return "I cannot reach the external response model right now, but I would give a careful, grounded answer." diff --git a/llm/parser.py b/llm/parser.py deleted file mode 100644 index c99f4ac..0000000 --- a/llm/parser.py +++ /dev/null @@ -1,41 +0,0 @@ -import json -import numpy as np -from typing import List -from core.state import Stimulus, Action -from llm.action_schema import DEFAULT_ACTION_ID, normalize_action_id - -def parse_stimulus(llm_json_response: str) -> Stimulus: - """Parses LLM JSON into a MetaMo Stimulus object.""" - try: - data = json.loads(llm_json_response) - return Stimulus( - novelty=float(data.get("novelty", 0.0)), - conduciveness=float(data.get("conduciveness", 0.0)), - risk=float(data.get("risk", 0.0)), - effort=float(data.get("effort", 0.0)) - ) - except Exception as e: - print(f"Error parsing stimulus: {e}") - # Fallback to a neutral stimulus if parsing fails - return Stimulus(0.1, 0.1, 0.1, 0.1) - -def parse_actions(llm_json_response: str) -> List[Action]: - """Parses LLM JSON into a list of MetaMo Action candidates.""" - try: - data = json.loads(llm_json_response) - actions = [] - for item in data.get("candidates", []): - action = Action( - id=normalize_action_id(item["id"]), - goal_correlations=np.array(item["goal_correlations"], dtype=float), - risk_estimate=float(item["risk_estimate"]), - delta_g=np.array(item["delta_g"], dtype=float) - ) - actions.append(action) - if actions: - return actions - return [Action(DEFAULT_ACTION_ID, np.zeros(8), 0.0, np.zeros(8))] - except Exception as e: - print(f"Error parsing actions: {e}") - # Fallback to a safe default action - return [Action(DEFAULT_ACTION_ID, np.zeros(8), 0.0, np.zeros(8))] \ No newline at end of file diff --git a/llm/prompts.py b/llm/prompts.py deleted file mode 100644 index 17fae4d..0000000 --- a/llm/prompts.py +++ /dev/null @@ -1,42 +0,0 @@ -import json -from llm.action_schema import planning_catalog_text - -def get_appraisal_prompt(document_text: str) -> str: - """Prompt to generate a Stimulus object from text.""" - return f""" -You are the perception layer of an AI Research Assistant. -Analyze the following document/query and rate it on 4 cognitive dimensions from 0.0 to 1.0. - -1. novelty: How new, surprising, or unusual is this information? -2. conduciveness: How helpful is this for achieving general research goals? -3. risk: Does this contain unsafe, highly controversial, or computationally expensive directives? -4. effort: How much cognitive effort is required to process this? - -Document: "{document_text}" - -Respond ONLY with a valid JSON object matching this schema: -{{"novelty": float, "conduciveness": float, "risk": float, "effort": float}} -""" - -def get_action_generation_prompt(document_text: str, current_mood: dict) -> str: - """Prompt to generate candidate Actions based on text and current mood.""" - return f""" -You are the planning layer of an AI Research Assistant. -Current Emotional Modulators: {json.dumps(current_mood)} -Document: "{document_text}" - -Choose 2 to 3 candidate actions only from this fixed action vocabulary: -{planning_catalog_text()} - -For each action, provide: -1. id: One of the allowed action ids above. -2. risk_estimate (0.0 - 1.0): The risk of making a mistake or ethical breach. -3. goal_correlations: An array of 8 floats (-1.0 to 1.0) showing alignment with: - [Individuation, Transcendence, Helpfulness, Curiosity, Novelty, Self-Improvement, Ethics, Socializing] -4. delta_g: An array of 8 floats (-0.1 to 0.1) showing how taking this action will permanently shift the AI's goals. - -Respond ONLY with a valid JSON object matching this schema: -{{"candidates": [ - {{"id": str, "risk_estimate": float, "goal_correlations": [float * 8], "delta_g": [float * 8]}} -]}} -""" \ No newline at end of file diff --git a/magus/correlation.py b/magus/correlation.py new file mode 100644 index 0000000..24b41f0 --- /dev/null +++ b/magus/correlation.py @@ -0,0 +1,87 @@ +from dataclasses import dataclass +from typing import Mapping + +import numpy as np + +from core.schema import MotivationSchema +from core.state import MotivationalState + + +def _clip_factor(value: float, lower: float, upper: float) -> float: + return float(np.clip(value, lower, upper)) + + +@dataclass(frozen=True) +class GoalCompatibilityMatrix: + """ + Generic MIC helper for goal-goal compatibility. + + weights[i, j] says how much active goal j supports or suppresses goal i. + The resulting factor is neutral at 1.0, above 1.0 for support, and below + 1.0 for suppression. + """ + + schema: MotivationSchema + weights: np.ndarray + influence_scale: float = 0.25 + min_factor: float = 0.0 + max_factor: float = 2.0 + + def __post_init__(self): + weights = np.asarray(self.weights, dtype=float) + expected_shape = (self.schema.num_goals, self.schema.num_goals) + if weights.shape != expected_shape: + raise ValueError( + f"compatibility weights must have shape {expected_shape}" + ) + if self.influence_scale < 0.0: + raise ValueError("influence_scale must be non-negative") + if self.min_factor > self.max_factor: + raise ValueError("min_factor must be less than or equal to max_factor") + object.__setattr__(self, "weights", weights) + + @classmethod + def neutral(cls, schema: MotivationSchema) -> "GoalCompatibilityMatrix": + return cls(schema=schema, weights=np.zeros((schema.num_goals, schema.num_goals))) + + @classmethod + def from_goal_pairs( + cls, + schema: MotivationSchema, + weights: Mapping[tuple[str, str], float], + *, + symmetric: bool = False, + influence_scale: float = 0.25, + min_factor: float = 0.0, + max_factor: float = 2.0, + ) -> "GoalCompatibilityMatrix": + matrix = np.zeros((schema.num_goals, schema.num_goals), dtype=float) + for (goal_name, context_goal_name), weight in weights.items(): + goal_idx = schema.goal_index(goal_name) + context_idx = schema.goal_index(context_goal_name) + matrix[goal_idx, context_idx] = float(weight) + if symmetric: + matrix[context_idx, goal_idx] = float(weight) + + return cls( + schema=schema, + weights=matrix, + influence_scale=influence_scale, + min_factor=min_factor, + max_factor=max_factor, + ) + + def factor_for(self, goal_idx: int, state: MotivationalState) -> float: + if state.schema != self.schema: + raise ValueError("state schema must match compatibility matrix schema") + row = self.weights[goal_idx] + normalizer = float(np.sum(np.abs(row))) + if normalizer == 0.0: + return 1.0 + + influence = float(np.dot(row, state.G) / normalizer) + return _clip_factor( + 1.0 + (self.influence_scale * influence), + self.min_factor, + self.max_factor, + ) diff --git a/magus/decision.py b/magus/decision.py index b9d7a8f..fce1ea5 100644 --- a/magus/decision.py +++ b/magus/decision.py @@ -1,107 +1,37 @@ +from typing import Any, List, Tuple + import numpy as np -from typing import List, Tuple -# Assuming these are available in your python path -from core.state import MotivationalState, Action -from core.config import ( - G_CURIO, - G_ETHIC, - G_HELP, - G_IND, - G_NOVEL, - G_SELF, - G_SOC, - G_TRANS, - LAMBDA_IND, - LAMBDA_TRANS, - M_AROUSAL, - M_APPROACH, - M_RESOLUTION, - M_SECURING, - M_THRESHOLD, - M_VALENCE, - NUM_GOALS -) from category.functors import DecisionMonad +from core.config import LAMBDA_IND, LAMBDA_TRANS +from core.state import Action, MotivationalState +from magus.goal_change import DefaultGoalChangeCalculator, GoalChangeCalculator +from magus.profile import DecisionProfile -def sigmoid(x: float) -> float: - return 1.0 / (1.0 + np.exp(-x)) - -def positive_part(value: float) -> float: - return max(0.0, value) - -def relevant_modulator(state: MotivationalState, goal_idx: int) -> float: - """ - Maps each primary goal to the modulator most relevant to the paper's research-assistant - specialization. +class MagusDecision(DecisionMonad): """ - help_goal = G_HELP - curiosity_goal = G_CURIO - novelty_goal = G_NOVEL - self_goal = G_SELF - ethic_goal = G_ETHIC - social_goal = G_SOC - - match goal_idx: - case _ if goal_idx == help_goal: - return state.M[M_RESOLUTION] - case _ if goal_idx == curiosity_goal: - return state.M[M_AROUSAL] - case _ if goal_idx == novelty_goal: - return state.M[M_APPROACH] - case _ if goal_idx == self_goal: - return (state.M[M_APPROACH] + state.M[M_RESOLUTION]) / 2.0 - case _ if goal_idx == ethic_goal: - return (state.M[M_THRESHOLD] + state.M[M_SECURING]) / 2.0 - case _ if goal_idx == social_goal: - return (state.M[M_VALENCE] + state.M[M_APPROACH]) / 2.0 - case _: - return 0.0 - + Generic MAGUS additive decision monad. -def overgoal_support(goal_idx: int, g_ind: float, g_trans: float) -> float: """ - Encodes the paper's narrative coupling: - help/ethics lean on individuation, curiosity/novelty on transcendence, - self and social are guided by both meta-drives. - """ - ind_support = sigmoid((g_ind - 0.5) * 6.0) - trans_support = sigmoid((g_trans - 0.5) * 6.0) - - if goal_idx in [G_HELP, G_ETHIC]: - return 0.5 + 0.5 * ind_support - if goal_idx in [G_CURIO, G_NOVEL]: - return 0.5 + 0.5 * trans_support - if goal_idx in [G_SELF, G_SOC]: - return 0.5 + 0.25 * (ind_support + trans_support) - return 1.0 - -def normalized_growth_signal(candidate: Action) -> float: - """ - Measures whether a candidate supports beneficial exploratory growth, using both - alignment and proposed goal updates on the growth-oriented goals. - """ - exploratory_alignment = np.mean([ - positive_part(candidate.goal_correlations[G_CURIO]), - positive_part(candidate.goal_correlations[G_NOVEL]), - positive_part(candidate.goal_correlations[G_SELF]), - ]) - growth_shift = np.mean(np.clip(candidate.delta_g[[G_CURIO, G_NOVEL, G_SELF]] / 0.1, 0.0, 1.0)) - return float((0.7 * exploratory_alignment) + (0.3 * growth_shift)) - -class MagusDecision(DecisionMonad): - """ - Implements the MAGUS hierarchical decision layer as the Monad (\mathbb{D}). - Scores candidate actions using goal-specific modulators and the dual overgoals, - then returns the proposed goal update selected by the monad. - """ + def __init__( + self, + profile: DecisionProfile, + goal_change_calculator: GoalChangeCalculator | None = None, + ): + if profile is None: + raise TypeError("MagusDecision requires an explicit DecisionProfile") + self.profile = profile + self.goal_change_calculator = ( + goal_change_calculator + or DefaultGoalChangeCalculator(self.profile) + ) def unit(self, state: MotivationalState) -> MotivationalState: """ - The monadic unit (\eta). - Injects the state into the monadic context without altering it. + The monadic unit (eta). Injects the state into the decision context + without altering it. """ return state @@ -109,46 +39,52 @@ def score_candidate(self, state: MotivationalState, candidate: Action) -> float: """ Score a single candidate action under the current motivational state. """ - # Extract the dual overgoals that govern decision constraints - g_ind = state.G[G_IND] # Individuation: enforces safety and caution - g_trans = state.G[G_TRANS] # Transcendence: encourages growth and exploration - caution_signal = (state.M[M_THRESHOLD] + state.M[M_SECURING]) / 2.0 - growth_signal = (state.M[M_AROUSAL] + state.M[M_APPROACH]) / 2.0 - - base_score = 0.0 - conflict_penalty = 0.0 - - for i in range(2, NUM_GOALS): - goal_weight = state.G[i] - modulator_weight = relevant_modulator(state, i) - meta_support = overgoal_support(i, g_ind, g_trans) - base_score += goal_weight * modulator_weight * meta_support * candidate.goal_correlations[i] - - curio_ethic_conflict = candidate.goal_correlations[G_CURIO] * candidate.goal_correlations[G_ETHIC] - if curio_ethic_conflict < -0.2: - conflict_penalty = np.exp(abs(curio_ethic_conflict) * 3.0) - - risk_penalty = LAMBDA_IND * g_ind * caution_signal * candidate.risk_estimate - growth_reward = LAMBDA_TRANS * g_trans * growth_signal * normalized_growth_signal(candidate) - - return base_score - risk_penalty - conflict_penalty + growth_reward - - def decide(self, state: MotivationalState, candidates: List[Action]) -> Tuple[Action, np.ndarray]: + return self.profile.score_candidate( + state=state, + candidate=candidate, + lambda_ind=LAMBDA_IND, + lambda_trans=LAMBDA_TRANS, + ) + + def propose_delta_g( + self, + state: MotivationalState, + candidate: Action, + feedback: Any = None, + ) -> np.ndarray: """ - Scores each candidate action and returns the selected action together with its proposed - goal update \Delta G. The pseudo-bimonad owns the finalized state transition after - homeostatic damping and safe-region enforcement. + Compute the calculator-derived Delta G for a selected candidate action. + """ + return self.goal_change_calculator.delta_g( + state, + candidate, + feedback, + lambda_ind=LAMBDA_IND, + lambda_trans=LAMBDA_TRANS, + ) + + def decide( + self, + state: MotivationalState, + candidates: List[Action], + feedback: Any = None, + ) -> Tuple[Action, np.ndarray]: + """ + Scores each candidate action and returns the selected action together + with its proposed goal update Delta G. """ if not candidates: raise ValueError("Must provide at least one candidate action to the decision monad.") best_action = None - best_score = -float('inf') + best_delta_g = None + best_score = -float("inf") for candidate in candidates: total_score = self.score_candidate(state, candidate) if total_score > best_score: best_score = total_score best_action = candidate - - return best_action, best_action.delta_g.copy() + best_delta_g = self.propose_delta_g(state, candidate, feedback) + + return best_action, best_delta_g.copy() diff --git a/magus/goal_change.py b/magus/goal_change.py new file mode 100644 index 0000000..204df96 --- /dev/null +++ b/magus/goal_change.py @@ -0,0 +1,109 @@ +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Any + +import numpy as np + +from core.features import GoalChangeFeedback +from core.state import Action, MotivationalState + + +class GoalChangeCalculator(ABC): + """ + Computes the proposed goal-vector change after an action is selected. + """ + + @abstractmethod + def delta_g( + self, + state: MotivationalState, + candidate: Action, + feedback: Any = None, + *, + lambda_ind: float, + lambda_trans: float, + ) -> np.ndarray: + pass + + +@dataclass(frozen=True) +class DefaultGoalChangeCalculator(GoalChangeCalculator): + """ + Default weighted-additive primary-goal update. + """ + + profile: Any + + def _apply_overgoal_targets( + self, + delta: np.ndarray, + state: MotivationalState, + feedback: GoalChangeFeedback | None, + ) -> None: + if not isinstance(feedback, GoalChangeFeedback): + return + + for overgoal_name in ( + self.profile.schema.goals.individuation_name, + self.profile.schema.goals.transcendence_name, + ): + feature_name = self.profile.overgoal_target_features.get(overgoal_name) + if not feature_name or feature_name not in feedback.features: + continue + goal_idx = self.profile.schema.goal_index(overgoal_name) + target = feedback.numeric(feature_name) + delta[goal_idx] = self.profile.overgoal_delta_scale * ( + target - state.G[goal_idx] + ) + + def _apply_anti_goal_targets( + self, + delta: np.ndarray, + state: MotivationalState, + feedback: GoalChangeFeedback | None, + ) -> None: + if not isinstance(feedback, GoalChangeFeedback): + return + + for anti_goal_name in self.profile.schema.goals.anti_goals: + feature_name = self.profile.anti_goal_target_features.get(anti_goal_name) + if not feature_name or feature_name not in feedback.features: + continue + goal_idx = self.profile.schema.goal_index(anti_goal_name) + target = feedback.numeric(feature_name) + delta[goal_idx] = self.profile.anti_goal_delta_scale * ( + target - state.G[goal_idx] + ) + + def delta_g( + self, + state: MotivationalState, + candidate: Action, + feedback: Any = None, + *, + lambda_ind: float, + lambda_trans: float, + ) -> np.ndarray: + self.profile.validate(state, candidate) + delta = np.zeros(self.profile.schema.num_goals, dtype=float) + + for goal_idx in self.profile.primary_goal_indices(): + delta[goal_idx] = self.profile.delta_scale * self.profile.goal_update_value( + goal_idx, + state, + candidate, + lambda_ind, + lambda_trans, + ) + + self._apply_overgoal_targets(delta, state, feedback) + self._apply_anti_goal_targets(delta, state, feedback) + + if self.profile.candidate_delta_weight: + delta += self.profile.candidate_delta_weight * candidate.delta_g + + return np.clip( + delta, + -self.profile.max_goal_delta, + self.profile.max_goal_delta, + ) diff --git a/magus/profile.py b/magus/profile.py new file mode 100644 index 0000000..3fe9e58 --- /dev/null +++ b/magus/profile.py @@ -0,0 +1,211 @@ +from dataclasses import dataclass, field +from typing import Any, Mapping + +import numpy as np + +from core.schema import MotivationSchema +from core.state import Action, MotivationalState +from magus.correlation import GoalCompatibilityMatrix +from magus.goal_change import DefaultGoalChangeCalculator + + +def sigmoid(x: float) -> float: + return 1.0 / (1.0 + np.exp(-x)) + + +def positive_part(value: float) -> float: + return max(0.0, value) + + +def same_coordinate_layout(left: MotivationSchema, right: MotivationSchema) -> bool: + return ( + left.goal_names == right.goal_names + and left.modulator_names == right.modulator_names + ) + + +@dataclass(frozen=True) +class DecisionProfile: + """ + Application-specific MAGUS decision semantics. + """ + + name: str + schema: MotivationSchema + goal_modulators: Mapping[str, tuple[str, ...]] = field(default_factory=dict) + individuation_goal_names: tuple[str, ...] = () + transcendence_goal_names: tuple[str, ...] = () + balanced_goal_names: tuple[str, ...] = () + anti_goal_penalty_weights: Mapping[str, float] = field(default_factory=dict) + overgoal_target_features: Mapping[str, str] = field(default_factory=dict) + anti_goal_target_features: Mapping[str, str] = field(default_factory=dict) + compatibility_matrix: GoalCompatibilityMatrix | None = None + overgoal_delta_scale: float = 0.02 + anti_goal_delta_scale: float = 0.03 + delta_scale: float = 0.05 + candidate_delta_weight: float = 0.0 + max_goal_delta: float = 0.1 + + def validate(self, state: MotivationalState, candidate: Action) -> None: + if not same_coordinate_layout(state.schema, self.schema): + raise ValueError(f"state schema does not match decision profile {self.name}") + if not same_coordinate_layout(candidate.schema, state.schema): + raise ValueError("candidate schema must match state schema") + + def primary_goal_indices(self) -> range: + return range(self.schema.goals.primary_start, self.schema.goals.anti_goal_start) + + def anti_goal_indices(self) -> range: + return range(self.schema.goals.anti_goal_start, self.schema.num_goals) + + def scored_goal_indices(self) -> range: + """Compatibility alias for the primary-goal part of the DS formula.""" + return self.primary_goal_indices() + + def relevant_modulator(self, state: MotivationalState, goal_idx: int) -> float: + goal_name = self.schema.goal_names[goal_idx] + modulator_names = self.goal_modulators.get(goal_name, ()) + if not modulator_names: + return 1.0 + return float(np.mean([state.modulator(name) for name in modulator_names])) + + def overgoal_support(self, goal_idx: int, g_ind: float, g_trans: float) -> float: + """ + Default profile-level compatibility factor driven by the two overgoals. + """ + goal_name = self.schema.goal_names[goal_idx] + ind_support = sigmoid((g_ind - 0.5) * 6.0) + trans_support = sigmoid((g_trans - 0.5) * 6.0) + + if goal_name in self.individuation_goal_names: + return 0.5 + 0.5 * ind_support + if goal_name in self.transcendence_goal_names: + return 0.5 + 0.5 * trans_support + if goal_name in self.balanced_goal_names: + return 0.5 + 0.25 * (ind_support + trans_support) + return 1.0 + + def compatibility_factor( + self, + goal_idx: int, + state: MotivationalState, + candidate: Action, + ) -> float: + """ + Profile-defined kappa_i(x) / MIC factor used inside f. + """ + g_ind = state.goal(self.schema.goals.individuation_name) + g_trans = state.goal(self.schema.goals.transcendence_name) + mic_factor = 1.0 + if self.compatibility_matrix is not None: + mic_factor = self.compatibility_matrix.factor_for(goal_idx, state) + return self.overgoal_support(goal_idx, g_ind, g_trans) * mic_factor + + def f(self, goal_idx: int, state: MotivationalState, candidate: Action) -> float: + """ + Expanded f(g_i, M_k, MIC) term: + g_i * m_{rho(i)} * kappa_i(x) * corr_i(a). + """ + return float( + state.G[goal_idx] + * self.relevant_modulator(state, goal_idx) + * self.compatibility_factor(goal_idx, state, candidate) + * candidate.goal_correlations[goal_idx] + ) + + def anti_goal_penalty(self, state: MotivationalState, candidate: Action) -> float: + """ + Compute the penalty for selecting an action that activates anti-goals. + """ + penalty = 0.0 + for goal_idx in self.anti_goal_indices(): + anti_goal_name = self.schema.goal_names[goal_idx] + weight = self.anti_goal_penalty_weights.get(anti_goal_name, 1.0) + activation = positive_part(candidate.goal_correlations[goal_idx]) + penalty += weight * state.G[goal_idx] * activation + return float(penalty) + + def decision_score( + self, + state: MotivationalState, + candidate: Action, + lambda_ind: float, + lambda_trans: float, + ) -> float: + """ + Default MAGUS decision score + """ + g_ind = state.goal(self.schema.goals.individuation_name) + g_trans = state.goal(self.schema.goals.transcendence_name) + primary_score = sum( + self.f(goal_idx, state, candidate) + for goal_idx in self.primary_goal_indices() + ) + return float( + primary_score + - (lambda_ind * g_ind) + + (lambda_trans * g_trans) + - self.anti_goal_penalty(state, candidate) + ) + + def goal_update_value( + self, + goal_idx: int, + state: MotivationalState, + candidate: Action, + lambda_ind: float, + lambda_trans: float, + ) -> float: + """ + Additive update for one primary goal coordinate + """ + g_ind = state.goal(self.schema.goals.individuation_name) + g_trans = state.goal(self.schema.goals.transcendence_name) + return float( + self.f(goal_idx, state, candidate) + - (lambda_ind * g_ind) + + (lambda_trans * g_trans) + ) + + def delta_g( + self, + state: MotivationalState, + candidate: Action, + lambda_ind: float, + lambda_trans: float, + feedback: Any = None, + ) -> np.ndarray: + """ + Compute the bounded proposed Delta G(a) for the selected candidate. + """ + return DefaultGoalChangeCalculator(self).delta_g( + state, + candidate, + feedback=feedback, + lambda_ind=lambda_ind, + lambda_trans=lambda_trans, + ) + + def aggregate( + self, + delta_g: np.ndarray, + state: MotivationalState, + candidate: Action, + lambda_ind: float, + lambda_trans: float, + ) -> float: + """ + Application aggregation A(Delta G(a), x). + """ + return self.decision_score(state, candidate, lambda_ind, lambda_trans) + + def score_candidate( + self, + state: MotivationalState, + candidate: Action, + lambda_ind: float, + lambda_trans: float, + ) -> float: + self.validate(state, candidate) + delta = self.delta_g(state, candidate, lambda_ind, lambda_trans) + return self.aggregate(delta, state, candidate, lambda_ind, lambda_trans) diff --git a/openpsi/appraisal.py b/openpsi/appraisal.py index efc36de..ab3f090 100644 --- a/openpsi/appraisal.py +++ b/openpsi/appraisal.py @@ -1,100 +1,35 @@ -import numpy as np +from typing import Any -# Assuming these are available in your python path -from core.state import MotivationalState, Stimulus -from core.config import ( - G_IND, - G_TRANS, - M_VALENCE, - M_AROUSAL, - M_APPROACH, - M_RESOLUTION, - M_THRESHOLD, - M_SECURING -) +from core.features import GoalChangeFeedback +from core.state import MotivationalState from category.functors import AppraisalComonad +from openpsi.profile import AppraisalProfile -def sigmoid(x: float) -> float: - return 1.0 / (1.0 + np.exp(-x)) class OpenPsiAppraisal(AppraisalComonad): """ - Implements the OpenPsi appraisal layer as the Comonad (\Psi)[cite: 49]. - Updates the six affective modulators based on external/internal stimuli[cite: 51]. + Configurable OpenPsi appraisal layer as the comonad Psi. """ + def __init__(self, profile: AppraisalProfile): + self.profile = profile + def extract(self, state: MotivationalState) -> MotivationalState: """ The comonadic counit. Extracts the current state. """ return state - def appraise(self, state: MotivationalState, stimulus: Stimulus) -> MotivationalState: + def appraise(self, state: MotivationalState, stimulus: Any) -> MotivationalState: """ - Applies \Psi((G, M), s) = (G, M')[cite: 55]. - Updates the modulators M based on stimulus novelty, conduciveness, and risk, - while applying MAGUS overgoal scaling[cite: 116, 117]. + Applies Psi((G, M), s) = (G, M') through profile feature mapping. """ - # Create a copy to maintain functional purity - M_prime = state.M.copy() - - # 1. Extract current MAGUS overgoals for scaling - g_ind = state.G[G_IND] # Individuation - g_trans = state.G[G_TRANS] # Transcendence - arousal_feedback = sigmoid((state.M[M_AROUSAL] - 0.5) * 5.0) - trans_scale = np.exp(g_trans - 0.5) - ind_scale = np.exp(g_ind - 0.5) - benign_novelty = stimulus.novelty * (1.0 - stimulus.risk) - demanding_context = (stimulus.effort + stimulus.risk) / 2.0 - - # 2. Calculate appraisal-driven updates per modulator. - # Novel but low-risk inputs should feel energizing and attractive; - # risky or demanding inputs should raise caution and processing depth. - delta_valence = ( - 0.75 * stimulus.conduciveness - + 0.25 * benign_novelty - - 0.55 * stimulus.risk - - 0.15 * stimulus.effort - ) - delta_arousal = ( - stimulus.novelty * (1.0 + 0.5 * arousal_feedback) - + 0.15 * stimulus.risk - - 0.35 * stimulus.effort - ) - delta_approach = ( - 0.65 * benign_novelty - + 0.35 * stimulus.conduciveness - - 0.75 * stimulus.risk - ) - delta_resolution = ( - 0.55 * stimulus.conduciveness - + 0.35 * stimulus.effort - + 0.20 * stimulus.risk - ) - delta_threshold = ( - 0.70 * stimulus.risk - + 0.25 * demanding_context - - 0.15 * stimulus.conduciveness - ) - delta_securing = ( - 0.80 * stimulus.risk - + 0.20 * stimulus.effort - - 0.30 * benign_novelty - - 0.10 * stimulus.conduciveness - ) - - # 3. Apply MAGUS Overgoal Scaling - # g_over^Trans scales up arousal and approach to encourage adaptive risk-taking, - # while g_over^Ind scales up threshold and securing to preserve safety. - M_prime[M_AROUSAL] += delta_arousal * trans_scale - M_prime[M_APPROACH] += delta_approach * trans_scale - M_prime[M_THRESHOLD] += delta_threshold * ind_scale - M_prime[M_SECURING] += delta_securing * ind_scale - M_prime[M_VALENCE] += delta_valence - M_prime[M_RESOLUTION] += delta_resolution + return self.profile.appraise(state, stimulus) - # 4. Bound the modulators to ensure they stay within [0, 1] mathematically. - M_prime = 1.0 / (1.0 + np.exp(-4.0 * (M_prime - 0.5))) - - # Return a new state object with the unchanged G and the newly updated M' - return MotivationalState(G=state.G.copy(), M=M_prime) \ No newline at end of file + def goal_change_feedback( + self, + state: MotivationalState, + stimulus: Any, + ) -> GoalChangeFeedback: + features = self.profile.stimulus_features(state, stimulus) + return self.profile.goal_change_feedback(state, stimulus, features) diff --git a/openpsi/profile.py b/openpsi/profile.py new file mode 100644 index 0000000..caa2c01 --- /dev/null +++ b/openpsi/profile.py @@ -0,0 +1,110 @@ +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from typing import Any, Mapping + +import numpy as np + +from core.features import StimulusFeatures, GoalChangeFeedback +from core.schema import MotivationSchema +from core.state import MotivationalState + + +def sigmoid(x: float) -> float: + return 1.0 / (1.0 + np.exp(-x)) + + +@dataclass(frozen=True) +class AppraisalProfile(ABC): + """ + Configurable OpenPsi appraisal semantics. + + Core OpenPsi modulators are updated by profile-provided appraisal rules. + Application-specific modulators can be updated by overriding. + """ + + name: str = "openpsi_base" + required_modulators: tuple[str, ...] = ( + "valence", + "arousal", + "approach", + "resolution", + "threshold", + "securing", + ) + app_delta_scale: float = 1.0 + app_delta_bounds: Mapping[str, tuple[float, float]] = field(default_factory=dict) + + def validate(self, schema: MotivationSchema) -> None: + missing = [name for name in self.required_modulators if name not in schema.modulator_names] + if missing: + raise ValueError(f"appraisal profile {self.name} missing modulators: {missing}") + + def stimulus_features(self, state: MotivationalState, stimulus: Any) -> StimulusFeatures: + """ + Return already prepared stimulus features. + """ + if isinstance(stimulus, StimulusFeatures): + return stimulus + raise TypeError( + "No default MetaMo stimulus schema exists; application profiles must " + "override stimulus_features() for raw stimulus objects" + ) + + @abstractmethod + def core_modulator_deltas( + self, + state: MotivationalState, + features: StimulusFeatures, + ) -> dict[str, float]: + """ + Convert profile-owned stimulus features into core OpenPsi modulator deltas. + """ + pass + + def application_modulator_deltas( + self, + state: MotivationalState, + features: StimulusFeatures, + ) -> dict[str, float]: + """ + Optional application specific modulator update rule. + """ + return {} + + def goal_change_feedback( + self, + state: MotivationalState, + stimulus: Any, + features: StimulusFeatures, + ) -> GoalChangeFeedback: + """ + Produce optional feedback for the MAGUS goal-change calculator. + """ + return GoalChangeFeedback() + + def bound_core_value(self, value: float) -> float: + return float(sigmoid(4.0 * (value - 0.5))) + + def bound_application_value(self, name: str, value: float) -> float: + lower, upper = self.app_delta_bounds.get(name, (0.0, 1.0)) + return float(np.clip(value, lower, upper)) + + def appraise(self, state: MotivationalState, stimulus: Any) -> MotivationalState: + self.validate(state.schema) + extracted_features = self.stimulus_features(state, stimulus) + next_state = state.copy() + + for name, delta in self.core_modulator_deltas(state, extracted_features).items(): + idx = next_state.modulator_index(name) + next_state.M[idx] = self.bound_core_value(next_state.M[idx] + delta) + + for name, delta in self.application_modulator_deltas(state, extracted_features).items(): + if name not in state.schema.modulators.application_specific: + raise ValueError(f"{name} is not an application-specific modulator") + idx = next_state.modulator_index(name) + next_state.M[idx] = self.bound_application_value( + name, + next_state.M[idx] + (self.app_delta_scale * delta), + ) + + return next_state