Engram Memory

Memory Management with Engram

Add persistent, long-term memory to AI applications using Engram, Weaviate's managed memory server for LLM agents. Engram automatically extracts, consolidates, and stores memories from raw text or conversations, and retrieves them with vector, keyword, or hybrid search.

Use this cookbook when the user wants their app to remember things across sessions: user preferences, profiles, past interactions, or lessons learned by an agent (continual learning).

Read if integrating with an agent or chatbot:

• Basic Agent Cookbook — for the RouterAgent and tool design patterns used below.

• Query Agent Chatbot — for the chatbot integration pattern.

Docs to reference if needed:

• Engram docs (overview): https://docs.weaviate.io/engram

• Quickstart: https://docs.weaviate.io/engram/quickstart

• Concepts (memories, topics, scopes, groups, pipelines): https://docs.weaviate.io/engram/concepts

• Guides (store, search, manage, run status): https://docs.weaviate.io/engram/guides

• Engram deep dive blog: https://weaviate.io/blog/engram-deep-dive

Core Rules

• Use uv for Python project/dependency management, and let it manage the virtual environment (uv venv).

• Do not manually author pyproject.toml or uv.lock; let uv generate/update them.

• Use this install set: uv add weaviate-engram python-dotenv (see Installation — Engram is a separate package, not part of weaviate-client)

• If combining with an agent (recommended pairing), also follow the install set from Basic Agent: uv add dspy

• Customise this cookbook to the users specification, ask them for details if not given.

• Engram is a managed service accessed via Weaviate Cloud. If the user does not have an Engram project, direct them to Weaviate Cloud to create one and generate an Engram API key.

• Do not build a hand-rolled memory system (e.g. a raw Weaviate collection with manual inserts) when the user asks for memory — Engram handles extraction, deduplication, and consolidation out of the box. The manual approach in Agentic RAG is only for users who explicitly want full control or self-hosting.

Env Rules

Mandatory:

• ENGRAM_API_KEY — Engram API key from Weaviate Cloud (format eng_...). This carries the project identity; WEAVIATE_URL/WEAVIATE_API_KEY are not needed for Engram itself.

Optional (only when combining with an agent/chatbot):

• An LLM provider API key (e.g. OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY)

Engram Overview

Key concepts (use these terms with the user):

Topics and groups are configured per project in the Weaviate Cloud console (starter templates exist for personalization and continual learning). Code interacts with them by name — do not assume you can create topics from the SDK; ask the user what topics their project defines, or assume the template default (e.g. UserKnowledge).

