windlass.interfaces.memory

memory

The memory interface.

Memory is what turns a stateless model call into a conversation. Windlass separates two concerns that are often conflated:

• Conversation memory — the recent transcript, replayed into the prompt. Buffer, sliding-window and summarising strategies live here.
• Long-term memory — durable facts recalled by relevance rather than recency, backed by a vector store.

Both implement this interface, so an agent can hold one, the other, or a composite of both without knowing the difference.

Implementers override :meth:Memory.aadd and :meth:Memory.aget.

Example

from windlass.providers.memory.conversation import BufferMemory from windlass.core.types import Message m = BufferMemory() m.add(Message.user("hi")) len(m.get()) 1

Memory

Memory(
    *,
    max_messages: int | None = None,
    return_system: bool = False,
    name: str | None = None,
    **config: Any
)

Bases: Component

Abstract conversation / long-term memory.

Memory is keyed by thread_id so one instance can serve many concurrent users — an agent handling a hundred chat sessions needs one memory object, not a hundred.

Parameters:

Name	Type	Description	Default
max_messages	int | None	Ceiling on how many messages :meth:aget returns. None means unlimited.	None
return_system	bool	Whether system messages are included in recall. Usually False: the system prompt is supplied by the agent, not the history.	False
name	str | None	Component name for traces.	None
**config	Any	Strategy-specific options.	{}

Example

Implementing a memory takes two methods::

class NullMemory(Memory):
    provider_name = "null"

    async def aadd(self, messages, *, thread_id="default"): ...
    async def aget(self, *, thread_id="default", query=None):
        return []

Source code in src\windlass\interfaces\memory.py

def __init__(
    self,
    *,
    max_messages: int | None = None,
    return_system: bool = False,
    name: str | None = None,
    **config: Any,
) -> None:
    super().__init__(
        name=name or self.provider_name,
        max_messages=max_messages,
        return_system=return_system,
        **config,
    )
    self.max_messages = max_messages
    self.return_system = return_system

aadd abstractmethod async

aadd(messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD) -> None

Record one or more messages.

Parameters:

Name	Type	Description	Default
messages	Message | Sequence[Message]	A message or a sequence of them.	required
thread_id	str	Conversation this belongs to.	DEFAULT_THREAD

Raises:

Type	Description
MemoryError_	When the backend cannot persist the messages.

Source code in src\windlass\interfaces\memory.py

@abc.abstractmethod
async def aadd(
    self, messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD
) -> None:
    """Record one or more messages.

    Args:
        messages: A message or a sequence of them.
        thread_id: Conversation this belongs to.

    Raises:
        WindlassMemoryError: When the backend cannot persist the messages.
    """

aget abstractmethod async

aget(
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None
) -> list[Message]

Recall messages for a thread.

Parameters:

Name	Type	Description	Default
thread_id	str	Conversation to recall.	DEFAULT_THREAD
query	str | None	Current user input. Semantic memories use it to rank recall; recency-based memories ignore it.	None
limit	int | None	Override for :attr:max_messages.	None

Returns:

Type	Description
list[Message]	Messages in chronological order, oldest first.

Source code in src\windlass\interfaces\memory.py

@abc.abstractmethod
async def aget(
    self,
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None,
) -> list[Message]:
    """Recall messages for a thread.

    Args:
        thread_id: Conversation to recall.
        query: Current user input. Semantic memories use it to rank recall;
            recency-based memories ignore it.
        limit: Override for :attr:`max_messages`.

    Returns:
        Messages in chronological order, oldest first.
    """

aclear async

aclear(*, thread_id: str | None = None) -> None

Forget a thread, or everything when thread_id is None.

Parameters:

Name	Type	Description	Default
thread_id	str | None	Thread to clear. None clears every thread.	None

Source code in src\windlass\interfaces\memory.py

async def aclear(self, *, thread_id: str | None = None) -> None:
    """Forget a thread, or everything when ``thread_id`` is ``None``.

    Args:
        thread_id: Thread to clear. ``None`` clears every thread.
    """

athreads async

athreads() -> list[str]

Return the thread ids this memory knows about.

Source code in src\windlass\interfaces\memory.py

async def athreads(self) -> list[str]:
    """Return the thread ids this memory knows about."""
    return []

add

add(messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD) -> None

Blocking :meth:aadd.

Source code in src\windlass\interfaces\memory.py

def add(
    self, messages: Message | Sequence[Message], *, thread_id: str = DEFAULT_THREAD
) -> None:
    """Blocking :meth:`aadd`."""
    run_sync(self.aadd(messages, thread_id=thread_id))

get

get(
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None
) -> list[Message]

Blocking :meth:aget.

Source code in src\windlass\interfaces\memory.py

def get(
    self,
    *,
    thread_id: str = DEFAULT_THREAD,
    query: str | None = None,
    limit: int | None = None,
) -> list[Message]:
    """Blocking :meth:`aget`."""
    return run_sync(self.aget(thread_id=thread_id, query=query, limit=limit))

clear

clear(*, thread_id: str | None = None) -> None

Blocking :meth:aclear.

Source code in src\windlass\interfaces\memory.py

def clear(self, *, thread_id: str | None = None) -> None:
    """Blocking :meth:`aclear`."""
    run_sync(self.aclear(thread_id=thread_id))

threads

threads() -> list[str]

Blocking :meth:athreads.

Source code in src\windlass\interfaces\memory.py

def threads(self) -> list[str]:
    """Blocking :meth:`athreads`."""
    return run_sync(self.athreads())