Memory API

memory

Memory subsystem for orqest agents.

Provides a pluggable memory protocol (MemoryStore) with a local SQLite backend. Three cognitive memory kinds are first-class:

• semantic — what's true (facts, summaries)
• episodic — what happened (sessions, traces)
• procedural — how to do things (skills via :class:Skill)

MemoryConfig dataclass

MemoryConfig(
    backend: Literal["local", "supabase"] = "local",
    local_db_path: str = "~/.orqest/memory.db",
    embedding_model: str = "all-MiniLM-L6-v2",
    embedding_dim: int = 384,
    supabase_url: str | None = None,
    supabase_key: str | None = None,
    semantic: PerKindConfig = PerKindConfig(),
    episodic: PerKindConfig = PerKindConfig(),
    procedural: PerKindConfig = PerKindConfig(),
    tool: PerKindConfig = (
        lambda: PerKindConfig(
            ttl_days=None,
            version_on_edit=True,
            decay_on_failure=0.5,
        )
    )(),
)

Immutable configuration for the memory subsystem.

PerKindConfig dataclass

PerKindConfig(
    decay_on_failure: float = 0.7,
    prune_below: float = 0.1,
    ttl_days: int | None = None,
    version_on_edit: bool = False,
)

Per-kind reliability / retention / versioning policy.

decay_on_failure class-attribute instance-attribute

decay_on_failure: float = 0.7

Reliability multiplier applied on a failed-recall report.

prune_below class-attribute instance-attribute

prune_below: float = 0.1

Reliability floor below which an entry is pruned after decay.

ttl_days class-attribute instance-attribute

ttl_days: int | None = None

Retention window. None means keep forever; otherwise entries older than this are deleted by :meth:LocalMemoryStore.prune_expired.

version_on_edit class-attribute instance-attribute

version_on_edit: bool = False

When True, storing a procedural entry whose structured_content.name matches an already-stored skill bumps the new entry's version one past the highest stored version and keeps the prior rows — an audit trail.

LocalMemoryStore

LocalMemoryStore(
    db_path: str | Path | None = None,
    *,
    config: MemoryConfig | None = None,
    embedder: EmbedderT | None = None,
    strategies: dict[str, RetrievalStrategy] | None = None,
)

SQLite-backed memory store with optional FTS5 full-text search.

Per-kind retrieval strategies are configurable via the strategies constructor argument; the default table provides Semantic / Episodic / Procedural strategies. Override individual entries to inject custom behavior (e.g. a fuzzy judge for procedural recall).

Initialize the store.

Parameters:

Name	Type	Description	Default
db_path	str | Path | None	SQLite file path (created lazily). When omitted, config.local_db_path is used.	None
config	MemoryConfig | None	Memory subsystem configuration. Supplies the per-kind policy read by :meth:update_reliability, :meth:store, and :meth:prune_expired. Defaults to :class:MemoryConfig defaults.	None
embedder	EmbedderT | None	Optional Callable[[str], list[float]] (sync or async). When supplied, :meth:store embeds entry content and semantic recall ranks by cosine similarity instead of FTS5 keyword match. Orqest stays embedding-model-neutral — the consumer wires the embedder.	None
strategies	dict[str, RetrievalStrategy] | None	Optional override of the per-kind retrieval table. Defaults to default_strategy_table(embedder=...) (Semantic / Episodic / Procedural). Unknown memory_type values fall back to the "semantic" strategy at recall time. When you pass an explicit table, wiring the embedder into it is yours.	None

Source code in orqest/memory/local.py

