Resonance

emotional_memory.resonance.ResonanceConfig

Bases: BaseModel

Parameters for resonance link construction and spreading activation.

candidate_multiplier: int = 3 class-attribute instance-attribute

Pre-filter multiplier for encode-side resonance building.

When the store has more than max_links * candidate_multiplier entries, search_by_embedding is used to narrow candidates before scoring, keeping resonance-link construction sub-linear in store size.

causal_semantic_threshold: float = Field(default=0.6, ge=0.0, le=1.0) class-attribute instance-attribute

Minimum semantic similarity for causal link classification.

causal_temporal_threshold: float = Field(default=0.7, ge=0.0, le=1.0) class-attribute instance-attribute

Minimum temporal proximity for causal link classification.

contrastive_temporal_threshold: float = Field(default=0.5, ge=0.0, le=1.0) class-attribute instance-attribute

Minimum temporal proximity for contrastive link classification.

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

Minimum absolute valence difference for contrastive link classification.

hebbian_increment: float = Field(default=0.05, ge=0.0, le=0.5) class-attribute instance-attribute

Strength increment applied to links between co-retrieved memories.

"Neurons that fire together wire together" (Hebb, 1949). Set to 0.0 to disable Hebbian strengthening.

max_links: int = 5 class-attribute instance-attribute

Maximum resonance links per memory (both forward and backward).

propagation_hops: int = Field(default=2, ge=1, le=5) class-attribute instance-attribute

Number of hops for spreading activation (Collins & Loftus, 1975).

1 = direct neighbours only (one-hop check, minimal overhead). 2 = two hops, capturing indirect associations with strength decay. Higher values reach more distant associations but with diminishing returns since each hop multiplies by link strength (< 1.0).

temporal_half_life_seconds: float = Field(default=3600.0, gt=0) class-attribute instance-attribute

Half-life for temporal proximity decay. Must be positive.

threshold: float = 0.3 class-attribute instance-attribute

Minimum composite resonance score to create a link.

emotional_memory.resonance.build_resonance_links(new_memory: Memory, candidates: list[Memory], config: ResonanceConfig) -> list[ResonanceLink]

Build resonance links from new_memory to the most resonant candidates.

Excludes self-links. Returns up to config.max_links links above threshold, sorted by strength descending.

Note: This function only builds forward links (new_memory → candidate). The caller is responsible for creating the corresponding backward links on each target memory (candidate → new_memory) to maintain a bidirectional associative network.

Source code in src/emotional_memory/resonance.py

