Memory Consolidation

Merge duplicates, promote durable facts, and prune stale entries under a loss-bounded guard

Memory consolidation runs a scheduled, off-hot-path pass that merges duplicate memories, promotes durable facts, and prunes stale entries — refusing any rewrite that would drop more than a set fraction of your store.

Long-lived agents capture memories inline during a turn, so the store accretes near-duplicate, never-pruned entries that degrade recall and grow token cost. Consolidation is the background maintenance pass that cleans it up safely.

Quick Start

1. Check a result under the loss guard

   Construct a ConsolidationResult, read loss_fraction, and ask the guard whether a pass drops too much.

   from praisonaiagents.memory import ConsolidationResult
   
   result = ConsolidationResult(entries_before=100, entries_after=60)
   
   print(result.loss_fraction)        # 0.4
   print(result.exceeds_loss(0.25))   # True  → too lossy, reject it
   print(result.exceeds_loss(0.5))    # False → within budget

2. Wire a consolidator to an Agent

   Implement MemoryConsolidationProtocol and hand it your agent’s memory on a schedule — the agent keeps chatting while consolidation runs in the background.

   from praisonaiagents import Agent
   from praisonaiagents.memory import ConsolidationResult, MemoryConsolidationProtocol
   
   class MyConsolidator:
       def consolidate(self, memory, *, max_loss_fraction=0.25):
           before = memory.get_all_memories()
           # merge/promote/prune here, tracking surviving originals as `retained`
           after, retained = before, len(before)
           result = ConsolidationResult(
               entries_before=len(before),
               entries_after=len(after),
               retained_originals=retained,
           )
           if result.exceeds_loss(max_loss_fraction):
               result.rejected = True
               result.reason = "loss guard tripped"
               return result          # store left untouched
           return result
   
   agent = Agent(name="Support", instructions="Help the user.", memory=True)
   
   # Later, off the reply path (e.g. a nightly job):
   consolidator = MyConsolidator()
   result = consolidator.consolidate(agent._memory_instance)
   print(result.merged, result.pruned, result.rejected)

How It Works

Consolidation is background maintenance: the user never invokes it mid-conversation — a scheduler runs it off-turn, and the loss guard decides whether the rewrite is applied.

The protocols are @runtime_checkable, so isinstance works at runtime, and both consolidate and aconsolidate carry a resolvable ConsolidationResult return annotation for introspection.

Configuration Options

ConsolidationResult reports what a pass did and carries the loss-guard math. No feature_configs.py entry exists — this is a protocol, not a config class — so these are the fields the dataclass exposes.

Methods:

Loss-Guard Semantics

The loss guard is the invariant that keeps a bad rewrite from wiping your memory.

Originals-based vs net-count loss. loss_fraction normally divides the drop in entries by entries_before. When you set retained_originals, it instead measures how many original entries survived — which catches destructive rewrites that keep the count the same.

from praisonaiagents.memory import ConsolidationResult

# Net-count loss: 100 → 60 removes 40%.
ConsolidationResult(entries_before=100, entries_after=60).loss_fraction  # 0.4

# Equal-size but destructive: 100 in, 100 out, but zero originals kept → 100% loss.
r = ConsolidationResult(entries_before=100, entries_after=100, retained_originals=0)
r.loss_fraction        # 1.0
r.exceeds_loss(0.25)   # True

# Partial retention: 100 → 90, 80 originals survive → 20% loss.
r = ConsolidationResult(entries_before=100, entries_after=90, retained_originals=80)
r.loss_fraction        # 0.2
r.exceeds_loss(0.25)   # False

Empty-store safety. An empty store reports 0.0, never a divide-by-zero, and growth never reports negative loss.

ConsolidationResult(entries_before=0, entries_after=0).loss_fraction    # 0.0
ConsolidationResult(entries_before=10, entries_after=15).loss_fraction  # 0.0

Fail-closed threshold validation. An invalid threshold can never silently permit a destructive rewrite — exceeds_loss raises ValueError on NaN, inf, out-of-range, or non-numeric input.

result = ConsolidationResult(entries_before=100, entries_after=60)

for bad in (float("nan"), float("inf"), -0.1, 1.5, True, "x"):
    try:
        result.exceeds_loss(bad)
    except ValueError as e:
        print("rejected:", bad)

Common Patterns

Implement the sync protocol and reject rewrites that trip the guard:

from praisonaiagents.memory import ConsolidationResult

class SyncConsolidator:
    def consolidate(self, memory, *, max_loss_fraction=0.25):
        before = memory.get_all_memories()
        after, retained = merge_and_prune(before)   # your logic
        result = ConsolidationResult(
            entries_before=len(before),
            entries_after=len(after),
            retained_originals=retained,
        )
        if result.exceeds_loss(max_loss_fraction):
            result.rejected = True
            result.reason = "loss guard tripped"
            return result
        apply(after)                                 # persist the rewrite
        return result

Use the async protocol when your store or LLM calls are awaitable:

from praisonaiagents.memory import ConsolidationResult

class AsyncConsolidator:
    async def aconsolidate(self, memory, *, max_loss_fraction=0.25):
        before = await memory.aget_all_memories()
        after, retained = await amerge_and_prune(before)
        result = ConsolidationResult(
            entries_before=len(before),
            entries_after=len(after),
            retained_originals=retained,
        )
        if result.exceeds_loss(max_loss_fraction):
            result.rejected = True
            result.reason = "loss guard tripped"
            return result
        await aapply(after)
        return result

Run consolidation off the hot path behind a scheduler — the actual scheduler wiring lives in praisonai-plugins:

# Pseudo-code — praisonai-plugins wires this to a cron like "0 3 * * *".
def nightly_consolidation(agent, consolidator):
    result = consolidator.consolidate(
        agent._memory_instance,
        max_loss_fraction=0.25,
    )
    log.info("consolidation", extra=result.context)
    return result

Verify structural typing at runtime — both protocols are @runtime_checkable:

from praisonaiagents.memory import (
    MemoryConsolidationProtocol,
    AsyncMemoryConsolidationProtocol,
)

isinstance(SyncConsolidator(), MemoryConsolidationProtocol)        # True
isinstance(AsyncConsolidator(), AsyncMemoryConsolidationProtocol)  # True

Sync vs Async

Pick the protocol that matches how your memory store and consolidation logic run.

Layering

Consolidation is built in three layers so the core stays dependency-free.

Best Practices

Related