EmotionalMemory (sync)

emotional_memory.engine.EmotionalMemory(store: MemoryStore, embedder: Embedder, appraisal_engine: AppraisalEngine | None = None, config: EmotionalMemoryConfig | None = None, state_store: AffectiveStateStore | None = None, query_classifier: QueryClassifier | None = None)

Emotional memory system for LLMs based on Affective Field Theory.

The sync engine is intended for single-threaded use; concurrent encode() / observe() from multiple threads can interleave affective-state updates. Use :class:AsyncEmotionalMemory for concurrent async workloads.

Usage::

store = InMemoryStore()
embedder = MyEmbedder()          # implements Embedder protocol
engine = EmotionalMemory(store, embedder)

# Encode with automatic appraisal (if engine provided) or manual affect
engine.set_affect(CoreAffect(valence=0.8, arousal=0.6))
memory = engine.encode("Just completed a difficult project successfully!")

# Retrieve — emotionally weighted by current state
results = engine.retrieve("challenging work accomplishment")

Source code in src/emotional_memory/engine.py

def __init__(
    self,
    store: MemoryStore,
    embedder: Embedder,
    appraisal_engine: AppraisalEngine | None = None,
    config: EmotionalMemoryConfig | None = None,
    state_store: AffectiveStateStore | None = None,
    query_classifier: QueryClassifier | None = None,
) -> None:
    self._store = store
    self._embedder = embedder
    self._appraisal_engine = appraisal_engine
    self._config = config or EmotionalMemoryConfig()
    self._state_store = state_store
    self._state = self._load_initial_state()
    qc_cfg = self._config.retrieval.query_classifier
    if qc_cfg is not None and qc_cfg.mode != "disabled":
        if qc_cfg.mode == "heuristic":
            self._query_classifier: QueryClassifier | None = HeuristicQueryClassifier()
        else:
            self._query_classifier = query_classifier
    else:
        self._query_classifier = None

__len__() -> int

Return the number of memories in the store.

Source code in src/emotional_memory/engine.py

def __len__(self) -> int:
    """Return the number of memories in the store."""
    return len(self._store)

clear_persisted_state() -> None

Remove the persisted affective-state snapshot, if configured.

Source code in src/emotional_memory/engine.py

def clear_persisted_state(self) -> None:
    """Remove the persisted affective-state snapshot, if configured."""
    if self._state_store is not None:
        self._state_store.clear()

close() -> None

Release resources held by store, state store, and appraisal engine.

Calls close() on each collaborator that exposes it (e.g. SQLiteStore, LLMAppraisalEngine with an httpx-backed LLM). Safe on collaborators without cleanup. The embedder is not owned by the engine and remains the caller's responsibility.

Source code in src/emotional_memory/engine.py

def close(self) -> None:
    """Release resources held by store, state store, and appraisal engine.

    Calls ``close()`` on each collaborator that exposes it (e.g.
    ``SQLiteStore``, ``LLMAppraisalEngine`` with an httpx-backed LLM).
    Safe on collaborators without cleanup. The embedder is not owned by
    the engine and remains the caller's responsibility.
    """
    close = getattr(self._store, "close", None)
    if callable(close):
        close()
    state_close = getattr(self._state_store, "close", None)
    if callable(state_close):
        state_close()
    appraisal_close = getattr(self._appraisal_engine, "close", None)
    if callable(appraisal_close):
        appraisal_close()

delete(memory_id: str) -> None

Remove a memory from the store.

Source code in src/emotional_memory/engine.py

def delete(self, memory_id: str) -> None:
    """Remove a memory from the store."""
    self._store.delete(memory_id)

elaborate(memory_id: str) -> Memory | None

Run full appraisal on a fast-path (pending) memory and blend core_affect.

Dual-path slow path (LeDoux, 1996): runs the full Scherer appraisal on the memory's content, blends the appraised affect with the raw fast-path affect, updates consolidation_strength, clears pending_appraisal, and opens the reconsolidation window.

Parameters:

Name	Type	Description	Default
memory_id	str	ID of the memory to elaborate.	required

Returns:

Type	Description
Memory | None	The updated Memory, or None if the memory does not exist,
Memory | None	is not pending appraisal, or no appraisal engine is configured.

Source code in src/emotional_memory/engine.py

