Conversation Memory

ConversationMemory gives any pattern a persistent, searchable short-term memory. Add past interactions, retrieve recent history, keyword-search for relevant context, and compose with retrieval pipelines.

Quick start

from brain.patterns import ConversationMemory, ReActAgent

memory = ConversationMemory(max_recent=50)
agent = ReActAgent(tools={"search": my_search})

# Turn 1
result = agent.run("What is OAuth2?")
memory.add(task="What is OAuth2?", answer=result.answer)

# Turn 2 — inject relevant memory into the prompt
context = memory.search("authentication token", top_k=1)
task = f"Context: {context[0]}\n\nQuestion: What token format is used?"
result2 = agent.run(task)
memory.add(task=task, answer=result2.answer)

SQLite persistence

Omit db_path for in-memory only; pass a path to persist across sessions:

from pathlib import Path
from brain.patterns import ConversationMemory

# Writes to disk — survives process restarts
memory = ConversationMemory(
    db_path=Path("~/.secondbrain/memory.db").expanduser(),
    max_recent=200,
)

Keyword search

memory.add("OAuth2 question", "OAuth2 is an authorization framework.")
memory.add("JWT question", "JWT tokens carry signed claims.")
memory.add("RAG question", "RAG combines retrieval with generation.")

results = memory.search("token authentication", top_k=2)
# → ["OAuth2 is an authorization...", "JWT tokens carry..."]

Search ranks entries by term overlap between the query and stored task + answer text. No embeddings required.

as_retriever() — compose with RAGAgent

Mix memory chunks with a corpus retriever:

from brain.patterns import RAGAgent, ConversationMemory

memory = ConversationMemory()
memory.add("auth question", "OAuth2 is used for all API calls.")

def corpus_retriever(query: str, top_k: int = 3) -> list[str]:
    return ["The API uses OAuth2.", "Rate limit: 100 req/min."][:top_k]

# memory_weight controls the fraction of top_k slots filled from memory
hybrid = memory.as_retriever(corpus_retriever, memory_weight=0.4)

agent = RAGAgent(retriever=hybrid, top_k=5)
result = agent.run("What auth method is used?")

With memory_weight=0.4 and top_k=5, up to 2 slots come from memory and 3 from the corpus.

Multi-turn conversation loop

from brain.patterns import ReActAgent, ConversationMemory

memory = ConversationMemory(max_recent=20)
agent = ReActAgent(tools={"search": my_search})

for user_message in conversation_stream:
    # Retrieve relevant past context
    past = memory.search(user_message, top_k=1)
    context = f"Conversation history:\n{past[0]}\n\n" if past else ""
    result = agent.run(f"{context}{user_message}")
    memory.add(task=user_message, answer=result.answer)
    print(result.answer)

API Reference

brain.patterns.memory.ConversationMemory

ConversationMemory(max_recent: int = 50, db_path: str | Path | None = None, table: str = 'pattern_memory')

Stores agent interactions and retrieves relevant past context.

Parameters:

Name	Type	Description	Default
max_recent	int	Maximum number of entries to keep in the in-memory buffer. Older entries are evicted (FIFO). Does not affect SQLite storage.	50
db_path	str | Path | None	If set, persist entries to this SQLite database file. Entries survive process restarts and are loaded on init.	None
table	str	SQLite table name (default: "pattern_memory").	'pattern_memory'

Source code in brain/patterns/memory.py

def __init__(
    self,
    max_recent: int = 50,
    db_path: str | Path | None = None,
    table: str = "pattern_memory",
) -> None:
    self.max_recent = max_recent
    self._entries: list[MemoryEntry] = []
    self._db_path = Path(db_path) if db_path else None
    self._table = table

    if self._db_path is not None:
        self._init_db()
        self._load_from_db()

add

add(task: str, answer: str, metadata: dict[str, Any] | None = None) -> MemoryEntry

Store a task → answer interaction.

Parameters:

Name	Type	Description	Default
task	str	The question or instruction that was run.	required
answer	str	The agent's final answer.	required
metadata	dict[str, Any] | None	Optional extra data attached to the entry.	None

Returns:

Type	Description
MemoryEntry	The created MemoryEntry.

Source code in brain/patterns/memory.py

def add(
    self,
    task: str,
    answer: str,
    metadata: dict[str, Any] | None = None,
) -> MemoryEntry:
    """Store a task → answer interaction.

    Args:
        task: The question or instruction that was run.
        answer: The agent's final answer.
        metadata: Optional extra data attached to the entry.

    Returns:
        The created ``MemoryEntry``.
    """
    entry = MemoryEntry(task=task, answer=answer, metadata=metadata or {})
    self._entries.append(entry)

    # Evict oldest entries beyond the buffer limit
    if len(self._entries) > self.max_recent:
        self._entries = self._entries[-self.max_recent :]

    if self._db_path is not None:
        self._persist(entry)

    return entry

recent

