Memory Backend

The MemoryBackend provides persistent semantic storage for AXON programs, enabling remember and recall operations.

remember(data) → [Memory Backend] → Storage
recall(query) → [Memory Backend] → Retrieved Results

Overview

The memory layer stores semantic values with metadata, enabling:

Cross-execution persistence — Data survives between program runs
Semantic retrieval — Query by meaning, not just keys
Scoped storage — Namespace isolation for different memory definitions
Decay policies — Automatic expiration of stale data

Architecture

Abstract Interface

from abc import ABC, abstractmethod

class MemoryBackend(ABC):
    """Abstract base class for semantic memory storage."""

    @abstractmethod
    async def store(
        self,
        key: str,
        value: Any,
        metadata: dict[str, Any] | None = None,
    ) -> MemoryEntry:
        """Store a value in semantic memory."""
        ...

    @abstractmethod
    async def retrieve(
        self,
        query: str,
        top_k: int = 5,
        scope: str | None = None,
    ) -> list[MemoryEntry]:
        """Retrieve values from semantic memory."""
        ...

    @abstractmethod
    async def clear(self, scope: str | None = None) -> int:
        """Clear stored entries."""
        ...

MemoryEntry

@dataclass(frozen=True)
class MemoryEntry:
    """A single value stored in semantic memory."""
    key: str                           # Storage key
    value: Any                         # The stored value
    metadata: dict[str, Any] = field(default_factory=dict)
    score: float = 0.0                 # Relevance score (retrieval only)
    timestamp: float = 0.0             # Unix timestamp

Example:

MemoryEntry(
    key="contract_type",
    value="Non-Disclosure Agreement",
    metadata={"source": "Extract", "confidence": 0.95},
    timestamp=1704067200.0,
)

InMemoryBackend (Default)

Overview

The InMemoryBackend is a dict-based implementation for testing and simple use cases. Features:

Fast in-process storage
Substring-based retrieval (no vector embeddings)
Tracer integration
Scoped namespacing

Limitations:

No persistence across process restarts
Simple substring matching (not semantic)
No scalability for large datasets

Implementation

class InMemoryBackend(MemoryBackend):
    """Dict-based memory backend for testing and simple use cases."""

    def __init__(self, tracer: Tracer | None = None) -> None:
        self._store: dict[str, MemoryEntry] = {}
        self._tracer = tracer

    async def store(
        self,
        key: str,
        value: Any,
        metadata: dict[str, Any] | None = None,
    ) -> MemoryEntry:
        if not key:
            raise ValueError("Memory key must not be empty")

        entry = MemoryEntry(
            key=key,
            value=value,
            metadata=metadata or {},
            timestamp=time.time(),
        )
        self._store[key] = entry

        if self._tracer:
            self._tracer.emit(
                TraceEventType.MEMORY_WRITE,
                data={"key": key, "value_type": type(value).__name__},
            )

        return entry

    async def retrieve(
        self,
        query: str,
        top_k: int = 5,
        scope: str | None = None,
    ) -> list[MemoryEntry]:
        candidates: list[MemoryEntry] = []
        query_lower = query.lower()

        for entry in self._store.values():
            # Scope filter
            if scope and entry.metadata.get("scope") != scope:
                continue

            # Score by match quality
            score = 0.0
            if entry.key.lower() == query_lower:
                score = 1.0  # Exact match
            elif query_lower in entry.key.lower():
                score = 0.7  # Key contains query
            elif query_lower in str(entry.value).lower():
                score = 0.4  # Value contains query

            if score > 0:
                scored_entry = MemoryEntry(
                    key=entry.key,
                    value=entry.value,
                    metadata=entry.metadata,
                    score=score,
                    timestamp=entry.timestamp,
                )
                candidates.append(scored_entry)

        # Sort by score descending, then by timestamp descending
        candidates.sort(key=lambda e: (-e.score, -e.timestamp))

        results = candidates[:top_k]

        if self._tracer:
            self._tracer.emit(
                TraceEventType.MEMORY_READ,
                data={
                    "query": query,
                    "results_count": len(results),
                    "top_k": top_k,
                },
            )

        return results

    async def clear(self, scope: str | None = None) -> int:
        if scope is None:
            count = len(self._store)
            self._store.clear()
            return count

        # Scope-filtered clear
        keys_to_remove = [
            k for k, v in self._store.items()
            if v.metadata.get("scope") == scope
        ]
        for key in keys_to_remove:
            del self._store[key]
        return len(keys_to_remove)