def elaborate(self, memory_id: str) -> Memory | None:
    """Run full appraisal on a fast-path (pending) memory and blend core_affect.

    Dual-path slow path (LeDoux, 1996): runs the full Scherer appraisal
    on the memory's content, blends the appraised affect with the raw
    fast-path affect, updates consolidation_strength, clears
    pending_appraisal, and opens the reconsolidation window.

    Args:
        memory_id: ID of the memory to elaborate.

    Returns:
        The updated Memory, or None if the memory does not exist,
        is not pending appraisal, or no appraisal engine is configured.
    """
    with traced_span("emotional_memory.elaborate", {"memory_id": memory_id}):
        mem = self._store.get(memory_id)
        if mem is None:
            return None
        result = self._elaborate_with_memory(mem)
    return result

elaborate_pending() -> list[Memory]

Elaborate all memories with pending_appraisal=True.

Convenience method to run the slow appraisal path on all fast-path memories in the store. Each memory is elaborated in sequence.

Returns:

Type	Description
list[Memory]	List of updated Memory objects (only those that were elaborated).

Source code in src/emotional_memory/engine.py

def elaborate_pending(self) -> list[Memory]:
    """Elaborate all memories with pending_appraisal=True.

    Convenience method to run the slow appraisal path on all fast-path
    memories in the store. Each memory is elaborated in sequence.

    Returns:
        List of updated Memory objects (only those that were elaborated).
    """
    results = []
    for mem in self._store.list_all():
        if mem.tag.pending_appraisal:
            updated = self._elaborate_with_memory(mem)
            if updated is not None:
                results.append(updated)
    return results

encode(content: str, appraisal: AppraisalVector | None = None, metadata: dict[str, Any] | None = None) -> Memory

Encode content into emotional memory.

Pipeline

1. Compute/use appraisal → derive CoreAffect
2. Update AffectiveState (core_affect, momentum, mood)
3. Build EmotionalTag (snapshot of all 5 layers)
4. Embed content
5. Store memory
6. Build resonance links against existing memories
7. Update stored memory with resonance links

Source code in src/emotional_memory/engine.py

def encode(
    self,
    content: str,
    appraisal: AppraisalVector | None = None,
    metadata: dict[str, Any] | None = None,
) -> Memory:
    """Encode content into emotional memory.

    Pipeline:
      1. Compute/use appraisal → derive CoreAffect
      2. Update AffectiveState (core_affect, momentum, mood)
      3. Build EmotionalTag (snapshot of all 5 layers)
      4. Embed content
      5. Store memory
      6. Build resonance links against existing memories
      7. Update stored memory with resonance links
    """
    check_content_length(content, self._config.max_content_length)
    with traced_span("emotional_memory.encode", {"content_length": len(content)}):
        now = datetime.now(tz=UTC)
        logger.debug("encode start: content_len=%d", len(content))

        # Steps 1-3: resolve affect, update state, and build EmotionalTag.
        tag, _ = self._build_tag(
            content,
            appraisal,
            metadata,
            now=now,
            allow_fast_path=True,
        )

        # Step 4: embed
        with traced_span("emotional_memory.embed", {"content_length": len(content)}):
            embedding = self._embedder.embed(content)
        if embedding and bool(np.isnan(np.asarray(embedding)).any()):
            warnings.warn(
                f"Embedder returned NaN values for content (len={len(content)}). "
                "Semantic retrieval will be degraded for this memory.",
                stacklevel=2,
            )

        # Step 5: create and store memory (without resonance links yet)
        memory = Memory.create(
            content=content, tag=tag, embedding=embedding, metadata=metadata
        )
        self._store.save(memory)
        logger.debug(
            "encode stored: id=%s valence=%.3f arousal=%.3f cs=%.3f",
            memory.id,
            tag.core_affect.valence,
            tag.core_affect.arousal,
            tag.consolidation_strength,
        )

        # Step 6: build resonance links — pre-filter when store is large (G2)
        if self._config.enable_resonance:
            resonance_limit = (
                self._config.resonance.max_links * self._config.resonance.candidate_multiplier
            )
            if len(self._store) > resonance_limit and memory.embedding is not None:
                with traced_span("emotional_memory.store.search_by_embedding"):
                    candidates = self._store.search_by_embedding(
                        memory.embedding, resonance_limit
                    )
            else:
                candidates = self._store.list_all()
            links = build_resonance_links(memory, candidates, self._config.resonance)

            if links:
                updated_tag = tag.model_copy(update={"resonance_links": links})
                memory = memory.model_copy(update={"tag": updated_tag})
                self._store.update(memory)
                logger.debug("encode resonance: id=%s links=%d", memory.id, len(links))

                self._add_bidirectional_links(memory, links)

    return memory

encode_batch(contents: list[str], metadata: list[dict[str, Any]] | None = None) -> list[Memory]

