Memory API Reference¶

Complete API reference for Memory protocol, MemoryEntry, ConversationMemory, and VectorMemory.

`MemoryEntry`¶

MemoryEntry `dataclass` ¶

MemoryEntry(
    content: str, metadata: dict[str, Any] = dict(), score: float = 0.0
)

A memory entry.

@dataclass
class MemoryEntry:
    """A memory entry."""
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    score: float = 0.0

A single memory record returned by search queries.

Parameters¶

Parameter	Type	Default	Description
`content`	`str`	required	The text content of the memory.
`metadata`	`dict[str, Any]`	`{}`	Arbitrary metadata (e.g., `{"role": "user"}`, `{"source": "doc.pdf"}`).
`score`	`float`	`0.0`	Relevance score assigned by the search method (higher = more relevant).

Usage¶

from flux.memory.base import MemoryEntry

entry = MemoryEntry(
    content="The user prefers dark mode.",
    metadata={"source": "conversation"},
    score=0.95,
)

`Memory` (Protocol)¶

Memory ¶

Bases: Protocol

Protocol for long-term memory storage.

Methods:¶

search `async` ¶

search(query: str, limit: int = 5) -> list[MemoryEntry]

Search memory for relevant entries.

Source code in flux\memory\base.py

async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:
    """Search memory for relevant entries."""
    ...

store `async` ¶

store(content: str, metadata: dict[str, Any] | None = None) -> None

Store a memory entry.

Source code in flux\memory\base.py

async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:
    """Store a memory entry."""
    ...

clear `async` ¶

clear() -> None

Clear all memories.

Source code in flux\memory\base.py

async def clear(self) -> None:
    """Clear all memories."""
    ...

@runtime_checkable
class Memory(Protocol):
    """Protocol for long-term memory storage."""

    async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:
        """Search memory for relevant entries."""
        ...

    async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:
        """Store a memory entry."""
        ...

    async def clear(self) -> None:
        """Clear all memories."""
        ...

The Memory protocol defines the interface for long-term memory backends. Memory stores information across conversations and can be searched by relevance.

Methods¶

`search`¶

async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:

Search memory for entries relevant to the query.

Parameter	Type	Default	Description
`query`	`str`	required	The search query.
`limit`	`int`	`5`	Maximum number of results to return.

Returns: list[MemoryEntry] — entries ordered by relevance (most relevant first).

`store`¶

async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:

Store a new memory entry.

Parameter	Type	Default	Description
`content`	`str`	required	The text content to remember.
`metadata`	`dict[str, Any] \\| None`	`None`	Optional metadata to attach.

`clear`¶

async def clear(self) -> None:

Remove all stored memories.

Implementing a Custom Memory¶

from typing import Any
from flux.memory.base import MemoryEntry

class PostgresMemory:
    async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:
        # Your implementation (e.g., pgvector semantic search)
        ...

    async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:
        # Your implementation
        ...

    async def clear(self) -> None:
        # Your implementation
        ...

`ConversationMemory`¶

ConversationMemory ¶

ConversationMemory(session: Any)

Memory backed by conversation history (wraps a Session).

Source code in flux\memory\conversation.py

def __init__(self, session: Any) -> None:
    self._session = session

Methods:¶

search `async` ¶

search(query: str, limit: int = 5) -> list[MemoryEntry]

Naive substring search over stored messages.

Source code in flux\memory\conversation.py

async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:
    """Naive substring search over stored messages."""
    messages = await self._session.get_messages()
    query_lower = query.lower()
    results: list[MemoryEntry] = []

    for msg in messages:
        content = msg.get("content", "")
        if query_lower in content.lower():
            results.append(
                MemoryEntry(
                    content=content,
                    metadata={"role": msg.get("role", "unknown")},
                )
            )

    return results[:limit]

store `async` ¶

store(content: str, metadata: dict[str, Any] | None = None) -> None

Store a memory as a user message.

Source code in flux\memory\conversation.py

async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:
    """Store a memory as a user message."""
    msg: dict[str, Any] = {"role": "user", "content": content}
    if metadata:
        msg["metadata"] = metadata
    await self._session.add_messages([msg])

class ConversationMemory:
    """Memory backed by conversation history (wraps a Session)."""

    def __init__(self, session: Any) -> None:
        self._session = session

A Memory implementation that wraps a Session. Search performs naive substring matching over stored conversation messages.

Parameters¶

Parameter	Type	Default	Description
`session`	`Any`	required	A `Session` instance to use as the backing store.

Methods¶

`search`¶

async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:

Performs case-insensitive substring matching over all stored messages. Returns up to limit matching entries.

Parameter	Type	Default	Description
`query`	`str`	required	The search query.
`limit`	`int`	`5`	Maximum results.

Returns: list[MemoryEntry] — each entry's metadata includes {"role": "<message_role>"}.

`store`¶

async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:

Store a memory as a user message in the backing session.

Parameter	Type	Default	Description
`content`	`str`	required	The text to store.
`metadata`	`dict[str, Any] \\| None`	`None`	Optional metadata attached to the message.

`clear`¶

async def clear(self) -> None:

Delegates to session.clear(). Removes all messages from the backing session.

Usage¶

from flux.sessions.in_memory import InMemorySession
from flux.memory.conversation import ConversationMemory

session = InMemorySession()
memory = ConversationMemory(session)

