{ "nbformat": 4, "nbformat_minor": 5, "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.13.0" }, "blog_metadata": { "topic": "Your agents are only as good as their memory: what EvoLib gets right about enterprise learning loops", "slug": "your-agents-are-only-as-good-as-their-memory-what-evolib-get", "generated_by": "LinkedIn Post Generator + Azure OpenAI", "generated_at": "2026-08-04T14:27:10.596Z" } }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Your agents are only as good as their memory: what EvoLib gets right about enterprise learning loops\n", "\n", "This notebook turns the blog post into a hands-on validation of a core enterprise AI claim: agent memory is not just an AI feature, it is a governed data lifecycle. We will implement simple Python examples for memory write gating, policy-aware retrieval, learning-loop promotion, governance gap reporting, stale-memory review checks, and executive scorecards. The goal is to make the ideas testable: provenance, approval, retention, auditability, and revocation readiness." ] }, { "cell_type": "code", "metadata": {}, "source": [ "%pip install -q pandas" ], "execution_count": null, "outputs": [] }, { "cell_type": "code", "metadata": {}, "source": [ "from dataclasses import dataclass, asdict\n", "from datetime import date, datetime, timedelta\n", "import hashlib\n", "import json\n", "import pandas as pd" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Governance gate flow\n", "\n", "The blog argues that memory writes should never go directly into retrieval. Instead, the agent proposes a memory candidate, a governance gate validates required metadata and approval state, an immutable audit event is appended, and only then is the record indexed for future retrieval.\n", "\n", "Below we encode that flow as a simple Python structure so it can be inspected and reused in later cells." ] }, { "cell_type": "code", "metadata": {}, "source": [ "governance_gate_flow = {\n", " \"Agent proposes memory write\": \"Governance gate\",\n", " \"Governance gate\": {\n", " \"Required metadata present?\": {\n", " \"No\": \"Reject + reason\",\n", " \"Yes\": {\n", " \"Approval state valid?\": {\n", " \"No\": \"Reject + reason\",\n", " \"Yes\": [\n", " \"Append immutable audit event\",\n", " \"Index approved memory\",\n", " \"Retrieval with policy filters\"\n", " ]\n", " }\n", " }\n", " }\n", " }\n", "}\n", "\n", "governance_gate_flow" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Governed memory-write gate\n", "\n", "This example implements the minimum controls for durable enterprise memory: provenance, ownership, classification, retention, approval, and immutable audit. If a record is missing required metadata or fails policy checks, it should be rejected before indexing." ] }, { "cell_type": "code", "metadata": {}, "source": [ "# Governed memory-write gate with provenance, ownership, classification, retention, approval, and immutable audit.\n", "from dataclasses import dataclass, asdict\n", "from datetime import date\n", "import hashlib, json\n", "\n", "@dataclass\n", "class MemoryRecord:\n", " content: str\n", " provenance: str\n", " owner: str\n", " classification: str\n", " retention_until: str\n", " approval_state: str\n", "\n", "def validate(record: MemoryRecord) -> None:\n", " allowed = {\"public\", \"internal\", \"confidential\"}\n", " assert all(asdict(record).values()), \"missing required field\"\n", " assert record.classification in allowed, \"invalid classification\"\n", " assert date.fromisoformat(record.retention_until) >= date.today(), \"expired retention\"\n", " assert record.approval_state == \"approved\", \"not approved\"\n", "\n", "def append_audit(record: MemoryRecord) -> dict:\n", " payload = json.dumps(asdict(record), sort_keys=True).encode()\n", " return {\"event\": \"memory_write_approved\", \"hash\": hashlib.sha256(payload).hexdigest()}\n", "\n", "record = MemoryRecord(\n", " \"Runbook: rotate keys quarterly\",\n", " \"ticket:CHG-1042\",\n", " \"secops\",\n", " \"internal\",\n", " \"2027-12-31\",\n", " \"approved\"\n", ")\n", "validate(record)\n", "audit_event = append_audit(record)\n", "index_doc = {\"record\": asdict(record), \"audit\": audit_event}\n", "print(index_doc)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Validate rejection behavior\n", "\n", "A governed memory system is defined as much by what it rejects as by what it accepts. This quick test shows how invalid records fail when approval or retention requirements are not met." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from dataclasses import dataclass, asdict\n", "from datetime import date\n", "\n", "@dataclass\n", "class MemoryRecord:\n", " content: str\n", " provenance: str\n", " owner: str\n", " classification: str\n", " retention_until: str\n", " approval_state: str\n", "\n", "def validate(record: MemoryRecord) -> None:\n", " allowed = {\"public\", \"internal\", \"confidential\"}\n", " assert all(asdict(record).values()), \"missing required field\"\n", " assert record.classification in allowed, \"invalid classification\"\n", " assert date.fromisoformat(record.retention_until) >= date.today(), \"expired retention\"\n", " assert record.approval_state == \"approved\", \"not approved\"\n", "\n", "bad_records = [\n", " MemoryRecord(\"Draft note\", \"ticket:1\", \"ops\", \"internal\", \"2027-01-01\", \"pending\"),\n", " MemoryRecord(\"Old workaround\", \"ticket:2\", \"ops\", \"internal\", \"2020-01-01\", \"approved\"),\n", " MemoryRecord(\"\", \"ticket:3\", \"ops\", \"internal\", \"2027-01-01\", \"approved\")\n", "]\n", "\n", "results = []\n", "for r in bad_records:\n", " try:\n", " validate(r)\n", " results.append((r.provenance, \"accepted\"))\n", " except AssertionError as e:\n", " results.append((r.provenance, f\"rejected: {e}\"))\n", "\n", "results" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Retrieval with policy filters\n", "\n", "The blog emphasizes that retrieval relevance alone is not enough. Retrieval must also respect approval state, retention windows, and authorization boundaries such as classification level. This example filters memory records so agents only learn from approved, in-retention content allowed for the requested audience." ] }, { "cell_type": "code", "metadata": {}, "source": [ "# Retrieval applies policy filters so agents only learn from approved, in-retention memory.\n", "from datetime import date\n", "\n", "memory_index = [\n", " {\"content\": \"Use vendor SSO\", \"classification\": \"internal\", \"owner\": \"iam\", \"retention_until\": \"2027-01-01\", \"approval_state\": \"approved\"},\n", " {\"content\": \"Temporary workaround\", \"classification\": \"confidential\", \"owner\": \"ops\", \"retention_until\": \"2024-01-01\", \"approval_state\": \"approved\"},\n", " {\"content\": \"Draft migration note\", \"classification\": \"internal\", \"owner\": \"platform\", \"retention_until\": \"2027-05-01\", \"approval_state\": \"pending\"},\n", "]\n", "\n", "def retrieve(query: str, max_classification: str = \"internal\"):\n", " rank = {\"public\": 0, \"internal\": 1, \"confidential\": 2}\n", " today = date.today()\n", " return [\n", " m[\"content\"] for m in memory_index\n", " if query.lower() in m[\"content\"].lower()\n", " and m[\"approval_state\"] == \"approved\"\n", " and date.fromisoformat(m[\"retention_until\"]) >= today\n", " and rank[m[\"classification\"]] <= rank[max_classification]\n", " ]\n", "\n", "print(retrieve(\"SSO\"))" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Inspect retrieval outcomes across policy levels\n", "\n", "To make the retrieval logic easier to validate, this cell runs multiple queries and classification ceilings. It demonstrates that expired, pending, or over-classified records do not appear even if they are textually relevant." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from datetime import date, timedelta\n", "\n", "future_1 = (date.today() + timedelta(days=400)).isoformat()\n", "future_2 = (date.today() + timedelta(days=200)).isoformat()\n", "past_1 = (date.today() - timedelta(days=30)).isoformat()\n", "\n", "memory_index = [\n", " {\"content\": \"Use vendor SSO\", \"classification\": \"internal\", \"owner\": \"iam\", \"retention_until\": future_1, \"approval_state\": \"approved\"},\n", " {\"content\": \"Temporary workaround for SSO outage\", \"classification\": \"confidential\", \"owner\": \"ops\", \"retention_until\": future_2, \"approval_state\": \"approved\"},\n", " {\"content\": \"Legacy SSO note\", \"classification\": \"internal\", \"owner\": \"platform\", \"retention_until\": past_1, \"approval_state\": \"approved\"},\n", " {\"content\": \"Draft SSO migration note\", \"classification\": \"internal\", \"owner\": \"platform\", \"retention_until\": future_2, \"approval_state\": \"pending\"},\n", "]\n", "\n", "def retrieve(query: str, max_classification: str = \"internal\"):\n", " rank = {\"public\": 0, \"internal\": 1, \"confidential\": 2}\n", " today = date.today()\n", " return [\n", " m for m in memory_index\n", " if query.lower() in m[\"content\"].lower()\n", " and m[\"approval_state\"] == \"approved\"\n", " and date.fromisoformat(m[\"retention_until\"]) >= today\n", " and rank[m[\"classification\"]] <= rank[max_classification]\n", " ]\n", "\n", "scenarios = {\n", " \"internal ceiling\": retrieve(\"SSO\", \"internal\"),\n", " \"confidential ceiling\": retrieve(\"SSO\", \"confidential\"),\n", " \"public ceiling\": retrieve(\"SSO\", \"public\")\n", "}\n", "\n", "scenarios" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Sequence of governed promotion\n", "\n", "The post also describes the learning loop as a sequence: submit candidate memory with metadata, validate provenance and policy fields, append an immutable audit event, and then make the record retrieval-ready. The next cell represents that sequence in executable Python data." ] }, { "cell_type": "code", "metadata": {}, "source": [ "sequence_steps = [\n", " {\"actor\": \"Agent\", \"action\": \"Submit memory candidate + metadata\", \"target\": \"Governance Gate\"},\n", " {\"actor\": \"Governance Gate\", \"action\": \"Validate provenance/owner/classification/retention/approval\", \"target\": \"Governance Gate\"},\n", " {\"actor\": \"Governance Gate\", \"action\": \"Append approval event\", \"target\": \"Immutable Audit Log\"},\n", " {\"actor\": \"Immutable Audit Log\", \"action\": \"Return event hash\", \"target\": \"Governance Gate\"},\n", " {\"actor\": \"Governance Gate\", \"action\": \"Index record + audit hash\", \"target\": \"Memory Index\"},\n", " {\"actor\": \"Memory Index\", \"action\": \"Return retrieval-ready status\", \"target\": \"Agent\"}\n", "]\n", "\n", "pd.DataFrame(sequence_steps)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Minimal enterprise learning loop\n", "\n", "This compact example captures feedback from an incident, converts it into a candidate memory, checks required fields and approval state, and promotes it into a memory index if valid. It demonstrates the blog's point that feedback is not durable memory until it passes a governance gate." ] }, { "cell_type": "code", "metadata": {}, "source": [ "# Minimal enterprise learning loop: capture feedback, propose memory, gate it, then retrieve it later.\n", "from datetime import date, timedelta\n", "\n", "feedback = {\"incident\": \"INC-77\", \"lesson\": \"Pin API version for billing client\", \"owner\": \"finops\"}\n", "candidate = {\n", " \"content\": feedback[\"lesson\"],\n", " \"provenance\": f\"incident:{feedback['incident']}\",\n", " \"owner\": feedback[\"owner\"],\n", " \"classification\": \"internal\",\n", " \"retention_until\": (date.today() + timedelta(days=365)).isoformat(),\n", " \"approval_state\": \"approved\",\n", "}\n", "required = {\"content\", \"provenance\", \"owner\", \"classification\", \"retention_until\", \"approval_state\"}\n", "approved = required.issubset(candidate) and candidate[\"approval_state\"] == \"approved\"\n", "memory_index = [candidate] if approved else []\n", "print(memory_index[0][\"content\"] if memory_index else \"rejected\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Extend the learning loop with feedback classification\n", "\n", "The article recommends classifying human feedback before promotion: immediate task correction, knowledge correction, policy escalation, or product-quality signal. This example routes each type differently so only the right feedback becomes a durable memory candidate." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from datetime import date, timedelta\n", "\n", "feedback_events = [\n", " {\"type\": \"immediate_task_correction\", \"incident\": \"INC-10\", \"lesson\": \"Retry once before failing\", \"owner\": \"ops\"},\n", " {\"type\": \"knowledge_correction\", \"incident\": \"INC-11\", \"lesson\": \"Pin API version for billing client\", \"owner\": \"finops\"},\n", " {\"type\": \"policy_escalation\", \"incident\": \"INC-12\", \"lesson\": \"Do not expose customer export links in chat\", \"owner\": \"risk\"},\n", " {\"type\": \"product_quality_signal\", \"incident\": \"INC-13\", \"lesson\": \"Answer was slow under load\", \"owner\": \"platform\"},\n", "]\n", "\n", "def route_feedback(event):\n", " if event[\"type\"] == \"immediate_task_correction\":\n", " return {\"route\": \"session_local\", \"promote\": False}\n", " if event[\"type\"] == \"knowledge_correction\":\n", " candidate = {\n", " \"content\": event[\"lesson\"],\n", " \"provenance\": f\"incident:{event['incident']}\",\n", " \"owner\": event[\"owner\"],\n", " \"classification\": \"internal\",\n", " \"retention_until\": (date.today() + timedelta(days=365)).isoformat(),\n", " \"approval_state\": \"approved\",\n", " }\n", " return {\"route\": \"memory_candidate\", \"promote\": True, \"candidate\": candidate}\n", " if event[\"type\"] == \"policy_escalation\":\n", " return {\"route\": \"human_review\", \"promote\": False}\n", " return {\"route\": \"telemetry_only\", \"promote\": False}\n", "\n", "routed = [route_feedback(e) for e in feedback_events]\n", "routed" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Governance gap export in Python\n", "\n", "The original post included a PowerShell example for exporting memory records that are missing owner, retention, or review metadata. Since this notebook is Python-first, the next cell reproduces that inventory pattern using pandas and writes a CSV file." ] }, { "cell_type": "code", "metadata": {}, "source": [ "records = [\n", " {\"Id\": \"mem-001\", \"Owner\": \"secops\", \"RetentionUntil\": \"2027-12-31\", \"LastReviewed\": \"2026-06-01\"},\n", " {\"Id\": \"mem-002\", \"Owner\": \"\", \"RetentionUntil\": \"2027-03-01\", \"LastReviewed\": \"2026-05-15\"},\n", " {\"Id\": \"mem-003\", \"Owner\": \"ops\", \"RetentionUntil\": \"\", \"LastReviewed\": \"\"},\n", "]\n", "\n", "df = pd.DataFrame(records)\n", "gaps = df[\n", " df[\"Owner\"].fillna(\"\").str.strip().eq(\"\") |\n", " df[\"RetentionUntil\"].fillna(\"\").str.strip().eq(\"\") |\n", " df[\"LastReviewed\"].fillna(\"\").str.strip().eq(\"\")\n", "][[\"Id\", \"Owner\", \"RetentionUntil\", \"LastReviewed\"]]\n", "\n", "path = \"memory-governance-gaps.csv\"\n", "gaps.to_csv(path, index=False)\n", "print(f\"Exported {len(gaps)} records to {path}\")\n", "gaps" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Stale memory review detection in Python\n", "\n", "Another operational control from the post is flagging memory that has not been reviewed within policy. The next cell converts the PowerShell stale-review check into Python and identifies records older than the allowed review age." ] }, { "cell_type": "code", "metadata": {}, "source": [ "max_age_days = 180\n", "today = datetime.today()\n", "records = [\n", " {\"Id\": \"mem-101\", \"LastReviewed\": \"2026-07-01\", \"Owner\": \"platform\"},\n", " {\"Id\": \"mem-102\", \"LastReviewed\": \"2025-12-15\", \"Owner\": \"iam\"},\n", "]\n", "\n", "df = pd.DataFrame(records)\n", "df[\"LastReviewed\"] = pd.to_datetime(df[\"LastReviewed\"])\n", "df[\"AgeDays\"] = (today - df[\"LastReviewed\"]).dt.days\n", "stale = df[df[\"AgeDays\"] > max_age_days][[\"Id\", \"Owner\", \"LastReviewed\", \"AgeDays\"]]\n", "stale" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## End-to-end governed learning loop\n", "\n", "The blog closes the loop from human feedback and incidents to candidate memory, metadata enrichment, governance review, immutable audit, indexed enterprise memory, and improved agent behavior. This cell simulates that loop with a small in-memory pipeline." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from datetime import date, timedelta\n", "import hashlib\n", "import json\n", "\n", "feedback_sources = [\n", " {\"source_type\": \"incident\", \"source_id\": \"INC-77\", \"lesson\": \"Pin API version for billing client\", \"owner\": \"finops\", \"classification\": \"internal\", \"approval_state\": \"approved\"},\n", " {\"source_type\": \"ticket\", \"source_id\": \"CHG-1042\", \"lesson\": \"Rotate keys quarterly\", \"owner\": \"secops\", \"classification\": \"internal\", \"approval_state\": \"approved\"},\n", " {\"source_type\": \"chat\", \"source_id\": \"CHAT-9\", \"lesson\": \"One-off workaround\", \"owner\": \"ops\", \"classification\": \"confidential\", \"approval_state\": \"pending\"},\n", "]\n", "\n", "def enrich_metadata(item):\n", " return {\n", " \"content\": item[\"lesson\"],\n", " \"provenance\": f\"{item['source_type']}:{item['source_id']}\",\n", " \"owner\": item[\"owner\"],\n", " \"classification\": item[\"classification\"],\n", " \"retention_until\": (date.today() + timedelta(days=365)).isoformat(),\n", " \"approval_state\": item[\"approval_state\"],\n", " }\n", "\n", "def approve_and_audit(candidate):\n", " required = {\"content\", \"provenance\", \"owner\", \"classification\", \"retention_until\", \"approval_state\"}\n", " if not required.issubset(candidate):\n", " return None\n", " if candidate[\"approval_state\"] != \"approved\":\n", " return None\n", " payload = json.dumps(candidate, sort_keys=True).encode()\n", " audit = {\"event\": \"memory_write_approved\", \"hash\": hashlib.sha256(payload).hexdigest()}\n", " return {\"record\": candidate, \"audit\": audit}\n", "\n", "candidates = [enrich_metadata(x) for x in feedback_sources]\n", "indexed_memory = [doc for c in candidates if (doc := approve_and_audit(c)) is not None]\n", "\n", "indexed_memory" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Executive scorecard metrics\n", "\n", "The post recommends measuring governed-memory coverage and learning-loop quality instead of vanity metrics like total embeddings or raw retrieval counts. This example computes a compact scorecard from sample memory records and operational events." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from datetime import date, timedelta\n", "\n", "future = (date.today() + timedelta(days=365)).isoformat()\n", "past = (date.today() - timedelta(days=10)).isoformat()\n", "\n", "memory_records = [\n", " {\"id\": \"m1\", \"provenance\": \"incident:1\", \"owner\": \"ops\", \"classification\": \"internal\", \"retention_until\": future, \"review_state\": \"current\", \"approval_state\": \"approved\", \"retrievals\": 10, \"approved_retrievals\": 10},\n", " {\"id\": \"m2\", \"provenance\": \"\", \"owner\": \"secops\", \"classification\": \"internal\", \"retention_until\": future, \"review_state\": \"current\", \"approval_state\": \"approved\", \"retrievals\": 5, \"approved_retrievals\": 5},\n", " {\"id\": \"m3\", \"provenance\": \"ticket:3\", \"owner\": \"\", \"classification\": \"confidential\", \"retention_until\": future, \"review_state\": \"expired\", \"approval_state\": \"approved\", \"retrievals\": 2, \"approved_retrievals\": 1},\n", " {\"id\": \"m4\", \"provenance\": \"chat:4\", \"owner\": \"platform\", \"classification\": \"\", \"retention_until\": past, \"review_state\": \"expired\", \"approval_state\": \"pending\", \"retrievals\": 0, \"approved_retrievals\": 0},\n", "]\n", "\n", "ops_events = {\n", " \"review_latencies_days\": [2, 5, 3],\n", " \"correction_latencies_days\": [1, 4],\n", " \"rollback_attempts\": 5,\n", " \"rollback_successes\": 4,\n", " \"policy_denials\": 7,\n", "}\n", "\n", "df = pd.DataFrame(memory_records)\n", "total = len(df)\n", "scorecard = {\n", " \"governed_memory_coverage\": {\n", " \"complete_provenance_pct\": round((df[\"provenance\"].fillna(\"\").str.strip() != \"\").mean() * 100, 1),\n", " \"assigned_owner_pct\": round((df[\"owner\"].fillna(\"\").str.strip() != \"\").mean() * 100, 1),\n", " \"sensitivity_classification_pct\": round((df[\"classification\"].fillna(\"\").str.strip() != \"\").mean() * 100, 1),\n", " \"retention_metadata_pct\": round((df[\"retention_until\"].fillna(\"\").str.strip() != \"\").mean() * 100, 1),\n", " \"current_review_state_pct\": round((df[\"review_state\"] == \"current\").mean() * 100, 1),\n", " },\n", " \"learning_loop_quality\": {\n", " \"review_latency_days_avg\": round(sum(ops_events[\"review_latencies_days\"]) / len(ops_events[\"review_latencies_days\"]), 2),\n", " \"correction_latency_days_avg\": round(sum(ops_events[\"correction_latencies_days\"]) / len(ops_events[\"correction_latencies_days\"]), 2),\n", " \"rollback_success_rate_pct\": round(ops_events[\"rollback_successes\"] / ops_events[\"rollback_attempts\"] * 100, 1),\n", " \"policy_denial_rate_per_total_records\": round(ops_events[\"policy_denials\"] / total, 2),\n", " \"retrievals_linked_to_approved_records_pct\": round(df[\"approved_retrievals\"].sum() / max(df[\"retrievals\"].sum(), 1) * 100, 1),\n", " \"stale_memory_rate_pct\": round((df[\"review_state\"] == \"expired\").mean() * 100, 1),\n", " }\n", "}\n", "\n", "scorecard" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Revocation and rollback test\n", "\n", "A key executive question from the post is whether one bad learned behavior can be removed quickly, with proof of what changed. This final code example simulates revoking a memory record, appending an audit event, and verifying that retrieval no longer returns the revoked content." ] }, { "cell_type": "code", "metadata": {}, "source": [ "from datetime import date, timedelta\n", "import hashlib\n", "import json\n", "\n", "memory_index = [\n", " {\n", " \"id\": \"mem-201\",\n", " \"content\": \"Use vendor SSO\",\n", " \"classification\": \"internal\",\n", " \"owner\": \"iam\",\n", " \"retention_until\": (date.today() + timedelta(days=365)).isoformat(),\n", " \"approval_state\": \"approved\",\n", " \"status\": \"active\"\n", " },\n", " {\n", " \"id\": \"mem-202\",\n", " \"content\": \"Rotate keys quarterly\",\n", " \"classification\": \"internal\",\n", " \"owner\": \"secops\",\n", " \"retention_until\": (date.today() + timedelta(days=365)).isoformat(),\n", " \"approval_state\": \"approved\",\n", " \"status\": \"active\"\n", " }\n", "]\n", "\n", "audit_log = []\n", "\n", "def retrieve(query: str):\n", " today = date.today()\n", " return [\n", " m[\"content\"] for m in memory_index\n", " if query.lower() in m[\"content\"].lower()\n", " and m[\"approval_state\"] == \"approved\"\n", " and m[\"status\"] == \"active\"\n", " and date.fromisoformat(m[\"retention_until\"]) >= today\n", " ]\n", "\n", "def revoke_memory(record_id: str, reason: str):\n", " for m in memory_index:\n", " if m[\"id\"] == record_id:\n", " m[\"status\"] = \"revoked\"\n", " event = {\n", " \"event\": \"memory_revoked\",\n", " \"record_id\": record_id,\n", " \"reason\": reason,\n", " \"date\": date.today().isoformat()\n", " }\n", " event[\"hash\"] = hashlib.sha256(json.dumps(event, sort_keys=True).encode()).hexdigest()\n", " audit_log.append(event)\n", " return event\n", " return None\n", "\n", "before = retrieve(\"SSO\")\n", "revocation_event = revoke_memory(\"mem-201\", \"Incorrect standing policy\")\n", "after = retrieve(\"SSO\")\n", "\n", "{\"before\": before, \"revocation_event\": revocation_event, \"after\": after, \"audit_log\": audit_log}" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Next Steps\n", "\n", "This notebook validated the blog's central argument: enterprise agent memory must be governed like data, not treated as an opaque AI convenience. The examples showed how to gate writes, enforce policy-aware retrieval, classify feedback, export governance gaps, detect stale records, compute executive metrics, and revoke bad learned behavior with audit evidence.\n", "\n", "Next steps:\n", "1. Replace the in-memory lists with your actual memory store or data platform.\n", "2. Add identity-aware authorization checks tied to agent and user roles.\n", "3. Version memory records explicitly and track supersession links.\n", "4. Add review workflows, deletion SLAs, and rollback runbooks.\n", "5. Measure governed-memory coverage before scaling any enterprise learning loop." ] } ] }