Quick Start

Get up and running with engrava in 5 minutes.

Installation

pip install engrava

For vector search and the bundled walkthrough you also need a local embedding encoder — install the embeddings-local extra:

pip install 'engrava[embeddings-local]'

The extra pulls sentence-transformers and torch and downloads a small (~30–90 MB) encoder model on first use. The encoder is not a language model: it turns text into a fixed-size vector. There are no API keys for this local setup, and the walkthrough needs no network after the first download. Engrava itself does not call a language model.

Installing the extra is necessary but not sufficient: a store only embeds and ranks on vectors once an embedding provider is passed to it, so a store built without one skips the vector signal however the extra is installed. See Configuration for wiring one up.

Run the bundled walkthrough

The repository ships a single-file walkthrough that exercises the end-to-end ingest → dream → query flow on a small demo dataset:

python examples/quickstart.py        # 5-minute end-to-end tour

quickstart.py boots an in-memory store, ingests a handful of percepts (things the agent learned about the user) plus two utterances (replies the agent already produced), runs one dreaming consolidation cycle, and queries via hybrid search. The expected top result for the shipped query is My favorite color is teal..

What is dreaming?

Dreaming is engrava’s offline consolidation step. Between interactions the engine reviews the thoughts that have proven durable — confirmed and revisited over time — and groups related memories into REFLECTION nodes: deterministic, structural summaries that record which observations were grouped and the keywords distilled from them. It uses no language-model call and touches no network. Dreaming is deliberately conservative: a brand-new store of one-off facts has nothing to consolidate yet; REFLECTIONs emerge as memories accumulate and repeat over an agent’s lifetime.

Self-anchored identity

Every thought carries structured metadata that pins its origin. The package exposes three small helpers in engrava.metadata:

Helper Use for
percept() Input arriving from outside (user message, document)
utterance() The agent’s own output sent to the world
thought() The agent’s internal cognition (reflection, plan) 
from engrava import percept, utterance, thought

percept(source_id="user-42", label="user")
# -> {'perspective': 'percept', 'source': {'is_self': False, 'confidence': 'high', 'id': 'user-42', 'label': 'user'}, 'lang': 'en', 'content_type': 'natural_language'}

utterance()
# -> {'perspective': 'utterance', 'source': {'is_self': True, 'confidence': 'high'}, 'lang': 'en', 'content_type': 'natural_language'}

The helpers are pure functions: same arguments always return an equal dictionary, and the returned value carries no shared state. Callers who want a different shape are free to pass a literal dictionary instead — the helpers exist to remove a class of typo-driven shape mismatches at the call site.

Create a Store

import asyncio
import aiosqlite
from engrava import SqliteEngravaCore

async def main() -> None:
    # SqliteEngravaCore wraps an open aiosqlite connection.
    # Use ":memory:" for experimentation, or a file path to persist.
    async with aiosqlite.connect(":memory:") as conn:
        conn.row_factory = aiosqlite.Row
        store = SqliteEngravaCore(conn)
        await store.ensure_schema()
        print("Store ready!")

asyncio.run(main())

The rest of this page assumes you are inside the async with block above, so store and conn are in scope. For a configuration-driven alternative, use await SqliteEngravaCore.from_config("engrava.yaml") (it opens and owns the connection for you).

Store and search a memory — the short way

remember() and recall() are the two-call path for getting started: store a piece of text, then search for it. No IDs to generate, no record to assemble.

await store.remember("User prefers concise answers")
await store.remember("User works in Berlin")

result = await store.recall("what does the user prefer?")
for thought_id, score in result.results:
    record = await store.get_thought(thought_id)
    if record is not None:
        print(f"{record.essence}  (score: {score:.3f})")

remember() stores the text as a thought (generating its ID for you) and returns the stored ThoughtRecord. recall() runs the same hybrid search as search_hybrid() and returns the ranked results — on the store built above that is keyword and priority ranking: no embedding provider is configured, so the vector signal is skipped and its weight redistributed across the active ones. See Hybrid Search for the full signal model.

