Appraisal

See also: Appraisal Schema — plug in custom theories (OCC, GRID, domain-specific).

Protocols and Data Types

emotional_memory.appraisal.AppraisalVector

Bases: BaseModel

Multi-component appraisal of an event (Scherer CPM).

to_core_affect() -> CoreAffect

Map appraisal dimensions to valence-arousal space.

Delegates to the Scherer CPM schema's projection (SCHERER_CPM_SCHEMA.project_to_core_affect) so the recalibrated mapping weights live in a single place (Addendum O). See appraisal_schema._scherer_project for the formula and provenance.

Source code in src/emotional_memory/appraisal.py

def to_core_affect(self) -> CoreAffect:
    """Map appraisal dimensions to valence-arousal space.

    Delegates to the Scherer CPM schema's projection
    (``SCHERER_CPM_SCHEMA.project_to_core_affect``) so the recalibrated mapping
    weights live in a single place (Addendum O). See
    ``appraisal_schema._scherer_project`` for the formula and provenance.
    """
    return SCHERER_CPM_SCHEMA.project_to_core_affect(
        {
            "novelty": self.novelty,
            "goal_relevance": self.goal_relevance,
            "coping_potential": self.coping_potential,
            "norm_congruence": self.norm_congruence,
            "self_relevance": self.self_relevance,
        }
    )

emotional_memory.appraisal.AppraisalEngine

Bases: Protocol

Protocol for computing an appraisal vector from an event description.

Implementations may call an LLM, use rule-based heuristics, or any other strategy. The library ships LLMAppraisalEngine (LLM-backed), KeywordAppraisalEngine (rule-based), and StaticAppraisalEngine (fixed vector, for testing).

Return type is AppraisalVector | GenericAppraisalVector: AppraisalVector for the default Scherer CPM schema, GenericAppraisalVector when a custom AppraisalSchema is used.

emotional_memory.appraisal.StaticAppraisalEngine(vector: AppraisalVector | None = None)

Returns a fixed AppraisalVector regardless of input. For testing.

Source code in src/emotional_memory/appraisal.py

def __init__(self, vector: AppraisalVector | None = None) -> None:
    self._vector = vector or AppraisalVector.neutral()

emotional_memory.appraisal.consolidation_strength(arousal: float, mood_arousal: float) -> float

Arousal-modulated consolidation strength (Yerkes-Dodson inverted-U).

Effective arousal blends encoding arousal (70%) with background mood arousal (30%). Peak consolidation near effective_arousal=0.7.

A floor of 0.1 ensures even calm, mundane memories have a minimum consolidation chance — otherwise events with effective_arousal < 0.2 would get zero strength and be immediately prunable.

Returns a value in [0.1, 1.0].

Source code in src/emotional_memory/appraisal.py

def consolidation_strength(arousal: float, mood_arousal: float) -> float:
    """Arousal-modulated consolidation strength (Yerkes-Dodson inverted-U).

    Effective arousal blends encoding arousal (70%) with background
    mood arousal (30%). Peak consolidation near effective_arousal=0.7.

    A floor of 0.1 ensures even calm, mundane memories have a minimum
    consolidation chance — otherwise events with effective_arousal < 0.2
    would get zero strength and be immediately prunable.

    Returns a value in [0.1, 1.0].
    """
    effective = 0.7 * max(0.0, min(1.0, arousal)) + 0.3 * max(0.0, min(1.0, mood_arousal))
    # Inverted parabola with peak=1.0 at x=0.7
    raw = -4.0 * (effective - 0.7) ** 2 + 1.0
    return max(0.1, min(1.0, raw))

LLM-backed Engine

emotional_memory.appraisal_llm.LLMAppraisalEngine(llm: LLMCallable, config: LLMAppraisalConfig | None = None, fallback: AppraisalVector | None = None)

AppraisalEngine backed by any LLM via a provider-agnostic callable.

Thread-safety: the LRU cache is protected by a threading.Lock. Concurrent appraise() calls with the same key will each call the LLM independently (no stampede protection), but the cache dict is always mutated under the lock.

Source code in src/emotional_memory/appraisal_llm.py

