`memory.retrieval`¶

memory.retrieval ¶

Wiring for blended recall-then-rerank over the MEMORY tree.

Memory-side counterpart to :mod:zettelkasten.retrieval. It builds the channel maps the pure scoring core (:mod:memory.reranking) consumes and adapts them to the memory tree's data shapes:

an importance_map — per-entry undirected degree centrality over the tree's has_child / related / invalidated_by edges (log1p-normalized to [0, 1]), restricted to POSITIVE scores so an isolated leaf carries no importance signal (absence, never a penalty);
a recency_map — an exponential-decay score over each entry's updated_at/created_at age, with a boost for entries touched in the current session (_session_accessed / _session_modified); and
a get_vector accessor backed by the entry-embeddings sidecar (:func:memory.storage.load_embeddings_sidecar), so the reranker's MMR diversity pass has real vectors and tolerates missing ones.

The memory tree is a SINGLE graph (unlike the zettelkasten's per-box graphs), so every candidate uses a constant source_graph ("") and the maps are keyed by the bare entry id — the reranker's bare-note_id fallback resolves them.

Non-punitive by default. :func:memory_rerank_config ships with w_importance == w_recency == 0 and mmr_lambda == 1.0, which the reranker guarantees reproduces pure similarity/keyword-RRF order. Turning the weights on (MEMORY_RERANK_W_IMPORTANCE / _W_RECENCY / _MMR_LAMBDA) is an opt-in. Every step degrades gracefully: a missing sidecar, empty session, or unreadable record never breaks retrieval, it just drops that channel's signal.

memory_rerank_config ¶

memory_rerank_config() -> RerankConfig

Build the memory reranker config from MEMORY_RERANK_* env vars.

Defaults are the non-punitive identity: w_importance = w_recency = 0 and mmr_lambda = 1.0 (pure similarity order, MMR off). Calendar recency is off for memory (entries carry no chronology distinct from created_at).

Env knobs

MEMORY_RERANK_W_SIM (default 1.0) MEMORY_RERANK_W_IMPORTANCE (default 0.0) MEMORY_RERANK_W_RECENCY (default 0.0) MEMORY_RERANK_MMR_LAMBDA (default 1.0 = MMR off) MEMORY_RERANK_OVERFETCH (default 3) MEMORY_RECENCY_HALF_LIFE_DAYS (default 30) — consumed by :func:build_recency_map, carried here for one-object policy.

Source code in memory/retrieval.py

def memory_rerank_config() -> RerankConfig:
    """Build the memory reranker config from ``MEMORY_RERANK_*`` env vars.

    Defaults are the non-punitive identity: ``w_importance = w_recency = 0`` and
    ``mmr_lambda = 1.0`` (pure similarity order, MMR off). Calendar recency is off
    for memory (entries carry no chronology distinct from ``created_at``).

    Env knobs:
        MEMORY_RERANK_W_SIM (default 1.0)
        MEMORY_RERANK_W_IMPORTANCE (default 0.0)
        MEMORY_RERANK_W_RECENCY (default 0.0)
        MEMORY_RERANK_MMR_LAMBDA (default 1.0 = MMR off)
        MEMORY_RERANK_OVERFETCH (default 3)
        MEMORY_RECENCY_HALF_LIFE_DAYS (default 30) — consumed by
            :func:`build_recency_map`, carried here for one-object policy.
    """
    overfetch = int(_env_float("MEMORY_RERANK_OVERFETCH", 3.0)) or 3
    return RerankConfig(
        w_sim=_env_float("MEMORY_RERANK_W_SIM", 1.0),
        w_importance=_env_float("MEMORY_RERANK_W_IMPORTANCE", 0.0),
        w_recency=_env_float("MEMORY_RERANK_W_RECENCY", 0.0),
        w_calendar_recency=0.0,
        mmr_lambda=_env_float("MEMORY_RERANK_MMR_LAMBDA", 1.0),
        overfetch=max(1, overfetch),
        calendar_half_life=_env_float("MEMORY_RECENCY_HALF_LIFE_DAYS", 30.0),
    )

rerank_disabled ¶

rerank_disabled(cfg: RerankConfig) -> bool

True when the config reproduces pure similarity order (no blend, no MMR).

Memory has no calendar channel, so it is disabled iff both boost weights are zero. (MMR is honored only when a boost is active — see :func:rerank_entries — matching the zettelkasten "no silent diversification" contract.)

Source code in memory/retrieval.py

def rerank_disabled(cfg: RerankConfig) -> bool:
    """True when the config reproduces pure similarity order (no blend, no MMR).

    Memory has no calendar channel, so it is disabled iff both boost weights are
    zero. (MMR is honored only when a boost is active — see :func:`rerank_entries`
    — matching the zettelkasten "no silent diversification" contract.)
    """
    return cfg.w_importance == 0.0 and cfg.w_recency == 0.0

build_importance_map ¶

build_importance_map(entries: list[dict]) -> dict[str, float]

Positive per-entry degree-centrality importance, keyed by entry id.

log1p of undirected degree, max-normalized to [0, 1]. Zero-degree (isolated) entries are DROPPED so they are absent from the importance channel rather than uniformly ranked last — preserving the non-punitive contract. The log is monotonic in degree so it does not change the channel's ordering (RRF is rank-based); anti-entrenchment comes from the importance WEIGHT and MMR.

Only entry ids are scored; non-entry edge endpoints (project roots) are dropped since they are never search candidates.

Source code in memory/retrieval.py

def build_importance_map(entries: list[dict]) -> dict[str, float]:
    """Positive per-entry degree-centrality importance, keyed by entry id.

    ``log1p`` of undirected degree, max-normalized to ``[0, 1]``. Zero-degree
    (isolated) entries are DROPPED so they are absent from the importance channel
    rather than uniformly ranked last — preserving the non-punitive contract. The
    log is monotonic in degree so it does not change the channel's ordering (RRF
    is rank-based); anti-entrenchment comes from the importance WEIGHT and MMR.

    Only entry ids are scored; non-entry edge endpoints (project roots) are
    dropped since they are never search candidates.
    """
    ids = {e.get("id") for e in entries if e.get("id")}
    degree = _undirected_degree(entries)
    scored = {
        nid: math.log1p(d) for nid, d in degree.items() if nid in ids and d > 0
    }
    normalized = _max_normalize(scored)
    return {nid: val for nid, val in normalized.items() if val > 0.0}

build_recency_map ¶

build_recency_map(entries: list[dict], *, touched_ids: 'Iterable[str] | None' = None, now: 'datetime | None' = None, half_life_days: float = 30.0) -> dict[str, float]

Exponential-decay recency per entry, keyed by entry id, in (0, 1].

Score is 0.5 ** (age_days / half_life_days) over the entry's updated_at (falling back to created_at) age — chronologically newer or more-recently-edited entries score higher. Entries touched in the current session (touched_ids: the union of _session_accessed and _session_modified) are boosted to 1.0 so work-in-progress surfaces first.

Non-punitive: recency is a boost channel gated by w_recency (0 by default). An entry with no parseable timestamp AND not session-touched is OMITTED (absent from the channel, never penalized). half_life_days <= 0 disables the age term (only session-touched entries score).

Source code in memory/retrieval.py

def build_recency_map(
    entries: list[dict],
    *,
    touched_ids: "Iterable[str] | None" = None,
    now: "datetime | None" = None,
    half_life_days: float = 30.0,
) -> dict[str, float]:
    """Exponential-decay recency per entry, keyed by entry id, in ``(0, 1]``.

    Score is ``0.5 ** (age_days / half_life_days)`` over the entry's
    ``updated_at`` (falling back to ``created_at``) age — chronologically newer
    or more-recently-edited entries score higher. Entries touched in the current
    session (``touched_ids``: the union of ``_session_accessed`` and
    ``_session_modified``) are boosted to ``1.0`` so work-in-progress surfaces
    first.

    Non-punitive: recency is a boost channel gated by ``w_recency`` (0 by
    default). An entry with no parseable timestamp AND not session-touched is
    OMITTED (absent from the channel, never penalized). ``half_life_days <= 0``
    disables the age term (only session-touched entries score).
    """
    touched = set(touched_ids or ())
    now = now or datetime.now(timezone.utc)
    rec: dict[str, float] = {}
    for e in entries:
        eid = e.get("id")
        if not eid:
            continue
        score = 0.0
        if half_life_days > 0:
            ts = _parse_ts(e.get("updated_at")) or _parse_ts(e.get("created_at"))
            if ts is not None:
                age_days = max(0.0, (now - ts).total_seconds() / 86400.0)
                score = 0.5 ** (age_days / half_life_days)
        if eid in touched:
            score = 1.0
        if score > 0.0:
            rec[eid] = score
    return rec

build_get_vector ¶

build_get_vector(vectors: dict[str, list[float]] | None) -> 'Callable[[str, str], list[float] | None]'

A (source_graph, note_id) -> vector | None accessor for the MMR pass.

Backed by the entry-embeddings sidecar map ({entry_id: vector}). Returns None for an unembedded entry or a missing map, which the reranker treats as "no diversity contribution" rather than an error. source_graph is ignored (memory is a single graph).

Source code in memory/retrieval.py

def build_get_vector(
    vectors: dict[str, list[float]] | None,
) -> "Callable[[str, str], list[float] | None]":
    """A ``(source_graph, note_id) -> vector | None`` accessor for the MMR pass.

    Backed by the entry-embeddings sidecar map (``{entry_id: vector}``). Returns
    ``None`` for an unembedded entry or a missing map, which the reranker treats
    as "no diversity contribution" rather than an error. ``source_graph`` is
    ignored (memory is a single graph).
    """
    vecs = vectors or {}

    def _get_vector(source_graph: str, note_id: str) -> "list[float] | None":
        return vecs.get(note_id)

    return _get_vector

rerank_entries ¶

rerank_entries(candidates: list[Any], entries: list[dict], *, cfg: RerankConfig | None = None, vectors: dict[str, list[float]] | None = None, touched_ids: 'Iterable[str] | None' = None, now: 'datetime | None' = None, top_k: 'int | None' = None) -> list[dict]

Rerank similarity-recalled memory candidates, returning breakdown dicts.

candidates are :class:memory.reranking.Candidate (or (id, similarity, source_graph[, keyword_tier]) tuples). entries is the full entry-record set (used to build the importance + recency channels over the whole tree, since a candidate's degree depends on neighbors outside the recalled set). Returns the reranker's per-candidate breakdown list (best-first), truncated to top_k.

Honors the disabled contract: when :func:rerank_disabled, it forces mmr_lambda = 1.0 and passes empty channel maps so the output is pure similarity order (dedup preserved), regardless of the configured lambda — mirroring :func:zettelkasten.retrieval.rerank_hits.

Source code in memory/retrieval.py

def rerank_entries(
    candidates: list[Any],
    entries: list[dict],
    *,
    cfg: RerankConfig | None = None,
    vectors: dict[str, list[float]] | None = None,
    touched_ids: "Iterable[str] | None" = None,
    now: "datetime | None" = None,
    top_k: "int | None" = None,
) -> list[dict]:
    """Rerank similarity-recalled memory candidates, returning breakdown dicts.

    ``candidates`` are :class:`memory.reranking.Candidate` (or
    ``(id, similarity, source_graph[, keyword_tier])`` tuples). ``entries`` is the
    full entry-record set (used to build the importance + recency channels over
    the whole tree, since a candidate's degree depends on neighbors outside the
    recalled set). Returns the reranker's per-candidate breakdown list
    (best-first), truncated to ``top_k``.

    Honors the disabled contract: when :func:`rerank_disabled`, it forces
    ``mmr_lambda = 1.0`` and passes empty channel maps so the output is pure
    similarity order (dedup preserved), regardless of the configured lambda —
    mirroring :func:`zettelkasten.retrieval.rerank_hits`.
    """
    if not candidates:
        return []
    cfg = cfg or memory_rerank_config()

    if rerank_disabled(cfg):
        pure_cfg = replace(cfg, mmr_lambda=1.0)
        results = rerank(
            candidates, importance_map={}, recency_map={}, cfg=pure_cfg,
            get_vector=None, limit=top_k,
        )
        return results[:top_k] if top_k is not None else results

    importance_map = build_importance_map(entries)
    recency_map = build_recency_map(
        entries, touched_ids=touched_ids, now=now,
        half_life_days=cfg.calendar_half_life,
    )
    get_vector = build_get_vector(vectors) if cfg.mmr_lambda < 1.0 else None

    results = rerank(
        candidates,
        importance_map=importance_map,
        recency_map=recency_map,
        cfg=cfg,
        get_vector=get_vector,
        limit=top_k,
    )
    return results[:top_k] if top_k is not None else results

memory.retrieval¶