Encode multiple contents using a single embed_batch call.

Uses embed_batch() for efficient batched embedding (G6), then encodes each item sequentially so that AffectiveState evolves naturally. Resonance links are built against the store state at each step, meaning later items in the batch can link to earlier ones. Per-item appraisal is computed up front in parallel (see _appraise_many).

Source code in src/emotional_memory/engine.py

def encode_batch(
    self,
    contents: list[str],
    metadata: list[dict[str, Any]] | None = None,
) -> list[Memory]:
    """Encode multiple contents using a single embed_batch call.

    Uses embed_batch() for efficient batched embedding (G6), then encodes
    each item sequentially so that AffectiveState evolves naturally.
    Resonance links are built against the store state at each step, meaning
    later items in the batch can link to earlier ones. Per-item appraisal is
    computed up front in parallel (see ``_appraise_many``).
    """
    if metadata is not None and len(metadata) != len(contents):
        raise ValueError(
            f"metadata length ({len(metadata)}) must match contents length ({len(contents)})"
        )
    with traced_span("emotional_memory.encode_batch", {"batch_size": len(contents)}):
        with traced_span(
            "emotional_memory.embed", {"content_length": sum(len(c) for c in contents)}
        ):
            embeddings = self._embedder.embed_batch(contents)
        # Dual-path: skip appraisal on fast path. Appraise the whole batch in
        # parallel up front, then evolve state sequentially below.
        use_fast_path = self._config.dual_path_encoding and self._appraisal_engine is not None
        appraisals = self._appraise_many(contents, metadata, use_fast_path)
        results = []
        for i, (content, embedding) in enumerate(zip(contents, embeddings, strict=True)):
            meta = metadata[i] if metadata else None
            now = datetime.now(tz=UTC)
            computed_appraisal: AppraisalVector | GenericAppraisalVector | None = appraisals[i]

            if computed_appraisal is not None:
                new_affect = computed_appraisal.to_core_affect()
            else:
                new_affect = self._state.core_affect

            self._state = self._state.update(
                new_affect,
                now=now,
                mood_alpha=self._config.mood_alpha,
                mood_decay=self._config.mood_decay,
            )
            self._persist_state()

            stored_appraisal_b: AppraisalVector | None = (
                computed_appraisal if isinstance(computed_appraisal, AppraisalVector) else None
            )
            cs = consolidation_strength(new_affect.arousal, self._state.mood.arousal)
            tag = make_emotional_tag(
                core_affect=self._state.core_affect,
                momentum=self._state.momentum,
                mood=self._state.mood,
                consolidation_strength=cs,
                appraisal=stored_appraisal_b,
            )

            # Mark pending_appraisal for fast-path memories
            if use_fast_path:
                tag = tag.model_copy(update={"pending_appraisal": True})

            # Auto-categorize: attach Plutchik EmotionLabel
            if self._config.auto_categorize:
                tag = label_tag(tag)

            if embedding and bool(np.isnan(np.asarray(embedding)).any()):
                warnings.warn(
                    f"Embedder returned NaN values for content[{i}] (len={len(content)}). "
                    "Semantic retrieval will be degraded for this memory.",
                    stacklevel=2,
                )
            memory = Memory.create(
                content=content, tag=tag, embedding=embedding, metadata=meta
            )
            self._store.save(memory)

            # Layer-5 ablation: mirror the single-encode guard so batch-encoded
            # memories are not silently given resonance links when disabled.
            if self._config.enable_resonance:
                resonance_limit = (
                    self._config.resonance.max_links
                    * self._config.resonance.candidate_multiplier
                )
                if len(self._store) > resonance_limit and memory.embedding is not None:
                    candidates = self._store.search_by_embedding(
                        memory.embedding, resonance_limit
                    )
                else:
                    candidates = self._store.list_all()
                links = build_resonance_links(memory, candidates, self._config.resonance)
                if links:
                    updated_tag = tag.model_copy(update={"resonance_links": links})
                    memory = memory.model_copy(update={"tag": updated_tag})
                    self._store.update(memory)

                    self._add_bidirectional_links(memory, links)

            results.append(memory)
    return results

export_memories() -> list[dict[str, Any]]

Export all memories as a list of JSON-serialisable dicts.

Suitable for backup or migration between store backends. The output can be restored with import_memories().

Source code in src/emotional_memory/engine.py

def export_memories(self) -> list[dict[str, Any]]:
    """Export all memories as a list of JSON-serialisable dicts.

    Suitable for backup or migration between store backends.  The output
    can be restored with ``import_memories()``.
    """
    return [m.model_dump(mode="json") for m in self._store.list_all()]