AXON Integration

Memory Declaration

memory LongTermKnowledge {
  store: persistent
  backend: vector_db
  retrieval: semantic
  decay: none
}

Maps to:

IRMemory(
    name="LongTermKnowledge",
    store="persistent",
    backend="vector_db",
    retrieval="semantic",
    decay="none",
)

Remember Statement

remember(ContractSummary) -> LongTermKnowledge

Runtime execution:

await memory.store(
    key="contract_summary",
    value=contract_summary,
    metadata={"scope": "LongTermKnowledge", "source": "Summarize"},
)

Recall Statement

recall("contract types") from LongTermKnowledge

Runtime execution:

results = await memory.retrieve(
    query="contract types",
    top_k=5,
    scope="LongTermKnowledge",
)

Retrieval Scoring

InMemoryBackend Scoring

The default backend uses simple substring matching:

Match Type	Score	Example
Exact key match	1.0	query=“contract_type” → key=“contract_type”
Key contains query	0.7	query=“contract” → key=“contract_type”
Value contains query	0.4	query=“NDA” → value=“Non-Disclosure Agreement”
No match	0.0	Excluded from results

Sorting:

Score (descending)
Timestamp (descending) — newer entries preferred

Example

# Stored entries
await memory.store("contract_type", "NDA")
await memory.store("contract_date", "2024-01-15")
await memory.store("parties", "Acme Corp, Beta LLC")

# Query
results = await memory.retrieve("contract", top_k=2)

# Results (sorted by score)
[
    MemoryEntry(key="contract_type", value="NDA", score=0.7),
    MemoryEntry(key="contract_date", value="2024-01-15", score=0.7),
]

Scope Filtering

Memory entries can be scoped to namespaces:

Storing with Scope

await memory.store(
    key="research_summary",
    value="Quantum computing trends...",
    metadata={"scope": "ResearchKnowledge"},
)

await memory.store(
    key="legal_precedent",
    value="Smith v. Jones...",
    metadata={"scope": "LegalKnowledge"},
)

Retrieving with Scope

# Only search ResearchKnowledge
results = await memory.retrieve(
    query="quantum",
    scope="ResearchKnowledge",
)
# Returns: [research_summary]

# Search all scopes
results = await memory.retrieve(
    query="knowledge",
    scope=None,
)
# Returns: [research_summary, legal_precedent] (if "knowledge" in values)

Clearing by Scope

# Clear only ResearchKnowledge
count = await memory.clear(scope="ResearchKnowledge")
# Removes: research_summary
# Keeps: legal_precedent

Trace Integration

All memory operations are traced:

Memory Write

tracer.emit(
    TraceEventType.MEMORY_WRITE,
    data={
        "key": "contract_type",
        "value_type": "str",
    },
)

Memory Read

tracer.emit(
    TraceEventType.MEMORY_READ,
    data={
        "query": "contract",
        "results_count": 2,
        "top_k": 5,
    },
)

Example Trace:

{
  "event_type": "memory_write",
  "timestamp": 1704067200.5,
  "data": {
    "key": "contract_summary",
    "value_type": "dict"
  }
},
{
  "event_type": "memory_read",
  "timestamp": 1704067205.3,
  "data": {
    "query": "contract types",
    "results_count": 3,
    "top_k": 5
  }
}

Future Backends

Vector DB Backends (Planned)

Phase 4+ expansion will add semantic retrieval via vector databases:

Pinecone Backend

class PineconeBackend(MemoryBackend):
    """Vector database backend using Pinecone."""
    
    def __init__(self, api_key: str, index_name: str):
        import pinecone
        pinecone.init(api_key=api_key)
        self.index = pinecone.Index(index_name)
    
    async def store(self, key, value, metadata):
        # Generate embedding
        embedding = await self._embed(str(value))
        # Store in Pinecone
        self.index.upsert([(key, embedding, metadata)])
    
    async def retrieve(self, query, top_k, scope):
        # Generate query embedding
        query_embedding = await self._embed(query)
        # Semantic search
        results = self.index.query(
            query_embedding,
            top_k=top_k,
            filter={"scope": scope} if scope else None,
        )
        return [self._to_memory_entry(r) for r in results]

