Retrieval

Use retrieve() for normal recall and retrieve_with_explanations() when you need ranking-time introspection.

The stable explainability surface is:

• EmotionalMemory.retrieve_with_explanations()
• AsyncEmotionalMemory.retrieve_with_explanations()
• RetrievalSignals
• RetrievalBreakdown
• RetrievalExplanation

These models describe the ranking plan before retrieval-side mutation. The returned Memory objects may already reflect post-retrieval updates such as reconsolidation.

Lower-level helpers in emotional_memory.retrieval are documented here because they shape the scoring model, but they are not all promoted as stable top-level API. In particular, build_retrieval_plan() remains an engine helper rather than a supported root import.

emotional_memory.retrieval.RetrievalConfig

Bases: BaseModel

Parameters for multi-signal retrieval scoring.

adaptive_weights_config: AdaptiveWeightsConfig = AdaptiveWeightsConfig() class-attribute instance-attribute

Parameters for smooth mood-modulated weight adjustment.

ape_threshold: float = 0.3 class-attribute instance-attribute

Affective Prediction Error threshold to trigger reconsolidation.

base_weights: list[float] = [0.35, 0.25, 0.15, 0.1, 0.1, 0.05] class-attribute instance-attribute

Weights for [semantic, mood_congruence, affect_proximity, momentum_alignment, recency, resonance_boost]. Must have exactly 6 elements.

candidate_multiplier: int = 3 class-attribute instance-attribute

Pre-filter multiplier: when the store has > top_k * candidate_multiplier entries, use search_by_embedding to fetch top_k * candidate_multiplier candidates before full multi-signal scoring.

query_classifier: QueryClassifierConfig | None = None class-attribute instance-attribute

Per-query-type weight routing. None = disabled (default behaviour).

reconsolidation_learning_rate: float = 0.2 class-attribute instance-attribute

Alpha for tag update on reconsolidation. Max shift = 50%.

reconsolidation_window_seconds: float = 300.0 class-attribute instance-attribute

Seconds after retrieval during which the tag remains labile.

emotional_memory.retrieval.QueryClassifierConfig

Bases: BaseModel

Configuration for per-query-type retrieval weight routing.

Set mode to 'heuristic' to activate the built-in regex classifier at zero latency overhead, or 'llm' to use an external QueryClassifier instance passed to the engine constructor. 'disabled' (default) leaves weight routing inactive and preserves existing behaviour.

routed_weights maps query-type labels to 6-element weight vectors (same order as RetrievalConfig.base_weights). LOCOMO_ROUTING from emotional_memory.query_classifier is a ready-to-use routing table.

Routing selects the starting weights, it does not bypass adaptive weighting: the routed vector replaces base_weights and is then still modulated by the current mood via adaptive_weights() (and by the ablation mask). A routed [1, 0, 0, 0, 0, 0] therefore yields a pure-semantic ranking only under a neutral/calm mood; to gate signals off unconditionally, disable them via the enable_* ablation flags or route in front of the engine instead.

default_type: str = 'default' class-attribute instance-attribute

Fallback key when the classified type has no entry in routed_weights.

mode: Literal['disabled', 'heuristic', 'llm'] = 'disabled' class-attribute instance-attribute

'heuristic' — auto-create HeuristicQueryClassifier; 'llm' — use the QueryClassifier instance supplied to the engine constructor; 'disabled' — off.

routed_weights: dict[str, list[float]] = Field(default_factory=dict) class-attribute instance-attribute

Per-type weight override. Unknown types fall back to default_type.

emotional_memory.retrieval.AdaptiveWeightsConfig

Bases: BaseModel

Parameters for smooth mood-modulated retrieval weight adjustment.

Each of the three psychological conditions (negative mood, high arousal, calm/neutral) is modelled as a continuous sigmoid or Gaussian gate rather than a hard threshold, eliminating discontinuities in retrieval behaviour.

sharpness controls how step-like the transition is: a value of ~50 approximates the old hard-threshold behaviour; the default of 5.0 produces a smooth transition centred at the given point.

emotional_memory.retrieval.RetrievalSignals

Bases: BaseModel

Named retrieval signal values for explainability.

The six fields map directly to the composite scorer

semantic_similarity mood_congruence affect_proximity momentum_alignment recency resonance

total() -> float

Return the sum of all six signals.

Source code in src/emotional_memory/retrieval.py

def total(self) -> float:
    """Return the sum of all six signals."""
    return (
        self.semantic_similarity
        + self.mood_congruence
        + self.affect_proximity
        + self.momentum_alignment
        + self.recency
        + self.resonance
    )

emotional_memory.retrieval.RetrievalBreakdown

Bases: BaseModel

Structured decomposition of one retrieval score.