get(memory_id: str) -> Memory | None

Look up a single memory by ID, or None if not found.

Source code in src/emotional_memory/engine.py

def get(self, memory_id: str) -> Memory | None:
    """Look up a single memory by ID, or None if not found."""
    return self._store.get(memory_id)

get_current_mood(now: datetime | None = None) -> MoodField

Return the mood field regressed to now without modifying state.

Useful for read-only mood inspection between encode/retrieve calls. If mood_decay is not configured, returns the frozen mood.

Source code in src/emotional_memory/engine.py

def get_current_mood(self, now: datetime | None = None) -> MoodField:
    """Return the mood field regressed to ``now`` without modifying state.

    Useful for read-only mood inspection between encode/retrieve calls.
    If ``mood_decay`` is not configured, returns the frozen mood.
    """
    if now is None:
        now = datetime.now(tz=UTC)
    if self._config.mood_decay is not None:
        return self._state.mood.regress(now, self._config.mood_decay)
    return self._state.mood

get_state() -> AffectiveState

Return a copy of the current affective state.

Source code in src/emotional_memory/engine.py

def get_state(self) -> AffectiveState:
    """Return a copy of the current affective state."""
    return self._state.model_copy()

import_memories(data: list[dict[str, Any]], *, overwrite: bool = False) -> int

Import memories from a list of dicts produced by export_memories().

Parameters:

Name	Type	Description	Default
data	list[dict[str, Any]]	List of memory dicts (as returned by export_memories()).	required
overwrite	bool	When True, existing memories with the same ID are replaced. When False (default), duplicates are skipped silently.	False

Returns:

Type	Description
int	Number of memories actually written.

Source code in src/emotional_memory/engine.py

def import_memories(self, data: list[dict[str, Any]], *, overwrite: bool = False) -> int:
    """Import memories from a list of dicts produced by ``export_memories()``.

    Args:
        data:      List of memory dicts (as returned by ``export_memories()``).
        overwrite: When True, existing memories with the same ID are replaced.
                   When False (default), duplicates are skipped silently.

    Returns:
        Number of memories actually written.
    """
    written = 0
    for item in data:
        memory = Memory.model_validate(item)
        exists = self._store.get(memory.id) is not None
        if exists and not overwrite:
            continue
        if exists:
            self._store.update(memory)
        else:
            self._store.save(memory)
        written += 1
    return written

list_all() -> list[Memory]

Return all memories in the store.

Source code in src/emotional_memory/engine.py

def list_all(self) -> list[Memory]:
    """Return all memories in the store."""
    return self._store.list_all()

load_state(data: dict[str, Any]) -> None

Restore a previously saved affective state.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	A dict produced by a previous save_state() call.	required

Source code in src/emotional_memory/engine.py

def load_state(self, data: dict[str, Any]) -> None:
    """Restore a previously saved affective state.

    Args:
        data: A dict produced by a previous ``save_state()`` call.
    """
    self._state = AffectiveState.restore(data)
    self._persist_state()

observe(content: str, appraisal: AppraisalVector | None = None, metadata: dict[str, Any] | None = None) -> EmotionalTag

Update affective state from content without storing a retrievable memory.

Source code in src/emotional_memory/engine.py

def observe(
    self,
    content: str,
    appraisal: AppraisalVector | None = None,
    metadata: dict[str, Any] | None = None,
) -> EmotionalTag:
    """Update affective state from content without storing a retrievable memory."""
    check_content_length(content, self._config.max_content_length)
    with traced_span("emotional_memory.observe", {"content_length": len(content)}):
        now = datetime.now(tz=UTC)
        tag, _ = self._build_tag(
            content,
            appraisal,
            metadata,
            now=now,
            allow_fast_path=False,
        )
    return tag

persist_state() -> dict[str, Any]

Return and, when configured, persist the current affective state.

Source code in src/emotional_memory/engine.py

def persist_state(self) -> dict[str, Any]:
    """Return and, when configured, persist the current affective state."""
    self._persist_state()
    return self.save_state()

prune(threshold: float = 0.05) -> int

Remove memories whose effective consolidation strength has fallen below threshold.

Parameters:

Name	Type	Description	Default
threshold	float	Minimum acceptable strength in [0, 1]. Memories whose compute_effective_strength() is below this value are deleted. Defaults to 0.05 (5%).	0.05

Returns:

Type	Description
int	Number of memories removed.

Source code in src/emotional_memory/engine.py

