cortexm/api/long_recall.py
Rama main · réplica de lectura del repositorio original.
"""Long-context recall — "memory past 20 steps".
This is the killer feature the user asked for explicitly:
> "use my token to submit the core of whatever a user needs
> should be the core attempt to get memory and recall past 20 steps"
Problem: every mainstream LLM has a context window that scrolls.
After ~20 turns of conversation (~8k tokens for a 32k-window model),
the early turns have scrolled out of the prompt. The LLM forgets
what was said at turn 1 by turn 25. This is the #1 pain point for
anyone building agents on top of an LLM — the model literally
cannot remember what it committed to two turns ago, let alone last
session.
cortexm's answer:
Every fact the extractor derives from each turn is persisted to
the bi-temporal Trace with a step number (run_id encodes the
session, agent_id encodes the agent, the chunk's `created_at`
encodes the step order). On retrieval, we DON'T just query by
relevance — we BIAS toward facts whose step is about to scroll
out of the LLM's window. A fact from step 5 in a 30-turn
conversation is far more valuable to surface now than a fact from
step 28 — the LLM still has step 28 in its prompt.
This is the asymmetric retrieval insight: the closer a fact is to
scrolling out, the more we should boost it. The standard cosine
score ranks by similarity; we multiply by a step-distance decay
that peaks at the LLM's window edge.
Concrete API:
m.recall_step(query, user_id="alice", current_step=30,
window=20, k=12)
→ returns the top-k facts RELEVANT to the query AND in danger of
scrolling out of the LLM's window.
m.stepped_context_block(query, user_id="alice", current_step=30,
window=20, k=12)
→ returns a ready-to-inject markdown context block:
## Recalled memory (steps 1–10 about to scroll out)
- **[step 5]** Alice works at Google (conf=0.92, valid_from=2026-08-01)
- **[step 8]** Alice's birthday is 1990-05-12 (conf=0.88)
- **[step 11]** Alice prefers Python (conf=0.85)
...
This is the "memory past 20 steps" UX. Drop it into your agent's
system prompt template, and the LLM never forgets — even if it
physically scrolled the early turns out of its window.
Lean and simple: ~200 LoC, pure Python, μ=0 (no LLM, deterministic
step-distance decay multiplied onto the existing VSA fusion score).
"""
from __future__ import annotations
import datetime as _dt
import math
from datetime import datetime, timezone
from typing import Any
from cortexm.util import parse_ts
def _step_for_fact(fact, store) -> int | None:
"""Best-effort step number for a fact. We use chunk.created_at
ordinal within the (user_id, agent_id, run_id) scope — the order
in which the chunk entered the Trace. If the store doesn't expose
a step field directly, we synthesize one from created_at ordering.
Returns None if no chunk is attached (machine-derived facts
with no source — these don't have a meaningful step number).
"""
if not fact.source_id:
return None
try:
chunk = store.get_chunk(fact.source_id)
if not chunk:
return None
# if the schema already has a step column, use it
step = chunk.get("step") if isinstance(chunk, dict) else None
if step is not None:
return int(step)
# otherwise use created_at ordering — query the store for
# the rank of this chunk's created_at among same-scope chunks
ca = chunk.get("created_at") if isinstance(chunk, dict) else None
if not ca:
return None
try:
row = store.conn.execute(
"SELECT COUNT(*) AS n FROM chunks "
"WHERE user_id=? AND created_at < ?",
(fact.user_id, ca)).fetchone()
return int(row["n"]) + 1 if row else None
except Exception:
return None
except Exception:
return None
def _step_distance_boost(step: int | None, current_step: int,
window: int) -> float:
"""Asymmetric retrieval: boost facts close to scrolling out of
the LLM's window.
If current_step = 30 and window = 20, the LLM still sees steps
11..30. Step 10 is ABOUT to scroll out — boost it most. Step 1
has already scrolled out — boost it strongly too. Step 25 is
still in the LLM's prompt — boost it less (the LLM already has it).
The boost function is a Gaussian centered at (current_step -
window), so it peaks at the window edge:
boost(step) = 1 + peak * exp(-(step - (current_step - window))^2
/ (2 * sigma^2))
Plus a constant floor of 1.0 (we never zero-out a fact — the
underlying VSA score still ranks it; we only nudge ordering).
"""
if step is None:
return 1.0 # no step info → no boost, no penalty
if current_step <= 0 or window <= 0:
return 1.0
edge = max(0, current_step - window)
sigma = max(1.0, window / 3.0) # spread = 1/3 of window
peak = 0.6 # max +60% boost at the window edge
# Gaussian centered at the edge, but also weighted for already-
# scrolled-out facts (step < edge) — they get full peak boost
# because the LLM has zero access to them now.
if step <= edge:
return 1.0 + peak
return 1.0 + peak * math.exp(
-((step - edge) ** 2) / (2 * sigma * sigma))
def recall_step(memory, query: str, *, user_id: str | None = None,
agent_id: str | None = None, run_id: str | None = None,
current_step: int = 0, window: int = 20,
k: int = 12) -> dict:
"""Asymmetric retrieval: top-k facts RELEVANT to the query AND in
danger of scrolling out of the LLM window.
Pipeline:
1. Run the standard reader.search() — produces top-K candidates
ranked by VSA + symbolic fusion.
2. For each candidate fact, compute the step-distance boost.
3. Re-rank by (fusion_score * boost). Top-k wins.
Returns a dict shaped like ``m.search()`` but with extra fields
per memory:
- step: int | None
- step_distance_boost: float
- scrolled_out: bool (True if step <= current_step - window)
"""
user_id = user_id or memory.config.default_user_id
# over-fetch then re-rank — we want enough candidates so the
# step-distance re-ordering has room to surface scrolled-out facts
raw_k = max(k * 4, 24)
result = memory.reader.search(query, user_id=user_id, agent_id=agent_id,
run_id=run_id, k=raw_k)
scored: list[tuple[float, Any, int | None, float, bool]] = []
for f in result.facts:
step = _step_for_fact(f, memory.store)
boost = _step_distance_boost(step, current_step, window)
# underlying fusion score (cosine + symbolic + chunk_recall)
fs = float(getattr(f, "score", 0.0) or
getattr(f, "fusion_score", 0.0) or 0.0)
# if the reader didn't attach a score, fall back to confidence
if fs <= 0.0:
fs = float(getattr(f, "confidence", 0.0) or 0.0)
new_score = fs * boost
scrolled_out = step is not None and current_step > 0 and step <= (
current_step - window)
scored.append((new_score, f, step, boost, scrolled_out))
scored.sort(key=lambda x: -x[0])
top = scored[:k]
out_memories = []
for new_score, f, step, boost, scrolled_out in top:
out_memories.append({
"id": f.id,
"memory": f"{f.subject} | {f.relation} | {f.value}",
"step": step,
"step_distance_boost": round(boost, 3),
"scrolled_out": scrolled_out,
"fusion_score": round(new_score, 4),
"confidence": float(getattr(f, "confidence", 0.0) or 0.0),
"valid_from": str(f.valid_from) if f.valid_from else None,
"valid_to": str(f.valid_to) if f.valid_to else None,
"source_snippet": _snippet(memory, f),
})
return {
"query": query,
"user_id": user_id,
"current_step": current_step,
"window": window,
"results": out_memories,
"context_block": _format_context_block(
out_memories, current_step, window),
"llm_calls": 0,
}
def _snippet(memory, f) -> str:
if not f.source_id:
return ""
chunk = memory.store.get_chunk(f.source_id)
if chunk and chunk.get("text"):
return chunk["text"][:160]
return ""
def _format_context_block(mems: list[dict], current_step: int,
window: int) -> str:
"""Render a markdown context block ready to inject into the LLM
system prompt. Grouped by 'scrolled out' vs 'in window' so the
model sees the structure clearly."""
if not mems:
return ""
edge = max(0, current_step - window)
scrolled = [m for m in mems if m["scrolled_out"]]
in_win = [m for m in mems if not m["scrolled_out"]]
lines = []
if scrolled:
lines.append(f"## Recalled memory (steps 1–{edge} — scrolled "
f"out of context window)")
for m in scrolled:
lines.append(f"- **[step {m['step']}]** {m['memory']} "
f"(conf={m['confidence']:.2f}, "
f"valid_from={m.get('valid_from', 'n/a')})")
if m.get("source_snippet"):
lines.append(f" > …{m['source_snippet'][:120]}")
if in_win:
lines.append(f"\n## Active memory (steps {edge+1}–{current_step} "
f"— still in window, surfaced for relevance)")
for m in in_win:
lines.append(f"- **[step {m['step']}]** {m['memory']} "
f"(conf={m['confidence']:.2f})")
return "\n".join(lines)
def stepped_context_block(memory, query: str, *,
user_id: str | None = None,
current_step: int = 0,
window: int = 20, k: int = 12) -> str:
"""Convenience wrapper — returns just the context_block string."""
return recall_step(memory, query, user_id=user_id,
current_step=current_step, window=window,
k=k)["context_block"]