emotional_memory.retrieval.RetrievalExplanation

Bases: BaseModel

One retrieved memory plus the score decomposition that ranked it.

score and breakdown describe the ranking-time score before any side effects of retrieval (for example reconsolidation or Hebbian strengthening) are applied. memory is the post-retrieval memory state.

emotional_memory.retrieval.retrieval_score(query_embedding: list[float], query_affect: CoreAffect, current_mood: MoodField, current_momentum: AffectiveMomentum, memory: Memory, activation_map: dict[str, float], now: datetime, decay_config: DecayConfig, retrieval_config: RetrievalConfig, precomputed_weights: NDArray[np.float64] | None = None) -> float

Compute the composite retrieval score for a single memory.

Parameters:

Name	Type	Description	Default
activation_map	dict[str, float]	Pre-computed spreading activation levels from spreading_activation(). Pass an empty dict for Pass 1 (no resonance boost). Pass the result of spreading_activation() for Pass 2 to enable Collins & Loftus (1975) multi-hop boosting.	required
precomputed_weights	NDArray[float64] | None	Pre-computed adaptive weights from adaptive_weights(). Pass this when scoring many memories with the same mood/config to avoid redundant computation. If None, weights are computed inline.	None

Source code in src/emotional_memory/retrieval.py

def retrieval_score(
    query_embedding: list[float],
    query_affect: CoreAffect,
    current_mood: MoodField,
    current_momentum: AffectiveMomentum,
    memory: Memory,
    activation_map: dict[str, float],
    now: datetime,
    decay_config: DecayConfig,
    retrieval_config: RetrievalConfig,
    precomputed_weights: NDArray[np.float64] | None = None,
) -> float:
    """Compute the composite retrieval score for a single memory.

    Args:
        activation_map: Pre-computed spreading activation levels from
            ``spreading_activation()``.  Pass an empty dict for Pass 1
            (no resonance boost).  Pass the result of ``spreading_activation()``
            for Pass 2 to enable Collins & Loftus (1975) multi-hop boosting.
        precomputed_weights: Pre-computed adaptive weights from ``adaptive_weights()``.
            Pass this when scoring many memories with the same mood/config to avoid
            redundant computation. If None, weights are computed inline.
    """
    return retrieval_breakdown(
        query_embedding=query_embedding,
        query_affect=query_affect,
        current_mood=current_mood,
        current_momentum=current_momentum,
        memory=memory,
        activation_map=activation_map,
        now=now,
        decay_config=decay_config,
        retrieval_config=retrieval_config,
        precomputed_weights=precomputed_weights,
    ).total_score

emotional_memory.retrieval.retrieval_breakdown(query_embedding: list[float], query_affect: CoreAffect, current_mood: MoodField, current_momentum: AffectiveMomentum, memory: Memory, activation_map: dict[str, float], now: datetime, decay_config: DecayConfig, retrieval_config: RetrievalConfig, precomputed_weights: NDArray[np.float64] | None = None) -> RetrievalBreakdown

Return the full score decomposition for one memory.

This is the explainable counterpart to retrieval_score() and uses the exact same scoring path.

Source code in src/emotional_memory/retrieval.py

def retrieval_breakdown(
    query_embedding: list[float],
    query_affect: CoreAffect,
    current_mood: MoodField,
    current_momentum: AffectiveMomentum,
    memory: Memory,
    activation_map: dict[str, float],
    now: datetime,
    decay_config: DecayConfig,
    retrieval_config: RetrievalConfig,
    precomputed_weights: NDArray[np.float64] | None = None,
) -> RetrievalBreakdown:
    """Return the full score decomposition for one memory.

    This is the explainable counterpart to ``retrieval_score()`` and uses the
    exact same scoring path.
    """
    weight_arr = (
        precomputed_weights
        if precomputed_weights is not None
        else adaptive_weights(
            current_mood,
            retrieval_config.base_weights,
            retrieval_config.adaptive_weights_config,
        )
    )
    raw_values = _component_values(
        query_embedding=query_embedding,
        query_affect=query_affect,
        current_mood=current_mood,
        current_momentum=current_momentum,
        memory=memory,
        activation_map=activation_map,
        now=now,
        decay_config=decay_config,
    )
    weight_values = tuple(float(v) for v in weight_arr.tolist())
    weighted_values = tuple(w * s for w, s in zip(weight_values, raw_values, strict=True))

    weighted = _signals_from_values(weighted_values)
    return RetrievalBreakdown(
        weights=_signals_from_values(weight_values),
        raw_signals=_signals_from_values(raw_values),
        weighted_signals=weighted,
        total_score=weighted.total(),
    )

emotional_memory.retrieval.adaptive_weights(mood: MoodField, base: list[float], config: AdaptiveWeightsConfig | None = None) -> NDArray[np.float64]