def build_resonance_links(
    new_memory: Memory,
    candidates: list[Memory],
    config: ResonanceConfig,
) -> list[ResonanceLink]:
    """Build resonance links from new_memory to the most resonant candidates.

    Excludes self-links. Returns up to config.max_links links above threshold,
    sorted by strength descending.

    Note: This function only builds *forward* links (new_memory → candidate).
    The caller is responsible for creating the corresponding *backward* links
    on each target memory (candidate → new_memory) to maintain a bidirectional
    associative network.
    """
    # Separate memories that have embeddings from those that do not.
    # Semantic similarities for the former are computed in a single
    # vectorised numpy operation instead of a Python loop.
    cands = [m for m in candidates if m.id != new_memory.id]
    if not cands:
        return []

    new_emb = new_memory.embedding
    if new_emb is not None:
        emb_cands = [m for m in cands if m.embedding is not None]
        if emb_cands:
            matrix = np.array([m.embedding for m in emb_cands], dtype=np.float64)
            q = np.array(new_emb, dtype=np.float64)
            q_norm = np.linalg.norm(q)
            if q_norm > 0:
                row_norms = np.linalg.norm(matrix, axis=1)
                safe_norms = np.where(row_norms > 0, row_norms, 1.0)
                sims = (matrix @ q) / (safe_norms * q_norm)
            else:
                sims = np.zeros(len(emb_cands))
            sem_sim_map: dict[str, float] = {m.id: float(sims[i]) for i, m in enumerate(emb_cands)}
        else:
            sem_sim_map = {}
    else:
        sem_sim_map = {}

    links: list[tuple[float, ResonanceLink]] = []

    for mem in cands:
        sem_sim = sem_sim_map.get(mem.id, 0.0)

        emo_sim = _emotional_similarity(new_memory.tag.core_affect, mem.tag.core_affect)
        temp_prox = temporal_proximity(
            new_memory.tag.timestamp,
            mem.tag.timestamp,
            config.temporal_half_life_seconds,
        )

        score = (
            config.semantic_weight * sem_sim
            + config.emotional_weight * emo_sim
            + config.temporal_weight * temp_prox
        )

        if score < config.threshold:
            continue

        # Causal: target (existing memory) preceded source (new memory) in time.
        # Since new_memory is just being encoded, existing memories almost always
        # satisfy this condition — gated further by temporal_prox and semantic_sim.
        target_precedes_source = mem.tag.timestamp < new_memory.tag.timestamp
        link_type = _classify_link_type(
            sem_sim,
            emo_sim,
            temp_prox,
            source_affect=new_memory.tag.core_affect,
            target_affect=mem.tag.core_affect,
            target_precedes_source=target_precedes_source,
            config=config,
        )
        link = ResonanceLink(
            source_id=new_memory.id,
            target_id=mem.id,
            strength=score,
            link_type=link_type,
        )
        links.append((score, link))

    # heapq.nlargest is O(n log k) vs O(n log n) for full sort when k << n
    top = heapq.nlargest(config.max_links, links, key=lambda t: t[0])
    return [lnk for _, lnk in top]

emotional_memory.resonance.spreading_activation(seed_ids: set[str], candidates: list[Memory], hops: int = 2) -> dict[str, float]

Compute activation levels via spreading activation from seed memories.

Collins & Loftus (1975): activation spreads from seed nodes through associative links with multiplicative strength decay at each hop. Because each link has strength < 1.0, activation diminishes naturally with distance — no explicit hop-discount factor is needed.

The adjacency is built from every memory's stored resonance_links, which include both forward (new→old) and backward (old→new) links when bidirectional link creation is used at encode time.

Parameters:

Name	Type	Description	Default
seed_ids	set[str]	IDs of initially activated memories (e.g. Pass 1 top-k).	required
candidates	list[Memory]	All candidate memories in the retrieval pool.	required
hops	int	Number of propagation steps. 1 = direct neighbours only.	2

Returns:

Type	Description
dict[str, float]	Mapping memory_id → activation level in (0, 1] for non-seed memories
dict[str, float]	that were reached by activation spreading. Seeds are excluded from
dict[str, float]	the returned map (they are already in the top-k).

Source code in src/emotional_memory/resonance.py

def spreading_activation(
    seed_ids: set[str],
    candidates: list[Memory],
    hops: int = 2,
) -> dict[str, float]:
    """Compute activation levels via spreading activation from seed memories.

    Collins & Loftus (1975): activation spreads from seed nodes through
    associative links with multiplicative strength decay at each hop.
    Because each link has strength < 1.0, activation diminishes naturally
    with distance — no explicit hop-discount factor is needed.

    The adjacency is built from every memory's stored resonance_links,
    which include both forward (new→old) and backward (old→new) links
    when bidirectional link creation is used at encode time.

    Args:
        seed_ids: IDs of initially activated memories (e.g. Pass 1 top-k).
        candidates: All candidate memories in the retrieval pool.
        hops: Number of propagation steps. 1 = direct neighbours only.

    Returns:
        Mapping memory_id → activation level in (0, 1] for non-seed memories
        that were reached by activation spreading. Seeds are excluded from
        the returned map (they are already in the top-k).
    """
    if not seed_ids or not candidates:
        return {}

    # Build adjacency: memory_id → list of (target_id, strength)
    # Only includes memories in candidates (retrieval pool).
    candidate_ids: set[str] = {m.id for m in candidates}
    adj: dict[str, list[tuple[str, float]]] = {}
    for mem in candidates:
        for link in mem.tag.resonance_links:
            if link.target_id in candidate_ids:
                adj.setdefault(mem.id, []).append((link.target_id, link.strength))

    if not adj:
        return {}

    # Spreading activation via BFS with activation levels.
    # Each node accumulates the maximum activation reaching it (not the sum),
    # preventing artificial inflation from multiple converging paths.
    current: dict[str, float] = {sid: 1.0 for sid in seed_ids if sid in candidate_ids}
    accumulated: dict[str, float] = {}

    for _ in range(hops):
        if not current:
            break
        next_wave: dict[str, float] = {}
        for node_id, node_activation in current.items():
            for target_id, strength in adj.get(node_id, []):
                if target_id in seed_ids:
                    continue  # do not feed activation back into seeds
                spread = node_activation * strength
                if spread > next_wave.get(target_id, 0.0):
                    next_wave[target_id] = spread
        # Merge into accumulated, taking maximum
        for tid, val in next_wave.items():
            new_val = min(1.0, val)
            if new_val > accumulated.get(tid, 0.0):
                accumulated[tid] = new_val
        current = next_wave

    return accumulated

