"""Read-only, replaceable language model backend contracts. This module defines the core protocols and data types for the Language Organ, a controlled, replaceable interface between Brain-5D and external language models. The Language Organ is designed as a read-only adapter that: - Translates between symbolic text and subsymbolic SignalFrames - Never mutates Brain-5D state directly - Is replaceable (NullBackend, LlamaCppBackend, etc.) - Is fault-tolerant (failures do not stop the simulation) Design Principles: 1. The Language Organ is an optional component, not a core dependency. 2. It operates asynchronously and does not block the simulation loop. 3. It has no direct access to synapse weights or network structure. 4. All outputs are treated as data, never as runtime commands. Example: >>> from src.language_organ import NullBackend, LanguageRequest >>> backend = NullBackend() >>> request = LanguageRequest( ... request_id="req_001", ... purpose="translate", ... text="Hello, world!" ... ) >>> response = backend.infer(request) >>> print(response.text) Hello, world! """ from __future__ import annotations from dataclasses import asdict, dataclass from enum import StrEnum from typing import Any, Protocol from src.signal_processing.models import SignalFrame # ============================================================================ # Request / Response Models # ============================================================================ class LanguageOrganMode(StrEnum): """Experimental treatment applied by the language organ.""" LINGUISTIC_TRANSPORT = "LINGUISTIC_TRANSPORT" SEMANTIC_AUGMENTATION = "SEMANTIC_AUGMENTATION" @dataclass(frozen=True, slots=True) class LanguageRequest: """A bounded request to the optional language organ. This represents a self-contained, immutable request for language processing. It is intentionally limited to text and optional SignalFrame data to prevent any mutation of Brain-5D state. Attributes: request_id: Unique identifier for this request (for traceability). purpose: The purpose of the request (e.g., 'translate', 'monitor', 'encode', 'decode', 'describe'). text: The input text to process. signal_frame: Optional SignalFrame for context-aware processing. """ request_id: str purpose: str text: str signal_frame: SignalFrame | None = None mode: LanguageOrganMode = LanguageOrganMode.LINGUISTIC_TRANSPORT def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" # Explicitly annotate result as dict[str, Any] to allow mixed types result: dict[str, Any] = { "request_id": self.request_id, "purpose": self.purpose, "text": self.text, "mode": self.mode.value, } if self.signal_frame is not None: result["signal_frame"] = asdict(self.signal_frame) return result @dataclass(frozen=True, slots=True) class LanguageResponse: """Language-organ result. It is data, never a runtime command. This response contains only text and metadata. It must never contain commands that could mutate the Brain-5D network state. Attributes: request_id: ID of the original request (for traceability). text: The response text generated by the language model. backend_name: Name of the backend that generated the response. success: Whether the inference was successful. error: Error message if success is False, otherwise None. """ request_id: str text: str backend_name: str success: bool error: str | None = None def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" # Explicitly annotate result as dict[str, Any] to allow mixed types result: dict[str, Any] = { "request_id": self.request_id, "text": self.text, "backend_name": self.backend_name, "success": self.success, } if self.error is not None: result["error"] = self.error return result # ============================================================================ # Backend Protocol # ============================================================================ class LanguageModelBackend(Protocol): """Backend contract that deliberately exposes no Brain-5D runtime object. This protocol defines the minimal interface that any language model backend must implement. It is designed to be: 1. Replaceable – Backends can be swapped at runtime (NullBackend, LlamaCppBackend, etc.) 2. Read-only – No Brain-5D state mutation 3. Fault-tolerant – Errors are returned in the response, not raised 4. Asynchronous-capable – Inference should be non-blocking where possible The backend is intentionally decoupled from the Brain-5D runtime to prevent accidental mutation of network state. All communication is through immutable request/response objects. Example: >>> class MyBackend(LanguageModelBackend): ... @property ... def name(self) -> str: ... return "my_backend" ... def infer(self, request: LanguageRequest) -> LanguageResponse: ... return LanguageResponse( ... request_id=request.request_id, ... text=f"Processed: {request.text}", ... backend_name=self.name, ... success=True, ... ) """ @property def name(self) -> str: """Return the backend identifier. This should be a short, descriptive string that identifies the backend implementation (e.g., 'null', 'qwen_0.5b', 'granite_nano'). """ ... def infer(self, request: LanguageRequest) -> LanguageResponse: """Process a language request and return text data only. This method must never mutate Brain-5D state. It receives a self-contained request and returns a self-contained response. Args: request: The language request to process. Returns: A LanguageResponse containing the result or error. Raises: This method should never raise exceptions. All errors should be captured and returned in the response with success=False. """ ... # ============================================================================ # Module Exports # ============================================================================ __all__ = [ "LanguageModelBackend", "LanguageOrganMode", "LanguageRequest", "LanguageResponse", ]