def __init__(
    self,
    db_path: str | Path | None = None,
    *,
    config: MemoryConfig | None = None,
    embedder: EmbedderT | None = None,
    strategies: dict[str, RetrievalStrategy] | None = None,
) -> None:
    """Initialize the store.

    Args:
        db_path: SQLite file path (created lazily). When omitted,
            ``config.local_db_path`` is used.
        config: Memory subsystem configuration. Supplies the per-kind
            policy read by :meth:`update_reliability`, :meth:`store`,
            and :meth:`prune_expired`. Defaults to :class:`MemoryConfig`
            defaults.
        embedder: Optional ``Callable[[str], list[float]]`` (sync or
            async). When supplied, :meth:`store` embeds entry content
            and semantic recall ranks by cosine similarity instead of
            FTS5 keyword match. Orqest stays embedding-model-neutral —
            the consumer wires the embedder.
        strategies: Optional override of the per-kind retrieval table.
            Defaults to ``default_strategy_table(embedder=...)`` (Semantic
            / Episodic / Procedural). Unknown ``memory_type`` values fall
            back to the ``"semantic"`` strategy at recall time. When you
            pass an explicit table, wiring the embedder into it is yours.
    """
    self._config = config or MemoryConfig()
    self._embedder = embedder
    resolved_path = (
        db_path if db_path is not None else self._config.local_db_path
    )
    self._db_path = Path(resolved_path).expanduser()
    self._db: aiosqlite.Connection | None = None
    self._fts_available: bool = False
    self._strategies = strategies or default_strategy_table(embedder=embedder)

store async

store(entry: MemoryEntry) -> str

Persist a memory entry. Returns the entry id.

When the procedural per-kind policy has version_on_edit=True and a procedural entry's structured_content.name matches an existing stored skill, the new entry's version is bumped to one past the highest stored version — the prior rows are kept (entries are keyed by id, so this is always an INSERT, never an overwrite-by-name).

Source code in orqest/memory/local.py

