Memory

Three orthogonal layers, each with its own contract:

1. Conversation management — what to do with the message history within a single run (keep, window, summarize) when it grows past the model's context window.
2. Cross-thread store — durable key-value scoped by namespace, used for long-term memory entries the agent reads at session start and writes at session end.
3. Long-term memory manager — the policy layer that decides what to extract from a run and when to retrieve, sitting on top of any BaseStore backend.

For checkpointing (state persistence between runs), see Checkpointers — those backends live in tulip.memory.backends, with S3-compatible object storage as a production target. Note those are checkpointer backends (per-thread state); the cross-thread store layer below is a separate contract.

For the concepts, start with Long-term memory, Conversation management and the cross-thread store.

Conversation management

ConversationManager

Bases: ABC

Base class for conversation management strategies.

Conversation managers handle how message history is maintained, including trimming, summarization, and other memory management.

apply abstractmethod

apply(messages: list[Message]) -> list[Message]

Apply conversation management to messages.

Parameters:

Name	Type	Description	Default
messages	list[Message]	Full message history	required

Returns:

Type	Description
list[Message]	Managed message list (potentially trimmed/summarized)

Source code in .sdk/src/tulip/memory/conversation.py

@abstractmethod
def apply(self, messages: list[Message]) -> list[Message]:
    """
    Apply conversation management to messages.

    Args:
        messages: Full message history

    Returns:
        Managed message list (potentially trimmed/summarized)
    """
    ...

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async version of apply. Supports async summarization functions.

Default implementation delegates to the synchronous apply(). Override in subclasses that need async operations (e.g., LLM summarization).

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async version of apply. Supports async summarization functions.

    Default implementation delegates to the synchronous apply().
    Override in subclasses that need async operations (e.g., LLM summarization).
    """
    return self.apply(messages)

NullManager

Bases: ConversationManager

Null conversation manager - keeps all messages unchanged.

Use this when you want to preserve the entire conversation history without any modifications. Suitable for short conversations or when full history is required.

apply

apply(messages: list[Message]) -> list[Message]

Return messages unchanged.

Source code in .sdk/src/tulip/memory/conversation.py

def apply(self, messages: list[Message]) -> list[Message]:
    """Return messages unchanged."""
    return messages.copy()

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async version of apply. Supports async summarization functions.

Default implementation delegates to the synchronous apply(). Override in subclasses that need async operations (e.g., LLM summarization).

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async version of apply. Supports async summarization functions.

    Default implementation delegates to the synchronous apply().
    Override in subclasses that need async operations (e.g., LLM summarization).
    """
    return self.apply(messages)

SlidingWindowManager

SlidingWindowManager(window_size: int = 20, preserve_system: bool = True, preserve_first_user: bool = True, slide_step: int = 1)

Bases: ConversationManager

Sliding window conversation manager - keeps last N messages.

Preserves the system message (if present) and the last N messages from the conversation. This is a simple and effective strategy for managing conversation length.

When the window would retain no user turn at all, the opening request is re-attached. That case is an agent loop — assistant/tool all the way down — where the first user message is the task, and losing it leaves the model working from a role description and a wall of tool output with no statement of what it was asked to do. It reads as drift rather than as an error, and on Qwen-family chat templates it fails outright with No user query found in messages.. Twenty tool round-trips reach the default window, so this is the ordinary case for a tool-using agent, not an edge one.

A chat is left alone: while any user turn survives in the window, the newest ones are the live context and the first is not privileged.

Only the leading system messages (the instructions, a memory block right after them) are kept at the front. A system note the loop added mid-run stays where it was written and leaves with the turns around it: moving it to the front would change the start of every request after it, and a provider's prompt cache serves only a byte-identical prefix.

For the same reason the window can slide in steps. With slide_step=1 the oldest message goes on every request once the window is full, so the history after the system prompt is different every time and nothing past it is ever served from cache. A larger step drops that many messages at once and then leaves the start alone until the window fills again: the window holds between window_size - slide_step + 1 and window_size messages, and all but one request in slide_step reuses the cache.

Parameters:

Name	Type	Description	Default
window_size	int	Maximum number of messages to keep (excluding system message)	20
preserve_system	bool	Whether to preserve the leading system messages at the start	True
preserve_first_user	bool	Whether to re-attach the opening user turn when the window would otherwise contain none	True
slide_step	int	How many messages the window drops at a time once full.	1

Source code in .sdk/src/tulip/memory/conversation.py

def __init__(
    self,
    window_size: int = 20,
    preserve_system: bool = True,
    preserve_first_user: bool = True,
    slide_step: int = 1,
):
    if window_size < 1:
        raise ValueError("window_size must be at least 1")
    if not 1 <= slide_step <= window_size:
        raise ValueError("slide_step must be between 1 and window_size")
    self.window_size = window_size
    self.preserve_system = preserve_system
    self.preserve_first_user = preserve_first_user
    self.slide_step = slide_step

apply

apply(messages: list[Message]) -> list[Message]

Apply sliding window to messages.

Keeps the system message (if preserve_system is True) and the last window_size messages, re-attaching the opening user turn when the window would otherwise retain none (see the class docstring).

Source code in .sdk/src/tulip/memory/conversation.py

