limbic_system

Limbic System: shard-based respiration model with emotion classification.

Manages per-channel (local shard) and global (heart) neurochemical state in Redis DB12 using an inhale/exhale respiration cycle.

v3 update: - Expanded baseline to full NCM node set from ncm_limbic_index.yaml - Emotion classification from accumulated chemical vector - Per-turn metabolic decay toward baseline - Context injection formatting for prompt recency bias

class limbic_system.LimbicSystem(redis_client=None, openrouter_api_key=None, openrouter_client=None, cache_redis_client=None)

Bases: object

Shard-based limbic respiration backed by Redis DB12.

Parameters:

• redis_client – An redis.asyncio.Redis instance pointed at DB12.

• openrouter_api_key (Optional[str]) – When provided, a CueVariantCache is constructed and wired into the cascade engine so that cascade cue and reason strings are gradually replaced with LLM-generated variants (cached permanently in Redis).

__init__(redis_client=None, openrouter_api_key=None, openrouter_client=None, cache_redis_client=None)

Initialize the instance.

Parameters:

• redis_client – Redis connection client (DB12 for limbic shards).

• openrouter_api_key (Optional[str]) – The openrouter api key value.

• openrouter_client – Shared OpenRouterClient for connection pooling. Passed through to SemanticTriggerMatcher.

• cache_redis_client – Redis client for NCM caches (DB0). When None, falls back to redis_client (DB12).

async inhale(channel_id)

RESPIRATION PHASE 1: INHALE

Fetches local shard + global heart, mixes via osmosis, runs homeostatic regulation, returns the effective state.

Return type:

Dict[str, Any]

Parameters:

channel_id (str)

async exhale(channel_id, stimulus_delta, apply_decay=True, user_message='', star_reply='', user_id='')

RESPIRATION PHASE 2: EXHALE

Updates local shard with new delta, applies metabolic decay, and pulses global heart.

Return type:

Dict[str, Any]

Parameters:

• channel_id (str)

• stimulus_delta (Dict[str, float])

• apply_decay (bool)

• user_message (str)

• star_reply (str)

• user_id (str)

async scan_triggers(text)

Scan text for emotional triggers using semantic matching.

Uses SemanticTriggerMatcher when available and loaded; falls back transparently to the exact-word scan_text_for_triggers from ncm_delta_parser.

Returns a list of (emotion_name, delta_vector) tuples.

Return type:

List[tuple]

Parameters:

text (str)

static classify_dominant_emotions(vector, top_n=3)

Score all emotions against the current vector and return top N.

Uses the delta vectors from the recursion index as templates. Scoring is a weighted dot product: how closely the current chemical state matches each emotion’s delta profile.

Return type:

List[Dict[str, Any]]

Parameters:

• vector (Dict[str, float])

• top_n (int)

static metabolic_decay(vector, tau=0.95, baseline=None)

Exponential decay toward baseline.

V(t+1) = V(t) * tau + baseline * (1 - tau)

At tau=0.95: - After 5 turns: ~77% of delta remains - After 14 turns: ~50% remains - After 45 turns: ~10% remains

Return type:

Dict[str, float]

Parameters:

• vector (Dict[str, float])

• tau (float)

• baseline (Dict[str, float] | None)

static format_context_injection(vector, cues, dominant, cascade_cues=None, rdf_output=None, user_read='', self_reflection=None)

Format the limbic state for injection into the system prompt.

Produces a compact dict that goes near the end of the prompt for maximum recency bias influence.

Return type:

Dict[str, Any]

Parameters:

• vector (Dict[str, float])

• cues (List[str])

• dominant (List[Dict[str, Any]])

• cascade_cues (List[str] | None)

• rdf_output (Dict[str, Any] | None)

• user_read (str)

• self_reflection (Dict[str, Any] | None)