async def store(self, entry: MemoryEntry) -> str:
    """Persist a memory entry. Returns the entry id.

    When the procedural per-kind policy has ``version_on_edit=True`` and
    a procedural entry's ``structured_content.name`` matches an existing
    stored skill, the new entry's ``version`` is bumped to one past the
    highest stored version — the prior rows are kept (entries are keyed
    by id, so this is always an INSERT, never an overwrite-by-name).
    """
    try:
        db = await self._ensure_db()
        structured = await self._maybe_version(db, entry)
        embedding = entry.embedding or await self._embed(entry.content)
        await db.execute(
            """INSERT OR REPLACE INTO memories
               (id, content, structured_content, memory_type, source_agent,
                confidence, embedding, metadata, created_at, last_accessed,
                access_count, reliability_score)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (
                entry.id,
                entry.content,
                json.dumps(structured) if structured is not None else None,
                entry.memory_type,
                entry.source_agent,
                entry.confidence,
                json.dumps(embedding) if embedding else None,
                json.dumps(entry.metadata),
                entry.created_at.isoformat(),
                entry.last_accessed.isoformat(),
                entry.access_count,
                entry.reliability_score,
            ),
        )
        await db.commit()
    except Exception:
        logger.warning("Failed to store memory entry {id}", id=entry.id)
    return entry.id

recall async

recall(
    query: str,
    *,
    k: int = 5,
    filters: MemoryFilter | None = None,
) -> list[MemoryEntry]

Retrieve entries matching the query, applying optional filters.

Dispatches to the strategy keyed by filters.memory_type. Unknown / missing memory_type → "semantic" strategy (preserves v0.0.1 behavior).

Source code in orqest/memory/local.py

async def recall(
    self,
    query: str,
    *,
    k: int = 5,
    filters: MemoryFilter | None = None,
) -> list[MemoryEntry]:
    """Retrieve entries matching the query, applying optional filters.

    Dispatches to the strategy keyed by ``filters.memory_type``.
    Unknown / missing memory_type → ``"semantic"`` strategy
    (preserves v0.0.1 behavior).
    """
    try:
        db = await self._ensure_db()
        kind = (filters.memory_type if filters else None) or "semantic"
        strategy = self._strategies.get(kind, self._strategies.get("semantic"))
        if strategy is None:
            return []

        rows = await strategy.recall(
            db,
            query,
            k=k,
            filters=filters,
            fts_available=self._fts_available,
        )

        entries: list[MemoryEntry] = []
        now = datetime.now()
        for row in rows:
            entry = _row_to_entry(row)
            entries.append(entry)
            # Update access metadata
            await db.execute(
                """UPDATE memories
                   SET last_accessed = ?, access_count = access_count + 1
                   WHERE id = ?""",
                (now.isoformat(), entry.id),
            )
        await db.commit()
    except Exception as exc:
        logger.warning(
            "Failed to recall memories for query {q!r}: {e}",
            q=query,
            e=exc,
        )
        return []
    else:
        return entries

forget async

forget(entry_id: str) -> None

Remove a memory entry by id. No error if not found.

Source code in orqest/memory/local.py

async def forget(self, entry_id: str) -> None:
    """Remove a memory entry by id. No error if not found."""
    try:
        db = await self._ensure_db()
        await db.execute("DELETE FROM memories WHERE id = ?", (entry_id,))
        await db.commit()
    except Exception:
        logger.warning("Failed to forget memory {id}", id=entry_id)

update_reliability async

update_reliability(entry_id: str, *, success: bool) -> None

Decay an entry's reliability on a failed-recall report.

success is a no-op — reliability only decays. On failure the entry's reliability is multiplied by the per-kind decay_on_failure factor, and the entry is pruned if it drops below the per-kind prune_below floor (see :class:PerKindConfig).

Source code in orqest/memory/local.py

async def update_reliability(
    self, entry_id: str, *, success: bool
) -> None:
    """Decay an entry's reliability on a failed-recall report.

    ``success`` is a no-op — reliability only decays. On failure the
    entry's reliability is multiplied by the per-kind
    ``decay_on_failure`` factor, and the entry is pruned if it drops
    below the per-kind ``prune_below`` floor (see :class:`PerKindConfig`).
    """
    if success:
        return

    try:
        db = await self._ensure_db()
        cursor = await db.execute(
            "SELECT memory_type FROM memories WHERE id = ?", (entry_id,)
        )
        row = await cursor.fetchone()
        if row is None:
            return
        policy = self._policy_for(row["memory_type"])
        await db.execute(
            "UPDATE memories SET reliability_score = reliability_score * ? "
            "WHERE id = ?",
            (policy.decay_on_failure, entry_id),
        )
        # Prune entries that have decayed below the reliability floor.
        await db.execute(
            "DELETE FROM memories WHERE id = ? AND reliability_score < ?",
            (entry_id, policy.prune_below),
        )
        await db.commit()
    except Exception:
        logger.warning(
            "Failed to update reliability for {id}", id=entry_id
        )

prune_expired async

prune_expired() -> int

Delete entries older than their per-kind ttl_days.

Best-effort maintenance: a kind whose :class:PerKindConfig.ttl_days is None is never pruned. Errors are logged, never raised — the method returns the number of rows deleted (0 on failure).

Source code in orqest/memory/local.py

async def prune_expired(self) -> int:
    """Delete entries older than their per-kind ``ttl_days``.

    Best-effort maintenance: a kind whose :class:`PerKindConfig.ttl_days`
    is ``None`` is never pruned. Errors are logged, never raised — the
    method returns the number of rows deleted (``0`` on failure).
    """
    pruned = 0
    try:
        db = await self._ensure_db()
        now = datetime.now()
        for kind in ("semantic", "episodic", "procedural"):
            ttl_days = getattr(self._config, kind).ttl_days
            if ttl_days is None:
                continue
            cutoff = (now - timedelta(days=ttl_days)).isoformat()
            cursor = await db.execute(
                "DELETE FROM memories "
                "WHERE memory_type = ? AND created_at < ?",
                (kind, cutoff),
            )
            pruned += max(0, cursor.rowcount)
        await db.commit()
    except Exception:
        logger.warning("prune_expired failed")
    return pruned

count async

count() -> int

Return the total number of stored entries.

Source code in orqest/memory/local.py

async def count(self) -> int:
    """Return the total number of stored entries."""
    try:
        db = await self._ensure_db()
        cursor = await db.execute("SELECT COUNT(*) FROM memories")
        row = await cursor.fetchone()
        return row[0] if row else 0
    except Exception:
        logger.warning("Failed to count memories")
        return 0

list_recent async

list_recent(
    *, memory_type: str | None = None, limit: int = 50
) -> list[MemoryEntry]

Return the most recently stored entries, newest first.

Browse-style enumeration that complements :meth:recall (which is query-driven). Used by consumer surfaces that want to render a "memory inspector" view without issuing a search. Filters by memory_type when supplied; None returns every kind.

Best-effort like the other read paths — returns [] on any SQLite failure rather than raising.

Source code in orqest/memory/local.py

async def list_recent(
    self,
    *,
    memory_type: str | None = None,
    limit: int = 50,
) -> list[MemoryEntry]:
    """Return the most recently stored entries, newest first.

    Browse-style enumeration that complements :meth:`recall` (which
    is query-driven). Used by consumer surfaces that want to render
    a "memory inspector" view without issuing a search. Filters
    by ``memory_type`` when supplied; ``None`` returns every kind.

    Best-effort like the other read paths — returns ``[]`` on any
    SQLite failure rather than raising.
    """
    try:
        db = await self._ensure_db()
        sql = "SELECT * FROM memories"
        params: tuple[Any, ...] = ()
        if memory_type:
            sql += " WHERE memory_type = ?"
            params = (memory_type,)
        sql += " ORDER BY created_at DESC LIMIT ?"
        params = (*params, max(1, min(int(limit), 500)))
        cursor = await db.execute(sql, params)
        rows = await cursor.fetchall()
        return [_row_to_entry(row) for row in rows]
    except Exception:
        logger.warning(
            "list_recent failed (memory_type={mt!r})", mt=memory_type
        )
        return []

close async

close() -> None

Close the underlying aiosqlite connection.

Source code in orqest/memory/local.py

async def close(self) -> None:
    """Close the underlying aiosqlite connection."""
    if self._db is not None:
        await self._db.close()
        self._db = None

MemoryEntry

A single unit of stored knowledge.

structured_content class-attribute instance-attribute

structured_content: dict[str, Any] | None = None

Typed payload for non-text memory (e.g. Skill for procedural). When memory_type == "procedural" and this field is set, it must validate against the :class:Skill schema. Validation is gated to procedural entries to keep the legacy semantic/episodic paths untouched.

MemoryFilter

Query-time constraints for memory recall.

skill_name class-attribute instance-attribute

skill_name: str | None = None

Exact-match on structured_content.name — only applies when memory_type == "procedural". No-op otherwise.

skill_min_version class-attribute instance-attribute

skill_min_version: int | None = None

Filter procedural entries to version >= skill_min_version.

MemoryStore

Protocol for pluggable memory backends.

Implementations must provide async store, recall, forget, update_reliability, and count operations.

store async

store(entry: MemoryEntry) -> str

Persist a memory entry and return its id.

Source code in orqest/memory/store.py

async def store(self, entry: MemoryEntry) -> str:
    """Persist a memory entry and return its id."""
    ...

recall async

recall(
    query: str,
    *,
    k: int = 5,
    filters: MemoryFilter | None = None,
) -> list[MemoryEntry]

Retrieve the top-k entries matching the query and optional filters.

Source code in orqest/memory/store.py

async def recall(
    self,
    query: str,
    *,
    k: int = 5,
    filters: MemoryFilter | None = None,
) -> list[MemoryEntry]:
    """Retrieve the top-k entries matching the query and optional filters."""
    ...

forget async

forget(entry_id: str) -> None

Remove a memory entry by id. No error if not found.

Source code in orqest/memory/store.py

async def forget(self, entry_id: str) -> None:
    """Remove a memory entry by id. No error if not found."""
    ...

update_reliability async

update_reliability(entry_id: str, *, success: bool) -> None

Adjust an entry's reliability score based on outcome.

Source code in orqest/memory/store.py

async def update_reliability(self, entry_id: str, *, success: bool) -> None:
    """Adjust an entry's reliability score based on outcome."""
    ...

count async

count() -> int

Return the total number of stored entries.

Source code in orqest/memory/store.py

async def count(self) -> int:
    """Return the total number of stored entries."""
    ...

Skill

Procedural memory content — a learned tool sequence with outcome.

Stored inside :class:MemoryEntry.structured_content when memory_type == "procedural". The trigger field is the natural-language phrase the agent matches against to decide whether to invoke the skill; it is also written verbatim into :class:MemoryEntry.content so FTS5 still indexes it.

SkillExample

A worked example of a successful Skill invocation.

ToolCallSpec

One step in a Skill's tool_sequence.

EpisodicStrategy

Time-windowed search ordered by created_at DESC.

Same content matching as Semantic, but the ordering reflects the "what happened" framing of episodic memory. A future extension may honor filters.metadata.session_id for session-scoped recall.

ProceduralStrategy

ProceduralStrategy(fuzzy_judge: FuzzyJudge | None = None)

Trigger-match retrieval for skills.

Algorithm

1. Lower-case the query.
2. SQL: select procedural rows whose structured_content -> '$.trigger' (lower-cased) equals the query OR contains the query as a substring.
3. Honor filters.skill_name (exact match on structured_content -> '$.name') and filters.skill_min_version (version numeric compare).
4. Order by reliability_score DESC, version DESC, last_accessed DESC.
5. If exact-match returns rows, return up to k.
6. Else if fuzzy_judge is configured, ask it to pick from the top 20 candidate triggers and return judged matches.
7. Else return [].

The fuzzy judge is injected, not built-in — Orqest core stays LLM-judge-neutral; the consumer wires one if desired.

Source code in orqest/memory/strategies.py

def __init__(self, fuzzy_judge: FuzzyJudge | None = None) -> None:
    self._judge = fuzzy_judge

RetrievalStrategy

One strategy per memory_type. The store dispatches to it.

SemanticStrategy

SemanticStrategy(embedder: EmbedderT | None = None)

Content retrieval — embedding cosine, or FTS5 / LIKE keyword match.

With no embedder this is identical to the v0.0.1 keyword behavior, ordered by recency of access. With an embedder, recall embeds the query, brute-force scores every stored vector by cosine similarity, and returns the top-k — fine for the local SQLite store; the pgvector backend is the path to scale.

Store the optional embedder; None selects the FTS5 path.

Source code in orqest/memory/strategies.py

def __init__(self, embedder: EmbedderT | None = None) -> None:
    """Store the optional embedder; ``None`` selects the FTS5 path."""
    self._embedder = embedder

default_strategy_table

default_strategy_table(
    embedder: EmbedderT | None = None,
) -> dict[str, RetrievalStrategy]

The default per-kind retrieval table used by :class:LocalMemoryStore.

When embedder is supplied, :class:SemanticStrategy ranks by embedding cosine similarity; otherwise it falls back to FTS5 / LIKE.

Source code in orqest/memory/strategies.py

def default_strategy_table(
    embedder: EmbedderT | None = None,
) -> dict[str, RetrievalStrategy]:
    """The default per-kind retrieval table used by :class:`LocalMemoryStore`.

    When ``embedder`` is supplied, :class:`SemanticStrategy` ranks by
    embedding cosine similarity; otherwise it falls back to FTS5 / LIKE.
    """
    return {
        "semantic": SemanticStrategy(embedder=embedder),
        "episodic": EpisodicStrategy(),
        "procedural": ProceduralStrategy(),
        "tool": ToolStrategy(),
    }