Adding a memory system¶

A memory system is added as a plugin: a separate Python package that the framework discovers when it is installed in the same environment. You do not edit the core package. The working reference is the example plugin in memory_plugins/heams_memory_example/.

How the pieces fit:

1. Your package defines a config class, a memory-system class and a MemoryPlugin descriptor.
2. Its pyproject.toml exposes the descriptor through the heams.memory_systems entry point, so heams finds it once the package is installed.
3. A YAML file under configs/memory_systems/ selects it by its type.
4. A run config pairs that YAML with an agent and a benchmark, and heams run evaluates it.

1. Import from the public API only¶

Import everything from heams.memory_systems.api. Other modules under heams.memory_systems are internal and may change.

from heams.memory_systems.api import (
    BaseMemoryConfig,
    BaseMemorySystem,
    MemoryEntry,
    MemoryEventTarget,
    MemoryOperationResult,
    MemoryPlugin,
    MemoryRuntimeContext,
    render_tool_memory_entry,
)

2. Define the config¶

Subclass BaseMemoryConfig and give it a type literal. The type is how YAML files select your system, so make it unique and namespaced (for example acme.graph_memory, not graph_memory).

from typing import Literal

from pydantic import Field


class AcmeMemoryConfig(BaseMemoryConfig):
    type: Literal["acme.memory"] = "acme.memory"
    max_entries: int = Field(default=100, gt=0)

Every config inherits these fields:

Unknown fields are rejected, so typos in YAML fail early. If your backend keeps files on disk, override with_runtime_storage(storage_dir) to return a copy of the config that points your storage at the per-sample storage_dir (the HippoRAG, mem0 and milamem plugins do this).

3. Implement the memory system¶

Subclass BaseMemorySystem. Keep the keyword-only constructor arguments and pass them to super().__init__; the framework uses them for storage and event logging.

from pathlib import Path
from typing import Any


class AcmeMemorySystem(BaseMemorySystem):
    config: AcmeMemoryConfig

    def __init__(
        self,
        config: AcmeMemoryConfig,
        *,
        storage_dir: Path,
        event_sink: MemoryEventTarget | None = None,
        run_context: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(
            config,
            storage_dir=storage_dir,
            event_sink=event_sink,
            run_context=run_context,
        )
        self._entries: list[MemoryEntry] = []

    def _retain_impl(
        self, content: str, metadata: dict[str, Any]
    ) -> MemoryOperationResult:
        self._entries.append(MemoryEntry(content=content, metadata=metadata))
        return MemoryOperationResult(
            value="Saved to memory.",
            result={"write_count": 1},
        )

    def _recall_impl(self, query: str, top_k: int) -> MemoryOperationResult:
        matches = [e for e in self._entries if query.casefold() in e.content.casefold()]
        items = [render_tool_memory_entry(e) for e in matches[-top_k:]]
        return MemoryOperationResult(
            value=items,
            result={
                "returned_items": items,
                "result_count": len(items),
                "output_bytes": self._memory_size_proxy(items),
            },
        )

Methods you implement¶

Do not replace the public retain, recall, delete and consolidate methods. They are what the agent's memory tools and the benchmark call, and the base class wraps your _*_impl methods to time them, log an event for each call, and turn an exception into an error event plus a failure message for the agent. If you need to wrap one, for example to hold a lock as the Letta plugin does, call super() so the timing and logging still happen.

You also do not implement set_retention_policy, which the Retention policy benchmark uses. Set privacy_config in the memory-system YAML instead; the framework then wraps your system in a privacy layer that provides it.

MemoryOperationResult¶

result keys¶

Some keys are read by the framework's metrics and trace viewer; the rest are conventions that keep plugins consistent.

The request side (content size, query, top_k, run context) is recorded by the base class; you do not need to add it.

Helpers¶

4. Expose the plugin¶

Create a factory and a MemoryPlugin descriptor in your package:

def _create_memory(
    config: BaseMemoryConfig, runtime: MemoryRuntimeContext
) -> BaseMemorySystem:
    return AcmeMemorySystem(
        config=config,
        storage_dir=runtime.storage_dir,
        event_sink=runtime.event_sink,
        run_context=runtime.run_context,
    )


plugin = MemoryPlugin(
    type_name="acme.memory",
    config_model=AcmeMemoryConfig,
    create=_create_memory,
)

The framework calls create once per sample, with a fresh storage_dir, so state never leaks between samples. type_name must match the config's type and be unique among installed plugins; a duplicate stops discovery with an error.

The contract is synchronous: each operation returns when it is done. Entry points load and run installed Python code; they are not a security sandbox, so only install plugins you trust.

Declare the entry point in your package's pyproject.toml:

[project.entry-points."heams.memory_systems"]
acme_memory = "acme_memory.plugin:plugin"

Then install the package into the framework's environment, for example:

cd python_package
uv pip install -e ../path/to/acme_memory

make setup (and uv sync) reset the environment to the lock file, so reinstall your package afterwards. Plugins bundled in this repository are wired as extras in python_package/pyproject.toml instead.

5. Add a config file¶

Create a YAML under configs/memory_systems/. backend_config is validated against your config class:

id: acme_memory
backend_config:
  type: acme.memory
  name: acme.memory
  top_k: 5
  max_entries: 100

Reference it from a run config's memory_system_config_file, as the configs under configs/runs/ do.

6. Check it¶

cd python_package
uv run heams memory list
uv run heams memory test

memory list shows every discovered plugin type. memory test creates each plugin in a temporary directory, retains a sentinel and recalls it, and reports pass, failure or setup-required. setup-required means the plugin's config cannot be built without settings (for example a model or an API key); test such plugins with a real run config instead. memory test does not exercise delete or consolidate, so cover those in your own tests.

To run your plugin on a benchmark, copy configs/runs/examples/compare_memory_systems.yaml, point one experiment's memory_system_config_file at your YAML, and run it as described in Running experiments.

For a complete, working version of steps 1 to 6, including a real _delete_impl and _consolidate_impl, read memory_plugins/heams_memory_example/src/heams_memory_example/plugin.py.

7. Wrapping an existing memory system¶

Most plugins wrap a library or service that already stores and searches memories. The bundled plugins show the patterns below; each row names the one to read. HippoRAG (memory_plugins/heams_memory_hipporag/) is the smallest and the best place to start; mem0 (memory_plugins/heams_memory_mem0/) adds more configuration; Letta (memory_plugins/heams_memory_letta/) wraps a running service.

Ingestion hints¶

Benchmarks that prime memory with a conversation (LoCoMo, LongMemEval and BEAM) add ingest_* keys to each MemoryEntry's metadata, so a plugin can rebuild the conversation instead of storing turns one by one:

They are hints for ingestion, not facts about the memory: HippoRAG drops them from the metadata of the passages it builds while priming. Entries without them should be stored one by one, which is what the default prime_memory_entries does.