Chroma Backend

class ChromaBackend(MemoryBackend):
    """Local vector database backend using Chroma."""
    
    def __init__(self, persist_directory: str):
        import chromadb
        self.client = chromadb.Client(Settings(
            persist_directory=persist_directory,
        ))
        self.collection = self.client.get_or_create_collection("axon_memory")
    
    async def store(self, key, value, metadata):
        self.collection.add(
            ids=[key],
            documents=[str(value)],
            metadatas=[metadata],
        )
    
    async def retrieve(self, query, top_k, scope):
        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            where={"scope": scope} if scope else None,
        )
        return [self._to_memory_entry(r) for r in results]

Usage Example

Setup

from axon.runtime.memory_backend import InMemoryBackend
from axon.runtime.tracer import Tracer

tracer = Tracer()
memory = InMemoryBackend(tracer=tracer)

Store

await memory.store(
    key="contract_analysis_summary",
    value={
        "parties": ["Acme Corp", "Beta LLC"],
        "risk_score": 0.35,
        "key_findings": "Standard NDA with no unusual clauses",
    },
    metadata={
        "scope": "LegalKnowledge",
        "source": "AnalyzeContract",
        "confidence": 0.92,
    },
)

Retrieve

results = await memory.retrieve(
    query="contract risk",
    top_k=3,
    scope="LegalKnowledge",
)

for entry in results:
    print(f"Key: {entry.key}")
    print(f"Score: {entry.score}")
    print(f"Value: {entry.value}")
    print(f"Metadata: {entry.metadata}")

Clear

# Clear specific scope
count = await memory.clear(scope="LegalKnowledge")
print(f"Cleared {count} entries from LegalKnowledge")

# Clear all
count = await memory.clear()
print(f"Cleared {count} total entries")

Decay Policies (Planned)

Future memory backends will support automatic decay:

memory ShortTermCache {
  store: ephemeral
  backend: in_memory
  decay: 1h  # Expire after 1 hour
}

memory WeeklyKnowledge {
  store: persistent
  backend: vector_db
  decay: 7d  # Expire after 7 days
}

Implementation (planned):

if memory_def.decay:
    decay_seconds = parse_duration(memory_def.decay)
    if time.time() - entry.timestamp > decay_seconds:
        # Entry expired, exclude from results
        continue

Next Steps

Executor

See how memory integrates into execution

Tracer

Understand memory operation tracing

Architecture Overview

Review the complete AXON architecture

Compiler

Backends

Runtime

​Memory Backend

​Overview

​Architecture

​Abstract Interface

​MemoryEntry

​InMemoryBackend (Default)

​Overview

​Implementation

​AXON Integration

​Memory Declaration

​Remember Statement

​Recall Statement

​Retrieval Scoring

​InMemoryBackend Scoring

​Example

​Scope Filtering

​Storing with Scope

​Retrieving with Scope

​Clearing by Scope

​Trace Integration

​Memory Write

​Memory Read

​Future Backends

​Vector DB Backends (Planned)

​Pinecone Backend

​Chroma Backend

​Usage Example

​Setup

​Store

​Retrieve

​Clear

​Decay Policies (Planned)

​Next Steps

Executor

Tracer

Architecture Overview

Build docs developers (and LLMs) love

Memory Backend

Overview

Architecture

Abstract Interface

MemoryEntry

InMemoryBackend (Default)

Overview

Implementation

AXON Integration

Memory Declaration

Remember Statement

Recall Statement

Retrieval Scoring

InMemoryBackend Scoring

Example

Scope Filtering

Storing with Scope

Retrieving with Scope

Clearing by Scope

Trace Integration

Memory Write

Memory Read

Future Backends

Vector DB Backends (Planned)

Pinecone Backend

Chroma Backend

Usage Example

Setup

Store

Retrieve

Clear

Decay Policies (Planned)

Next Steps