emotional_memory.resonance.hebbian_strengthen(memory: Memory, co_retrieved_ids: set[str], increment: float) -> list[ResonanceLink]

Strengthen links to co-retrieved memories (Hebb, 1949).

"Neurons that fire together wire together" — when two linked memories are both retrieved in the same query, their associative link is incrementally strengthened up to a maximum of 1.0.

Parameters:

Name	Type	Description	Default
memory	Memory	The memory whose links are being updated.	required
co_retrieved_ids	set[str]	IDs of the other memories retrieved in the same query (excluding the memory itself).	required
increment	float	Strength increment per co-retrieval event.	required

Returns:

Type	Description
list[ResonanceLink]	Updated resonance_links list. Returns the original list unchanged
list[ResonanceLink]	if no links were strengthened (avoids unnecessary copies).

Source code in src/emotional_memory/resonance.py

def hebbian_strengthen(
    memory: Memory,
    co_retrieved_ids: set[str],
    increment: float,
) -> list[ResonanceLink]:
    """Strengthen links to co-retrieved memories (Hebb, 1949).

    "Neurons that fire together wire together" — when two linked memories
    are both retrieved in the same query, their associative link is
    incrementally strengthened up to a maximum of 1.0.

    Args:
        memory: The memory whose links are being updated.
        co_retrieved_ids: IDs of the other memories retrieved in the same
            query (excluding the memory itself).
        increment: Strength increment per co-retrieval event.

    Returns:
        Updated resonance_links list. Returns the original list unchanged
        if no links were strengthened (avoids unnecessary copies).
    """
    if not co_retrieved_ids or increment == 0.0:
        return list(memory.tag.resonance_links)

    updated: list[ResonanceLink] = []
    changed = False
    for link in memory.tag.resonance_links:
        if link.target_id in co_retrieved_ids:
            new_strength = min(1.0, link.strength + increment)
            if new_strength != link.strength:
                updated.append(link.model_copy(update={"strength": new_strength}))
                changed = True
                continue
        updated.append(link)

    return updated if changed else list(memory.tag.resonance_links)

emotional_memory.resonance.temporal_proximity(t1: datetime, t2: datetime, half_life_seconds: float = 3600.0) -> float

Exponential decay of temporal proximity.

proximity = exp(-|delta_t| * ln(2) / half_life) Returns 1.0 for simultaneous events, approaches 0 for distant events.

Source code in src/emotional_memory/resonance.py

def temporal_proximity(t1: datetime, t2: datetime, half_life_seconds: float = 3600.0) -> float:
    """Exponential decay of temporal proximity.

    proximity = exp(-|delta_t| * ln(2) / half_life)
    Returns 1.0 for simultaneous events, approaches 0 for distant events.
    """
    if half_life_seconds <= 0:
        return 0.0
    delta = abs((t1 - t2).total_seconds())
    return math.exp(-delta * math.log(2) / half_life_seconds)