def prune(self, threshold: float = 0.05) -> int:
    """Remove memories whose effective consolidation strength has fallen below *threshold*.

    Args:
        threshold: Minimum acceptable strength in [0, 1].  Memories whose
                   ``compute_effective_strength()`` is below this value are deleted.
                   Defaults to 0.05 (5%).

    Returns:
        Number of memories removed.
    """
    from emotional_memory.decay import compute_effective_strength_batch

    validate_prune_threshold(threshold)
    with traced_span("emotional_memory.prune", {"threshold": threshold}):
        now = datetime.now(tz=UTC)
        mems = self._store.list_all()
        if mems:
            strengths = compute_effective_strength_batch(
                [m.tag for m in mems], now, self._config.decay
            )
            to_delete = [m.id for m, s in zip(mems, strengths, strict=True) if s < threshold]
        else:
            to_delete = []
        for memory_id in to_delete:
            self._store.delete(memory_id)
    return len(to_delete)

reset_state() -> None

Reset the runtime affective state to its initial baseline.

Source code in src/emotional_memory/engine.py

def reset_state(self) -> None:
    """Reset the runtime affective state to its initial baseline."""
    self._state = AffectiveState.initial()
    self._persist_state()

restore_persisted_state() -> bool

Load the last persisted affective state from the configured state store.

Source code in src/emotional_memory/engine.py

def restore_persisted_state(self) -> bool:
    """Load the last persisted affective state from the configured state store."""
    if self._state_store is None:
        return False
    persisted = self._state_store.load()
    if persisted is None:
        return False
    self._state = persisted.model_copy()
    return True

retrieve(query: str, top_k: int = 5, *, query_affect: CoreAffect | None = None, precomputed_weights: NDArray[np.float64] | None = None) -> list[Memory]

Retrieve the top-k most relevant memories for the query.

Scoring uses all 6 AFT signals with mood-adaptive weights.

Two-pass strategy for spreading activation (G1): Pass 1 — score all candidates with empty active_ids to get an initial top-k ranking; their IDs become the active set. Pass 2 — re-score with the active set, allowing resonance links to boost memories associated with the initially top-ranked ones.

Pre-filter (G2): when the store is large, use search_by_embedding to narrow candidates to top_k * candidate_multiplier before full scoring.

Reconsolidation (D1): only triggered if the memory was last retrieved within the reconsolidation_window_seconds lability window.

Query affect: by default the affect-proximity signal (s3) uses the current runtime affective state (self._state.core_affect). Pass an explicit query_affect to score against the query's own affect instead — e.g. the appraised affect of the query text — without mutating the runtime state. This is the production-reachable path validated in Addendum T (retrieve-time query appraisal); see retrieve_with_query_appraisal.

Source code in src/emotional_memory/engine.py

def retrieve(
    self,
    query: str,
    top_k: int = 5,
    *,
    query_affect: CoreAffect | None = None,
    precomputed_weights: NDArray[np.float64] | None = None,
) -> list[Memory]:
    """Retrieve the top-k most relevant memories for the query.

    Scoring uses all 6 AFT signals with mood-adaptive weights.

    Two-pass strategy for spreading activation (G1):
      Pass 1 — score all candidates with empty active_ids to get an initial
               top-k ranking; their IDs become the active set.
      Pass 2 — re-score with the active set, allowing resonance links to
               boost memories associated with the initially top-ranked ones.

    Pre-filter (G2): when the store is large, use search_by_embedding to
    narrow candidates to top_k * candidate_multiplier before full scoring.

    Reconsolidation (D1): only triggered if the memory was last retrieved
    within the reconsolidation_window_seconds lability window.

    Query affect: by default the affect-proximity signal (s3) uses the
    current runtime affective state (``self._state.core_affect``). Pass an
    explicit ``query_affect`` to score against the query's own affect instead
    — e.g. the appraised affect of the query text — without mutating the
    runtime state. This is the production-reachable path validated in
    Addendum T (retrieve-time query appraisal); see
    ``retrieve_with_query_appraisal``.
    """
    if top_k < 1:
        raise ValueError(f"top_k must be >= 1, got {top_k}")
    check_content_length(query, self._config.max_content_length, label="query")
    with traced_span(
        "emotional_memory.retrieve",
        {"query_length": len(query), "top_k": top_k, "store_size": len(self._store)},
    ):
        now = datetime.now(tz=UTC)
        logger.debug(
            "retrieve start: query_len=%d top_k=%d store=%d",
            len(query),
            top_k,
            len(self._store),
        )
        with traced_span("emotional_memory.embed", {"content_length": len(query)}):
            query_embedding = self._embedder.embed(query)

        # G2 — pre-filter candidates via embedding search when store is large
        rc = self._config.retrieval
        candidate_limit = top_k * rc.candidate_multiplier
        if len(self._store) > candidate_limit:
            with traced_span(
                "emotional_memory.store.search_by_embedding", {"limit": candidate_limit}
            ):
                candidates = self._store.search_by_embedding(query_embedding, candidate_limit)
        else:
            candidates = self._store.list_all()

        if not candidates:
            return []

        _spreading_fn = (
            (lambda *a, **kw: {})
            if not self._config.enable_resonance
            else spreading_activation
        )
        plan = build_retrieval_plan(
            query_embedding=query_embedding,
            query_affect=query_affect or self._state.core_affect,
            current_mood=self._state.mood,
            current_momentum=self._state.momentum,
            candidates=candidates,
            top_k=top_k,
            now=now,
            decay_config=self._config.decay,
            retrieval_config=rc,
            propagation_hops=self._config.resonance.propagation_hops,
            spreading_activation_fn=_spreading_fn,
            precomputed_weights=(
                self._effective_retrieval_weights(query)
                if precomputed_weights is None
                else precomputed_weights
            ),
        )
        top = [item.memory for item in plan.pass2[:top_k]]
        result = self._apply_retrieval_updates(top, now)
        logger.debug("retrieve done: returned=%d candidates=%d", len(result), len(candidates))
    return result