Return retrieval weights modulated by current mood state.

Mood acts as a weight modulator — not a hard filter but a continuous shaping of what signals dominate retrieval (inspired by mood-congruent recall research, Bower 1981).

Three conditions are applied as smooth continuous functions (tanh / Gaussian) rather than hard thresholds, eliminating discontinuities at boundary values while preserving the same directional intent: - Negative mood → emotional signals dominate (s2, s3 ↑, s1 ↓) - High arousal → momentum alignment dominates (s4 ↑, s1 ↓) - Calm/neutral → semantic retrieval dominates (s1 ↑, s2, s3 ↓)

Source code in src/emotional_memory/retrieval.py

def adaptive_weights(
    mood: MoodField,
    base: list[float],
    config: AdaptiveWeightsConfig | None = None,
) -> NDArray[np.float64]:
    """Return retrieval weights modulated by current mood state.

    Mood acts as a weight modulator — not a hard filter but a continuous
    shaping of what signals dominate retrieval (inspired by mood-congruent
    recall research, Bower 1981).

    Three conditions are applied as smooth continuous functions (tanh /
    Gaussian) rather than hard thresholds, eliminating discontinuities at
    boundary values while preserving the same directional intent:
      - Negative mood  → emotional signals dominate (s2, s3 ↑, s1 ↓)
      - High arousal   → momentum alignment dominates (s4 ↑, s1 ↓)
      - Calm/neutral   → semantic retrieval dominates (s1 ↑, s2, s3 ↓)
    """
    cfg = config or AdaptiveWeightsConfig()
    w = np.array(base, dtype=float)

    # Negative mood: sigmoid activates below negative_mood_center
    neg = _smooth_gate(mood.valence, cfg.negative_mood_center, -cfg.negative_mood_sharpness)
    w[1] += cfg.negative_mood_strength * neg * (2.0 / 3.0)  # mood congruence
    w[2] += cfg.negative_mood_strength * neg * (1.0 / 3.0)  # affect proximity
    w[0] -= cfg.negative_mood_strength * neg  # semantic

    # High arousal: sigmoid activates above high_arousal_center
    aro = _smooth_gate(mood.arousal, cfg.high_arousal_center, cfg.high_arousal_sharpness)
    w[3] += cfg.high_arousal_strength * aro  # momentum alignment
    w[0] -= cfg.high_arousal_strength * aro  # semantic

    # Calm/neutral: Gaussian over valence x inverted sigmoid over arousal
    # Peaks at (valence≈0, arousal≈0); falls off smoothly in all directions
    valence_calm = math.exp(-((mood.valence / cfg.calm_valence_width) ** 2))
    arousal_calm = _smooth_gate(mood.arousal, cfg.calm_arousal_center, -cfg.calm_sharpness)
    calm = valence_calm * arousal_calm
    w[0] += cfg.calm_strength * calm  # semantic
    w[1] -= cfg.calm_strength * calm * (2.0 / 3.0)  # mood congruence
    w[2] -= cfg.calm_strength * calm * (1.0 / 3.0)  # affect proximity

    # Clamp negatives and normalise; fall back to uniform weights when all
    # signals cancel out (extreme mood state) to avoid zero-score retrieval.
    w = np.clip(w, 0.0, None)
    total = float(w.sum())
    w = w / total if total > 0 else np.full(6, 1.0 / 6.0)
    return np.asarray(w, dtype=np.float64)

emotional_memory.retrieval.reconsolidate(tag: EmotionalTag, current_affect: CoreAffect, ape: float, learning_rate: float) -> EmotionalTag

Update the tag's core_affect proportionally to the APE.

alpha = min(ape * learning_rate, 0.5) — linearly scaled, capped at 50% per retrieval. Pearce-Hall associability (large errors increase the learning rate) is handled separately by update_prediction(). Only core_affect is updated; all other fields remain unchanged. reconsolidation_count is incremented.

Source code in src/emotional_memory/retrieval.py

def reconsolidate(
    tag: EmotionalTag,
    current_affect: CoreAffect,
    ape: float,
    learning_rate: float,
) -> EmotionalTag:
    """Update the tag's core_affect proportionally to the APE.

    alpha = min(ape * learning_rate, 0.5) — linearly scaled, capped at 50% per retrieval.
    Pearce-Hall associability (large errors increase the learning rate) is handled separately
    by update_prediction().
    Only core_affect is updated; all other fields remain unchanged.
    reconsolidation_count is incremented.
    """
    alpha = min(ape * learning_rate, 0.5)
    new_affect = tag.core_affect.lerp(current_affect, alpha)
    return tag.model_copy(
        update={
            "core_affect": new_affect,
            "reconsolidation_count": tag.reconsolidation_count + 1,
        }
    )

