Memory Management Skill

You are an expert in AI agent memory architectures, implementing the patterns used by

Anthropic's Claude agents, OpenAI's Assistants API, and MemGPT research.

You write production-ready memory systems in British English.

IMI colours: Navy #1A1A2E · Teal #0F3D66 · Gold #E2B95A

IMI fan segments: Tribal · Passionate · Casual · Distant · Corporate

Part 1 — mem0: Managed Memory Layer

# pip install mem0ai

from mem0 import Memory
import os

# ── Initialise mem0 with OpenAI backend ──────────────────────────────────────
config = {
    "llm": {
        "provider": "openai",
        "config": {"model": "gpt-4o-mini", "api_key": os.getenv("OPENAI_API_KEY")}
    },
    "embedder": {
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    },
    "vector_store": {
        "provider": "chroma",
        "config": {"collection_name": "imi_agent_memory", "path": "./memory_store"}
    }
}

memory = Memory.from_config(config)


def remember(user_id: str, content: str, metadata: dict | None = None):
    """Store a memory for a user."""
    result = memory.add(content, user_id=user_id, metadata=metadata or {})
    return result


def recall(user_id: str, query: str, limit: int = 5) -> list[dict]:
    """Retrieve relevant memories for a query."""
    results = memory.search(query, user_id=user_id, limit=limit)
    return results


def get_all_memories(user_id: str) -> list[dict]:
    """Get all stored memories for a user."""
    return memory.get_all(user_id=user_id)


def update_memory(memory_id: str, new_content: str):
    """Update an existing memory."""
    memory.update(memory_id, new_content)


def forget(memory_id: str):
    """Delete a specific memory."""
    memory.delete(memory_id)


def forget_user(user_id: str):
    """Delete all memories for a user (GDPR right to erasure)."""
    memory.delete_all(user_id=user_id)


# ── IMI-specific memory usage ─────────────────────────────────────────────────
def imi_agent_with_memory(user_id: str, query: str) -> str:
    """Agent that uses mem0 to personalise responses."""
    import anthropic
    client = anthropic.Anthropic()

    # Retrieve relevant memories
    memories = recall(user_id, query, limit=5)
    memory_context = "\n".join([f"- {m['memory']}" for m in memories]) if memories else "No prior context."

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        system=(
            "You are an IMI sports fan intelligence research assistant.\n"
            f"Known context about this user:\n{memory_context}\n"
            "Use this context to personalise your response."
        ),
        messages=[{"role": "user", "content": query}]
    )

    answer = response.content[0].text

    # Store new memory from this interaction
    remember(user_id, f"User asked about: {query[:100]}", {"type": "query"})

    return answer

Part 2 — Zep: Conversation Memory + Entity Extraction

# pip install zep-python

from zep_python import ZepClient, Message, Memory as ZepMemory
from zep_python.user import CreateUserRequest
from zep_python.memory import Session
import os, uuid

zep = ZepClient(api_key=os.getenv("ZEP_API_KEY"))


def create_zep_session(user_id: str) -> str:
    """Create a new Zep session for a user."""
    session_id = str(uuid.uuid4())
    zep.memory.add_session(Session(session_id=session_id, user_id=user_id))
    return session_id


def add_conversation_turn(session_id: str, user_msg: str, assistant_msg: str):
    """Add a conversation turn to Zep memory."""
    messages = [
        Message(role="user", role_type="user", content=user_msg),
        Message(role="assistant", role_type="assistant", content=assistant_msg)
    ]
    zep.memory.add(session_id, messages=messages)


def get_zep_context(session_id: str) -> dict:
    """
    Get Zep memory context — includes:
    - summary: AI-generated conversation summary
    - facts: extracted facts from conversation
    - entities: named entities (brands, people, orgs)
    """
    memory = zep.memory.get(session_id)
    return {
        "summary": memory.summary.content if memory.summary else "",
        "facts": [f.fact for f in (memory.facts or [])],
        "context": memory.context or ""
    }


def zep_search_memory(session_id: str, query: str, limit: int = 5) -> list[dict]:
    """Semantic search across conversation history."""
    results = zep.memory.search(session_id, query, limit=limit)
    return [
        {"content": r.message.content, "score": r.score}
        for r in results if r.message
    ]

Part 3 — SQLite Conversation History (No Dependencies)

import sqlite3, json, datetime
from pathlib import Path
from dataclasses import dataclass

@dataclass
class ConversationTurn:
    role: str
    content: str
    timestamp: str
    metadata: dict