retrieve_query_gated(query: str, top_k: int = 5, *, tau: float | None = None) -> list[Memory]

Retrieve with the Addendum Y query-affect gate (Branch A).

Appraises the query text, then routes to semantic-only retrieval when |valence| < tau (default query_affect_gate_tau, 0.2), else to the retrieve-time query-appraisal path (retrieve_with_query_appraisal).

This is a validated safe wrapper: it recovers the neutral-query component of the off-regime penalty without mutating runtime affective state on the affect-carrying path.

Raises:

Type	Description
RuntimeError	if no appraisal engine is configured.

Source code in src/emotional_memory/engine.py

def retrieve_query_gated(
    self, query: str, top_k: int = 5, *, tau: float | None = None
) -> list[Memory]:
    """Retrieve with the Addendum Y query-affect gate (Branch A).

    Appraises the query text, then routes to semantic-only retrieval when
    ``|valence| < tau`` (default ``query_affect_gate_tau``, 0.2), else to
    the retrieve-time query-appraisal path (``retrieve_with_query_appraisal``).

    This is a validated safe wrapper: it recovers the neutral-query component
    of the off-regime penalty without mutating runtime affective state on the
    affect-carrying path.

    Raises:
        RuntimeError: if no appraisal engine is configured.
    """
    if self._appraisal_engine is None:
        raise RuntimeError(
            "retrieve_query_gated requires an appraisal_engine; "
            "none is configured. Pass appraisal_engine=... to EmotionalMemory."
        )
    effective_tau = self._config.query_affect_gate_tau if tau is None else tau
    query_affect = self._appraisal_engine.appraise(query).to_core_affect()
    if is_query_affect_neutral(query_affect.valence, effective_tau):
        return self.retrieve(query, top_k=top_k, precomputed_weights=semantic_only_weights())
    return self.retrieve(query, top_k=top_k, query_affect=query_affect)

retrieve_with_explanations(query: str, top_k: int = 5, *, query_affect: CoreAffect | None = None, precomputed_weights: NDArray[np.float64] | None = None) -> list[RetrievalExplanation]

Retrieve memories plus a structured score decomposition.

Ranking is computed by the same two-pass retrieval pipeline used by retrieve(). The returned score and breakdown reflect the ranking state before retrieval-side updates such as reconsolidation and Hebbian strengthening; the returned memory objects reflect the post- retrieval stored state.

query_affect overrides the affect-proximity signal source exactly as in retrieve() (default: current runtime state).

Source code in src/emotional_memory/engine.py