def __init__(
    self,
    llm: LLMCallable,
    config: LLMAppraisalConfig | None = None,
    fallback: AppraisalVector | None = None,
) -> None:
    self._llm = llm
    self._config = config or LLMAppraisalConfig()
    self._schema: AppraisalSchema = self._config.appraisal_schema or SCHERER_CPM_SCHEMA
    self._fallback = fallback or AppraisalVector.neutral()
    # Count of appraisals that returned the neutral fallback (LLM/parse error).
    # Lets benchmark harnesses detect silent degradation without disabling the
    # fallback; guarded by ``_cache_lock``.
    self._fallback_count = 0
    self._cache: OrderedDict[str, AppraisalVector | GenericAppraisalVector] = OrderedDict()
    self._cache_lock = threading.Lock()

    # Effective prompt: user-set system_prompt in config overrides schema default.
    # Detect via model_fields_set so the schema prompt wins when config is default.
    if "system_prompt" in self._config.model_fields_set:
        self._effective_prompt: str = self._config.system_prompt
    else:
        self._effective_prompt = self._schema.system_prompt

    self._json_schema: dict[str, Any] = self._schema.to_json_schema()

fallback_count: int property

Number of appraisals that returned the neutral fallback (LLM/parse error).

A non-zero value after a measurement run signals silent degradation: some results are neutral fallbacks, not genuine appraisals. Reset with reset_fallback_count().

appraise(event_text: str, context: dict[str, Any] | None = None) -> AppraisalVector | GenericAppraisalVector

Appraise event_text via the LLM.

Returns an AppraisalVector for the default Scherer CPM schema or a GenericAppraisalVector when a custom AppraisalSchema is configured.

Results are cached by (schema, event_text, context) key. On error, if fallback_on_error is True, returns the neutral fallback vector instead of propagating the exception.

Source code in src/emotional_memory/appraisal_llm.py

def appraise(
    self, event_text: str, context: dict[str, Any] | None = None
) -> AppraisalVector | GenericAppraisalVector:
    """Appraise *event_text* via the LLM.

    Returns an ``AppraisalVector`` for the default Scherer CPM schema or a
    ``GenericAppraisalVector`` when a custom ``AppraisalSchema`` is configured.

    Results are cached by (schema, event_text, context) key.  On error, if
    ``fallback_on_error`` is True, returns the neutral fallback vector
    instead of propagating the exception.
    """
    cache_key = self._make_cache_key(event_text, context, self._schema.name)

    if self._config.cache_size > 0:
        with self._cache_lock:
            if cache_key in self._cache:
                self._cache.move_to_end(cache_key)
                logger.debug("appraise cache hit: key=%s", cache_key[:8])
                return self._cache[cache_key]

    prompt = self._build_prompt(event_text, context)
    vector: AppraisalVector | GenericAppraisalVector
    try:
        raw = self._llm(prompt, self._json_schema)
    except Exception as exc:
        if self._config.fallback_on_error:
            logger.warning("appraise: LLM call failed, using fallback: %s", exc)
            vector = self._fallback
            self._record_fallback()
        else:
            raise
    else:
        try:
            data = self._extract_json(raw)
            missing = [d.name for d in self._schema.dimensions if d.name not in data]
            if missing:
                # Do NOT silently neutral-fill absent dimensions: a partial
                # response is a degraded result, not a genuine neutral one.
                raise ValueError(f"LLM response missing required dimensions: {missing}")
            coerced = coerce_appraisal_dimensions(data, self._schema)
            if self._schema is SCHERER_CPM_SCHEMA:
                vector = AppraisalVector(**coerced)
            else:
                vector = GenericAppraisalVector(dimensions=coerced, schema=self._schema)
        except Exception as exc:
            if self._config.fallback_on_error:
                # Warn, not debug: a requested appraisal was silently replaced
                # with the neutral fallback. Leaving this at debug hid the
                # Addendum X2 silent-corruption for a whole measurement run.
                logger.warning(
                    "appraise: parse/validation error, using neutral fallback "
                    "(text_len=%d): %s",
                    len(event_text),
                    exc,
                    exc_info=True,
                )
                vector = self._fallback
                self._record_fallback()
            else:
                raise

    # Do not cache error fallbacks: a transient LLM/parse failure would
    # otherwise pin neutral for that key until eviction, masking recovery.
    if self._config.cache_size > 0 and vector is not self._fallback:
        with self._cache_lock:
            if len(self._cache) >= self._config.cache_size:
                self._cache.popitem(last=False)
            self._cache[cache_key] = vector

    return vector

clear_cache() -> None

Evict all cached appraisals.

Source code in src/emotional_memory/appraisal_llm.py

def clear_cache(self) -> None:
    """Evict all cached appraisals."""
    with self._cache_lock:
        self._cache.clear()

close() -> None

Release resources held by the LLM callable, if supported.

Delegates to llm.close() when the callable exposes it (e.g. the httpx-backed client from make_httpx_llm). Safe on callables without cleanup.

Source code in src/emotional_memory/appraisal_llm.py