A note on time. recall() leaves recency ranking off until it has a reference to decay against: an explicit current_cycle, a configured cycle_provider, or a recency_now instant. A cycle is a logical clock you own — increment it once per turn and pass it to recall(..., current_cycle=n) on read, and newer memories start ranking ahead of older ones. remember() stamps both cycle fields at 0; when you need to set the cycle on a write, build a ThoughtRecord with created_cycle=n and call create_thought() (shown below). Until one of the three is present the signal contributes nothing — engrava substitutes no clock of its own.

The rest of this page shows the full-control path: building a ThoughtRecord yourself, linking thoughts with edges, and querying with MindQL. Reach for it when you need to set fields remember() defaults for you (priority, thought type, metadata, and the cycle clock on writes).

Add Thoughts

import uuid
from engrava import ThoughtRecord, ThoughtType, Priority, LifecycleStatus

observation = ThoughtRecord(
    thought_id=str(uuid.uuid4()),
    thought_type=ThoughtType.OBSERVATION,
    essence="Python is great for AI agents",
    content="Python's async ecosystem and rich ML libraries make it ideal.",
    priority=Priority.P2,
    lifecycle_status=LifecycleStatus.ACTIVE,
    created_cycle=0,
    updated_cycle=0,
    source="human",
)
stored = await store.create_thought(observation)
print(f"Created thought: {stored.thought_id}")
from engrava import EdgeRecord, EdgeType

belief = ThoughtRecord(
    thought_id=str(uuid.uuid4()),
    thought_type=ThoughtType.BELIEF,
    essence="SQLite provides zero-config persistence",
    content="WAL mode enables concurrent reads with single-writer safety.",
    priority=Priority.P2,
    lifecycle_status=LifecycleStatus.ACTIVE,
    created_cycle=0,
    updated_cycle=0,
    source="human",
)
await store.create_thought(belief)

edge = await store.create_edge(
    EdgeRecord(
        edge_id=str(uuid.uuid4()),
        from_thought_id=observation.thought_id,
        to_thought_id=belief.thought_id,
        edge_type=EdgeType.ASSOCIATED,
        weight=0.8,
        created_cycle=0,
    )
)
print(f"Linked thoughts via edge: {edge.edge_id}")

Full-Text Search

# search_fts returns (thought_id, bm25_score) tuples — fetch the record for fields.
for thought_id, score in await store.search_fts("Python AI", top_k=5):
    record = await store.get_thought(thought_id)
    if record is not None:
        print(f"  [{record.priority.value}] {record.essence}  (score={score:.3f})")
from engrava import CallbackProvider

# Use any embedding function
provider = CallbackProvider(
    callback=lambda text: [0.1] * 384,  # Replace with real embeddings
    dimension=384,
    model_name="my-model",
)

# Store an embedding for an existing thought
vector = await provider.embed(observation.content)
await store.store_embedding(observation.thought_id, vector, model_name="my-model")

# Search by similarity — returns (thought_id, score) tuples
for thought_id, score in await store.search_similar(vector, top_k=5):
    record = await store.get_thought(thought_id)
    if record is not None:
        print(f"  {record.essence}  (score: {score:.3f})")

Query with MindQL

from engrava import MindQLExecutor, parse

# MindQLExecutor runs against an aiosqlite connection; parse the string first.
executor = MindQLExecutor(conn)

# Find observations
result = await executor.execute(
    parse("FIND thoughts WHERE thought_type = 'OBSERVATION' LIMIT 10")
)
print(f"Found {len(result.rows)} thoughts")

# Count active thoughts
result = await executor.execute(
    parse("COUNT thoughts WHERE lifecycle_status = 'ACTIVE'")
)
print(f"Active thoughts: {result.count}")

Use the CLI

# Database info
engrava --db my_thoughts.db info

# Run a MindQL query
engrava --db my_thoughts.db query "FIND thoughts WHERE thought_type = 'OBSERVATION' LIMIT 5"

# Back up your data
engrava --db my_thoughts.db snapshot -o backup.jsonl

# Restore from backup
engrava --db my_thoughts.db restore -i backup.jsonl

Next Steps