def retrieve_with_explanations(
    self,
    query: str,
    top_k: int = 5,
    *,
    query_affect: CoreAffect | None = None,
    precomputed_weights: NDArray[np.float64] | None = None,
) -> list[RetrievalExplanation]:
    """Retrieve memories plus a structured score decomposition.

    Ranking is computed by the same two-pass retrieval pipeline used by
    ``retrieve()``. The returned score and breakdown reflect the ranking
    state before retrieval-side updates such as reconsolidation and Hebbian
    strengthening; the returned ``memory`` objects reflect the post-
    retrieval stored state.

    ``query_affect`` overrides the affect-proximity signal source exactly as
    in ``retrieve()`` (default: current runtime state).
    """
    if top_k < 1:
        raise ValueError(f"top_k must be >= 1, got {top_k}")
    check_content_length(query, self._config.max_content_length, label="query")
    now = datetime.now(tz=UTC)
    logger.debug(
        "retrieve_with_explanations start: query_len=%d top_k=%d store=%d",
        len(query),
        top_k,
        len(self._store),
    )
    query_embedding = self._embedder.embed(query)

    rc = self._config.retrieval
    candidate_limit = top_k * rc.candidate_multiplier
    if len(self._store) > candidate_limit:
        candidates = self._store.search_by_embedding(query_embedding, candidate_limit)
    else:
        candidates = self._store.list_all()

    if not candidates:
        return []

    _spreading_fn = (
        (lambda *a, **kw: {}) if not self._config.enable_resonance else spreading_activation
    )
    plan = build_retrieval_plan(
        query_embedding=query_embedding,
        query_affect=query_affect or self._state.core_affect,
        current_mood=self._state.mood,
        current_momentum=self._state.momentum,
        candidates=candidates,
        top_k=top_k,
        now=now,
        decay_config=self._config.decay,
        retrieval_config=rc,
        propagation_hops=self._config.resonance.propagation_hops,
        spreading_activation_fn=_spreading_fn,
        precomputed_weights=(
            self._effective_retrieval_weights(query)
            if precomputed_weights is None
            else precomputed_weights
        ),
    )
    top_ranked = plan.pass2[:top_k]
    result = self._apply_retrieval_updates([item.memory for item in top_ranked], now)
    updated_by_id = {mem.id: mem for mem in result}
    pass1_ranks = {item.memory.id: index for index, item in enumerate(plan.pass1, start=1)}

    explanations = [
        RetrievalExplanation(
            memory=updated_by_id[item.memory.id],
            score=item.score,
            breakdown=item.breakdown,
            activation_level=plan.activation_map.get(item.memory.id, 0.0),
            pass1_rank=pass1_ranks.get(item.memory.id),
            pass2_rank=index,
            selected_as_seed=item.memory.id in plan.seed_ids,
            candidate_count=plan.candidate_count,
        )
        for index, item in enumerate(top_ranked, start=1)
    ]
    logger.debug(
        "retrieve_with_explanations done: returned=%d candidates=%d",
        len(explanations),
        len(candidates),
    )
    return explanations

retrieve_with_query_appraisal(query: str, top_k: int = 5) -> list[Memory]

Retrieve using the query's own appraised affect (Addendum T).

Appraises the query text with the configured appraisal engine and uses the resulting :class:CoreAffect as query_affect for retrieval, without mutating the runtime affective state. This is the production-reachable retrieve-time query-appraisal path: it does not require an oracle affect to be injected.

Pairs best with an appraisal engine using DIRECT_VAD_SCHEMA (the LLM rates valence/arousal/dominance directly), which Addendum V found estimates query affect more faithfully than the Scherer SEC projection.

Raises:

Type	Description
RuntimeError	if no appraisal engine is configured.

Source code in src/emotional_memory/engine.py

def retrieve_with_query_appraisal(self, query: str, top_k: int = 5) -> list[Memory]:
    """Retrieve using the query's own appraised affect (Addendum T).

    Appraises the query text with the configured appraisal engine and uses
    the resulting :class:`CoreAffect` as ``query_affect`` for retrieval,
    without mutating the runtime affective state. This is the
    production-reachable retrieve-time query-appraisal path: it does not
    require an oracle affect to be injected.

    Pairs best with an appraisal engine using ``DIRECT_VAD_SCHEMA`` (the LLM
    rates valence/arousal/dominance directly), which Addendum V found
    estimates query affect more faithfully than the Scherer SEC projection.

    Raises:
        RuntimeError: if no appraisal engine is configured.
    """
    if self._appraisal_engine is None:
        raise RuntimeError(
            "retrieve_with_query_appraisal requires an appraisal_engine; "
            "none is configured. Pass appraisal_engine=... to EmotionalMemory, "
            "or call retrieve(query, query_affect=...) with an explicit affect."
        )
    query_affect = self._appraisal_engine.appraise(query).to_core_affect()
    return self.retrieve(query, top_k=top_k, query_affect=query_affect)

save_state() -> dict[str, Any]

Serialise the current affective state for persistence.