def close(self) -> None:
    """Release resources held by the LLM callable, if supported.

    Delegates to ``llm.close()`` when the callable exposes it (e.g. the
    httpx-backed client from ``make_httpx_llm``). Safe on callables
    without cleanup.
    """
    close = getattr(self._llm, "close", None)
    if callable(close):
        close()

reset_fallback_count() -> None

Reset the fallback counter to zero.

Source code in src/emotional_memory/appraisal_llm.py

def reset_fallback_count(self) -> None:
    """Reset the fallback counter to zero."""
    with self._cache_lock:
        self._fallback_count = 0

emotional_memory.appraisal_llm.LLMAppraisalConfig

Bases: BaseModel

Configuration for LLMAppraisalEngine.

appraisal_schema: AppraisalSchema | None = None class-attribute instance-attribute

Appraisal schema to use. None defaults to SCHERER_CPM_SCHEMA.

cache_size: int = 128 class-attribute instance-attribute

Number of (text, context) pairs to cache. 0 disables the cache.

fallback_on_error: bool = True class-attribute instance-attribute

Return a neutral AppraisalVector instead of raising on LLM/parse errors.

Set to False in measurement/benchmark code that must not silently ingest a neutral fallback as if it were a genuine appraisal (the failure would be indistinguishable from a real neutral result). When True (default), the engine still records every fallback in LLMAppraisalEngine.fallback_count so aggregators can detect and exclude degraded runs without aborting mid-batch.

system_prompt: str = SCHERER_CPM_SCHEMA.system_prompt class-attribute instance-attribute

System prompt describing the appraisal task.

Defaults to the Scherer CPM schema prompt (single source of truth). When appraisal_schema is set and this field is left at its default, the engine uses appraisal_schema.system_prompt instead. Explicitly setting this field always takes precedence.

emotional_memory.appraisal_llm.LLMCallable

Bases: Protocol

Provider-agnostic LLM interface.

The user wraps their SDK of choice (openai, anthropic, local GGUF, …) in a callable matching this signature. The json_schema parameter is passed for informational purposes — structured-output enforcement is the caller's responsibility if the backend supports it.

__call__(prompt: str, json_schema: dict[str, Any]) -> str

Send prompt to the LLM, return a raw JSON string.

Source code in src/emotional_memory/appraisal_llm.py

def __call__(self, prompt: str, json_schema: dict[str, Any]) -> str:
    """Send ``prompt`` to the LLM, return a raw JSON string."""
    ...

Keyword-based Engine

emotional_memory.appraisal_llm.KeywordAppraisalEngine(rules: list[KeywordRule] | None = None)

Rule-based AppraisalEngine using keyword pattern matching.

Each rule contributes deltas from neutral baselines. After accumulation the results are averaged (when multiple rules fire) and the coping_potential baseline of 0.5 is re-applied. AppraisalVector validators clamp to range.

Scherer-5 specific: always returns an AppraisalVector and therefore does not honor a custom AppraisalSchema (e.g. DIRECT_VAD_SCHEMA).

Source code in src/emotional_memory/appraisal_llm.py

def __init__(self, rules: list[KeywordRule] | None = None) -> None:
    self._rules = rules if rules is not None else _DEFAULT_RULES

make_multilingual() -> KeywordAppraisalEngine classmethod

Return an engine with EN + IT keyword rules.

FR/ES rely on me5 embeddings; no language-specific keywords.

Source code in src/emotional_memory/appraisal_llm.py

@classmethod
def make_multilingual(cls) -> KeywordAppraisalEngine:
    """Return an engine with EN + IT keyword rules.

    FR/ES rely on me5 embeddings; no language-specific keywords.
    """
    return cls(rules=_DEFAULT_RULES + _ITALIAN_RULES)

emotional_memory.appraisal_llm.KeywordRule(pattern: str, novelty: float = 0.0, goal_relevance: float = 0.0, coping_potential: float = 0.0, norm_congruence: float = 0.0, self_relevance: float = 0.0)

A single keyword pattern contributing dimension score deltas.

All scores are deltas from the neutral baseline (0.0 for signed dims, 0.0 for coping_potential/self_relevance — the engine adds the baseline 0.5 for coping_potential after accumulation).

Source code in src/emotional_memory/appraisal_llm.py

def __init__(
    self,
    pattern: str,
    novelty: float = 0.0,
    goal_relevance: float = 0.0,
    coping_potential: float = 0.0,
    norm_congruence: float = 0.0,
    self_relevance: float = 0.0,
) -> None:
    self._re = re.compile(pattern, re.IGNORECASE)
    self.scores = {
        "novelty": novelty,
        "goal_relevance": goal_relevance,
        "coping_potential": coping_potential,
        "norm_congruence": norm_congruence,
        "self_relevance": self_relevance,
    }