Memory Hub

AgentHelm's Memory Hub provides a pluggable memory system with an SDK-first design. Zero Docker required for basic usage.

Quick Start

from agenthelm import MemoryHub, MemoryContext

# In-memory (default, ephemeral)
hub = MemoryHub()

# Short-term memory (key-value)
await hub.short_term.set("user:123:name", "Alice", ttl=3600)
name = await hub.short_term.get("user:123:name")

# Semantic memory (vector search)
await hub.semantic.store("User prefers dark mode.", metadata={"category": "preferences"})
results = await hub.semantic.search("What are the user's UI preferences?")

Architecture

The Memory Hub offers three tiers, automatically selected based on configuration:

Tier	Short-Term	Semantic	Use Case
In-Memory	dict + TTL	Qdrant :memory:	Development, testing
Local File	SQLite	Qdrant local	Single-node, no Docker
Network	Redis	Qdrant server	Production, scaling

Configuration

In-Memory (Default)

hub = MemoryHub()  # Zero config, ephemeral

Local Persistence

hub = MemoryHub(data_dir="./data")
# Creates:
#   ./data/short_term.db  (SQLite)
#   ./data/qdrant/        (Qdrant local)

Network (Production)

hub = MemoryHub(
    redis_url="redis://localhost:6379",
    qdrant_url="http://localhost:6333"
)

Session Context

Use MemoryContext for session-scoped operations with automatic key namespacing:

async with MemoryContext(hub, session_id="user-123") as ctx:
    # Keys are automatically prefixed with session ID
    await ctx.set("last_query", "What is AI?")

    # Store memories tagged with session
    await ctx.store_memory("User asked about AI basics.")

    # Search within session or globally
    results = await ctx.recall("AI questions", session_only=True)

# Session keys are cleaned up on exit (configurable)

Short-Term Memory API

BaseShortTermMemory

Bases: ABC

Abstract base class for short-term (key-value) memory.

Functions

close async

close()

Close any connections. Override if needed.

Source code in agenthelm/memory/base.py

async def close(self) -> None:
    """Close any connections. Override if needed."""
    pass

delete abstractmethod async

delete(key)

Delete a key.

Source code in agenthelm/memory/base.py

@abstractmethod
async def delete(self, key: str) -> None:
    """Delete a key."""
    ...

exists abstractmethod async

exists(key)

Check if key exists and is not expired.

Source code in agenthelm/memory/base.py

@abstractmethod
async def exists(self, key: str) -> bool:
    """Check if key exists and is not expired."""
    ...

get abstractmethod async

get(key)

Get a value by key. Returns None if not found or expired.

Source code in agenthelm/memory/base.py

@abstractmethod
async def get(self, key: str) -> Any | None:
    """Get a value by key. Returns None if not found or expired."""
    ...

get_many async

get_many(keys)

Get multiple keys. Default implementation calls get() for each.

Source code in agenthelm/memory/base.py

async def get_many(self, keys: list[str]) -> dict[str, Any]:
    """Get multiple keys. Default implementation calls get() for each."""
    result = {}
    for key in keys:
        value = await self.get(key)
        if value is not None:
            result[key] = value
    return result

set abstractmethod async

set(key, value, ttl=3600)

Set a value with optional TTL in seconds.

Source code in agenthelm/memory/base.py

@abstractmethod
async def set(self, key: str, value: Any, ttl: int = 3600) -> None:
    """Set a value with optional TTL in seconds."""
    ...

options: show_source: false members: - get - set - delete - exists - keys

Semantic Memory API

BaseSemanticMemory

Bases: ABC

Abstract base class for semantic (vector) memory.

Functions

close async

close()

Close any connections. Override if needed.

Source code in agenthelm/memory/base.py

async def close(self) -> None:
    """Close any connections. Override if needed."""
    pass

delete abstractmethod async

delete(ids)

Delete entries by ID.

Source code in agenthelm/memory/base.py

@abstractmethod
async def delete(self, ids: list[str]) -> None:
    """Delete entries by ID."""
    ...

search abstractmethod async

search(query, top_k=5, filter=None)

Search for similar texts. Returns ranked results.

Source code in agenthelm/memory/base.py

@abstractmethod
async def search(
    self,
    query: str,
    top_k: int = 5,
    filter: dict[str, Any] | None = None,
) -> list[SearchResult]:
    """Search for similar texts. Returns ranked results."""
    ...

store abstractmethod async

store(text, metadata=None, id=None)

Store text with optional metadata. Returns the ID.

Source code in agenthelm/memory/base.py

@abstractmethod
async def store(
    self,
    text: str,
    metadata: dict[str, Any] | None = None,
    id: str | None = None,
) -> str:
    """Store text with optional metadata. Returns the ID."""
    ...

store_many async

store_many(texts, metadatas=None)

Store multiple texts. Default implementation calls store() for each.

Source code in agenthelm/memory/base.py

async def store_many(
    self,
    texts: list[str],
    metadatas: list[dict[str, Any]] | None = None,
) -> list[str]:
    """Store multiple texts. Default implementation calls store() for each."""
    ids = []
    for i, text in enumerate(texts):
        meta = metadatas[i] if metadatas else None
        id = await self.store(text, meta)
        ids.append(id)
    return ids

options: show_source: false members: - store - search - delete - store_many

Backend Selection

AgentHelm automatically selects the appropriate backend:

# Explicit backend usage
from agenthelm.memory import InMemoryShortTermMemory, SqliteShortTermMemory, SemanticMemory

# In-memory short-term
memory = InMemoryShortTermMemory()

# SQLite short-term
memory = SqliteShortTermMemory(db_path="./data/cache.db")

# Semantic with mode selection
semantic = SemanticMemory(mode="memory")  # or "local" or "network"