A read-only client.groups API (get, list, returning each group's topics with their descriptions and scoping) is merged upstream but not in 1.0.1 — do not call it until a release includes it. Until then, ask the user rather than introspecting.

Installation

Engram ships as its own Python SDK, weaviate-engram. It does not come with weaviate-client (or weaviate-agents) — installing those does not give you Engram, and Engram does not require them.

uv add weaviate-engram

# or with pip
pip install weaviate-engram

Note the package name and import name differ: install weaviate-engram, import engram:

from engram import AsyncEngramClient

This guide targets weaviate-engram 1.0.x (current release 1.0.1), which requires Python 3.11–3.14. 1.0 is the first stable API — if the user is on a 0.x release, upgrade before following this guide, and pin the dependency (weaviate-engram>=1.0.1,<2) so a future major cannot silently change the surface.

Client Setup

Use the async client (AsyncEngramClient) — memory calls sit in request paths and event loops (FastAPI, chat backends, agents), so they should not block. A synchronous EngramClient with the identical surface (drop the await) exists for scripts and sync-only frameworks.

import os
from dotenv import load_dotenv
from engram import AsyncEngramClient

load_dotenv()
client = AsyncEngramClient(api_key=os.environ["ENGRAM_API_KEY"])

All client.memories.* and client.runs.* methods below are coroutines — call them with await from async code. See Async Client for general async patterns (lifecycle, FastAPI integration) that apply here too.

Close the client when you are done with it. Both clients hold an HTTP connection pool: await client.aclose() on the async client, client.close() on the sync one. Both are also context managers, which is the safer form for scripts and tests:

async with AsyncEngramClient(api_key=os.environ["ENGRAM_API_KEY"]) as client:
    ...

In a long-lived server, construct one client at startup and close it on shutdown (FastAPI lifespan) — never one per request.

EngramClient/AsyncEngramClient also accept base_url (default https://api.engram.weaviate.io) and timeout (default 30.0 seconds).

Error Handling

All SDK exceptions inherit from EngramError and import from either engram or engram.errors:

Two behaviours to handle deliberately.

Searching a user_id that has no memories yet raises APIError (HTTP 422, "user not found") instead of returning an empty result. A user whose memories were all deleted returns an empty result, not an error. Any code that recalls memories before the user's first add — which includes every chatbot's first message from a new user — must catch this and treat it as "no memories" (the integration patterns below do). The same error also appears transiently just after a user's first add while the tenant initializes; see Run Status.

AuthenticationError subclasses APIError, so a bare except APIError swallows a bad API key — and a misconfigured app then looks exactly like one with an empty memory store, silently and permanently. Always let authentication errors through:

from engram.errors import APIError, AuthenticationError

try:
    memories = await client.memories.search(query=q, user_id=user_id)
except AuthenticationError:
    raise                  # configuration error — never silently degrade
except APIError:
    memories = []          # this user genuinely has no memories yet

Storing Memories

client.memories.add(...) is fire-and-forget: it returns a run immediately and processes in the background. All three input types share optional parameters user_id, properties, and group (defaults to "default").

String input — raw text, Engram extracts the facts:

run = await client.memories.add(
    "The user prefers dark mode and works primarily in Python.",
    user_id="alice",
)
print(run.run_id, run.status)  # status: "running"

Conversation input — chat transcripts (OpenAI Chat Completions message format; roles user, assistant, system, tool, developer are supported):

run = await client.memories.add(
    [
        {"role": "user", "content": "I just moved to Berlin."},
        {"role": "assistant", "content": "Welcome to Berlin!"},
        {"role": "user", "content": "I prefer specialty coffee."},
    ],
    user_id="alice",
)

Pre-extracted input — you did the extraction yourself; items skip LLM extraction and go straight to transform + commit. Each item targets a topic by name:

from engram import PreExtractedInput, PreExtractedItem

run = await client.memories.add(
    PreExtractedInput(items=[
        PreExtractedItem(content="User prefers dark mode", topic="UserKnowledge"),
        PreExtractedItem(content="User works in Python", topic="UserKnowledge"),
    ]),
    user_id="alice",
)

Typed inputs — the plain str and list[dict] forms above are shorthands. The explicit models add fields the shorthands cannot express, most usefully timestamps, which matter when backfilling history so Engram reconciles facts in the order they actually happened rather than the order you uploaded them:

from datetime import datetime
from engram import StringInput, ConversationInput, MessageInput

run = await client.memories.add(
    StringInput(
        content="The user switched from Postgres to DuckDB.",
        created_at=datetime(2026, 3, 14, 9, 30),
    ),
    user_id="alice",
)

run = await client.memories.add(
    ConversationInput(
        messages=[MessageInput(role="user", content="I prefer specialty coffee.")],
        metadata={"source": "support-ticket"},
    ),
    user_id="alice",
)

MessageInput also carries tool_call_id, name, and tool_calls for agent transcripts that include tool calls.

The REST API accepts one further parameter the Python SDK does not expose: root, the pipeline root name, for projects with advanced multi-pipeline configurations. Call the REST API directly if a project needs it.

Run Status

Storage is async, so a search immediately after add may not see new memories. Block until processing finishes when tests or demos need read-after-write:

status = await client.runs.wait(run.run_id)
print(status.status)  # "completed"

# committed_operations groups results into created / updated / deleted lists,
# each entry carrying memory_id and committed_at. Prefer the three shorthand
# properties: committed_operations is None on a failed run, so reaching through
# it (status.committed_operations.created) raises AttributeError exactly when
# something went wrong — the shorthands return [] instead.
created_ids = [op.memory_id for op in status.memories_created]
#   also available: status.memories_updated, status.memories_deleted

Run statuses: running, in_buffer (paused at a buffer step waiting for a count/time trigger), completed, failed. Only completed and failed are terminal. runs.wait keeps polling anything else, so waiting on a run that parks at in_buffer always ends in EngramTimeoutError — never block a test on a buffered pipeline; use runs.get and accept in_buffer as a valid resting state. Default pipelines typically commit within ~5–15 seconds (conversation inputs trend slower) — a run stuck in running for minutes warrants checking for failed. runs.wait polls every interval=0.5 seconds with a default timeout=30.0 seconds and raises EngramTimeoutError past it; pass a larger timeout= when waiting in tests. In production chat loops, do not wait — fire and forget.

A completed run does not guarantee the memory is searchable yet. On a user's first add, the backing tenant may still be initializing, and a search issued right after the run completes can raise APIError or come back empty. Tests, demos, and seed scripts that need read-after-write should retry briefly:

import asyncio
from engram.errors import APIError, AuthenticationError

async def search_when_ready(**kwargs):
    """Search, tolerating a cold tenant right after a user's first add."""
    for attempt in range(5):
        try:
            return await client.memories.search(**kwargs)
        except AuthenticationError:
            raise
        except APIError:
            if attempt == 4:
                raise
            await asyncio.sleep(3)

This only affects the first write for a given scope — steady-state searches do not need it.

There is no single delete-all operation for a user. For deterministic cleanup (test teardown, "forget me" features), either collect memory_ids from committed_operations at write time, or list all of a user's memories with retrieval_config=FetchRetrieval(limit=N) (unranked listing — a query string is still required but unused) and delete by id.

Searching Memories

results = await client.memories.search(
    query="What programming language does the user prefer?",
    user_id="alice",
)
for memory in results:
    print(memory.content, memory.topic, memory.score)

Two things the examples don't show: search returns a SearchResults object (a Sequence of memories — iteration, len(), and indexing work), not a plain list, so annotate helpers accordingly. And results are filled up to limit even when weakly relevant: score is higher-is-better, but its range depends on retrieval type — hybrid is normalised roughly 0–1, while vector and BM25 scores run far lower. Apply a relevance cutoff before injecting memories into a prompt; ~0.5 is a reasonable starting point for hybrid (it would drop everything for vector/BM25 — tune per retrieval type and app).

Retrieval types — pass retrieval_config; hybrid is the default general-purpose choice:

from engram import VectorRetrieval, BM25Retrieval, HybridRetrieval, FetchRetrieval

results = await client.memories.search(
    query="user preferences",
    user_id="alice",
    retrieval_config=HybridRetrieval(limit=10),
    # or VectorRetrieval / BM25Retrieval / FetchRetrieval (unranked listing)
)

When the server-side default limit is fine, retrieval_config also accepts the bare strings "vector", "bm25", "hybrid", and "fetch" — retrieval_config="hybrid" is equivalent to HybridRetrieval(). Omitting retrieval_config entirely lets the server pick.

Filter by topic:

results = await client.memories.search(
    query="user preferences",
    topics=["UserKnowledge"],
    user_id="alice",
    retrieval_config=HybridRetrieval(limit=10),
)

Custom scope properties — narrow by e.g. conversation. At search time properties are optional filters (omitting one searches across all values); at add time every property the target topic is scoped by must be provided:

from engram import Topic

results = await client.memories.search(
    query="what did we decide?",
    user_id="alice",
    properties={"conversation_id": "abc-123"},
    topics=[
        "user_facts",                                                  # plain string: not conversation-scoped, ignores the filter
        Topic(name="conversation_summary"),                            # no properties: inherits the global filter
        Topic(name="messages", properties={"conversation_id": None}),  # None: clears the global filter for this topic only
    ],
)

Omitting topics searches every topic in the group.

Managing Memories

Get or delete an individual memory by ID (IDs come from search results or committed_operations):

memory = await client.memories.get(memory_id, user_id="alice", group="default")

await client.memories.delete(memory_id, user_id="alice", group="default")

Integration Pattern: Chatbot Memory

The standard pattern for chatbots: search memories before generating, add the exchange after responding.

from engram.errors import APIError, AuthenticationError

async def chat_turn(user_id: str, user_message: str, generate) -> str:
    # 1. Recall relevant memories for this message. A user_id that has never
    #    had an add (first ever message) raises APIError, not an empty result.
    #    AuthenticationError subclasses APIError, so re-raise it — otherwise a
    #    bad API key degrades into "this user has no memories" forever.
    try:
        memories = await client.memories.search(
            query=user_message,
            user_id=user_id,
            retrieval_config=HybridRetrieval(limit=5),
        )
    except AuthenticationError:
        raise
    except APIError:
        memories = []
    memory_context = "\n".join(f"- {m.content}" for m in memories if m.score >= 0.5)

    # 2. Generate with memories in context (any LLM framework)
    response = generate(user_message, memory_context)

    # 3. Store the exchange — the add call returns as soon as the run is
    #    accepted; processing happens server-side. Do not wait on the run.
    await client.memories.add(
        [
            {"role": "user", "content": user_message},
            {"role": "assistant", "content": response},
        ],
        user_id=user_id,
    )
    return response

generate is the user's generation function — e.g. the generate from Basic RAG with memory_context prepended to the context.

Two production notes. Conversation input extracts memories from the dialogue itself too (e.g. "the user asked for dinner ideas") — if recall gets polluted with meta-chatter over time, tighten the topic descriptions. And chat_turn discards the returned run — for a "forget me" feature, either record run.run_id/committed_operations out-of-band, or list-and-delete with FetchRetrieval (see Run Status).

Integration Pattern: Agent Memory Tools

Expose Engram as tools on the RouterAgent from Basic Agent, so the agent decides when to recall or remember:

from engram.errors import APIError, AuthenticationError

async def search_memories(query: str) -> str:
    """Search the user's long-term memory for facts, preferences, and past interactions.
    Use when answering may depend on something the user said in a previous session."""
    try:
        results = await client.memories.search(
            query=query,
            user_id=CURRENT_USER_ID,
            retrieval_config=HybridRetrieval(limit=5),
        )
    except AuthenticationError:
        raise  # configuration error — never report it to the model as "no memories"
    except APIError:  # user has no memories yet
        return "No relevant memories found."
    return "\n".join(m.content for m in results) or "No relevant memories found."


async def store_memory(fact: str) -> str:
    """Save an important fact about the user to long-term memory.
    Use when the user states a lasting preference, correction, or piece of personal context."""
    run = await client.memories.add(fact, user_id=CURRENT_USER_ID)
    return f"Memory stored (run {run.run_id})."


router = RouterAgent(model="<model_name>", tools=[search_memories, store_memory])

CURRENT_USER_ID should come from the app's session/auth context — set it per request, never hardcode it in multi-user apps.

The RouterAgent from Basic Agent calls tools synchronously (tool_result = tool_function(**tool_inputs)). With async tools, make get_response an async def and await coroutine tools:

import inspect

tool_result = tool_function(**tool_inputs)
if inspect.isawaitable(tool_result):
    tool_result = await tool_result

If the agent framework is strictly synchronous and cannot be adapted, use the sync EngramClient inside the tools instead.

For continual learning (project-wide agent memory, e.g. "filter on the genres property instead of doing a near-text query"), use a project-wide topic and omit user_id — lessons learned from one user's sessions then benefit all users. This is configured in the project's topics (the continual learning template).

Off-the-shelf Integrations

Before writing a custom integration, check whether the packaged one already fits — it needs no application code.

Claude Code plugin — gives Claude Code long-term, cross-session memory backed by Engram. It recalls relevant memories before each answer and stores each completed turn through hooks; the agent has no tools to call, and memory is best-effort and never blocks a session. Pick the Coding Assistant topic template when creating the project, then:

export ENGRAM_API_KEY=...   # in ~/.zshrc or ~/.bashrc

# inside a Claude Code session
/plugin marketplace add weaviate/engram-plugins
/plugin install engram@weaviate-engram

See weaviate/engram-plugins for optional customization.

The plugin reads ENGRAM_BASE_URL to target a staging or self-hosted endpoint; the SDK takes the same override as EngramClient(base_url=...).

REST API (non-Python stacks)

Same operations over HTTP, with Authorization: Bearer $ENGRAM_API_KEY:

# Store
curl -X POST https://api.engram.weaviate.io/v1/memories \
  -H "Authorization: Bearer $ENGRAM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"input": {"string": {"content": ["The user prefers dark mode."]}}, "user_id": "alice"}'

# Search
curl -X POST https://api.engram.weaviate.io/v1/memories/search \
  -H "Authorization: Bearer $ENGRAM_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"query": "What editor does the user prefer?", "user_id": "alice", "retrieval_config": {"retrieval_type": "hybrid", "limit": 5}}'

# Run status
curl https://api.engram.weaviate.io/v1/runs/{run-id} \
  -H "Authorization: Bearer $ENGRAM_API_KEY"

There is no official TypeScript/JavaScript SDK, so REST is the integration path for TypeScript/Next.js backends (see Frontend Interface) — call it from server-side routes only, never expose ENGRAM_API_KEY to the browser. The full endpoint list is at https://docs.weaviate.io/engram/api/rest.

User-specific Customisations

If not specified ask the user about these points before implementing:

Memory use case

• Personalization (per-user preferences/profile) → user-scoped topics, always pass user_id.

• Continual learning (agent improves from experience, shared across users) → project-wide topics, no user_id.

• Both → separate topics (or separate groups) in the same project; scoping is per topic.

Topics

What should the app remember? Each distinct category should be a topic with a clear natural-language description — descriptions directly control what the extraction LLM stores. For a single running summary or profile per user, use a bounded topic (max one memory per scope, updated in place) rather than accumulating facts.

Input type

• Chat app → conversation input per exchange.

• Event/log style data ("User viewed page X") → string input.

• The user already extracts facts with their own LLM logic → pre-extracted input.

Retrieval

Hybrid is the right default. Vector for purely conceptual recall, BM25 for exact terms/IDs. Ask what limit fits their context budget (5–10 is typical).

Troubleshooting

• APIError (422, user "..." not found) on search: nothing has ever been added for that user_id — catch it and treat as an empty result, as in the integration patterns. A user whose memories were all deleted returns an empty result, not this error.

• A bad API key behaves like an empty memory store: AuthenticationError subclasses APIError, so a bare except APIError swallows it. Catch AuthenticationError first and re-raise.

• Searches return nothing right after add: processing is asynchronous — await client.runs.wait(run.run_id) before searching, or check the run status for failed. If this is the first write for that user, the tenant can still be initializing even after the run reports completed; retry the search a few times with a short sleep (see Run Status).

• Memory counts don't match inputs: extraction may split one input into several memories or merge facts into existing ones — never assert exact counts in tests; assert on search results instead.

• Run stuck at in_buffer: the group's pipeline has a buffer step waiting for a count/time trigger; this is expected behaviour, not an error. Because in_buffer is not terminal, runs.wait will raise EngramTimeoutError on it — use runs.get instead when a pipeline buffers.

• Facts not being stored: only information matching a configured topic description is extracted — check the project's topics in the Weaviate Cloud console and tighten/broaden their descriptions.

• Scope errors on add: every scope property required by the target topic (including user_id for user-scoped topics) must be provided.

• AuthenticationError (401, .status_code == 401): ensure ENGRAM_API_KEY is set and is an Engram key (eng_...), not a Weaviate cluster API key. An empty or malformed key is not validated client-side — it fails on the first call. Note 403 raises plain APIError, not AuthenticationError.

• EngramTimeoutError from runs.wait: the default timeout is 30 s and conversation inputs can exceed it. The exception carries .run_id and .timeout. Pass a larger timeout=, or stop waiting and let the pipeline finish in the background. Check the run is not simply parked at in_buffer before raising the timeout.

• Unclosed-connection warnings at exit: close the client (await client.aclose() / client.close()) or use it as a context manager.

• ImportError on StringInput, MessageInput, or AuthenticationError: the project is on a 0.x weaviate-engram; upgrade to 1.0.1 or newer.

• For any other issues, refer to the official docs at https://docs.weaviate.io/engram and use web search extensively for troubleshooting.

Done Criteria

• Create test scripts to check store → wait → search → get → delete work end-to-end against the user's Engram project. Use a throwaway user_id (e.g. f"test-{uuid.uuid4().hex[:8]}"), and clean up by deleting the memory_ids from status.memories_created, or by listing the user's memories with FetchRetrieval and deleting by id. Wrap the first search in the retry helper from Run Status so a cold tenant does not fail the test. Tear down tests after completion, or create a proper test suite with pytest (requires install).

• Verify persistence the way a user would: seed a fact, await client.runs.wait(...), then search for it from a fresh process/session.

• Memory writes in the app are fire-and-forget; nothing in the request path blocks on runs.wait.

• Recall paths re-raise AuthenticationError instead of folding it into "no memories".

• The client is closed on shutdown (aclose/close, or a context manager), and is not constructed per request.

• User has completed specification of the app.