Memory Versioning and Rollback: Tracking Changes to Agent Knowledge Over Time

Why Memory Needs Version Control

Agent memory is mutable. User preferences change, facts get corrected, and tasks evolve. When the agent updates a memory — say, changing a user's preferred language from Python to Rust — the old value is typically overwritten and lost. If the update was wrong (the agent misinterpreted the user), there is no way to recover.

Memory versioning solves this by treating every change as a new version rather than an overwrite. Like git for agent knowledge, it lets you inspect the history of any memory, understand how knowledge evolved, and roll back mistakes.

Version-Controlled Memory Store

Each memory item has a unique key. Every write creates a new version with an incrementing version number. The current state is the latest version.

flowchart TD
    START["Memory Versioning and Rollback: Tracking Changes …"] --> A
    A["Why Memory Needs Version Control"]
    A --> B
    B["Version-Controlled Memory Store"]
    B --> C
    C["Change Tracking"]
    C --> D
    D["Rollback"]
    D --> E
    E["Audit Trails"]
    E --> F
    F["Practical Usage"]
    F --> G
    G["FAQ"]
    G --> DONE["Key Takeaways"]
    style START fill:#4f46e5,stroke:#4338ca,color:#fff
    style DONE fill:#059669,stroke:#047857,color:#fff

from dataclasses import dataclass, field
from datetime import datetime
from copy import deepcopy


@dataclass
class MemoryVersion:
    version: int
    content: str
    timestamp: datetime
    author: str = "agent"
    change_reason: str = ""
    metadata: dict = field(default_factory=dict)


@dataclass
class VersionedMemory:
    key: str
    versions: list[MemoryVersion] = field(default_factory=list)

    @property
    def current(self) -> MemoryVersion | None:
        return self.versions[-1] if self.versions else None

    @property
    def version_count(self) -> int:
        return len(self.versions)


class VersionedMemoryStore:
    def __init__(self, max_versions_per_key: int = 50):
        self.memories: dict[str, VersionedMemory] = {}
        self.max_versions = max_versions_per_key
        self.global_changelog: list[dict] = []

    def write(
        self,
        key: str,
        content: str,
        author: str = "agent",
        reason: str = "",
        metadata: dict | None = None,
    ) -> int:
        if key not in self.memories:
            self.memories[key] = VersionedMemory(key=key)

        mem = self.memories[key]
        version_num = mem.version_count + 1
        version = MemoryVersion(
            version=version_num,
            content=content,
            timestamp=datetime.now(),
            author=author,
            change_reason=reason,
            metadata=metadata or {},
        )
        mem.versions.append(version)

        # Trim old versions if needed
        if len(mem.versions) > self.max_versions:
            mem.versions = mem.versions[-self.max_versions:]

        # Log to global changelog
        self.global_changelog.append({
            "key": key,
            "version": version_num,
            "timestamp": version.timestamp.isoformat(),
            "author": author,
            "reason": reason,
        })

        return version_num

Change Tracking

The changelog provides a complete audit trail of every modification. You can query it to understand how knowledge evolved and who made each change.

def read(self, key: str) -> str | None:
    mem = self.memories.get(key)
    if mem and mem.current:
        return mem.current.content
    return None


def history(self, key: str) -> list[MemoryVersion]:
    mem = self.memories.get(key)
    return mem.versions if mem else []


def diff(self, key: str, v1: int, v2: int) -> dict | None:
    mem = self.memories.get(key)
    if not mem:
        return None

    ver1 = next(
        (v for v in mem.versions if v.version == v1), None
    )
    ver2 = next(
        (v for v in mem.versions if v.version == v2), None
    )
    if not ver1 or not ver2:
        return None

    return {
        "key": key,
        "from_version": v1,
        "to_version": v2,
        "old_content": ver1.content,
        "new_content": ver2.content,
        "changed_by": ver2.author,
        "reason": ver2.change_reason,
        "time_between": str(ver2.timestamp - ver1.timestamp),
    }

Rollback

Rollback creates a new version with the content from a previous version. It does not delete the intermediate versions — the history is preserved, and the rollback itself is tracked.

def rollback(
    self, key: str, to_version: int, reason: str = ""
) -> int | None:
    mem = self.memories.get(key)
    if not mem:
        return None

    target = next(
        (v for v in mem.versions if v.version == to_version),
        None,
    )
    if not target:
        return None

    rollback_reason = (
        reason or f"Rolled back to version {to_version}"
    )
    return self.write(
        key=key,
        content=target.content,
        author="system",
        reason=rollback_reason,
        metadata={"rolled_back_from": mem.current.version},
    )

Audit Trails

The global changelog lets you reconstruct exactly how the agent's knowledge changed over any time window. This is invaluable for debugging unexpected behavior.

def audit_trail(
    self,
    start: datetime | None = None,
    end: datetime | None = None,
    author: str | None = None,
) -> list[dict]:
    trail = self.global_changelog
    if start:
        trail = [
            e for e in trail
            if datetime.fromisoformat(e["timestamp"]) >= start
        ]
    if end:
        trail = [
            e for e in trail
            if datetime.fromisoformat(e["timestamp"]) <= end
        ]
    if author:
        trail = [e for e in trail if e["author"] == author]
    return trail

Practical Usage

store = VersionedMemoryStore()

# Initial knowledge
store.write(
    "user_language",
    "Python",
    author="onboarding",
    reason="User stated preference during setup",
)

# Agent updates based on conversation
store.write(
    "user_language",
    "Rust",
    author="conversation_agent",
    reason="User said they switched to Rust",
)

# Oops — agent misunderstood. Roll back.
store.rollback(
    "user_language",
    to_version=1,
    reason="Agent misinterpreted — user meant Rust for a side project only",
)

# Inspect the full history
for v in store.history("user_language"):
    print(f"v{v.version}: {v.content} ({v.change_reason})")
# v1: Python (User stated preference during setup)
# v2: Rust (User said they switched to Rust)
# v3: Python (Rolled back to version 1)

FAQ

How many versions should I keep per memory key?

Keep 20 to 50 versions for frequently updated keys. For rarely changed keys like user preferences, keep all versions. Use the max_versions parameter to cap storage. When trimming, always keep the first version so you can see the original value.

Does versioning add significant overhead?

The storage overhead is modest — each version is just a content string plus metadata. The write latency is negligible because it is an append operation. The main cost is in history queries, which scan the version list. With 50 versions per key, this is instant.

Should rollback require human approval?

For production agents handling sensitive data, yes. Implement a rollback request that an admin reviews before it executes. For development and testing, automatic rollback is fine. The audit trail provides accountability either way.

---

#MemoryVersioning #Rollback #AuditTrail #Python #AgenticAI #LearnAI #AIEngineering