def apply(self, messages: list[Message]) -> list[Message]:
    """
    Apply sliding window to messages.

    Keeps the system message (if preserve_system is True) and the last
    window_size messages, re-attaching the opening user turn when the window
    would otherwise retain none (see the class docstring).
    """
    if not messages:
        return []

    from tulip.core.messages import Role

    result: list[Message] = []
    non_system_messages: list[Message] = []

    leading = True
    for msg in messages:
        leading = leading and msg.role == Role.SYSTEM
        if leading and self.preserve_system:
            result.append(msg)
        else:
            non_system_messages.append(msg)

    # Keep only the last window_size messages
    before_count = len(non_system_messages)
    if before_count > self.window_size:
        anchor: Message | None = None
        if self.preserve_first_user:
            anchor = next(
                (m for m in non_system_messages if m.role == Role.USER),
                None,
            )

        # Drop whole steps, so the cut stays put until the window refills.
        excess = before_count - self.window_size
        drop = -(-excess // self.slide_step) * self.slide_step
        non_system_messages = non_system_messages[drop:]

        # The cut lands at a fixed offset, not a turn boundary, so it can
        # fall between an assistant's tool_calls and the tool messages
        # answering them. Those results are orphans the provider will
        # reject, so drop them here rather than leaving the agent loop to
        # strip them later.
        def _drop_leading_orphans(window: list[Message]) -> list[Message]:
            kept = {tc.id for m in window if m.role == Role.ASSISTANT for tc in m.tool_calls}
            start = 0
            while (
                start < len(window)
                and window[start].role == Role.TOOL
                and window[start].tool_call_id not in kept
            ):
                start += 1
            return window[start:]

        non_system_messages = _drop_leading_orphans(non_system_messages)

        # Re-attach the opening request only when the window kept no user
        # turn at all. In a chat the newest user turns are the live context
        # and the first one is not privileged; it is an agent loop —
        # assistant/tool all the way down — that ends up with no user turn
        # and therefore no statement of the task.
        #
        # Trimming orphans usually frees the slot; if it didn't, drop one
        # more step from the front (the cut the next slide would make
        # anyway) and re-normalise, since that drop can itself orphan the
        # results that followed it.
        if anchor is not None and not any(m.role == Role.USER for m in non_system_messages):
            if len(non_system_messages) >= self.window_size:
                non_system_messages = _drop_leading_orphans(
                    non_system_messages[self.slide_step :]
                )
            non_system_messages.insert(0, anchor)
        from tulip.observability.emit import (  # noqa: PLC0415
            EV_MEMORY_CONVERSATION_PRUNED,
            emit_sync,
        )

        emit_sync(
            EV_MEMORY_CONVERSATION_PRUNED,
            strategy="sliding_window",
            window_size=self.window_size,
            # Count what actually went, not what the raw slice implies:
            # orphan trimming and the re-attached anchor both move this.
            removed_count=before_count - len(non_system_messages),
        )

    result.extend(non_system_messages)
    return result

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async version of apply. Supports async summarization functions.

Default implementation delegates to the synchronous apply(). Override in subclasses that need async operations (e.g., LLM summarization).

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async version of apply. Supports async summarization functions.

    Default implementation delegates to the synchronous apply().
    Override in subclasses that need async operations (e.g., LLM summarization).
    """
    return self.apply(messages)

SummarizingManager

SummarizingManager(threshold: int = 30, keep_recent: int = 10, summarize_fn: Any | None = None, preserve_system: bool = True)

Bases: ConversationManager

Summarizing conversation manager - summarizes older messages.

When the conversation exceeds a threshold, older messages are summarized into a single summary message, preserving recent context.

This manager requires a summarization function to be provided, which can use an LLM to generate summaries.

Parameters:

Name	Type	Description	Default
threshold	int	Number of messages before summarization kicks in	30
keep_recent	int	Number of recent messages to always preserve	10
summarize_fn	Any | None	Async function that summarizes a list of messages	None
preserve_system	bool	Whether to preserve the system message	True

Source code in .sdk/src/tulip/memory/conversation.py

def __init__(
    self,
    threshold: int = 30,
    keep_recent: int = 10,
    summarize_fn: Any | None = None,
    preserve_system: bool = True,
):
    if threshold < 1:
        raise ValueError("threshold must be at least 1")
    if keep_recent < 1:
        raise ValueError("keep_recent must be at least 1")
    if keep_recent >= threshold:
        raise ValueError("keep_recent must be less than threshold")

    self.threshold = threshold
    self.keep_recent = keep_recent
    self.summarize_fn = summarize_fn
    self.preserve_system = preserve_system
    self._summary_cache: dict[int, str] = {}

apply

apply(messages: list[Message]) -> list[Message]

Apply summarization to messages.

If total messages exceed threshold, older messages are summarized. Note: If no summarize_fn is provided, falls back to a simple summary.

Source code in .sdk/src/tulip/memory/conversation.py

def apply(self, messages: list[Message]) -> list[Message]:
    """
    Apply summarization to messages.

    If total messages exceed threshold, older messages are summarized.
    Note: If no summarize_fn is provided, falls back to a simple summary.
    """
    if not messages:
        return []

    from tulip.core.messages import Message as Msg
    from tulip.core.messages import Role

    result: list[Msg] = []
    non_system_messages: list[Msg] = []
    system_message: Msg | None = None

    for msg in messages:
        if msg.role == Role.SYSTEM and self.preserve_system:
            system_message = msg
        else:
            non_system_messages.append(msg)

    # If under threshold, no summarization needed
    if len(non_system_messages) <= self.threshold:
        if system_message:
            result.append(system_message)
        result.extend(non_system_messages)
        return result

    # Split into messages to summarize and recent messages
    to_summarize = non_system_messages[: -self.keep_recent]
    recent = non_system_messages[-self.keep_recent :]

    # Generate summary
    summary_text = self._generate_summary(to_summarize)

    # Build result
    if system_message:
        result.append(system_message)

    # Add summary as a system message
    summary_message = Msg(
        role=Role.SYSTEM,
        content=f"[Summary of previous conversation ({len(to_summarize)} messages)]:\n{summary_text}",
    )
    result.append(summary_message)
    result.extend(recent)

    return result

async_apply async

async_apply(messages: list[Message]) -> list[Message]

Async apply with LLM summarization support.

If summarize_fn is an async function, it will be properly awaited. Falls back to synchronous _generate_summary() otherwise.

Source code in .sdk/src/tulip/memory/conversation.py

async def async_apply(self, messages: list[Message]) -> list[Message]:
    """
    Async apply with LLM summarization support.

    If summarize_fn is an async function, it will be properly awaited.
    Falls back to synchronous _generate_summary() otherwise.
    """
    if not messages:
        return []

    from tulip.core.messages import Message as Msg
    from tulip.core.messages import Role

    result: list[Msg] = []
    non_system_messages: list[Msg] = []
    system_message: Msg | None = None

    for msg in messages:
        if msg.role == Role.SYSTEM and self.preserve_system:
            system_message = msg
        else:
            non_system_messages.append(msg)

    if len(non_system_messages) <= self.threshold:
        if system_message:
            result.append(system_message)
        result.extend(non_system_messages)
        return result

    to_summarize = non_system_messages[: -self.keep_recent]
    recent = non_system_messages[-self.keep_recent :]

    # Use async summarize_fn if available
    import asyncio

    from tulip.observability.emit import (  # noqa: PLC0415
        EV_MEMORY_COMPACTOR_COMPLETED,
        EV_MEMORY_COMPACTOR_TRIGGERED,
        emit,
    )

    await emit(
        EV_MEMORY_COMPACTOR_TRIGGERED,
        strategy="summarizing",
        messages_before=len(non_system_messages),
        threshold=self.threshold,
    )
    import time as _time  # noqa: PLC0415

    _started = _time.perf_counter()

    if self.summarize_fn is not None and asyncio.iscoroutinefunction(self.summarize_fn):
        summary_text = await self.summarize_fn(to_summarize)
    else:
        summary_text = self._generate_summary(to_summarize)

    if system_message:
        result.append(system_message)

    summary_message = Msg(
        role=Role.SYSTEM,
        content=f"[Summary of previous conversation ({len(to_summarize)} messages)]:\n{summary_text}",
    )
    result.append(summary_message)
    result.extend(recent)
    await emit(
        EV_MEMORY_COMPACTOR_COMPLETED,
        strategy="summarizing",
        messages_before=len(non_system_messages),
        messages_after=len(result),
        summarized_count=len(to_summarize),
        duration_ms=(_time.perf_counter() - _started) * 1000,
    )
    return result

Cross-thread store

The cross-thread KV the long-term memory manager writes to. Three implementations ship:

Store	Backing	Reach for it when
InMemoryStore	in-process dict	tests, and anything that can lose memory on exit
HolographicStore	SQLite + FTS5 + holographic reduced representations	you want durable memory with no external infrastructure
PgMemory	Postgres + pgvector, per-tenant row-level security	multi-tenant, where isolation has to hold at the database

The drivers in tulip.memory.backends referenced by Checkpointers implement BaseCheckpointer (per-thread state), not BaseStore. The two layers are separate contracts and a backend for one is not a backend for the other.

BaseStore

Bases: ABC

Abstract base class for store implementations.

Provides common functionality and defines the interface.

capabilities property

capabilities: StoreCapabilities

Return capabilities. Override in subclasses.

put abstractmethod async

put(namespace: tuple[str, ...], key: str, value: Any, metadata: dict[str, Any] | None = None) -> None

Store a value.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def put(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Store a value."""
    ...

get abstractmethod async

get(namespace: tuple[str, ...], key: str) -> Any | None

Retrieve a value.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def get(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> Any | None:
    """Retrieve a value."""
    ...

delete abstractmethod async

delete(namespace: tuple[str, ...], key: str) -> bool

Delete a value.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def delete(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Delete a value."""
    ...

list_keys abstractmethod async

list_keys(namespace: tuple[str, ...], limit: int = 100) -> list[str]

List keys in namespace.

Source code in .sdk/src/tulip/memory/store.py

@abstractmethod
async def list_keys(
    self,
    namespace: tuple[str, ...],
    limit: int = 100,
) -> list[str]:
    """List keys in namespace."""
    ...

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

get_item async

get_item(namespace: tuple[str, ...], key: str) -> StoreItem | None

Get full item with metadata.

Source code in .sdk/src/tulip/memory/store.py

async def get_item(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> StoreItem | None:
    """Get full item with metadata."""
    # Default implementation - subclasses can override
    value = await self.get(namespace, key)
    if value is None:
        return None
    now = datetime.now(UTC)
    return StoreItem(
        namespace=namespace,
        key=key,
        value=value,
        metadata={},
        created_at=now,
        updated_at=now,
    )

search async

search(namespace: tuple[str, ...], query: str | None = None, limit: int = 10) -> list[StoreItem]

Search for items in namespace.

Optional method — backends advertise support via capabilities.search. Subclasses that support search must override this method; the default raises :class:StoreCapabilityError.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query	str | None	Search query (implementation-specific)	None
limit	int	Maximum results	10

Returns:

Type	Description
list[StoreItem]	List of matching items

Raises:

Type	Description
StoreCapabilityError	If the backend does not support search.

Source code in .sdk/src/tulip/memory/store.py

async def search(
    self,
    namespace: tuple[str, ...],
    query: str | None = None,
    limit: int = 10,
) -> list[StoreItem]:
    """
    Search for items in namespace.

    Optional method — backends advertise support via
    ``capabilities.search``. Subclasses that support search must
    override this method; the default raises
    :class:`StoreCapabilityError`.

    Args:
        namespace: Namespace to search in
        query: Search query (implementation-specific)
        limit: Maximum results

    Returns:
        List of matching items

    Raises:
        StoreCapabilityError: If the backend does not support search.
    """
    raise StoreCapabilityError(
        capability="search",
        store_class=type(self).__name__,
        hint=(
            "Choose a backend that advertises capabilities.search "
            "(e.g., InMemoryStore, OpenSearch, pgvector)."
        ),
    )

search_many async

search_many(namespaces: Sequence[tuple[str, ...]], query: str | None = None, limit: int = 10) -> list[list[StoreItem]]

:meth:search over several namespaces at once.

Returns one list per namespace, in the order given: [await search(ns, query, limit) for ns in namespaces], which is exactly what this default does. A backend that can answer them in one round trip overrides it (PgMemory does, in one statement); callers such as :class:~tulip.memory.manager.LLMMemoryManager use it so a recall over every memory type costs one store call instead of one per type.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support search.

Source code in .sdk/src/tulip/memory/store.py

async def search_many(
    self,
    namespaces: Sequence[tuple[str, ...]],
    query: str | None = None,
    limit: int = 10,
) -> list[list[StoreItem]]:
    """:meth:`search` over several namespaces at once.

    Returns one list per namespace, in the order given:
    ``[await search(ns, query, limit) for ns in namespaces]``, which is
    exactly what this default does. A backend that can answer them in one
    round trip overrides it (``PgMemory`` does, in one statement); callers
    such as :class:`~tulip.memory.manager.LLMMemoryManager` use it so a
    recall over every memory type costs one store call instead of one per
    type.

    Raises:
        StoreCapabilityError: If the backend does not support search.
    """
    return [await self.search(ns, query=query, limit=limit) for ns in namespaces]

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

put_batch async

put_batch(items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]) -> None

Store multiple items.

Backends that advertise capabilities.batch_operations should override this method to dispatch atomically. The default falls back to sequential put() calls so callers always get functional batching, just without atomicity guarantees.

Parameters:

Name	Type	Description	Default
items	list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]	List of (namespace, key, value, metadata) tuples	required

Source code in .sdk/src/tulip/memory/store.py

async def put_batch(
    self,
    items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]],
) -> None:
    """
    Store multiple items.

    Backends that advertise ``capabilities.batch_operations`` should
    override this method to dispatch atomically. The default falls
    back to sequential ``put()`` calls so callers always get
    functional batching, just without atomicity guarantees.

    Args:
        items: List of (namespace, key, value, metadata) tuples
    """
    # Sequential fallback — works for every backend, even those that
    # don't advertise batch_operations. Backends that *do* advertise
    # the capability override this method to dispatch in one
    # transaction / round-trip.
    for namespace, key, value, metadata in items:
        await self.put(namespace, key, value, metadata)

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

list_namespaces async

list_namespaces(prefix: tuple[str, ...] | None = None, limit: int = 100) -> list[tuple[str, ...]]

List all namespaces.

Optional method — see capabilities.list_namespaces.

Parameters:

Name	Type	Description	Default
prefix	tuple[str, ...] | None	Optional prefix to filter namespaces	None
limit	int	Maximum namespaces to return	100

Returns:

Type	Description
list[tuple[str, ...]]	List of namespace tuples

Raises:

Type	Description
StoreCapabilityError	If the backend does not support namespace listing.

Source code in .sdk/src/tulip/memory/store.py

async def list_namespaces(
    self,
    prefix: tuple[str, ...] | None = None,
    limit: int = 100,
) -> list[tuple[str, ...]]:
    """
    List all namespaces.

    Optional method — see ``capabilities.list_namespaces``.

    Args:
        prefix: Optional prefix to filter namespaces
        limit: Maximum namespaces to return

    Returns:
        List of namespace tuples

    Raises:
        StoreCapabilityError: If the backend does not support
            namespace listing.
    """
    raise StoreCapabilityError(
        capability="list_namespaces",
        store_class=type(self).__name__,
    )

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

close async

close() -> None

Close any resources.

Source code in .sdk/src/tulip/memory/store.py

async def close(self) -> None:
    """Close any resources."""

InMemoryStore

InMemoryStore()

Bases: BaseStore

In-memory store implementation.

Fast but not persistent - data is lost when process exits. Useful for testing and development.

Source code in .sdk/src/tulip/memory/store.py

def __init__(self) -> None:
    self._data: dict[str, StoreItem] = {}
    self._namespaces: set[tuple[str, ...]] = set()
    self._lock = asyncio.Lock()

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

search_many async

search_many(namespaces: Sequence[tuple[str, ...]], query: str | None = None, limit: int = 10) -> list[list[StoreItem]]

:meth:search over several namespaces at once.

Returns one list per namespace, in the order given: [await search(ns, query, limit) for ns in namespaces], which is exactly what this default does. A backend that can answer them in one round trip overrides it (PgMemory does, in one statement); callers such as :class:~tulip.memory.manager.LLMMemoryManager use it so a recall over every memory type costs one store call instead of one per type.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support search.

Source code in .sdk/src/tulip/memory/store.py

async def search_many(
    self,
    namespaces: Sequence[tuple[str, ...]],
    query: str | None = None,
    limit: int = 10,
) -> list[list[StoreItem]]:
    """:meth:`search` over several namespaces at once.

    Returns one list per namespace, in the order given:
    ``[await search(ns, query, limit) for ns in namespaces]``, which is
    exactly what this default does. A backend that can answer them in one
    round trip overrides it (``PgMemory`` does, in one statement); callers
    such as :class:`~tulip.memory.manager.LLMMemoryManager` use it so a
    recall over every memory type costs one store call instead of one per
    type.

    Raises:
        StoreCapabilityError: If the backend does not support search.
    """
    return [await self.search(ns, query=query, limit=limit) for ns in namespaces]

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

close async

close() -> None

Close any resources.

Source code in .sdk/src/tulip/memory/store.py

async def close(self) -> None:
    """Close any resources."""

NamespacedStore

NamespacedStore(store: BaseStore, namespace: tuple[str, ...])

Store wrapper with a fixed namespace prefix.

Makes it easier to work with a specific scope.

Example

user_store = NamespacedStore(store, ("users", user_id)) await user_store.put("preferences", {"theme": "dark"}) prefs = await user_store.get("preferences")

Source code in .sdk/src/tulip/memory/store.py

def __init__(
    self,
    store: BaseStore,
    namespace: tuple[str, ...],
):
    self._store = store
    self._namespace = namespace

scoped

scoped(*suffix: str) -> NamespacedStore

Create new wrapper with extended namespace.

Source code in .sdk/src/tulip/memory/store.py

def scoped(self, *suffix: str) -> NamespacedStore:
    """Create new wrapper with extended namespace."""
    return NamespacedStore(self._store, (*self._namespace, *suffix))

StoreContext

StoreContext(store: BaseStore, user_id: str | None = None, session_id: str | None = None)

Context object passed to nodes for store access.

Provides convenient methods for common memory operations.

Example

async def my_node(inputs, *, store: StoreContext): # Get user memory user_prefs = await store.get_user_memory("preferences")

# Remember something
await store.remember(
    "discussed_topic",
    {"topic": "python", "timestamp": now}
)

# Search memories
related = await store.search_memories("python")

Source code in .sdk/src/tulip/memory/store.py

def __init__(
    self,
    store: BaseStore,
    user_id: str | None = None,
    session_id: str | None = None,
):
    self._store = store
    self._user_id = user_id
    self._session_id = session_id

store property

store: BaseStore

Access the underlying store.

for_user

for_user(user_id: str | None = None) -> NamespacedStore

Get namespaced store for a user.

If user_id is not provided, uses the user_id from the context.

Source code in .sdk/src/tulip/memory/store.py

def for_user(self, user_id: str | None = None) -> NamespacedStore:
    """Get namespaced store for a user.

    If user_id is not provided, uses the user_id from the context.
    """
    uid = user_id or self._user_id
    if not uid:
        raise ValueError("user_id must be provided or set in context")
    return NamespacedStore(self._store, ("users", uid))

for_session

for_session(session_id: str | None = None) -> NamespacedStore

Get namespaced store for a session.

If session_id is not provided, uses the session_id from the context.

Source code in .sdk/src/tulip/memory/store.py

def for_session(self, session_id: str | None = None) -> NamespacedStore:
    """Get namespaced store for a session.

    If session_id is not provided, uses the session_id from the context.
    """
    sid = session_id or self._session_id
    if not sid:
        raise ValueError("session_id must be provided or set in context")
    return NamespacedStore(self._store, ("sessions", sid))

get_user_memory async

get_user_memory(key: str) -> Any | None

Get a memory for the current user.

Source code in .sdk/src/tulip/memory/store.py

async def get_user_memory(self, key: str) -> Any | None:
    """Get a memory for the current user."""
    if not self._user_id:
        return None
    return await self._store.get(("users", self._user_id, "memories"), key)

remember async

remember(key: str, value: Any, metadata: dict[str, Any] | None = None) -> None

Store a memory for the current user.

Source code in .sdk/src/tulip/memory/store.py

async def remember(
    self,
    key: str,
    value: Any,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Store a memory for the current user."""
    if not self._user_id:
        raise ValueError("No user_id set in store context")
    await self._store.put(
        ("users", self._user_id, "memories"),
        key,
        value,
        metadata,
    )

forget async

forget(key: str) -> bool

Delete a memory for the current user.

Source code in .sdk/src/tulip/memory/store.py

async def forget(self, key: str) -> bool:
    """Delete a memory for the current user."""
    if not self._user_id:
        return False
    return await self._store.delete(("users", self._user_id, "memories"), key)

search_memories async

search_memories(query: str, limit: int = 10) -> list[StoreItem]

Search user memories (full-text).

Source code in .sdk/src/tulip/memory/store.py

async def search_memories(
    self,
    query: str,
    limit: int = 10,
) -> list[StoreItem]:
    """Search user memories (full-text)."""
    if not self._user_id:
        return []
    return await self._store.search(
        ("users", self._user_id, "memories"),
        query,
        limit,
    )

remember_with_embedding async

remember_with_embedding(key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a memory with its embedding for semantic search.

Requires: capabilities.semantic_search = True

Parameters:

Name	Type	Description	Default
key	str	Memory key	required
value	Any	Value to store	required
embedding	list[float]	Vector embedding for semantic search	required
metadata	dict[str, Any] | None	Optional metadata	None

Example

embedding = await embedder.embed("User likes dark theme") await store.remember_with_embedding( "theme_pref", {"theme": "dark"}, embedding=embedding, )

Source code in .sdk/src/tulip/memory/store.py

async def remember_with_embedding(
    self,
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a memory with its embedding for semantic search.

    Requires: capabilities.semantic_search = True

    Args:
        key: Memory key
        value: Value to store
        embedding: Vector embedding for semantic search
        metadata: Optional metadata

    Example:
        embedding = await embedder.embed("User likes dark theme")
        await store.remember_with_embedding(
            "theme_pref",
            {"theme": "dark"},
            embedding=embedding,
        )
    """
    if not self._user_id:
        raise ValueError("No user_id set in store context")
    await self._store.put_with_embedding(
        ("users", self._user_id, "memories"),
        key,
        value,
        embedding,
        metadata,
    )

search_memories_semantic async

search_memories_semantic(query_embedding: list[float], limit: int = 10, threshold: float | None = None) -> list[SemanticSearchResult]

Search user memories by semantic similarity.

Requires: capabilities.semantic_search = True

Parameters:

Name	Type	Description	Default
query_embedding	list[float]	Vector to search with	required
limit	int	Maximum results	10
threshold	float | None	Minimum similarity score (0.0-1.0)	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult sorted by similarity

Example

query_vec = await embedder.embed("user preferences") results = await store.search_memories_semantic( query_embedding=query_vec, limit=5, threshold=0.7, ) for r in results: print(f"{r.item.key}: {r.score:.2f} similar")

Source code in .sdk/src/tulip/memory/store.py

async def search_memories_semantic(
    self,
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
) -> list[SemanticSearchResult]:
    """
    Search user memories by semantic similarity.

    Requires: capabilities.semantic_search = True

    Args:
        query_embedding: Vector to search with
        limit: Maximum results
        threshold: Minimum similarity score (0.0-1.0)

    Returns:
        List of SemanticSearchResult sorted by similarity

    Example:
        query_vec = await embedder.embed("user preferences")
        results = await store.search_memories_semantic(
            query_embedding=query_vec,
            limit=5,
            threshold=0.7,
        )
        for r in results:
            print(f"{r.item.key}: {r.score:.2f} similar")
    """
    if not self._user_id:
        return []
    return await self._store.search_by_embedding(
        ("users", self._user_id, "memories"),
        query_embedding,
        limit,
        threshold,
    )

get_global async

get_global(key: str) -> Any | None

Get a global value.

Source code in .sdk/src/tulip/memory/store.py

async def get_global(self, key: str) -> Any | None:
    """Get a global value."""
    return await self._store.get(("global",), key)

set_global async

set_global(key: str, value: Any, metadata: dict[str, Any] | None = None) -> None

Set a global value.

Source code in .sdk/src/tulip/memory/store.py

async def set_global(
    self,
    key: str,
    value: Any,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Set a global value."""
    await self._store.put(("global",), key, value, metadata)

StoreItem dataclass

StoreItem(namespace: tuple[str, ...], key: str, value: Any, metadata: dict[str, Any], created_at: datetime, updated_at: datetime, version: int = 1)

An item stored in the store.

Attributes:

Name	Type	Description
namespace	tuple[str, ...]	The namespace tuple
key	str	The key within namespace
value	Any	The stored value
metadata	dict[str, Any]	Optional metadata
created_at	datetime	Creation timestamp
updated_at	datetime	Last update timestamp
version	int	Version counter for optimistic locking

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

Source code in .sdk/src/tulip/memory/store.py

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary."""
    return {
        "namespace": list(self.namespace),
        "key": self.key,
        "value": self.value,
        "metadata": self.metadata,
        "created_at": self.created_at.isoformat(),
        "updated_at": self.updated_at.isoformat(),
        "version": self.version,
    }

from_dict classmethod

from_dict(d: dict[str, Any]) -> StoreItem

Create from dictionary.

Source code in .sdk/src/tulip/memory/store.py

@classmethod
def from_dict(cls, d: dict[str, Any]) -> StoreItem:
    """Create from dictionary."""
    return cls(
        namespace=tuple(d["namespace"]),
        key=d["key"],
        value=d["value"],
        metadata=d.get("metadata", {}),
        created_at=datetime.fromisoformat(d["created_at"]),
        updated_at=datetime.fromisoformat(d["updated_at"]),
        version=d.get("version", 1),
    )

StoreCapabilities dataclass

StoreCapabilities(search: bool = False, semantic_search: bool = False, embedding_dimension: int | None = None, ttl: bool = False, list_namespaces: bool = False, batch_operations: bool = False, transactions: bool = False)

Capabilities supported by a store implementation.

Use this to discover what features a store supports.

StoreCapabilityError

StoreCapabilityError(capability: str, store_class: str, hint: str | None = None)

Bases: NotImplementedError

Raised when a store operation is requested that the backend does not support.

Carries structured context (capability, store_class, hint) so callers can branch on the capability or surface a clean message.

Subclasses :class:NotImplementedError so legacy except NotImplementedError blocks continue to catch it — this is strictly a richer payload, not a behaviour change.

Example::

try:
    results = await store.search(("users", uid), query="...")
except StoreCapabilityError as exc:
    log.info(
        "store %s does not support %s; falling back",
        exc.store_class,
        exc.capability,
    )
    results = await store.list_keys(("users", uid))

Source code in .sdk/src/tulip/memory/store.py

def __init__(
    self,
    capability: str,
    store_class: str,
    hint: str | None = None,
) -> None:
    self.capability = capability
    self.store_class = store_class
    self.hint = hint
    msg = f"{store_class} does not support {capability!r}"
    if hint:
        msg = f"{msg}. {hint}"
    super().__init__(msg)

SemanticSearchResult dataclass

SemanticSearchResult(item: StoreItem, score: float, distance: float | None = None)

Result from semantic (vector similarity) search.

Attributes:

Name	Type	Description
item	StoreItem	The matching store item
score	float	Similarity score (0.0 to 1.0, higher is more similar)
distance	float | None	Raw distance metric (interpretation depends on distance type)

Durable store backends

Both shipped in 2.2.0.

HolographicStore

HolographicStore(path: str = ':memory:', *, dim: int = _DEFAULT_DIM)

Bases: BaseStore

Local SQLite + FTS5 + HRR store — zero external infra.

Every value is kept as a row plus an FTS5 index entry and (when numpy is present) an HRR phase vector. search ranks by HRR associative similarity, fused with FTS5 lexical match; without numpy it is FTS5-only. A single connection guarded by a lock keeps it safe across the store's threads.

Source code in .sdk/src/tulip/memory/store_backends/holographic.py

def __init__(self, path: str = ":memory:", *, dim: int = _DEFAULT_DIM) -> None:
    self._dim = dim
    self._lock = threading.Lock()
    self._conn = sqlite3.connect(path, check_same_thread=False)
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute(
        "CREATE TABLE IF NOT EXISTS items ("
        "ns TEXT, key TEXT, value TEXT, metadata TEXT, created_at TEXT, "
        "updated_at TEXT, version INTEGER, content TEXT, hrr BLOB, "
        "PRIMARY KEY (ns, key))"
    )
    self._conn.execute(
        "CREATE VIRTUAL TABLE IF NOT EXISTS items_fts "
        "USING fts5(ns UNINDEXED, key UNINDEXED, content)"
    )
    self._conn.commit()

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

get_item async

get_item(namespace: tuple[str, ...], key: str) -> StoreItem | None

Get full item with metadata.

Source code in .sdk/src/tulip/memory/store.py

async def get_item(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> StoreItem | None:
    """Get full item with metadata."""
    # Default implementation - subclasses can override
    value = await self.get(namespace, key)
    if value is None:
        return None
    now = datetime.now(UTC)
    return StoreItem(
        namespace=namespace,
        key=key,
        value=value,
        metadata={},
        created_at=now,
        updated_at=now,
    )

search_many async

search_many(namespaces: Sequence[tuple[str, ...]], query: str | None = None, limit: int = 10) -> list[list[StoreItem]]

:meth:search over several namespaces at once.

Returns one list per namespace, in the order given: [await search(ns, query, limit) for ns in namespaces], which is exactly what this default does. A backend that can answer them in one round trip overrides it (PgMemory does, in one statement); callers such as :class:~tulip.memory.manager.LLMMemoryManager use it so a recall over every memory type costs one store call instead of one per type.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support search.

Source code in .sdk/src/tulip/memory/store.py

async def search_many(
    self,
    namespaces: Sequence[tuple[str, ...]],
    query: str | None = None,
    limit: int = 10,
) -> list[list[StoreItem]]:
    """:meth:`search` over several namespaces at once.

    Returns one list per namespace, in the order given:
    ``[await search(ns, query, limit) for ns in namespaces]``, which is
    exactly what this default does. A backend that can answer them in one
    round trip overrides it (``PgMemory`` does, in one statement); callers
    such as :class:`~tulip.memory.manager.LLMMemoryManager` use it so a
    recall over every memory type costs one store call instead of one per
    type.

    Raises:
        StoreCapabilityError: If the backend does not support search.
    """
    return [await self.search(ns, query=query, limit=limit) for ns in namespaces]

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

put_batch async

put_batch(items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]) -> None

Store multiple items.

Backends that advertise capabilities.batch_operations should override this method to dispatch atomically. The default falls back to sequential put() calls so callers always get functional batching, just without atomicity guarantees.

Parameters:

Name	Type	Description	Default
items	list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]	List of (namespace, key, value, metadata) tuples	required

Source code in .sdk/src/tulip/memory/store.py

async def put_batch(
    self,
    items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]],
) -> None:
    """
    Store multiple items.

    Backends that advertise ``capabilities.batch_operations`` should
    override this method to dispatch atomically. The default falls
    back to sequential ``put()`` calls so callers always get
    functional batching, just without atomicity guarantees.

    Args:
        items: List of (namespace, key, value, metadata) tuples
    """
    # Sequential fallback — works for every backend, even those that
    # don't advertise batch_operations. Backends that *do* advertise
    # the capability override this method to dispatch in one
    # transaction / round-trip.
    for namespace, key, value, metadata in items:
        await self.put(namespace, key, value, metadata)

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

list_namespaces async

list_namespaces(prefix: tuple[str, ...] | None = None, limit: int = 100) -> list[tuple[str, ...]]

List all namespaces.

Optional method — see capabilities.list_namespaces.

Parameters:

Name	Type	Description	Default
prefix	tuple[str, ...] | None	Optional prefix to filter namespaces	None
limit	int	Maximum namespaces to return	100

Returns:

Type	Description
list[tuple[str, ...]]	List of namespace tuples

Raises:

Type	Description
StoreCapabilityError	If the backend does not support namespace listing.

Source code in .sdk/src/tulip/memory/store.py

async def list_namespaces(
    self,
    prefix: tuple[str, ...] | None = None,
    limit: int = 100,
) -> list[tuple[str, ...]]:
    """
    List all namespaces.

    Optional method — see ``capabilities.list_namespaces``.

    Args:
        prefix: Optional prefix to filter namespaces
        limit: Maximum namespaces to return

    Returns:
        List of namespace tuples

    Raises:
        StoreCapabilityError: If the backend does not support
            namespace listing.
    """
    raise StoreCapabilityError(
        capability="list_namespaces",
        store_class=type(self).__name__,
    )

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

PgMemory

PgMemory(dsn: str, *, table: str = 'tulip_memories', dim: int = _DEFAULT_PG_DIM, embedder: BaseEmbedding | None = None, encoding_cache: int = _DEFAULT_ENCODING_CACHE)

Bases: BaseStore

Postgres + pgvector memory store with per-tenant Row-Level Security.

namespace[0] is the tenant (the isolation boundary); the full namespace tuple scopes within a tenant (e.g. (tenant, "user", user_id)). The schema, RLS policy, and HNSW index are created on first use, so it works against a bare Postgres locally and a managed Aurora cluster identically.

Recall quality depends on the vector. Pass an embedder (any :class:~tulip.rag.embeddings.base.BaseEmbedding, e.g. brokered OpenAI text-embedding-3-small) for true semantic recall — the enterprise default. Without one it falls back to the HRR [cos φ, sin φ] encoding, which is lexical/associative (matches shared or hashed tokens), not trained semantics — fine offline, but it will not match paraphrases the way an embedding model does. Choose the embedder to match the value of recall.

Width and the ANN index. Without an embedder the pgvector column is 2·dim wide (the [cos φ, sin φ] pair), and pgvector will not build an HNSW index over more than :data:PGVECTOR_ANN_MAX_DIM dimensions. dim therefore defaults to 512 (a 1024-wide column) and a larger explicit dim is rejected at construction rather than at CREATE INDEX. An embedder wider than the limit is allowed — its width is the model's, not ours — but the table is then created without an ANN index and the constructor emits a RuntimeWarning saying so.

numpy. The HRR encoding needs numpy (the [pgvector] extra installs it). Without numpy and without an embedder nothing is wrong at write time, but rows are stored with no vector and recall degrades to a substring match; the constructor warns (RuntimeWarning) and the first degraded search logs a warning, so this never happens quietly.

Encoding cache. Encoding a text costs a pure-Python HRR pass (or an embedding call), and one recall often encodes the same words several times. Each store keeps the last encoding_cache encodings (0 turns the cache off), keyed by (tenant, text): an entry is only ever reused within the tenant that computed it, so the cache is not a surface two tenants share.

Several namespaces at once. :meth:search_many ranks several namespaces of ONE tenant in one transaction and one statement, encoding the query once. Namespaces of different tenants in one call are refused.

Source code in .sdk/src/tulip/memory/store_backends/postgresql.py

def __init__(
    self,
    dsn: str,
    *,
    table: str = "tulip_memories",
    dim: int = _DEFAULT_PG_DIM,
    embedder: BaseEmbedding | None = None,
    encoding_cache: int = _DEFAULT_ENCODING_CACHE,
) -> None:
    if dim < 1:
        raise ValueError(f"dim must be >= 1, got {dim}")
    if encoding_cache < 0:
        raise ValueError(f"encoding_cache must be >= 0, got {encoding_cache}")
    # asyncpg is an optional dependency imported lazily inside the pool
    # builder, so a missing package used to surface as a bare
    # ModuleNotFoundError from deep inside a coroutine on first *use* —
    # long after the mistake, and far from anything that explains it. Same
    # reasoning as the dim check below: a caller mistake belongs at
    # construction, in words that say what to install.
    if importlib.util.find_spec("asyncpg") is None:
        raise ImportError(
            "PgMemory needs the asyncpg driver, which ships as an optional "
            "dependency. Install it with:\n\n"
            "    pip install 'tulip-agents[pgvector]'\n\n"
            "(or `pip install asyncpg numpy` if you manage dependencies yourself)."
        )
    self._dsn = dsn
    self._table = _validate_ident(table, "table")
    self._dim = dim
    self._embedder = embedder
    # A real embedder fixes the column width; HRR [cos, sin] doubles `dim`.
    self._vdim = embedder.dimension if embedder is not None else 2 * dim
    # Fail here, not thousands of lines into a run with a raw driver error:
    # a `dim` we control that cannot carry an ANN index is a caller mistake.
    if embedder is None and self._vdim > PGVECTOR_ANN_MAX_DIM:
        raise ValueError(
            f"PgMemory(dim={dim}) is not indexable: the HRR encoding stores "
            f"[cos φ, sin φ], so the pgvector column would be 2·{dim} = "
            f"{self._vdim} dimensions wide, over pgvector's "
            f"{PGVECTOR_ANN_MAX_DIM}-dimension limit for an HNSW index "
            f"(CREATE INDEX would raise ProgramLimitExceededError). Pass "
            f"dim <= {PGVECTOR_ANN_MAX_DIM // 2} (the default is "
            f"{_DEFAULT_PG_DIM}), or pass an embedder."
        )
    # An embedder's width is *not* ours to choose (text-embedding-3-large is
    # 3072). Rather than reject a legitimate model we keep the column and skip
    # the index — but say so loudly, because a silent sequential scan is
    # exactly the failure this store had.
    self._ann = self._vdim <= PGVECTOR_ANN_MAX_DIM
    if not self._ann:
        warnings.warn(
            f"PgMemory: embedder dimension {self._vdim} exceeds pgvector's "
            f"{PGVECTOR_ANN_MAX_DIM}-dimension limit for an HNSW index, so "
            f"table {self._table!r} will be created WITHOUT an ANN index — "
            f"every search is a sequential scan. Use an embedder with "
            f"<= {PGVECTOR_ANN_MAX_DIM} dimensions (or a model that supports "
            f"shortening its output) for indexed recall.",
            RuntimeWarning,
            stacklevel=2,
        )
    # Without numpy (and without an embedder) there is no vector to store or
    # to rank by: recall silently became a substring match. Say so now.
    self._substring_only = embedder is None and _numpy() is None
    self._substring_warned = False
    if self._substring_only:
        warnings.warn(_SUBSTRING_ONLY, RuntimeWarning, stacklevel=2)
    self._encoding_cache = encoding_cache
    self._encodings: OrderedDict[tuple[str, str], str | None] = OrderedDict()
    self._pool: Pool | None = None
    # Serialises first use so two concurrent calls cannot each build a pool
    # (and each run schema creation).
    self._pool_lock = asyncio.Lock()

search_many async

search_many(namespaces: Sequence[tuple[str, ...]], query: str | None = None, limit: int = 10) -> list[list[StoreItem]]

:meth:search over several namespaces of ONE tenant, in one statement.

One transaction (the tenant pin and one SELECT), the query encoded once, and each namespace ranked on its own exactly as :meth:search would rank it (LATERAL per namespace), so the result is [search(ns, query, limit) for ns in namespaces] at the cost of one round trip.

Raises:

Type	Description
ValueError	The namespaces belong to more than one tenant (namespace[0]). A call never spans tenants.

Source code in .sdk/src/tulip/memory/store_backends/postgresql.py

async def search_many(
    self,
    namespaces: Sequence[tuple[str, ...]],
    query: str | None = None,
    limit: int = 10,
) -> list[list[StoreItem]]:
    """:meth:`search` over several namespaces of ONE tenant, in one statement.

    One transaction (the tenant pin and one ``SELECT``), the query encoded
    once, and each namespace ranked on its own exactly as :meth:`search`
    would rank it (``LATERAL`` per namespace), so the result is
    ``[search(ns, query, limit) for ns in namespaces]`` at the cost of one
    round trip.

    Raises:
        ValueError: The namespaces belong to more than one tenant
            (``namespace[0]``). A call never spans tenants.
    """
    if not namespaces:
        return []
    tenants = {self._tenant_of(ns) for ns in namespaces}
    if len(tenants) > 1:
        raise ValueError(
            f"search_many spans tenants {sorted(tenants)}: every namespace of one "
            "call must share namespace[0], the tenant"
        )
    tenant = tenants.pop()
    names = [_NS_SEP.join(ns) for ns in namespaces]
    cols = "key, value, metadata, created_at, updated_at, version"
    where = f"FROM {self._table} WHERE tenant=$1 AND ns=n.ns"
    args: list[Any] = [tenant, names]
    if not query:
        inner = f"SELECT {cols}, updated_at AS _rank {where} ORDER BY updated_at DESC"
        order = "n.idx, m._rank DESC"
    else:
        qvec = await self._vector(tenant, query)
        if qvec is None:
            self._warn_substring()
            inner = (
                f"SELECT {cols}, updated_at AS _rank {where} "
                "AND content ILIKE '%'||$3||'%' ORDER BY updated_at DESC"
            )
            order = "n.idx, m._rank DESC"
            args.append(query)
        else:
            inner = (
                f"SELECT {cols}, embedding <=> $3::vector AS _rank {where} "
                "AND embedding IS NOT NULL ORDER BY embedding <=> $3::vector"
            )
            order = "n.idx, m._rank"
            args.append(qvec)
    args.append(limit)
    sql = (
        f"SELECT n.idx, m.* FROM unnest($2::text[]) WITH ORDINALITY AS n(ns, idx) "
        f"CROSS JOIN LATERAL ({inner} LIMIT ${len(args)}) m ORDER BY {order}"
    )
    pool = await self._get_pool()
    async with pool.acquire() as conn, conn.transaction():
        await self._scoped(conn, tenant)
        rows = await conn.fetch(sql, *args)
    out: list[list[StoreItem]] = [[] for _ in namespaces]
    for r in rows:
        i = int(r["idx"]) - 1
        out[i].append(self._row_to_item(tuple(namespaces[i]), r))
    return out

exists async

exists(namespace: tuple[str, ...], key: str) -> bool

Check if a key exists.

Source code in .sdk/src/tulip/memory/store.py

async def exists(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> bool:
    """Check if a key exists."""
    value = await self.get(namespace, key)
    return value is not None

get_item async

get_item(namespace: tuple[str, ...], key: str) -> StoreItem | None

Get full item with metadata.

Source code in .sdk/src/tulip/memory/store.py

async def get_item(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> StoreItem | None:
    """Get full item with metadata."""
    # Default implementation - subclasses can override
    value = await self.get(namespace, key)
    if value is None:
        return None
    now = datetime.now(UTC)
    return StoreItem(
        namespace=namespace,
        key=key,
        value=value,
        metadata={},
        created_at=now,
        updated_at=now,
    )

put_with_embedding async

put_with_embedding(namespace: tuple[str, ...], key: str, value: Any, embedding: list[float], metadata: dict[str, Any] | None = None) -> None

Store a value with its embedding vector for semantic search.

Optional method — backends advertise support via capabilities.semantic_search. Subclasses that support semantic search must override this method.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required
value	Any	Value to store (must be JSON-serializable)	required
embedding	list[float]	Vector embedding (e.g., from OpenAI, Cohere, etc.)	required
metadata	dict[str, Any] | None	Optional metadata for filtering	None

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Example

Get embedding from your provider

embedding = await embedder.embed("User prefers dark theme")

await store.put_with_embedding( ("users", user_id, "memories"), "theme_preference", {"theme": "dark", "reason": "easier on eyes"}, embedding=embedding, metadata={"category": "preferences"} )

Source code in .sdk/src/tulip/memory/store.py

async def put_with_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
    value: Any,
    embedding: list[float],
    metadata: dict[str, Any] | None = None,
) -> None:
    """
    Store a value with its embedding vector for semantic search.

    Optional method — backends advertise support via
    ``capabilities.semantic_search``. Subclasses that support
    semantic search must override this method.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace
        value: Value to store (must be JSON-serializable)
        embedding: Vector embedding (e.g., from OpenAI, Cohere, etc.)
        metadata: Optional metadata for filtering

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.

    Example:
        # Get embedding from your provider
        embedding = await embedder.embed("User prefers dark theme")

        await store.put_with_embedding(
            ("users", user_id, "memories"),
            "theme_preference",
            {"theme": "dark", "reason": "easier on eyes"},
            embedding=embedding,
            metadata={"category": "preferences"}
        )
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Use put() for plain storage, or pick a vector-capable backend.",
    )

search_by_embedding async

search_by_embedding(namespace: tuple[str, ...], query_embedding: list[float], limit: int = 10, threshold: float | None = None, metadata_filter: dict[str, Any] | None = None) -> list[SemanticSearchResult]

Search for similar items using vector similarity.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to search in	required
query_embedding	list[float]	Vector to compare against stored embeddings	required
limit	int	Maximum results to return	10
threshold	float | None	Minimum similarity score (0.0-1.0), or None for no threshold	None
metadata_filter	dict[str, Any] | None	Optional metadata constraints	None

Returns:

Type	Description
list[SemanticSearchResult]	List of SemanticSearchResult, sorted by similarity (highest first)

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def search_by_embedding(
    self,
    namespace: tuple[str, ...],
    query_embedding: list[float],
    limit: int = 10,
    threshold: float | None = None,
    metadata_filter: dict[str, Any] | None = None,
) -> list[SemanticSearchResult]:
    """
    Search for similar items using vector similarity.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Namespace to search in
        query_embedding: Vector to compare against stored embeddings
        limit: Maximum results to return
        threshold: Minimum similarity score (0.0-1.0), or None for no threshold
        metadata_filter: Optional metadata constraints

    Returns:
        List of SemanticSearchResult, sorted by similarity (highest first)

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
        hint="Pick a vector-capable backend (e.g., pgvector, Qdrant).",
    )

get_embedding async

get_embedding(namespace: tuple[str, ...], key: str) -> list[float] | None

Get the embedding vector for a stored item.

Optional method — see capabilities.semantic_search.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Hierarchical namespace tuple	required
key	str	Key within the namespace	required

Returns:

Type	Description
list[float] | None	Embedding vector or None if not found / no embedding stored.

Raises:

Type	Description
StoreCapabilityError	If the backend does not support semantic search.

Source code in .sdk/src/tulip/memory/store.py

async def get_embedding(
    self,
    namespace: tuple[str, ...],
    key: str,
) -> list[float] | None:
    """
    Get the embedding vector for a stored item.

    Optional method — see ``capabilities.semantic_search``.

    Args:
        namespace: Hierarchical namespace tuple
        key: Key within the namespace

    Returns:
        Embedding vector or None if not found / no embedding stored.

    Raises:
        StoreCapabilityError: If the backend does not support
            semantic search.
    """
    raise StoreCapabilityError(
        capability="semantic_search",
        store_class=type(self).__name__,
    )

put_batch async

put_batch(items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]) -> None

Store multiple items.

Backends that advertise capabilities.batch_operations should override this method to dispatch atomically. The default falls back to sequential put() calls so callers always get functional batching, just without atomicity guarantees.

Parameters:

Name	Type	Description	Default
items	list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]]	List of (namespace, key, value, metadata) tuples	required

Source code in .sdk/src/tulip/memory/store.py

async def put_batch(
    self,
    items: list[tuple[tuple[str, ...], str, Any, dict[str, Any] | None]],
) -> None:
    """
    Store multiple items.

    Backends that advertise ``capabilities.batch_operations`` should
    override this method to dispatch atomically. The default falls
    back to sequential ``put()`` calls so callers always get
    functional batching, just without atomicity guarantees.

    Args:
        items: List of (namespace, key, value, metadata) tuples
    """
    # Sequential fallback — works for every backend, even those that
    # don't advertise batch_operations. Backends that *do* advertise
    # the capability override this method to dispatch in one
    # transaction / round-trip.
    for namespace, key, value, metadata in items:
        await self.put(namespace, key, value, metadata)

get_batch async

get_batch(keys: list[tuple[tuple[str, ...], str]]) -> dict[tuple[tuple[str, ...], str], Any]

Retrieve multiple items.

Parameters:

Name	Type	Description	Default
keys	list[tuple[tuple[str, ...], str]]	List of (namespace, key) tuples	required

Returns:

Type	Description
dict[tuple[tuple[str, ...], str], Any]	Dict mapping (namespace, key) to value

Source code in .sdk/src/tulip/memory/store.py

async def get_batch(
    self,
    keys: list[tuple[tuple[str, ...], str]],
) -> dict[tuple[tuple[str, ...], str], Any]:
    """
    Retrieve multiple items.

    Args:
        keys: List of (namespace, key) tuples

    Returns:
        Dict mapping (namespace, key) to value
    """
    results = {}
    for namespace, key in keys:
        value = await self.get(namespace, key)
        if value is not None:
            results[(namespace, key)] = value
    return results

list_namespaces async

list_namespaces(prefix: tuple[str, ...] | None = None, limit: int = 100) -> list[tuple[str, ...]]

List all namespaces.

Optional method — see capabilities.list_namespaces.

Parameters:

Name	Type	Description	Default
prefix	tuple[str, ...] | None	Optional prefix to filter namespaces	None
limit	int	Maximum namespaces to return	100

Returns:

Type	Description
list[tuple[str, ...]]	List of namespace tuples

Raises:

Type	Description
StoreCapabilityError	If the backend does not support namespace listing.

Source code in .sdk/src/tulip/memory/store.py

async def list_namespaces(
    self,
    prefix: tuple[str, ...] | None = None,
    limit: int = 100,
) -> list[tuple[str, ...]]:
    """
    List all namespaces.

    Optional method — see ``capabilities.list_namespaces``.

    Args:
        prefix: Optional prefix to filter namespaces
        limit: Maximum namespaces to return

    Returns:
        List of namespace tuples

    Raises:
        StoreCapabilityError: If the backend does not support
            namespace listing.
    """
    raise StoreCapabilityError(
        capability="list_namespaces",
        store_class=type(self).__name__,
    )

clear_namespace async

clear_namespace(namespace: tuple[str, ...]) -> int

Delete all items in a namespace.

Parameters:

Name	Type	Description	Default
namespace	tuple[str, ...]	Namespace to clear	required

Returns:

Type	Description
int	Number of items deleted

Source code in .sdk/src/tulip/memory/store.py

async def clear_namespace(
    self,
    namespace: tuple[str, ...],
) -> int:
    """
    Delete all items in a namespace.

    Args:
        namespace: Namespace to clear

    Returns:
        Number of items deleted
    """
    keys = await self.list_keys(namespace, limit=10000)
    count = 0
    for key in keys:
        if await self.delete(namespace, key):
            count += 1
    return count

Long-term memory manager

LLMMemoryManager is the default — it uses an auxiliary model to extract and categorize memories at session end and retrieves the top-k relevant entries at session start. NoopMemoryManager is the pass-through used in tests. Mem0MemoryManager (behind the mem0 extra, pip install "tulip-agents[mem0]") hands the finished conversation to the open-source mem0 library, which does its own extraction and keeps the memories in mem0's own vector store rather than a BaseStore.

BaseMemoryManager

Bases: ABC

Abstract base for long-term memory managers.

Subclasses implement :meth:extract to decide what is worth remembering; the base class handles retrieval, injection, and the session-start / session-end lifecycle hooks that the agent calls automatically.

Two concrete implementations are provided:

• :class:NoopMemoryManager — no-op, useful for testing.
• :class:LLMMemoryManager — persists to any :class:BaseStore backend; accepts an optional LLM-backed extraction function.

pending_extractions property

pending_extractions: int

Background extractions scheduled or running, not yet finished.

extract abstractmethod async

extract(messages: list[Message]) -> list[Memory]

Extract durable memories from a finished conversation.

Parameters:

Name	Type	Description	Default
messages	list[Message]	The full message history for the completed session.	required

Returns:

Type	Description
list[Memory]	List of :class:Memory objects to persist. May be empty.

Source code in .sdk/src/tulip/memory/manager.py

@abstractmethod
async def extract(self, messages: list[Message]) -> list[Memory]:
    """Extract durable memories from a finished conversation.

    Args:
        messages: The full message history for the completed session.

    Returns:
        List of :class:`Memory` objects to persist.  May be empty.
    """
    ...

retrieve abstractmethod async

retrieve(limit: int = 20) -> list[Memory]

Retrieve stored memories for injection at session start.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of memories to return.	20

Returns:

Type	Description
list[Memory]	List of :class:Memory objects, most recently updated first.

Source code in .sdk/src/tulip/memory/manager.py

@abstractmethod
async def retrieve(self, limit: int = 20) -> list[Memory]:
    """Retrieve stored memories for injection at session start.

    Args:
        limit: Maximum number of memories to return.

    Returns:
        List of :class:`Memory` objects, most recently updated first.
    """
    ...

save abstractmethod async

save(memories: list[Memory]) -> None

Persist a list of memories to the backing store.

Implementations should upsert by key — re-extracting the same fact updates the record rather than creating a duplicate.

Parameters:

Name	Type	Description	Default
memories	list[Memory]	Memories to persist.	required

Source code in .sdk/src/tulip/memory/manager.py

@abstractmethod
async def save(self, memories: list[Memory]) -> None:
    """Persist a list of memories to the backing store.

    Implementations should upsert by key — re-extracting the same
    fact updates the record rather than creating a duplicate.

    Args:
        memories: Memories to persist.
    """
    ...

retrieve_relevant async

retrieve_relevant(query: str | None, limit: int | None = None) -> list[Memory]

Retrieve the memories most relevant to query.

Called by :meth:on_session_start with the text of the current user turn. The default ignores query and delegates to :meth:retrieve, so subclasses written against the older retrieve(limit) contract keep working unchanged; managers whose backend can rank by relevance override this.

Parameters:

Name	Type	Description	Default
query	str | None	Text to rank memories against, usually the latest user message. None means "no query" (recency order).	required
limit	int | None	Maximum number of memories to return. None uses the manager's default.	None

Source code in .sdk/src/tulip/memory/manager.py

async def retrieve_relevant(self, query: str | None, limit: int | None = None) -> list[Memory]:
    """Retrieve the memories most relevant to ``query``.

    Called by :meth:`on_session_start` with the text of the current user
    turn. The default ignores ``query`` and delegates to :meth:`retrieve`,
    so subclasses written against the older ``retrieve(limit)`` contract
    keep working unchanged; managers whose backend can rank by relevance
    override this.

    Args:
        query: Text to rank memories against, usually the latest user
            message. ``None`` means "no query" (recency order).
        limit: Maximum number of memories to return. ``None`` uses the
            manager's default.
    """
    if limit is None:
        return await self.retrieve()
    return await self.retrieve(limit)

on_session_start async

on_session_start(state: AgentState) -> AgentState

Retrieve memories and inject them into the agent state.

Called by the agent runtime at the start of every invocation, after on_before_invocation hooks but before the first model call. The latest user message is used as the retrieval query, and the formatted block replaces any block injected by an earlier turn of the same (checkpointed) thread instead of being added next to it.

Parameters:

Name	Type	Description	Default
state	AgentState	Current agent state (just-created or loaded from checkpointer).	required

Returns:

Type	Description
AgentState	Possibly-modified state with memory context injected.

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_start(self, state: AgentState) -> AgentState:
    """Retrieve memories and inject them into the agent state.

    Called by the agent runtime at the start of every invocation,
    after ``on_before_invocation`` hooks but before the first model
    call. The latest user message is used as the retrieval query, and
    the formatted block *replaces* any block injected by an earlier turn
    of the same (checkpointed) thread instead of being added next to it.

    Args:
        state: Current agent state (just-created or loaded from
            checkpointer).

    Returns:
        Possibly-modified state with memory context injected.
    """
    memories = await self.retrieve_relevant(_latest_user_text(state))
    if not memories:
        # Still drop a block left over from an earlier turn: if the store
        # no longer holds those memories (deleted, erased, expired) they
        # must not keep reaching the model from the checkpoint.
        return _strip_memory_blocks(state)

    from tulip.observability.emit import emit  # noqa: PLC0415

    injected_state = _inject_memories_into_state(state, memories)

    await emit(
        "memory.manager.injected",
        memory_count=len(memories),
        types=[m.type.value for m in memories],
    )

    return injected_state

on_session_end async

on_session_end(state: AgentState) -> None

Extract memories from the finished session and save them.

Called by the agent runtime in the finally block of every invocation, after on_after_invocation hooks but before the final checkpoint.

With extract_mode="inline" (the default) extraction finishes before this returns, so its latency is part of the run. With extract_mode="background" it is scheduled as a tracked task and this returns at once: the run ends — and a streaming consumer sees its last event — without waiting for the extractor. Background jobs of one namespace run in submission order (two turns of one user never race their writes), at most :attr:max_concurrent_extractions run at once, and :meth:drain waits for all of them (call it, or Agent.drain_memory(), at graceful shutdown and in tests).

The injected memory block is removed before extraction, so recalled memories are not fed back to the extractor as if the user had just said them. A failing extractor or store is logged and reported as a memory.manager.extract_failed event rather than raised, in either mode: an auxiliary memory write must not cost the conversation its turn (inline), and must never surface in a run that already finished (background).

Parameters:

Name	Type	Description	Default
state	AgentState	Final agent state with the complete message history.	required

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_end(self, state: AgentState) -> None:
    """Extract memories from the finished session and save them.

    Called by the agent runtime in the ``finally`` block of every
    invocation, after ``on_after_invocation`` hooks but before the
    final checkpoint.

    With ``extract_mode="inline"`` (the default) extraction finishes before
    this returns, so its latency is part of the run. With
    ``extract_mode="background"`` it is scheduled as a tracked task and
    this returns at once: the run ends — and a streaming consumer sees its
    last event — without waiting for the extractor. Background jobs of one
    namespace run in submission order (two turns of one user never race
    their writes), at most :attr:`max_concurrent_extractions` run at once,
    and :meth:`drain` waits for all of them (call it, or
    ``Agent.drain_memory()``, at graceful shutdown and in tests).

    The injected memory block is removed before extraction, so recalled
    memories are not fed back to the extractor as if the user had just
    said them. A failing extractor or store is logged and reported as a
    ``memory.manager.extract_failed`` event rather than raised, in either
    mode: an auxiliary memory write must not cost the conversation its
    turn (inline), and must never surface in a run that already finished
    (background).

    Args:
        state: Final agent state with the complete message history.
    """
    messages = list(_strip_memory_blocks(state).messages)
    if self.extract_mode == "background":
        self._background().submit(
            self._extraction_order_key(), lambda: self._extract_and_save(messages)
        )
        return
    await self._extract_and_save(messages)

drain async

drain() -> None

Wait for every scheduled background extraction to finish.

Jobs submitted while draining are waited for too. Safe to call in "inline" mode (returns at once). Errors inside jobs were already logged and emitted; they are not raised here. Bound the wait with asyncio.timeout(...): cancelling drain leaves the jobs running.

Source code in .sdk/src/tulip/memory/manager.py

async def drain(self) -> None:
    """Wait for every scheduled background extraction to finish.

    Jobs submitted while draining are waited for too. Safe to call in
    ``"inline"`` mode (returns at once). Errors inside jobs were already
    logged and emitted; they are not raised here. Bound the wait with
    ``asyncio.timeout(...)``: cancelling ``drain`` leaves the jobs running.
    """
    bg: _BackgroundExtractions | None = self.__dict__.get("_bg_extractions")
    if bg is not None:
        await bg.drain()

LLMMemoryManager

LLMMemoryManager(store: BaseStore, *, extract_fn: ExtractFn | None = None, namespace_prefix: tuple[str, ...] = ('tulip_memory',), max_memories: int = 50, retrieve_limit: int = 20, extract_mode: ExtractMode = 'inline', max_concurrent_extractions: int = 4, namespace_resolver: NamespaceResolver | None = None)

Bases: BaseMemoryManager

Memory manager backed by any :class:~tulip.memory.store.BaseStore.

Extraction uses either a caller-supplied async function (extract_fn) or a built-in heuristic that pattern-matches common conversational signals (corrections, confirmations, role disclosures). Supply extract_fn for production use; the heuristic is adequate for demos and tests.

Parameters:

Name	Type	Description	Default
store	BaseStore	Any :class:~tulip.memory.store.BaseStore implementation. InMemoryStore works for development; use RedisBackend, PostgreSQLBackend, etc. for production.	required
extract_fn	ExtractFn | None	Optional async callable (messages: list[Message]) -> list[Memory]. When provided, the heuristic is bypassed.	None
namespace_prefix	tuple[str, ...]	Store namespace prefix. Scope per user or tenant by passing ("tenants", tenant_id). Default ("tulip_memory",).	('tulip_memory',)
max_memories	int	Hard cap on the total number of memories kept per type. Oldest entries are pruned when the limit is reached.	50
retrieve_limit	int	Maximum memories returned by :meth:retrieve.	20
extract_mode	ExtractMode	"inline" (default) runs extraction before the run ends; "background" schedules it after the turn's final event so a chat never waits for it — see :meth:BaseMemoryManager.on_session_end. Background mode needs a long-lived event loop (a server); call :meth:drain before it stops. Agent.run_sync drains for you.	'inline'
max_concurrent_extractions	int	Background extractions allowed to run at once. Jobs of one namespace always run one at a time, in order.	4
namespace_resolver	NamespaceResolver | None	(run: RunInfo) -> prefix choosing the namespace prefix PER RUN, from run.metadata (what agent.run(..., metadata=) passed). Lets one manager serve every user: its max_concurrent_extractions bound and :meth:drain then cover all of them. None from the resolver means NO memory for that run (e.g. an anonymous visitor): it neither reads nor writes memories — it never falls back to the shared namespace_prefix, which would pool every such run together. Return namespace_prefix explicitly to opt a run into it. A resolver that raises is treated the same way. Return a :class:MemoryScope to also recall read-only namespaces of the same tenant in the same call.	None

Example::

from tulip.memory.manager import LLMMemoryManager
from tulip.memory.store import InMemoryStore

manager = LLMMemoryManager(
    store=InMemoryStore(),
    extract_fn=my_llm_extractor,
    namespace_prefix=("users", user_id),
)

agent = Agent(model="anthropic:claude-sonnet-4-6", memory_manager=manager)

Source code in .sdk/src/tulip/memory/manager.py

def __init__(
    self,
    store: BaseStore,
    *,
    extract_fn: ExtractFn | None = None,
    namespace_prefix: tuple[str, ...] = ("tulip_memory",),
    max_memories: int = 50,
    retrieve_limit: int = 20,
    extract_mode: ExtractMode = "inline",
    max_concurrent_extractions: int = 4,
    namespace_resolver: NamespaceResolver | None = None,
) -> None:
    if extract_mode not in ("inline", "background"):
        raise ValueError(f"extract_mode must be 'inline' or 'background', got {extract_mode!r}")
    if max_concurrent_extractions < 1:
        raise ValueError("max_concurrent_extractions must be at least 1")
    self.extract_mode = extract_mode
    self.max_concurrent_extractions = max_concurrent_extractions
    self.store = store
    self.extract_fn = extract_fn
    self.namespace_prefix = namespace_prefix
    self.max_memories = max_memories
    self.retrieve_limit = retrieve_limit
    self.namespace_resolver = namespace_resolver

active_namespace property

active_namespace: tuple[str, ...]

The prefix the current call reads and writes under: the one a :meth:scoped block (or the run's resolver) set, else namespace_prefix.

active_recall property

active_recall: tuple[tuple[str, ...], ...]

Every prefix a recall in the current scope reads: the active namespace first, then the scope's read-only recall prefixes.

pending_extractions property

pending_extractions: int

Background extractions scheduled or running, not yet finished.

scoped

scoped(namespace_prefix: tuple[str, ...], *, recall: Sequence[tuple[str, ...]] = ()) -> Iterator[None]

Run :meth:retrieve / :meth:save / :meth:extract calls — and background jobs scheduled — inside the block under namespace_prefix.

recall adds read-only prefixes that :meth:retrieve also reads; writes still go to namespace_prefix alone. A recall prefix whose first element differs from namespace_prefix[0] (another tenant) is dropped with a warning.

Context-local: concurrent tasks each keep their own scope. The agent runtime does this for you when namespace_resolver is set.

Source code in .sdk/src/tulip/memory/manager.py

@contextmanager
def scoped(
    self,
    namespace_prefix: tuple[str, ...],
    *,
    recall: Sequence[tuple[str, ...]] = (),
) -> Iterator[None]:
    """Run :meth:`retrieve` / :meth:`save` / :meth:`extract` calls — and
    background jobs scheduled — inside the block under
    ``namespace_prefix``.

    ``recall`` adds read-only prefixes that :meth:`retrieve` also reads;
    writes still go to ``namespace_prefix`` alone. A recall prefix whose
    first element differs from ``namespace_prefix[0]`` (another tenant) is
    dropped with a warning.

    Context-local: concurrent tasks each keep their own scope. The agent
    runtime does this for you when ``namespace_resolver`` is set.
    """
    prefix = tuple(namespace_prefix)
    token = _ACTIVE_NAMESPACE.set((self, prefix, _same_tenant(prefix, recall)))
    try:
        yield
    finally:
        _ACTIVE_NAMESPACE.reset(token)

on_session_start async

on_session_start(state: AgentState) -> AgentState

Inject this run's memories, read from the run's namespace (and the read-only namespaces its :class:MemoryScope names) in one recall.

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_start(self, state: AgentState) -> AgentState:
    """Inject this run's memories, read from the run's namespace (and the
    read-only namespaces its :class:`MemoryScope` names) in one recall."""
    scope = self._resolve_namespace(state)
    if scope is None:
        return _strip_memory_blocks(state)
    with self.scoped(scope.namespace, recall=scope.recall):
        return await super().on_session_start(state)

on_session_end async

on_session_end(state: AgentState) -> None

Extract and save this run's memories into the run's namespace.

A background job is created inside the scope, so it inherits it; it shares this manager's one semaphore and one :meth:drain with every other namespace's jobs.

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_end(self, state: AgentState) -> None:
    """Extract and save this run's memories into the run's namespace.

    A background job is created inside the scope, so it inherits it; it
    shares this manager's one semaphore and one :meth:`drain` with every
    other namespace's jobs.
    """
    scope = self._resolve_namespace(state)
    if scope is None:
        return
    # Extraction writes to the run's own namespace only, never to a
    # recall namespace.
    with self.scoped(scope.namespace):
        await super().on_session_end(state)

extract async

extract(messages: list[Message]) -> list[Memory]

Extract memories from a message list.

Uses extract_fn when provided; otherwise applies the built-in heuristic.

Source code in .sdk/src/tulip/memory/manager.py

async def extract(self, messages: list[Message]) -> list[Memory]:
    """Extract memories from a message list.

    Uses ``extract_fn`` when provided; otherwise applies the
    built-in heuristic.
    """
    if self.extract_fn is not None:
        return await self.extract_fn(messages)
    return _heuristic_extract(messages)

retrieve async

retrieve(limit: int | None = None, *, query: str | None = None, namespaces: Sequence[tuple[str, ...]] | None = None) -> list[Memory]

Retrieve stored memories across every type.

Without query this returns the most recently updated memories. With query it first asks the store to rank each type's memories against it (BaseStore.search(namespace, query=...) — semantic on PgMemory / HolographicStore), interleaves the best matches of every type, then tops the list up with the most recent memories. The top-up is what keeps a store whose search only filters (the substring match of InMemoryStore) from injecting nothing just because no memory contains the whole user message. Stores that cannot search at all fall back to recency.

Every type of every namespace is read with one :meth:~tulip.memory.store.BaseStore.search_many call for the ranking and one for the recency top-up — one round trip each on PgMemory — instead of a search per type.

Parameters:

Name	Type	Description	Default
limit	int | None	Maximum memories returned. None uses retrieve_limit.	None
query	str | None	Optional text to rank memories against.	None
namespaces	Sequence[tuple[str, ...]] | None	Prefixes to recall from, in priority order. None reads the current scope: the active namespace and the scope's read-only recall prefixes (see :class:MemoryScope). All must share their first element (the tenant).	None

Raises:

Type	Description
ValueError	namespaces span more than one first element.

Source code in .sdk/src/tulip/memory/manager.py

async def retrieve(
    self,
    limit: int | None = None,
    *,
    query: str | None = None,
    namespaces: Sequence[tuple[str, ...]] | None = None,
) -> list[Memory]:
    """Retrieve stored memories across every type.

    Without ``query`` this returns the most recently updated memories.
    With ``query`` it first asks the store to rank each type's memories
    against it (``BaseStore.search(namespace, query=...)`` — semantic on
    ``PgMemory`` / ``HolographicStore``), interleaves the best matches of
    every type, then tops the list up with the most recent memories. The
    top-up is what keeps a store whose ``search`` only *filters* (the
    substring match of ``InMemoryStore``) from injecting nothing just
    because no memory contains the whole user message. Stores that cannot
    search at all fall back to recency.

    Every type of every namespace is read with one
    :meth:`~tulip.memory.store.BaseStore.search_many` call for the ranking
    and one for the recency top-up — one round trip each on ``PgMemory``
    — instead of a search per type.

    Args:
        limit: Maximum memories returned. ``None`` uses
            ``retrieve_limit``.
        query: Optional text to rank memories against.
        namespaces: Prefixes to recall from, in priority order. ``None``
            reads the current scope: the active namespace and the scope's
            read-only recall prefixes (see :class:`MemoryScope`). All must
            share their first element (the tenant).

    Raises:
        ValueError: ``namespaces`` span more than one first element.
    """
    top = self.retrieve_limit if limit is None else limit
    if top <= 0:
        return []
    if namespaces is None:
        prefixes = list(self.active_recall)
    else:
        prefixes = [tuple(p) for p in namespaces]
        if len({p[:1] for p in prefixes}) > 1:
            raise ValueError(
                "retrieve(namespaces=...) spans more than one tenant (first "
                "element); recall one tenant per call"
            )
    if not prefixes:
        return []
    targets = [(*p, t.value) for p in prefixes for t in MemoryType]

    ranked: list[Memory] = []
    if query and query.strip():
        per_ns = [
            _memories_from_items(items) for items in await self._ranked(targets, query, top)
        ]
        # Scores are not comparable across separate rankings, so merge
        # them rank-by-rank instead of by score.
        for rank in range(max((len(r) for r in per_ns), default=0)):
            ranked.extend(r[rank] for r in per_ns if rank < len(r))

    recent = await self._recent(top, targets)
    return _dedupe(ranked + recent)[:top]

retrieve_relevant async

retrieve_relevant(query: str | None, limit: int | None = None, *, namespaces: Sequence[tuple[str, ...]] | None = None) -> list[Memory]

Rank stored memories against query — see :meth:retrieve.

Source code in .sdk/src/tulip/memory/manager.py

async def retrieve_relevant(
    self,
    query: str | None,
    limit: int | None = None,
    *,
    namespaces: Sequence[tuple[str, ...]] | None = None,
) -> list[Memory]:
    """Rank stored memories against ``query`` — see :meth:`retrieve`."""
    return await self.retrieve(limit, query=query, namespaces=namespaces)

save async

save(memories: list[Memory]) -> None

Upsert memories into the backing store.

Memories with the same key and type overwrite the previous entry — no duplicates accumulate. Afterwards each touched type is pruned to max_memories entries, oldest first.

Source code in .sdk/src/tulip/memory/manager.py

async def save(self, memories: list[Memory]) -> None:
    """Upsert memories into the backing store.

    Memories with the same ``key`` and ``type`` overwrite the
    previous entry — no duplicates accumulate. Afterwards each touched
    type is pruned to ``max_memories`` entries, oldest first.
    """
    from datetime import UTC, datetime  # noqa: PLC0415

    now = datetime.now(UTC).isoformat()

    touched: set[MemoryType] = set()
    for memory in memories:
        ns = self._ns(memory.type)
        value = memory.to_store_value()
        # Copy, so stamping ``updated_at`` doesn't mutate the caller's Memory.
        value["metadata"] = {**value["metadata"], "updated_at": now}

        await self.store.put(
            ns,
            memory.key,
            value,
            metadata={"type": memory.type.value, "updated_at": now},
        )
        touched.add(memory.type)

    for memory_type in touched:
        await self._prune(memory_type)

drain async

drain() -> None

Wait for every scheduled background extraction to finish.

Jobs submitted while draining are waited for too. Safe to call in "inline" mode (returns at once). Errors inside jobs were already logged and emitted; they are not raised here. Bound the wait with asyncio.timeout(...): cancelling drain leaves the jobs running.

Source code in .sdk/src/tulip/memory/manager.py

async def drain(self) -> None:
    """Wait for every scheduled background extraction to finish.

    Jobs submitted while draining are waited for too. Safe to call in
    ``"inline"`` mode (returns at once). Errors inside jobs were already
    logged and emitted; they are not raised here. Bound the wait with
    ``asyncio.timeout(...)``: cancelling ``drain`` leaves the jobs running.
    """
    bg: _BackgroundExtractions | None = self.__dict__.get("_bg_extractions")
    if bg is not None:
        await bg.drain()

NoopMemoryManager

Bases: BaseMemoryManager

Pass-through memory manager — stores and retrieves nothing.

Useful as a test double or as a placeholder when you want the agent-wiring (memory_manager= kwarg) without actual persistence.

pending_extractions property

pending_extractions: int

Background extractions scheduled or running, not yet finished.

retrieve_relevant async

retrieve_relevant(query: str | None, limit: int | None = None) -> list[Memory]

Retrieve the memories most relevant to query.

Called by :meth:on_session_start with the text of the current user turn. The default ignores query and delegates to :meth:retrieve, so subclasses written against the older retrieve(limit) contract keep working unchanged; managers whose backend can rank by relevance override this.

Parameters:

Name	Type	Description	Default
query	str | None	Text to rank memories against, usually the latest user message. None means "no query" (recency order).	required
limit	int | None	Maximum number of memories to return. None uses the manager's default.	None

Source code in .sdk/src/tulip/memory/manager.py

async def retrieve_relevant(self, query: str | None, limit: int | None = None) -> list[Memory]:
    """Retrieve the memories most relevant to ``query``.

    Called by :meth:`on_session_start` with the text of the current user
    turn. The default ignores ``query`` and delegates to :meth:`retrieve`,
    so subclasses written against the older ``retrieve(limit)`` contract
    keep working unchanged; managers whose backend can rank by relevance
    override this.

    Args:
        query: Text to rank memories against, usually the latest user
            message. ``None`` means "no query" (recency order).
        limit: Maximum number of memories to return. ``None`` uses the
            manager's default.
    """
    if limit is None:
        return await self.retrieve()
    return await self.retrieve(limit)

drain async

drain() -> None

Wait for every scheduled background extraction to finish.

Jobs submitted while draining are waited for too. Safe to call in "inline" mode (returns at once). Errors inside jobs were already logged and emitted; they are not raised here. Bound the wait with asyncio.timeout(...): cancelling drain leaves the jobs running.

Source code in .sdk/src/tulip/memory/manager.py

async def drain(self) -> None:
    """Wait for every scheduled background extraction to finish.

    Jobs submitted while draining are waited for too. Safe to call in
    ``"inline"`` mode (returns at once). Errors inside jobs were already
    logged and emitted; they are not raised here. Bound the wait with
    ``asyncio.timeout(...)``: cancelling ``drain`` leaves the jobs running.
    """
    bg: _BackgroundExtractions | None = self.__dict__.get("_bg_extractions")
    if bg is not None:
        await bg.drain()

Memory dataclass

Memory(type: MemoryType, key: str, content: str, metadata: dict[str, Any] = dict())

A single durable memory entry.

Attributes:

Name	Type	Description
type	MemoryType	Semantic category (user / feedback / project / reference).
key	str	Stable logical name within the category. The store uses this as the key, so re-extracting the same fact under the same key updates the record rather than creating a duplicate.
content	str	The actual fact or rule, as a human-readable string.
metadata	dict[str, Any]	Arbitrary extra fields — confidence, source, why, how_to_apply, ISO timestamps, etc.

to_store_value

to_store_value() -> dict[str, Any]

Serialise to a JSON-compatible dict for the store.

Source code in .sdk/src/tulip/memory/manager.py

def to_store_value(self) -> dict[str, Any]:
    """Serialise to a JSON-compatible dict for the store."""
    return {
        "type": self.type.value,
        "key": self.key,
        "content": self.content,
        "metadata": self.metadata,
    }

from_store_value classmethod

from_store_value(value: dict[str, Any]) -> Memory

Deserialise from a store value dict.

Source code in .sdk/src/tulip/memory/manager.py

@classmethod
def from_store_value(cls, value: dict[str, Any]) -> Memory:
    """Deserialise from a store value dict."""
    return cls(
        type=MemoryType(value["type"]),
        key=value["key"],
        content=value["content"],
        metadata=value.get("metadata", {}),
    )

MemoryType

Bases: StrEnum

Semantic category for a stored memory.

Mem0MemoryManager

Mem0MemoryManager(*, client: Any = None, config: dict[str, Any] | None = None, user_id: str | None = None, agent_id: str | None = None, run_id: str | None = None, search_query: str | None = None, retrieve_limit: int = 20)

Bases: BaseMemoryManager

Memory manager backed by mem0.

Parameters:

Name	Type	Description	Default
client	Any	A pre-built mem0 async memory client (anything exposing async add / search / get_all). When None, one is created lazily from config.	None
config	dict[str, Any] | None	mem0 configuration dict passed to AsyncMemory.from_config. When both client and config are None, a default AsyncMemory() is created (OpenAI-backed).	None
user_id / agent_id / run_id		mem0 scope identifiers. At least one (typically user_id) should be set so memories are scoped to the right subject.	required
search_query	str | None	Optional query used by :meth:retrieve. When None, retrieval lists all memories in scope via get_all instead of a similarity search.	None
retrieve_limit	int	Maximum memories returned by :meth:retrieve.	20

Source code in .sdk/src/tulip/memory/managers/mem0.py

def __init__(
    self,
    *,
    client: Any = None,
    config: dict[str, Any] | None = None,
    user_id: str | None = None,
    agent_id: str | None = None,
    run_id: str | None = None,
    search_query: str | None = None,
    retrieve_limit: int = 20,
) -> None:
    self._client_override = client
    self._config = config
    self.user_id = user_id
    self.agent_id = agent_id
    self.run_id = run_id
    self.search_query = search_query
    self.retrieve_limit = retrieve_limit
    self._client: Any = None

pending_extractions property

pending_extractions: int

Background extractions scheduled or running, not yet finished.

extract async

extract(messages: list[Message]) -> list[Memory]

mem0 extracts internally during add — see :meth:on_session_end.

Source code in .sdk/src/tulip/memory/managers/mem0.py

async def extract(self, messages: list[Message]) -> list[Memory]:
    """mem0 extracts internally during ``add`` — see :meth:`on_session_end`."""
    return []

retrieve async

retrieve(limit: int = 20) -> list[Memory]

Retrieve in-scope memories from mem0 for injection.

Source code in .sdk/src/tulip/memory/managers/mem0.py

async def retrieve(self, limit: int = 20) -> list[Memory]:
    """Retrieve in-scope memories from mem0 for injection."""
    client = self._get_client()
    top = min(limit, self.retrieve_limit)

    if self.search_query:
        response = await self._scoped_read(client.search, self.search_query, limit=top)
    else:
        response = await self._scoped_read(client.get_all, limit=top)

    memories: list[Memory] = []
    for row in self._result_rows(response)[:top]:
        text = row.get("memory") or row.get("text") or ""
        if not text:
            continue
        metadata = {k: v for k, v in row.items() if k not in ("memory", "text")}
        memories.append(
            Memory(
                type=MemoryType.REFERENCE,
                key=str(row.get("id", text[:48])),
                content=text,
                metadata=metadata,
            )
        )
    return memories

save async

save(memories: list[Memory]) -> None

Persist explicit memories to mem0 (used for direct save calls).

Source code in .sdk/src/tulip/memory/managers/mem0.py

async def save(self, memories: list[Memory]) -> None:
    """Persist explicit memories to mem0 (used for direct ``save`` calls)."""
    if not memories:
        return
    client = self._get_client()
    scope = self._scope()
    for memory in memories:
        await client.add(memory.content, metadata=memory.metadata or None, **scope)

on_session_end async

on_session_end(state: Any) -> None

Hand the finished conversation to mem0 for extraction + storage.

Source code in .sdk/src/tulip/memory/managers/mem0.py

async def on_session_end(self, state: Any) -> None:
    """Hand the finished conversation to mem0 for extraction + storage."""
    mem0_messages = self._to_mem0_messages(list(state.messages))
    if not mem0_messages:
        return

    client = self._get_client()
    await client.add(mem0_messages, **self._scope())

    from tulip.observability.emit import emit  # noqa: PLC0415

    await emit("memory.manager.extracted", backend="mem0", message_count=len(mem0_messages))

retrieve_relevant async

retrieve_relevant(query: str | None, limit: int | None = None) -> list[Memory]

Retrieve the memories most relevant to query.

Called by :meth:on_session_start with the text of the current user turn. The default ignores query and delegates to :meth:retrieve, so subclasses written against the older retrieve(limit) contract keep working unchanged; managers whose backend can rank by relevance override this.

Parameters:

Name	Type	Description	Default
query	str | None	Text to rank memories against, usually the latest user message. None means "no query" (recency order).	required
limit	int | None	Maximum number of memories to return. None uses the manager's default.	None

Source code in .sdk/src/tulip/memory/manager.py

async def retrieve_relevant(self, query: str | None, limit: int | None = None) -> list[Memory]:
    """Retrieve the memories most relevant to ``query``.

    Called by :meth:`on_session_start` with the text of the current user
    turn. The default ignores ``query`` and delegates to :meth:`retrieve`,
    so subclasses written against the older ``retrieve(limit)`` contract
    keep working unchanged; managers whose backend can rank by relevance
    override this.

    Args:
        query: Text to rank memories against, usually the latest user
            message. ``None`` means "no query" (recency order).
        limit: Maximum number of memories to return. ``None`` uses the
            manager's default.
    """
    if limit is None:
        return await self.retrieve()
    return await self.retrieve(limit)

on_session_start async

on_session_start(state: AgentState) -> AgentState

Retrieve memories and inject them into the agent state.

Called by the agent runtime at the start of every invocation, after on_before_invocation hooks but before the first model call. The latest user message is used as the retrieval query, and the formatted block replaces any block injected by an earlier turn of the same (checkpointed) thread instead of being added next to it.

Parameters:

Name	Type	Description	Default
state	AgentState	Current agent state (just-created or loaded from checkpointer).	required

Returns:

Type	Description
AgentState	Possibly-modified state with memory context injected.

Source code in .sdk/src/tulip/memory/manager.py

async def on_session_start(self, state: AgentState) -> AgentState:
    """Retrieve memories and inject them into the agent state.

    Called by the agent runtime at the start of every invocation,
    after ``on_before_invocation`` hooks but before the first model
    call. The latest user message is used as the retrieval query, and
    the formatted block *replaces* any block injected by an earlier turn
    of the same (checkpointed) thread instead of being added next to it.

    Args:
        state: Current agent state (just-created or loaded from
            checkpointer).

    Returns:
        Possibly-modified state with memory context injected.
    """
    memories = await self.retrieve_relevant(_latest_user_text(state))
    if not memories:
        # Still drop a block left over from an earlier turn: if the store
        # no longer holds those memories (deleted, erased, expired) they
        # must not keep reaching the model from the checkpoint.
        return _strip_memory_blocks(state)

    from tulip.observability.emit import emit  # noqa: PLC0415

    injected_state = _inject_memories_into_state(state, memories)

    await emit(
        "memory.manager.injected",
        memory_count=len(memories),
        types=[m.type.value for m in memories],
    )

    return injected_state

drain async

drain() -> None

Wait for every scheduled background extraction to finish.

Jobs submitted while draining are waited for too. Safe to call in "inline" mode (returns at once). Errors inside jobs were already logged and emitted; they are not raised here. Bound the wait with asyncio.timeout(...): cancelling drain leaves the jobs running.

Source code in .sdk/src/tulip/memory/manager.py

async def drain(self) -> None:
    """Wait for every scheduled background extraction to finish.

    Jobs submitted while draining are waited for too. Safe to call in
    ``"inline"`` mode (returns at once). Errors inside jobs were already
    logged and emitted; they are not raised here. Bound the wait with
    ``asyncio.timeout(...)``: cancelling ``drain`` leaves the jobs running.
    """
    bg: _BackgroundExtractions | None = self.__dict__.get("_bg_extractions")
    if bg is not None:
        await bg.drain()

Delta checkpointing

Storage-efficient checkpointer that persists only the diff between consecutive states (the module targets ~77% storage reduction on long conversations; actual savings depend on workload). Layered on top of any DeltaStorage backend.

DeltaCheckpointer

DeltaCheckpointer(storage: DeltaStorage | None = None, max_chain_depth: int = 5, compression_level: int = 6)

Delta-based checkpointer for efficient state persistence.

This checkpointer stores only the differences between states, achieving significant storage savings. It maintains a chain of deltas pointing back to a full checkpoint.

Features: - Delta compression: Only store changes from parent state - Chain depth limiting: Force full checkpoint after N deltas - zlib compression: Additional compression of stored data - ~77% storage reduction in typical usage

Parameters:

Name	Type	Description	Default
storage	DeltaStorage | None	Storage backend for checkpoints	None
max_chain_depth	int	Maximum number of deltas before forcing full checkpoint	5
compression_level	int	zlib compression level (0-9, higher = better compression)	6

Source code in .sdk/src/tulip/memory/delta.py

def __init__(
    self,
    storage: DeltaStorage | None = None,
    max_chain_depth: int = 5,
    compression_level: int = 6,
):
    self.storage = storage or InMemoryDeltaStorage()
    self.max_chain_depth = max_chain_depth
    self.compression_level = compression_level
    self._state_cache: dict[str, dict[str, Any]] = {}  # Thread -> checkpoint_id -> state

save async

save(state: AgentState, thread_id: str, checkpoint_id: str | None = None) -> str

Save agent state with delta compression.

If a parent checkpoint exists and chain depth allows, only the delta from parent is stored. Otherwise, a full checkpoint is created.

Parameters:

Name	Type	Description	Default
state	AgentState	Current agent state	required
thread_id	str	Thread identifier	required
checkpoint_id	str | None	Optional specific checkpoint ID	None

Returns:

Type	Description
str	Checkpoint ID for the saved state

Source code in .sdk/src/tulip/memory/delta.py

async def save(
    self,
    state: AgentState,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> str:
    """
    Save agent state with delta compression.

    If a parent checkpoint exists and chain depth allows,
    only the delta from parent is stored. Otherwise, a full
    checkpoint is created.

    Args:
        state: Current agent state
        thread_id: Thread identifier
        checkpoint_id: Optional specific checkpoint ID

    Returns:
        Checkpoint ID for the saved state
    """
    checkpoint_id = checkpoint_id or uuid4().hex

    # Get current state as dict
    current_data = state.to_checkpoint()

    # Check for existing checkpoints
    existing = await self.storage.list_checkpoints(thread_id, limit=1)

    if not existing:
        # No parent - create full checkpoint
        checkpoint = await self._create_full_checkpoint(current_data, thread_id, checkpoint_id)
    else:
        parent_meta = existing[0]

        # Check if we should create a full checkpoint due to chain depth
        if parent_meta.chain_depth >= self.max_chain_depth:
            checkpoint = await self._create_full_checkpoint(
                current_data, thread_id, checkpoint_id
            )
        else:
            # Create delta checkpoint
            parent_checkpoint = await self.storage.retrieve(
                thread_id, parent_meta.checkpoint_id
            )
            if parent_checkpoint is None:
                # Parent not found, create full
                checkpoint = await self._create_full_checkpoint(
                    current_data, thread_id, checkpoint_id
                )
            else:
                # Load parent state and compute delta
                parent_data = await self._load_full_state(thread_id, parent_meta.checkpoint_id)
                if parent_data is None:
                    checkpoint = await self._create_full_checkpoint(
                        current_data, thread_id, checkpoint_id
                    )
                else:
                    checkpoint = await self._create_delta_checkpoint(
                        current_data,
                        parent_data,
                        thread_id,
                        checkpoint_id,
                        parent_meta.checkpoint_id,
                        parent_meta.chain_depth + 1,
                    )

    await self.storage.store(thread_id, checkpoint_id, checkpoint)

    # Update cache
    if thread_id not in self._state_cache:
        self._state_cache[thread_id] = {}
    self._state_cache[thread_id][checkpoint_id] = current_data

    return checkpoint_id

load async

load(thread_id: str, checkpoint_id: str | None = None) -> AgentState | None

Load agent state from checkpoint.

If loading a delta checkpoint, reconstructs full state by walking the delta chain back to a full checkpoint.

Parameters:

Name	Type	Description	Default
thread_id	str	Thread identifier	required
checkpoint_id	str | None	Specific checkpoint ID (latest if None)	None

Returns:

Type	Description
AgentState | None	Restored AgentState or None if not found

Source code in .sdk/src/tulip/memory/delta.py

async def load(
    self,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> AgentState | None:
    """
    Load agent state from checkpoint.

    If loading a delta checkpoint, reconstructs full state
    by walking the delta chain back to a full checkpoint.

    Args:
        thread_id: Thread identifier
        checkpoint_id: Specific checkpoint ID (latest if None)

    Returns:
        Restored AgentState or None if not found
    """
    from tulip.core.state import AgentState

    if checkpoint_id is None:
        # Get latest checkpoint
        checkpoints = await self.storage.list_checkpoints(thread_id, limit=1)
        if not checkpoints:
            return None
        checkpoint_id = checkpoints[0].checkpoint_id

    # Try cache first
    if thread_id in self._state_cache:
        if checkpoint_id in self._state_cache[thread_id]:
            return AgentState.from_checkpoint(self._state_cache[thread_id][checkpoint_id])

    # Load and reconstruct state
    state_data = await self._load_full_state(thread_id, checkpoint_id)
    if state_data is None:
        return None

    return AgentState.from_checkpoint(state_data)

list_checkpoints async

list_checkpoints(thread_id: str, limit: int = 10) -> list[str]

List available checkpoint IDs.

Parameters:

Name	Type	Description	Default
thread_id	str	Thread identifier	required
limit	int	Maximum number to return	10

Returns:

Type	Description
list[str]	List of checkpoint IDs, newest first

Source code in .sdk/src/tulip/memory/delta.py

async def list_checkpoints(
    self,
    thread_id: str,
    limit: int = 10,
) -> list[str]:
    """
    List available checkpoint IDs.

    Args:
        thread_id: Thread identifier
        limit: Maximum number to return

    Returns:
        List of checkpoint IDs, newest first
    """
    metadata_list = await self.storage.list_checkpoints(thread_id, limit)
    return [m.checkpoint_id for m in metadata_list]

get_metadata async

get_metadata(thread_id: str, checkpoint_id: str) -> CheckpointMetadata | None

Get metadata for a specific checkpoint.

Source code in .sdk/src/tulip/memory/delta.py

async def get_metadata(
    self,
    thread_id: str,
    checkpoint_id: str,
) -> CheckpointMetadata | None:
    """Get metadata for a specific checkpoint."""
    checkpoint = await self.storage.retrieve(thread_id, checkpoint_id)
    return checkpoint.metadata if checkpoint else None

delete async

delete(thread_id: str, checkpoint_id: str | None = None) -> bool

Delete checkpoint(s).

Source code in .sdk/src/tulip/memory/delta.py

async def delete(
    self,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> bool:
    """Delete checkpoint(s)."""
    # Clear cache
    if thread_id in self._state_cache:
        if checkpoint_id:
            self._state_cache[thread_id].pop(checkpoint_id, None)
        else:
            del self._state_cache[thread_id]

    return await self.storage.delete(thread_id, checkpoint_id)

get_storage_stats async

get_storage_stats(thread_id: str) -> dict[str, Any]

Get storage statistics for a thread.

Returns dict with: - total_checkpoints: Number of checkpoints - total_size: Uncompressed size - compressed_size: Compressed size - compression_ratio: Overall compression ratio - full_checkpoints: Number of full checkpoints - delta_checkpoints: Number of delta checkpoints

Source code in .sdk/src/tulip/memory/delta.py

async def get_storage_stats(
    self,
    thread_id: str,
) -> dict[str, Any]:
    """
    Get storage statistics for a thread.

    Returns dict with:
    - total_checkpoints: Number of checkpoints
    - total_size: Uncompressed size
    - compressed_size: Compressed size
    - compression_ratio: Overall compression ratio
    - full_checkpoints: Number of full checkpoints
    - delta_checkpoints: Number of delta checkpoints
    """
    metadata_list = await self.storage.list_checkpoints(thread_id, limit=1000)

    if not metadata_list:
        return {
            "total_checkpoints": 0,
            "total_size": 0,
            "compressed_size": 0,
            "compression_ratio": 1.0,
            "full_checkpoints": 0,
            "delta_checkpoints": 0,
        }

    total_size = sum(m.size_bytes for m in metadata_list)
    compressed_size = sum(m.compressed_size_bytes for m in metadata_list)
    full_count = sum(1 for m in metadata_list if m.is_full)
    delta_count = sum(1 for m in metadata_list if not m.is_full)

    return {
        "total_checkpoints": len(metadata_list),
        "total_size": total_size,
        "compressed_size": compressed_size,
        "compression_ratio": total_size / compressed_size if compressed_size > 0 else 1.0,
        "full_checkpoints": full_count,
        "delta_checkpoints": delta_count,
    }

DeltaCheckpoint dataclass

DeltaCheckpoint(metadata: CheckpointMetadata, data: bytes, is_delta: bool = True)

A delta checkpoint containing changes from parent.

compression_ratio property

compression_ratio: float

Calculate compression ratio (uncompressed / compressed).

CheckpointMetadata dataclass

CheckpointMetadata(checkpoint_id: str, thread_id: str, parent_id: str | None = None, is_full: bool = True, chain_depth: int = 0, created_at: datetime = (lambda: datetime.now(UTC))(), size_bytes: int = 0, compressed_size_bytes: int = 0)

Metadata for a checkpoint.

DeltaStorage

Bases: ABC

Abstract storage backend for delta checkpoints.

store abstractmethod async

store(thread_id: str, checkpoint_id: str, checkpoint: DeltaCheckpoint) -> None

Store a checkpoint.

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def store(
    self,
    thread_id: str,
    checkpoint_id: str,
    checkpoint: DeltaCheckpoint,
) -> None:
    """Store a checkpoint."""
    ...

retrieve abstractmethod async

retrieve(thread_id: str, checkpoint_id: str) -> DeltaCheckpoint | None

Retrieve a checkpoint.

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def retrieve(
    self,
    thread_id: str,
    checkpoint_id: str,
) -> DeltaCheckpoint | None:
    """Retrieve a checkpoint."""
    ...

list_checkpoints abstractmethod async

list_checkpoints(thread_id: str, limit: int = 10) -> list[CheckpointMetadata]

List checkpoint metadata, newest first.

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def list_checkpoints(
    self,
    thread_id: str,
    limit: int = 10,
) -> list[CheckpointMetadata]:
    """List checkpoint metadata, newest first."""
    ...

delete abstractmethod async

delete(thread_id: str, checkpoint_id: str | None = None) -> bool

Delete checkpoint(s).

Source code in .sdk/src/tulip/memory/delta.py

@abstractmethod
async def delete(
    self,
    thread_id: str,
    checkpoint_id: str | None = None,
) -> bool:
    """Delete checkpoint(s)."""
    ...

InMemoryDeltaStorage

InMemoryDeltaStorage()

Bases: DeltaStorage

In-memory storage for delta checkpoints (for testing).

Source code in .sdk/src/tulip/memory/delta.py

def __init__(self) -> None:
    self._storage: dict[str, dict[str, DeltaCheckpoint]] = {}

Registry

String-based checkpointer lookup — used when configuration passes a provider name (e.g. "s3", "redis") instead of an instance. Custom backends register themselves via register_checkpointer.

get_checkpointer

get_checkpointer(checkpointer_string: str, **kwargs: Any) -> BaseCheckpointer

Get a checkpointer from a string identifier.

Format: "provider" or "provider:config_hint"

The config_hint is provider-specific and is passed as a keyword argument.

Examples:

• "memory" -> MemoryCheckpointer
• "file:./checkpoints" -> FileCheckpointer(base_dir="./checkpoints")
• "redis:localhost:6379" -> RedisCheckpointer(url="redis://localhost:6379")
• "postgresql:mydb" -> PostgreSQLCheckpointer(database="mydb")
• "mysql:mydb" -> MySQLCheckpointer(database="mydb")
• "opensearch" -> OpenSearchCheckpointer()
• "s3:my-bucket" -> S3Backend(bucket="my-bucket")

Parameters:

Name	Type	Description	Default
checkpointer_string	str	Checkpointer identifier	required
**kwargs	Any	Provider-specific configuration	{}

Returns:

Type	Description
BaseCheckpointer	Checkpointer instance

Raises:

Type	Description
ValueError	If provider is unknown

Source code in .sdk/src/tulip/memory/registry.py

def get_checkpointer(checkpointer_string: str, **kwargs: Any) -> BaseCheckpointer:
    """
    Get a checkpointer from a string identifier.

    Format: "provider" or "provider:config_hint"

    The config_hint is provider-specific and is passed as a keyword argument.

    Examples:
        - "memory" -> MemoryCheckpointer
        - "file:./checkpoints" -> FileCheckpointer(base_dir="./checkpoints")
        - "redis:localhost:6379" -> RedisCheckpointer(url="redis://localhost:6379")
        - "postgresql:mydb" -> PostgreSQLCheckpointer(database="mydb")
        - "mysql:mydb" -> MySQLCheckpointer(database="mydb")
        - "opensearch" -> OpenSearchCheckpointer()
        - "s3:my-bucket" -> S3Backend(bucket="my-bucket")

    Args:
        checkpointer_string: Checkpointer identifier
        **kwargs: Provider-specific configuration

    Returns:
        Checkpointer instance

    Raises:
        ValueError: If provider is unknown
    """
    if ":" in checkpointer_string:
        provider, config_hint = checkpointer_string.split(":", 1)
    else:
        provider = checkpointer_string
        config_hint = None

    if provider not in _CHECKPOINTERS:
        available = list(_CHECKPOINTERS.keys())
        raise ValueError(
            f"Unknown checkpointer provider: '{provider}'. "
            f"Available providers: {available}. "
            f"Install optional dependencies or register a custom provider."
        )

    # Pass config_hint if provided
    if config_hint:
        kwargs["config_hint"] = config_hint

    return _CHECKPOINTERS[provider](**kwargs)

register_checkpointer

register_checkpointer(name: str, factory: Callable[..., BaseCheckpointer]) -> None

Register a checkpointer provider.

Parameters:

Name	Type	Description	Default
name	str	Provider name (e.g., "redis", "postgresql", "s3")	required
factory	Callable[..., BaseCheckpointer]	Factory function that takes kwargs and returns a checkpointer	required

Example

def my_factory(**kwargs) -> BaseCheckpointer: ... return MyCustomCheckpointer(**kwargs) register_checkpointer("custom", my_factory)

Source code in .sdk/src/tulip/memory/registry.py

def register_checkpointer(
    name: str,
    factory: Callable[..., BaseCheckpointer],
) -> None:
    """
    Register a checkpointer provider.

    Args:
        name: Provider name (e.g., "redis", "postgresql", "s3")
        factory: Factory function that takes kwargs and returns a checkpointer

    Example:
        >>> def my_factory(**kwargs) -> BaseCheckpointer:
        ...     return MyCustomCheckpointer(**kwargs)
        >>> register_checkpointer("custom", my_factory)
    """
    _CHECKPOINTERS[name] = factory

list_checkpointers

list_checkpointers() -> list[str]

List available checkpointer providers.

Returns:

Type	Description
list[str]	List of registered provider names

Source code in .sdk/src/tulip/memory/registry.py

def list_checkpointers() -> list[str]:
    """
    List available checkpointer providers.

    Returns:
        List of registered provider names
    """
    return list(_CHECKPOINTERS.keys())