await memory.store("User prefers TypeScript over JavaScript.")
await memory.store("User's project uses React 18.")

results = await memory.search("TypeScript")
for entry in results:
    print(entry.content)  # "User prefers TypeScript over JavaScript."
    print(entry.metadata)  # {"role": "user"}

Note

ConversationMemory performs substring matching, not semantic search. For semantic search, use VectorMemory or implement a custom memory with embeddings.

`VectorMemory`¶

VectorMemory ¶

VectorMemory()

Simple in-memory vector store using hash-based embeddings.

For production use, replace with real embeddings (e.g., sentence-transformers).

Source code in flux\memory\vector.py

def __init__(self) -> None:
    self._entries: list[tuple[list[float], MemoryEntry]] = []

Methods:¶

search `async` ¶

search(query: str, limit: int = 5) -> list[MemoryEntry]

Search by cosine similarity of hash-based embeddings.

Source code in flux\memory\vector.py

async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:
    """Search by cosine similarity of hash-based embeddings."""
    if not self._entries:
        return []

    query_vec = _text_to_vector(query)
    scored: list[tuple[float, MemoryEntry]] = []

    for vec, entry in self._entries:
        sim = _cosine_similarity(query_vec, vec)
        scored.append((sim, entry))

    scored.sort(key=lambda x: x[0], reverse=True)
    return [entry for _, entry in scored[:limit]]

store `async` ¶

store(content: str, metadata: dict[str, Any] | None = None) -> None

Store a memory entry.

Source code in flux\memory\vector.py

async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:
    """Store a memory entry."""
    vec = _text_to_vector(content)
    entry = MemoryEntry(
        content=content,
        metadata=metadata or {},
    )
    self._entries.append((vec, entry))

class VectorMemory:
    """Simple in-memory vector store using hash-based embeddings.
    For production use, replace with real embeddings (e.g., sentence-transformers).
    """

    def __init__(self) -> None:
        self._entries: list[tuple[list[float], MemoryEntry]] = []

A Memory implementation that uses simple hash-based vector embeddings for similarity search. Each stored text is converted to a 32-dimensional vector using character 3-gram (shingling) hashing and compared via cosine similarity.

Parameters¶

None. VectorMemory is initialized with an empty store.

Methods¶

`search`¶

async def search(self, query: str, limit: int = 5) -> list[MemoryEntry]:

Search by cosine similarity of hash-based embeddings. Returns the most similar entries.

Parameter	Type	Default	Description
`query`	`str`	required	The search query.
`limit`	`int`	`5`	Maximum number of results.

Returns: list[MemoryEntry] — entries ordered by similarity (most similar first).

`store`¶

async def store(self, content: str, metadata: dict[str, Any] | None = None) -> None:

Store a memory entry and compute its vector embedding.

Parameter	Type	Default	Description
`content`	`str`	required	The text to store.
`metadata`	`dict[str, Any] \\| None`	`None`	Optional metadata.

`clear`¶

async def clear(self) -> None:

Remove all stored entries and their embeddings.

Usage¶

from flux.memory.vector import VectorMemory

memory = VectorMemory()

await memory.store("Python is a programming language.", metadata={"topic": "python"})
await memory.store("JavaScript runs in the browser.", metadata={"topic": "javascript"})
await memory.store("Rust is known for memory safety.", metadata={"topic": "rust"})

results = await memory.search("programming language", limit=2)
for entry in results:
    print(f"{entry.content} (score: {entry.score:.2f})")

How Hash-Based Embeddings Work¶

VectorMemory uses a lightweight embedding technique:

Shingling: The input text is split into overlapping 3-character n-grams.
Hashing: Each shingle is hashed (MD5) and mapped to one of 32 dimensions.
Counting: The dimension value is incremented for each shingle that maps to it.
Normalization: The vector is L2-normalized.

Similarity is computed via cosine similarity between two normalized vectors.

Warning

Hash-based embeddings provide only basic similarity matching. They work well for exact or near-exact text overlap but do not capture semantic meaning. For production use, integrate real embedding models (e.g., sentence-transformers, OpenAI embeddings) and store vectors in a dedicated vector database (e.g., Pinecone, Weaviate, pgvector).

Memory API Reference¶

MemoryEntry¶

MemoryEntry dataclass ¶

Parameters¶

Usage¶

Memory (Protocol)¶

Memory ¶

Methods:¶

search async ¶

store async ¶

clear async ¶

Methods¶

search¶

store¶

clear¶

Implementing a Custom Memory¶

ConversationMemory¶

ConversationMemory ¶

Methods:¶

search async ¶

store async ¶

Parameters¶

Methods¶

search¶

store¶

clear¶

Usage¶

VectorMemory¶

VectorMemory ¶

Methods:¶

search async ¶

store async ¶

Parameters¶

Methods¶

search¶

store¶

clear¶

Usage¶

How Hash-Based Embeddings Work¶

`MemoryEntry`¶

MemoryEntry `dataclass` ¶

`Memory` (Protocol)¶

search `async` ¶

store `async` ¶

clear `async` ¶

`search`¶

`store`¶

`clear`¶

`ConversationMemory`¶

search `async` ¶

store `async` ¶

`search`¶

`store`¶

`clear`¶

`VectorMemory`¶

search `async` ¶

store `async` ¶

`search`¶

`store`¶

`clear`¶