perf: incremental profile rebuilds — fold new gists instead of re-digesting all
The profile pass map-reduced every session gist (~851) on every consolidation firing — the biggest redundant-work and MI50-heat source left after the eras fix. Now: skip when nothing's new, fold only the gists added since last build into the existing profile, and full-rebuild only when there's no profile, too much has accumulated to fold safely (>FOLD_LIMIT), on a periodic cadence (anti-drift), or when forced. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+58
-14
@@ -26,6 +26,19 @@ Organize under these headings: Poker Style, Leaks & Tendencies, Mental Game, \
|
||||
Personal Context, Working With Brian. Keep it tight — bullets, no fluff, no \
|
||||
repetition. Resolve contradictions toward the more recent/frequent signal."""
|
||||
|
||||
_FOLD_PROMPT = """Update Brian's existing profile with new facts from his most \
|
||||
recent sessions. Keep the same headings (Poker Style, Leaks & Tendencies, Mental \
|
||||
Game, Personal Context, Working With Brian). Integrate genuinely new durable facts, \
|
||||
strengthen or revise existing bullets where the new sessions confirm or contradict \
|
||||
them (favor the more recent signal), and drop nothing that's still true. Keep it \
|
||||
tight — bullets, no fluff, no repetition. Return the full updated profile."""
|
||||
|
||||
# A long gap (consolidation hasn't run in ages) folds too much at once to trust the
|
||||
# delta path; rebuild from scratch instead. And cross every Nth session do a full
|
||||
# rebuild regardless, so accumulated small folds can't fossilize stale facts.
|
||||
FOLD_LIMIT = 25
|
||||
FULL_REBUILD_EVERY = 100
|
||||
|
||||
|
||||
def _batch_texts(texts: list[str], budget: int) -> list[str]:
|
||||
"""Group texts into joined blocks under `budget` chars."""
|
||||
@@ -49,26 +62,57 @@ def _call(prompt: str, body: str, backend: Backend) -> str:
|
||||
return llm.complete(messages, backend=backend)
|
||||
|
||||
|
||||
def rebuild_profile(backend: Backend | None = None) -> str | None:
|
||||
"""Re-derive the profile from all current session gists and store it."""
|
||||
def _map_reduce(gists: list[str], backend: Backend) -> str:
|
||||
"""MAP: extract facts from batches of gists. REDUCE: fold to one fact list."""
|
||||
partials = [_call(_MAP_PROMPT, b, backend) for b in _batch_texts(gists, BATCH_CHARS)]
|
||||
while len(partials) > 1:
|
||||
partials = [_call(_REDUCE_PROMPT, g, backend) for g in _batch_texts(partials, BATCH_CHARS)]
|
||||
return partials[0]
|
||||
|
||||
|
||||
def _full_rebuild(gists: list[str], backend: Backend) -> str:
|
||||
"""Re-derive the whole profile from every gist (the expensive path)."""
|
||||
profile = _map_reduce(gists, backend)
|
||||
memory.set_profile(profile, len(gists))
|
||||
logbus.log("info", "profile rebuilt", sessions=len(gists), chars=len(profile))
|
||||
return profile
|
||||
|
||||
|
||||
def _fold(existing: str, new_gists: list[str], total: int, backend: Backend) -> str:
|
||||
"""Fold only the new session gists into the existing profile (the cheap path)."""
|
||||
facts = _map_reduce(new_gists, backend)
|
||||
body = f"EXISTING PROFILE:\n{existing}\n\nNEW FACTS FROM RECENT SESSIONS:\n{facts}"
|
||||
profile = _call(_FOLD_PROMPT, body, backend)
|
||||
memory.set_profile(profile, total)
|
||||
logbus.log("info", "profile folded", added=len(new_gists), total=total, chars=len(profile))
|
||||
return profile
|
||||
|
||||
|
||||
def rebuild_profile(backend: Backend | None = None, force: bool = False) -> str | None:
|
||||
"""Derive Brian's profile from session gists. Incremental by default: if a profile
|
||||
already exists, fold only the gists added since it was last built instead of
|
||||
re-digesting all of them every consolidation pass (the old behavior re-read ~851
|
||||
sessions each time — the biggest redundant-work / MI50-heat source). Falls back to
|
||||
a full rebuild when there's no profile yet, too much has accumulated to fold safely,
|
||||
on a periodic cadence (anti-drift), or when `force=True`."""
|
||||
backend = backend or config.load().summary_backend
|
||||
summaries = memory.list_summaries()
|
||||
if not summaries:
|
||||
return None
|
||||
total = len(summaries)
|
||||
existing = memory.get_profile()
|
||||
covered = memory.profile_sessions_covered()
|
||||
|
||||
# MAP: extract facts from batches of gists.
|
||||
blocks = _batch_texts([s.content for s in summaries], BATCH_CHARS)
|
||||
partials = [_call(_MAP_PROMPT, b, backend) for b in blocks]
|
||||
logbus.log("info", "profile map done", batches=len(partials), sessions=len(summaries))
|
||||
if existing and not force and 0 < covered <= total:
|
||||
new = total - covered
|
||||
if new == 0:
|
||||
logbus.log("info", "profile unchanged", sessions=total)
|
||||
return existing # nothing new since last build — skip entirely
|
||||
crosses_cadence = total // FULL_REBUILD_EVERY != covered // FULL_REBUILD_EVERY
|
||||
if new <= FOLD_LIMIT and not crosses_cadence:
|
||||
return _fold(existing, [s.content for s in summaries[covered:]], total, backend)
|
||||
|
||||
# REDUCE: fold partials together until one remains.
|
||||
while len(partials) > 1:
|
||||
partials = [_call(_REDUCE_PROMPT, g, backend) for g in _batch_texts(partials, BATCH_CHARS)]
|
||||
profile = partials[0]
|
||||
|
||||
memory.set_profile(profile, len(summaries))
|
||||
logbus.log("info", "profile rebuilt", sessions=len(summaries), chars=len(profile))
|
||||
return profile
|
||||
return _full_rebuild([s.content for s in summaries], backend)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
"""Profile derivation: fold only new gists into the existing profile (incremental).
|
||||
|
||||
The old pass re-digested all ~851 gists every consolidation; this checks the cheap
|
||||
delta path fires in steady state and the full rebuild fires only when it should.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
|
||||
import pytest
|
||||
|
||||
from lyra.memory import Summary
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def prof(monkeypatch):
|
||||
import lyra.profile as profile
|
||||
importlib.reload(profile)
|
||||
return profile
|
||||
|
||||
|
||||
def _wire(profile, monkeypatch, gists, covered, existing):
|
||||
"""Stub memory + the LLM passes; record which path ran."""
|
||||
state = {"stored_content": existing, "stored_covered": covered, "calls": []}
|
||||
|
||||
monkeypatch.setattr(profile.memory, "list_summaries",
|
||||
lambda: [Summary(f"s{i}", g, i, "t") for i, g in enumerate(gists)])
|
||||
monkeypatch.setattr(profile.memory, "get_profile", lambda: state["stored_content"])
|
||||
monkeypatch.setattr(profile.memory, "profile_sessions_covered", lambda: state["stored_covered"])
|
||||
|
||||
def set_profile(content, sessions_covered, profile_id="self"):
|
||||
state["stored_content"], state["stored_covered"] = content, sessions_covered
|
||||
monkeypatch.setattr(profile.memory, "set_profile", set_profile)
|
||||
|
||||
monkeypatch.setattr(profile, "_map_reduce",
|
||||
lambda gists, backend: state["calls"].append(("map_reduce", len(gists))) or "facts")
|
||||
monkeypatch.setattr(profile, "_call",
|
||||
lambda prompt, body, backend: state["calls"].append(("fold",)) or "folded profile")
|
||||
return state
|
||||
|
||||
|
||||
def test_no_profile_yet_does_full_rebuild(prof, monkeypatch):
|
||||
state = _wire(prof, monkeypatch, gists=["a", "b", "c"], covered=0, existing=None)
|
||||
out = prof.rebuild_profile(backend="local")
|
||||
assert state["calls"] == [("map_reduce", 3)] # mapped all three gists
|
||||
assert out == "facts" and state["stored_covered"] == 3
|
||||
|
||||
|
||||
def test_unchanged_skips_entirely(prof, monkeypatch):
|
||||
state = _wire(prof, monkeypatch, gists=["a", "b"], covered=2, existing="old profile")
|
||||
out = prof.rebuild_profile(backend="local")
|
||||
assert state["calls"] == [] # no LLM work at all
|
||||
assert out == "old profile"
|
||||
|
||||
|
||||
def test_small_delta_folds_only_new(prof, monkeypatch):
|
||||
state = _wire(prof, monkeypatch, gists=["a", "b", "c", "d"], covered=2, existing="old profile")
|
||||
out = prof.rebuild_profile(backend="local")
|
||||
assert state["calls"] == [("map_reduce", 2), ("fold",)] # mapped just the 2 new, then folded
|
||||
assert out == "folded profile" and state["stored_covered"] == 4
|
||||
|
||||
|
||||
def test_force_does_full_rebuild(prof, monkeypatch):
|
||||
state = _wire(prof, monkeypatch, gists=["a", "b", "c"], covered=3, existing="old profile")
|
||||
out = prof.rebuild_profile(backend="local", force=True)
|
||||
assert state["calls"] == [("map_reduce", 3)] # ignored the up-to-date profile
|
||||
assert out == "facts"
|
||||
|
||||
|
||||
def test_big_gap_falls_back_to_full_rebuild(prof, monkeypatch):
|
||||
gists = [str(i) for i in range(40)] # 30 new > FOLD_LIMIT
|
||||
state = _wire(prof, monkeypatch, gists=gists, covered=10, existing="old profile")
|
||||
out = prof.rebuild_profile(backend="local")
|
||||
assert state["calls"] == [("map_reduce", 40)] # full rebuild, not a giant fold
|
||||
assert out == "facts"
|
||||
|
||||
|
||||
def test_crossing_cadence_forces_full_rebuild(prof, monkeypatch):
|
||||
# covered=98, total=102 is a tiny delta, but it crosses the 100-session boundary.
|
||||
gists = [str(i) for i in range(102)]
|
||||
state = _wire(prof, monkeypatch, gists=gists, covered=98, existing="old profile")
|
||||
out = prof.rebuild_profile(backend="local")
|
||||
assert state["calls"] == [("map_reduce", 102)] # anti-drift full rebuild
|
||||
assert out == "facts"
|
||||
Reference in New Issue
Block a user