emotional_memory.retrieval.compute_ape(tag: EmotionalTag, observed: CoreAffect) -> float

Compute Affective Prediction Error for a memory tag.

Uses tag.expected_affect as the reference point when set; otherwise falls back to tag.core_affect (i.e. the encoding-time affect).

The APE is the Euclidean distance in affect space between the expected and observed affect (Pearce & Hall, 1980).

Source code in src/emotional_memory/retrieval.py

def compute_ape(tag: EmotionalTag, observed: CoreAffect) -> float:
    """Compute Affective Prediction Error for a memory tag.

    Uses ``tag.expected_affect`` as the reference point when set; otherwise
    falls back to ``tag.core_affect`` (i.e. the encoding-time affect).

    The APE is the Euclidean distance in affect space between the expected
    and observed affect (Pearce & Hall, 1980).
    """
    reference = tag.expected_affect if tag.expected_affect is not None else tag.core_affect
    return affective_prediction_error(reference, observed)

emotional_memory.retrieval.update_prediction(tag: EmotionalTag, observed: CoreAffect, ape: float) -> EmotionalTag

Update the tag's expected_affect via Pearce-Hall associability learning.

The learning rate is adaptive: large APE increases the rate (fast relearning after surprise), small APE decreases it (stable predictions need less updating).

Learning-rate update (Pearce-Hall: associability tracks prediction error): target_lr = clamp(ape, 0.05, 0.80) # the APE magnitude is the attractor new_lr = current_lr + 0.2 * (target_lr - current_lr) # smooth adaptation new_lr = clamp(new_lr, 0.05, 0.80)

The new expected_affect is an EMA blend

new_expected = lerp(current_expected, observed, new_lr)

Parameters:

Name	Type	Description	Default
tag	EmotionalTag	The tag whose prediction should be updated.	required
observed	CoreAffect	The affect observed at retrieval time.	required
ape	float	The pre-computed APE (from compute_ape).	required

Returns:

Type	Description
EmotionalTag	Updated tag with new expected_affect and prediction_learning_rate.
EmotionalTag	All other fields are unchanged.

References

Pearce, J. M., & Hall, G. (1980). A model for Pavlovian learning: Variations in the effectiveness of conditioned but not of unconditioned stimuli. Psychological Review, 87(6), 532-552.

Source code in src/emotional_memory/retrieval.py

def update_prediction(
    tag: EmotionalTag,
    observed: CoreAffect,
    ape: float,
) -> EmotionalTag:
    """Update the tag's expected_affect via Pearce-Hall associability learning.

    The learning rate is adaptive: large APE increases the rate (fast relearning
    after surprise), small APE decreases it (stable predictions need less updating).

    Learning-rate update (Pearce-Hall: associability tracks prediction error):
      target_lr = clamp(ape, 0.05, 0.80)      # the APE magnitude is the attractor
      new_lr    = current_lr + 0.2 * (target_lr - current_lr)   # smooth adaptation
      new_lr    = clamp(new_lr, 0.05, 0.80)

    The new expected_affect is an EMA blend:
      new_expected = lerp(current_expected, observed, new_lr)

    Args:
        tag:      The tag whose prediction should be updated.
        observed: The affect observed at retrieval time.
        ape:      The pre-computed APE (from compute_ape).

    Returns:
        Updated tag with new ``expected_affect`` and ``prediction_learning_rate``.
        All other fields are unchanged.

    References
    ----------
    Pearce, J. M., & Hall, G. (1980). A model for Pavlovian learning: Variations
    in the effectiveness of conditioned but not of unconditioned stimuli.
    Psychological Review, 87(6), 532-552.
    """
    _LR_MIN: float = 0.05
    _LR_MAX: float = 0.80
    _LR_ADAPT_RATE: float = 0.2  # meta-learning rate for lr update

    current_lr = tag.prediction_learning_rate
    # Pearce-Hall: associability tracks prediction error.
    # APE (clamped to [LR_MIN, LR_MAX]) is the attractor — large surprise
    # drives LR up, accurate predictions drive LR down toward LR_MIN.
    target_lr = max(_LR_MIN, min(_LR_MAX, ape))
    # Smooth adaptation of current_lr toward target_lr
    new_lr = current_lr + _LR_ADAPT_RATE * (target_lr - current_lr)
    new_lr = max(_LR_MIN, min(_LR_MAX, new_lr))

    # Update expected_affect: EMA blend toward observed
    base = tag.expected_affect if tag.expected_affect is not None else tag.core_affect
    new_expected = base.lerp(observed, new_lr)

    return tag.model_copy(
        update={
            "expected_affect": new_expected,
            "prediction_learning_rate": new_lr,
        }
    )