class ConversationHistory:
    """
    Lightweight SQLite-backed conversation history.
    No external dependencies — works offline.
    """

    def __init__(self, db_path: str = "./conversations.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS conversations (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    session_id TEXT NOT NULL,
                    user_id TEXT,
                    role TEXT NOT NULL,
                    content TEXT NOT NULL,
                    timestamp TEXT NOT NULL,
                    metadata TEXT DEFAULT '{}'
                )
            """)
            conn.execute("CREATE INDEX IF NOT EXISTS idx_session ON conversations(session_id)")
            conn.execute("CREATE INDEX IF NOT EXISTS idx_user ON conversations(user_id)")

    def add_turn(self, session_id: str, role: str, content: str,
                 user_id: str = None, metadata: dict = None):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO conversations (session_id, user_id, role, content, timestamp, metadata) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (session_id, user_id, role, content,
                 datetime.datetime.utcnow().isoformat(),
                 json.dumps(metadata or {}))
            )

    def get_session(self, session_id: str, last_n: int = 20) -> list[ConversationTurn]:
        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute(
                "SELECT role, content, timestamp, metadata FROM conversations "
                "WHERE session_id = ? ORDER BY id DESC LIMIT ?",
                (session_id, last_n)
            ).fetchall()
        turns = [ConversationTurn(r[0], r[1], r[2], json.loads(r[3])) for r in reversed(rows)]
        return turns

    def to_messages(self, session_id: str, last_n: int = 20) -> list[dict]:
        """Convert session history to LLM messages format."""
        turns = self.get_session(session_id, last_n)
        return [{"role": t.role, "content": t.content} for t in turns]

    def summarise_session(self, session_id: str) -> str:
        """Use Claude to summarise a conversation session."""
        import anthropic
        messages = self.to_messages(session_id)
        if not messages:
            return ""

        client = anthropic.Anthropic()
        transcript = "\n".join([f"{m['role'].upper()}: {m['content'][:200]}" for m in messages])

        resp = client.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"Summarise this IMI research conversation in 2-3 sentences:\n\n{transcript}"
            }]
        )
        return resp.content[0].text

    def delete_session(self, session_id: str):
        """Delete all turns in a session."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("DELETE FROM conversations WHERE session_id = ?", (session_id,))

    def delete_user_data(self, user_id: str):
        """GDPR: delete all data for a user."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("DELETE FROM conversations WHERE user_id = ?", (user_id,))

Part 4 — MemGPT-Style Virtual Context Manager

import json
from dataclasses import dataclass, field
from typing import Callable

@dataclass
class MemGPTContext:
    """
    MemGPT-inspired virtual context: splits memory into tiers.
    - in_context: recent messages in LLM window
    - archival: searchable long-term storage
    - core: always-present user/persona facts
    """
    session_id: str
    core_memory: dict = field(default_factory=dict)       # always in context
    in_context: list[dict] = field(default_factory=list)  # recent messages
    archival: list[dict] = field(default_factory=list)    # searchable archive
    max_context_tokens: int = 4000


def update_core_memory(ctx: MemGPTContext, key: str, value: str):
    """Update a core memory fact (always included in context)."""
    ctx.core_memory[key] = value


def archive_message(ctx: MemGPTContext, message: dict):
    """Move oldest in-context message to archival storage."""
    ctx.archival.append(message)


def search_archival(ctx: MemGPTContext, query: str, top_k: int = 3) -> list[dict]:
    """Simple keyword search over archival memory (swap for vector search in prod)."""
    query_lower = query.lower()
    scored = []
    for msg in ctx.archival:
        content = msg.get("content", "").lower()
        score = sum(1 for word in query_lower.split() if word in content)
        if score > 0:
            scored.append((score, msg))
    scored.sort(key=lambda x: x[0], reverse=True)
    return [msg for _, msg in scored[:top_k]]


def build_context_window(ctx: MemGPTContext, new_query: str) -> list[dict]:
    """
    Build the LLM message list from virtual context:
    core facts → archival recall → recent in-context messages → new query
    """
    messages = []

    # Core memory as system context
    if ctx.core_memory:
        core_str = "\n".join([f"{k}: {v}" for k, v in ctx.core_memory.items()])
        messages.append({"role": "system", "content": f"Known facts:\n{core_str}"})

    # Recalled archival memories
    recalled = search_archival(ctx, new_query)
    if recalled:
        recalled_str = "\n".join([f"[memory] {m['content'][:100]}" for m in recalled])
        messages.append({"role": "system", "content": f"Relevant past context:\n{recalled_str}"})

    # Recent in-context messages
    messages.extend(ctx.in_context[-10:])  # last 10 turns

    # New user message
    messages.append({"role": "user", "content": new_query})
    return messages


# ── IMI agent with MemGPT-style memory ───────────────────────────────────────
def imi_memgpt_agent(ctx: MemGPTContext, user_query: str) -> str:
    import anthropic
    client = anthropic.Anthropic()

    messages = build_context_window(ctx, user_query)

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=800,
        system="You are an IMI fan intelligence research assistant with long-term memory.",
        messages=[m for m in messages if m["role"] != "system"]
    )

    answer = response.content[0].text

    # Update context
    ctx.in_context.append({"role": "user", "content": user_query})
    ctx.in_context.append({"role": "assistant", "content": answer})

    # Archive old messages if context too long
    while len(ctx.in_context) > 20:
        archive_message(ctx, ctx.in_context.pop(0))

    return answer

Output Standards

• Memory tiers: in-context (fast) → semantic cache (medium) → archival vector (slow but complete)
• Session IDs: always UUID4, never reuse across users
• GDPR: always implement delete_user_data() for any persistent memory store
• Core memory: keep to <500 tokens — facts that must always be present
• Archival: use vector search (not keyword) in production for semantic recall
• British English throughout all stored content and summaries

pip install

pip install mem0ai zep-python anthropic openai chromadb

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)