recent(n: int | None = None) -> list[MemoryEntry]

Return the most recent n entries (newest last).

Parameters:

Name	Type	Description	Default
n	int | None	Number of entries to return. Defaults to all buffered entries.	None

Returns:

Type	Description
list[MemoryEntry]	List of MemoryEntry objects, oldest first.

Source code in brain/patterns/memory.py

def recent(self, n: int | None = None) -> list[MemoryEntry]:
    """Return the most recent *n* entries (newest last).

    Args:
        n: Number of entries to return. Defaults to all buffered entries.

    Returns:
        List of ``MemoryEntry`` objects, oldest first.
    """
    entries = self._entries
    if n is not None:
        entries = entries[-n:]
    return list(entries)

search

search(query: str, top_k: int = 3) -> list[str]

Simple keyword search over stored entries.

Scores each entry by how many query words appear in the task + answer text (case-insensitive). Returns the top-k matching plain-text chunks.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
top_k	int	Maximum number of results to return.	3

Returns:

Type	Description
list[str]	List of plain-text chunks ("Q: ...\nA: ...") ranked by relevance.

Source code in brain/patterns/memory.py

def search(self, query: str, top_k: int = 3) -> list[str]:
    """Simple keyword search over stored entries.

    Scores each entry by how many query words appear in the task + answer
    text (case-insensitive). Returns the top-k matching plain-text chunks.

    Args:
        query: The search query.
        top_k: Maximum number of results to return.

    Returns:
        List of plain-text chunks (``"Q: ...\\nA: ..."``) ranked by relevance.
    """
    if not self._entries:
        return []

    words = set(query.lower().split())
    scored: list[tuple[float, MemoryEntry]] = []
    for entry in self._entries:
        text = (entry.task + " " + entry.answer).lower()
        score = sum(1 for w in words if w in text)
        if score > 0:
            scored.append((score, entry))

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

as_retriever

as_retriever(base_retriever: Retriever | None = None, memory_weight: int = 1) -> Retriever

Return a retriever that combines memory hits with an optional base retriever.

Memory results are prepended to the base retriever's results. The total number of results is still bounded by top_k.

Parameters:

Name	Type	Description	Default
base_retriever	Retriever | None	Optional corpus retriever to call after memory lookup.	None
memory_weight	int	How many of the top_k slots to reserve for memory hits (default: 1). The rest go to base_retriever.	1

Returns:

Type	Description
Retriever	A Retriever callable matching the (query, top_k) -> list[str] signature.

Example::

retriever = mem.as_retriever(base_retriever=corpus_fn, memory_weight=2)
rag = RAGAgent(retriever=retriever)

Source code in brain/patterns/memory.py

def as_retriever(
    self,
    base_retriever: Retriever | None = None,
    memory_weight: int = 1,
) -> Retriever:
    """Return a retriever that combines memory hits with an optional base retriever.

    Memory results are prepended to the base retriever's results. The total
    number of results is still bounded by ``top_k``.

    Args:
        base_retriever: Optional corpus retriever to call after memory lookup.
        memory_weight: How many of the ``top_k`` slots to reserve for memory
            hits (default: 1). The rest go to ``base_retriever``.

    Returns:
        A ``Retriever`` callable matching the ``(query, top_k) -> list[str]`` signature.

    Example::

        retriever = mem.as_retriever(base_retriever=corpus_fn, memory_weight=2)
        rag = RAGAgent(retriever=retriever)
    """

    def _retriever(query: str, top_k: int = 3) -> list[str]:
        mem_hits = self.search(query, top_k=memory_weight)
        remaining = max(0, top_k - len(mem_hits))
        base_hits: list[str] = []
        if base_retriever is not None and remaining > 0:
            try:
                base_hits = base_retriever(query, remaining)
            except Exception as exc:  # noqa: BLE001
                logger.warning("ConversationMemory: base_retriever raised: %s", exc)
        return mem_hits + base_hits

    return _retriever

brain.patterns.memory.MemoryEntry dataclass

MemoryEntry(task: str, answer: str, timestamp: float = time(), metadata: dict[str, Any] = dict())

A single stored interaction.

Attributes:

Name	Type	Description
task	str	The task/question asked.
answer	str	The agent's answer.
timestamp	float	Unix timestamp when the entry was created.
metadata	dict[str, Any]	Arbitrary extra data (e.g. pattern name, provider).

to_text

to_text() -> str

Return a plain-text representation for use as a retrieval chunk.

Source code in brain/patterns/memory.py

def to_text(self) -> str:
    """Return a plain-text representation for use as a retrieval chunk."""
    return f"Q: {self.task}\nA: {self.answer}"

Sync vs Async

ConversationMemory is sync. For async pipelines, call memory.search() and memory.add() in a thread executor:

import asyncio
from brain.patterns import ConversationMemory

memory = ConversationMemory()

async def async_search(query: str) -> list[str]:
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, memory.search, query, 3)