Returns a JSON-serialisable dict that can be stored alongside the memory store and later restored with load_state(). The private momentum history is included so that momentum computation continues correctly after a restart.

Source code in src/emotional_memory/engine.py

def save_state(self) -> dict[str, Any]:
    """Serialise the current affective state for persistence.

    Returns a JSON-serialisable dict that can be stored alongside the
    memory store and later restored with ``load_state()``.  The private
    momentum history is included so that momentum computation continues
    correctly after a restart.
    """
    return self._state.snapshot()

set_affect(core_affect: CoreAffect) -> None

Manually inject a CoreAffect (e.g. from external appraisal).

Updates core_affect, momentum, and mood.

Source code in src/emotional_memory/engine.py

def set_affect(self, core_affect: CoreAffect) -> None:
    """Manually inject a CoreAffect (e.g. from external appraisal).

    Updates core_affect, momentum, and mood.
    """
    self._state = self._state.update(
        core_affect,
        mood_alpha=self._config.mood_alpha,
        mood_decay=self._config.mood_decay,
    )
    self._persist_state()

update_memory(memory_id: str, *, content: str | None = None, tag: EmotionalTag | None = None, metadata: dict[str, Any] | None = None) -> Memory

Replace selected fields of an existing memory.

None preserves a field. Changing content recomputes its embedding before the store is mutated, so embedding failures leave the original memory intact.

Source code in src/emotional_memory/engine.py

def update_memory(
    self,
    memory_id: str,
    *,
    content: str | None = None,
    tag: EmotionalTag | None = None,
    metadata: dict[str, Any] | None = None,
) -> Memory:
    """Replace selected fields of an existing memory.

    ``None`` preserves a field. Changing content recomputes its embedding
    before the store is mutated, so embedding failures leave the original
    memory intact.
    """
    memory = self._store.get(memory_id)
    if memory is None:
        raise KeyError(f"memory not found: {memory_id}")
    changes: dict[str, Any] = {}
    if content is not None:
        check_content_length(content, self._config.max_content_length)
        changes["content"] = content
        changes["embedding"] = self._embedder.embed(content)
    if tag is not None:
        changes["tag"] = tag
    if metadata is not None:
        changes["metadata"] = metadata
    updated = memory.model_copy(update=changes)
    self._store.update(updated)
    return updated

emotional_memory.engine.EmotionalMemoryConfig

Bases: BaseModel

appraisal_max_concurrency: int = Field(default=8, ge=1) class-attribute instance-attribute

Max concurrent appraisal calls in encode_batch(). Appraisal depends only on content/context, so a batch's appraisals are computed concurrently and then consumed by the order-preserving state-evolution loop (results are unchanged). Async uses an asyncio.Semaphore; sync uses a thread pool — the llm callable must be thread-safe (httpx.Client and KeywordAppraisalEngine are). Set to 1 for fully sequential appraisal.

auto_categorize: bool = False class-attribute instance-attribute

When True, run Plutchik categorization on encode and store EmotionLabel.

dual_path_encoding: bool = False class-attribute instance-attribute

When True, fast-path encoding skips appraisal; call elaborate() later to run full appraisal and blend core_affect (LeDoux, 1996).

elaboration_learning_rate: float = 0.7 class-attribute instance-attribute

Blend ratio used in elaborate(): 70% appraised / 30% raw affect.

enable_appraisal: bool = True class-attribute instance-attribute

When False, skip the appraisal engine call in _build_tag (ablation).

enable_momentum: bool = True class-attribute instance-attribute

When False, zero the momentum-alignment retrieval weight (ablation of Layer 2).

enable_mood_signal: bool = True class-attribute instance-attribute

When False, zero the mood-congruence retrieval weight (ablation of Layer 3).

enable_reconsolidation: bool = True class-attribute instance-attribute

When False, skip the APE-gated reconsolidation window at retrieval time (ablation of Pearce-Hall 1980 APE gate). Predictive-learning (update_prediction) still runs.

enable_resonance: bool = True class-attribute instance-attribute

When False, skip resonance link building and spreading activation (ablation of Layer 5).

max_content_length: int | None = Field(default=None, ge=1) class-attribute instance-attribute

Optional maximum length for content / query strings. None = unlimited.

mood_decay: MoodDecayConfig | None = None class-attribute instance-attribute

Time-based regression of mood toward baseline. None = disabled.

query_affect_gate_tau: float = Field(default=0.2, ge=0.0, le=2.0) class-attribute instance-attribute

Neutral-query threshold for retrieve_query_gated() (Addendum Y, default 0.2).