Base store

Simplified Async-First Base Store for TAF Framework

This module provides a clean, modern interface for memory stores with: - Async-first design for better performance - Core CRUD operations only - Message-specific convenience methods - Extensible for different backends (vector stores, managed services, etc.)

Classes:

Name	Description
BaseStore	Simplified async-first base class for memory stores in TAF.

Attributes:

Name	Type	Description
logger

Attributes

logger module-attribute

logger = getLogger(__name__)

Classes

BaseStore

Bases: ABC

Simplified async-first base class for memory stores in TAF.

This class provides a clean interface that supports: - Vector stores (Qdrant, Pinecone, Chroma, etc.) - Managed memory services (mem0, Zep, etc.) - Graph databases (Neo4j, etc.)

Key Design Principles: - Async-first for better performance - Core CRUD operations only - User and agent-centric operations - Extensible filtering and metadata

Methods:

Name	Description
adelete	Delete a memory by ID.
aforget_memory	Delete a memory by for a user or agent.
aget	Get a specific memory by ID.
aget_all	Get a specific memory by user_id.
arelease	Clean up any resources used by the store (override in subclasses if needed).
asearch	Search memories by content similarity.
asetup	Asynchronous setup method for checkpointer.
astore	Add a new memory.
aupdate	Update an existing memory.
delete	Synchronous wrapper for adelete that runs the async implementation.
forget_memory	Delete a memory by for a user or agent.
get	Synchronous wrapper for aget that runs the async implementation.
get_all	Synchronous wrapper for aget that runs the async implementation.
release	Clean up any resources used by the store (override in subclasses if needed).
search	Synchronous wrapper for asearch that runs the async implementation.
setup	Synchronous setup method for checkpointer.
store	Synchronous wrapper for astore that runs the async implementation.
update	Synchronous wrapper for aupdate that runs the async implementation.

Source code in agentflow/store/base_store.py

class BaseStore(ABC):
    """
    Simplified async-first base class for memory stores in TAF.

    This class provides a clean interface that supports:
    - Vector stores (Qdrant, Pinecone, Chroma, etc.)
    - Managed memory services (mem0, Zep, etc.)
    - Graph databases (Neo4j, etc.)

    Key Design Principles:
    - Async-first for better performance
    - Core CRUD operations only
    - User and agent-centric operations
    - Extensible filtering and metadata
    """

    def setup(self) -> Any:
        """
        Synchronous setup method for checkpointer.

        Returns:
            Any: Implementation-defined setup result.
        """
        return run_coroutine(self.asetup())

    async def asetup(self) -> Any:
        """
        Asynchronous setup method for checkpointer.

        Returns:
            Any: Implementation-defined setup result.
        """
        raise NotImplementedError

    # --- Core Memory Operations ---

    @abstractmethod
    async def astore(
        self,
        config: dict[str, Any],
        content: str | Message,
        memory_type: MemoryType = MemoryType.EPISODIC,
        category: str = "general",
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> str:
        """
        Add a new memory.

        Args:
            content: The memory content
            user_id: User identifier
            agent_id: Agent identifier
            memory_type: Type of memory (episodic, semantic, etc.)
            category: Memory category for organization
            metadata: Additional metadata
            **kwargs: Store-specific parameters

        Returns:
            Memory ID
        """
        raise NotImplementedError

    # --- Sync wrappers ---
    def store(
        self,
        config: dict[str, Any],
        content: str | Message,
        memory_type: MemoryType = MemoryType.EPISODIC,
        category: str = "general",
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> str:
        """Synchronous wrapper for `astore` that runs the async implementation."""
        return run_coroutine(
            self.astore(
                config,
                content,
                memory_type=memory_type,
                category=category,
                metadata=metadata,
                **kwargs,
            )
        )

    @abstractmethod
    async def asearch(
        self,
        config: dict[str, Any],
        query: str,
        memory_type: MemoryType | None = None,
        category: str | None = None,
        limit: int = 10,
        score_threshold: float | None = None,
        filters: dict[str, Any] | None = None,
        retrieval_strategy: RetrievalStrategy = RetrievalStrategy.SIMILARITY,
        distance_metric: DistanceMetric = DistanceMetric.COSINE,
        max_tokens: int = 4000,
        **kwargs,
    ) -> list[MemorySearchResult]:
        """
        Search memories by content similarity.

        Args:
            query: Search query
            user_id: Filter by user
            agent_id: Filter by agent
            memory_type: Filter by memory type
            category: Filter by category
            limit: Maximum results
            score_threshold: Minimum similarity score
            filters: Additional filters
            **kwargs: Store-specific parameters

        Returns:
            List of matching memories
        """
        raise NotImplementedError

    def search(
        self,
        config: dict[str, Any],
        query: str,
        memory_type: MemoryType | None = None,
        category: str | None = None,
        limit: int = 10,
        score_threshold: float | None = None,
        filters: dict[str, Any] | None = None,
        retrieval_strategy: RetrievalStrategy = RetrievalStrategy.SIMILARITY,
        distance_metric: DistanceMetric = DistanceMetric.COSINE,
        max_tokens: int = 4000,
        **kwargs,
    ) -> list[MemorySearchResult]:
        """Synchronous wrapper for `asearch` that runs the async implementation."""
        return run_coroutine(
            self.asearch(
                config,
                query,
                memory_type=memory_type,
                category=category,
                limit=limit,
                score_threshold=score_threshold,
                filters=filters,
                retrieval_strategy=retrieval_strategy,
                distance_metric=distance_metric,
                max_tokens=max_tokens,
                **kwargs,
            )
        )

    @abstractmethod
    async def aget(
        self,
        config: dict[str, Any],
        memory_id: str,
        **kwargs,
    ) -> MemorySearchResult | None:
        """Get a specific memory by ID."""
        raise NotImplementedError

    @abstractmethod
    async def aget_all(
        self,
        config: dict[str, Any],
        limit: int = 100,
        **kwargs,
    ) -> list[MemorySearchResult]:
        """Get a specific memory by user_id."""
        raise NotImplementedError

    def get(self, config: dict[str, Any], memory_id: str, **kwargs) -> MemorySearchResult | None:
        """Synchronous wrapper for `aget` that runs the async implementation."""
        return run_coroutine(self.aget(config, memory_id, **kwargs))

    def get_all(
        self,
        config: dict[str, Any],
        limit: int = 100,
        **kwargs,
    ) -> list[MemorySearchResult]:
        """Synchronous wrapper for `aget` that runs the async implementation."""
        return run_coroutine(self.aget_all(config, limit=limit, **kwargs))

    @abstractmethod
    async def aupdate(
        self,
        config: dict[str, Any],
        memory_id: str,
        content: str | Message,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> Any:
        """
        Update an existing memory.

        Args:
            memory_id: ID of memory to update
            content: New content (optional)
            metadata: New/additional metadata (optional)
            **kwargs: Store-specific parameters
        """
        raise NotImplementedError

    def update(
        self,
        config: dict[str, Any],
        memory_id: str,
        content: str | Message,
        metadata: dict[str, Any] | None = None,
        **kwargs,
    ) -> Any:
        """Synchronous wrapper for `aupdate` that runs the async implementation."""
        return run_coroutine(self.aupdate(config, memory_id, content, metadata=metadata, **kwargs))

    @abstractmethod
    async def adelete(
        self,
        config: dict[str, Any],
        memory_id: str,
        **kwargs,
    ) -> Any:
        """Delete a memory by ID."""
        raise NotImplementedError

    def delete(self, config: dict[str, Any], memory_id: str, **kwargs) -> None:
        """Synchronous wrapper for `adelete` that runs the async implementation."""
        return run_coroutine(self.adelete(config, memory_id, **kwargs))

    @abstractmethod
    async def aforget_memory(
        self,
        config: dict[str, Any],
        **kwargs,
    ) -> Any:
        """Delete a memory by for a user or agent."""
        raise NotImplementedError

    def forget_memory(
        self,
        config: dict[str, Any],
        **kwargs,
    ) -> Any:
        """Delete a memory by for a user or agent."""
        return run_coroutine(self.aforget_memory(config, **kwargs))

    # --- Cleanup and Resource Management ---

    async def arelease(self) -> None:
        """Clean up any resources used by the store (override in subclasses if needed)."""
        raise NotImplementedError

    def release(self) -> None:
        """Clean up any resources used by the store (override in subclasses if needed)."""
        return run_coroutine(self.arelease())

Functions

adelete abstractmethod async

adelete(config, memory_id, **kwargs)

Delete a memory by ID.

Source code in agentflow/store/base_store.py

@abstractmethod
async def adelete(
    self,
    config: dict[str, Any],
    memory_id: str,
    **kwargs,
) -> Any:
    """Delete a memory by ID."""
    raise NotImplementedError

aforget_memory abstractmethod async

aforget_memory(config, **kwargs)

Delete a memory by for a user or agent.

Source code in agentflow/store/base_store.py

@abstractmethod
async def aforget_memory(
    self,
    config: dict[str, Any],
    **kwargs,
) -> Any:
    """Delete a memory by for a user or agent."""
    raise NotImplementedError

aget abstractmethod async

aget(config, memory_id, **kwargs)

Get a specific memory by ID.

Source code in agentflow/store/base_store.py

@abstractmethod
async def aget(
    self,
    config: dict[str, Any],
    memory_id: str,
    **kwargs,
) -> MemorySearchResult | None:
    """Get a specific memory by ID."""
    raise NotImplementedError

aget_all abstractmethod async

aget_all(config, limit=100, **kwargs)

Get a specific memory by user_id.

Source code in agentflow/store/base_store.py

@abstractmethod
async def aget_all(
    self,
    config: dict[str, Any],
    limit: int = 100,
    **kwargs,
) -> list[MemorySearchResult]:
    """Get a specific memory by user_id."""
    raise NotImplementedError

arelease async

arelease()

Clean up any resources used by the store (override in subclasses if needed).

Source code in agentflow/store/base_store.py

async def arelease(self) -> None:
    """Clean up any resources used by the store (override in subclasses if needed)."""
    raise NotImplementedError

asearch abstractmethod async

asearch(config, query, memory_type=None, category=None, limit=10, score_threshold=None, filters=None, retrieval_strategy=RetrievalStrategy.SIMILARITY, distance_metric=DistanceMetric.COSINE, max_tokens=4000, **kwargs)

Search memories by content similarity.

Parameters:

Name	Type	Description	Default
query	str	Search query	required
user_id		Filter by user	required
agent_id		Filter by agent	required
memory_type	MemoryType | None	Filter by memory type	None
category	str | None	Filter by category	None
limit	int	Maximum results	10
score_threshold	float | None	Minimum similarity score	None
filters	dict[str, Any] | None	Additional filters	None
**kwargs		Store-specific parameters	{}

Returns:

Type	Description
list[MemorySearchResult]	List of matching memories

Source code in agentflow/store/base_store.py

@abstractmethod
async def asearch(
    self,
    config: dict[str, Any],
    query: str,
    memory_type: MemoryType | None = None,
    category: str | None = None,
    limit: int = 10,
    score_threshold: float | None = None,
    filters: dict[str, Any] | None = None,
    retrieval_strategy: RetrievalStrategy = RetrievalStrategy.SIMILARITY,
    distance_metric: DistanceMetric = DistanceMetric.COSINE,
    max_tokens: int = 4000,
    **kwargs,
) -> list[MemorySearchResult]:
    """
    Search memories by content similarity.

    Args:
        query: Search query
        user_id: Filter by user
        agent_id: Filter by agent
        memory_type: Filter by memory type
        category: Filter by category
        limit: Maximum results
        score_threshold: Minimum similarity score
        filters: Additional filters
        **kwargs: Store-specific parameters

    Returns:
        List of matching memories
    """
    raise NotImplementedError

asetup async

asetup()

Asynchronous setup method for checkpointer.

Returns:

Name	Type	Description
Any	Any	Implementation-defined setup result.

Source code in agentflow/store/base_store.py

async def asetup(self) -> Any:
    """
    Asynchronous setup method for checkpointer.

    Returns:
        Any: Implementation-defined setup result.
    """
    raise NotImplementedError

astore abstractmethod async

astore(config, content, memory_type=MemoryType.EPISODIC, category='general', metadata=None, **kwargs)

Add a new memory.

Parameters:

Name	Type	Description	Default
content	str | Message	The memory content	required
user_id		User identifier	required
agent_id		Agent identifier	required
memory_type	MemoryType	Type of memory (episodic, semantic, etc.)	EPISODIC
category	str	Memory category for organization	'general'
metadata	dict[str, Any] | None	Additional metadata	None
**kwargs		Store-specific parameters	{}

Returns:

Type	Description
str	Memory ID

Source code in agentflow/store/base_store.py

@abstractmethod
async def astore(
    self,
    config: dict[str, Any],
    content: str | Message,
    memory_type: MemoryType = MemoryType.EPISODIC,
    category: str = "general",
    metadata: dict[str, Any] | None = None,
    **kwargs,
) -> str:
    """
    Add a new memory.

    Args:
        content: The memory content
        user_id: User identifier
        agent_id: Agent identifier
        memory_type: Type of memory (episodic, semantic, etc.)
        category: Memory category for organization
        metadata: Additional metadata
        **kwargs: Store-specific parameters

    Returns:
        Memory ID
    """
    raise NotImplementedError

aupdate abstractmethod async

aupdate(config, memory_id, content, metadata=None, **kwargs)

Update an existing memory.

Parameters:

Name	Type	Description	Default
memory_id	str	ID of memory to update	required
content	str | Message	New content (optional)	required
metadata	dict[str, Any] | None	New/additional metadata (optional)	None
**kwargs		Store-specific parameters	{}

Source code in agentflow/store/base_store.py

@abstractmethod
async def aupdate(
    self,
    config: dict[str, Any],
    memory_id: str,
    content: str | Message,
    metadata: dict[str, Any] | None = None,
    **kwargs,
) -> Any:
    """
    Update an existing memory.

    Args:
        memory_id: ID of memory to update
        content: New content (optional)
        metadata: New/additional metadata (optional)
        **kwargs: Store-specific parameters
    """
    raise NotImplementedError

delete

delete(config, memory_id, **kwargs)

Synchronous wrapper for adelete that runs the async implementation.

Source code in agentflow/store/base_store.py

def delete(self, config: dict[str, Any], memory_id: str, **kwargs) -> None:
    """Synchronous wrapper for `adelete` that runs the async implementation."""
    return run_coroutine(self.adelete(config, memory_id, **kwargs))

forget_memory

forget_memory(config, **kwargs)

Delete a memory by for a user or agent.

Source code in agentflow/store/base_store.py

def forget_memory(
    self,
    config: dict[str, Any],
    **kwargs,
) -> Any:
    """Delete a memory by for a user or agent."""
    return run_coroutine(self.aforget_memory(config, **kwargs))

get

get(config, memory_id, **kwargs)

Synchronous wrapper for aget that runs the async implementation.

Source code in agentflow/store/base_store.py

def get(self, config: dict[str, Any], memory_id: str, **kwargs) -> MemorySearchResult | None:
    """Synchronous wrapper for `aget` that runs the async implementation."""
    return run_coroutine(self.aget(config, memory_id, **kwargs))

get_all

get_all(config, limit=100, **kwargs)

Synchronous wrapper for aget that runs the async implementation.

Source code in agentflow/store/base_store.py

def get_all(
    self,
    config: dict[str, Any],
    limit: int = 100,
    **kwargs,
) -> list[MemorySearchResult]:
    """Synchronous wrapper for `aget` that runs the async implementation."""
    return run_coroutine(self.aget_all(config, limit=limit, **kwargs))

release

release()

Clean up any resources used by the store (override in subclasses if needed).

Source code in agentflow/store/base_store.py

def release(self) -> None:
    """Clean up any resources used by the store (override in subclasses if needed)."""
    return run_coroutine(self.arelease())

search

search(config, query, memory_type=None, category=None, limit=10, score_threshold=None, filters=None, retrieval_strategy=RetrievalStrategy.SIMILARITY, distance_metric=DistanceMetric.COSINE, max_tokens=4000, **kwargs)

Synchronous wrapper for asearch that runs the async implementation.

Source code in agentflow/store/base_store.py

def search(
    self,
    config: dict[str, Any],
    query: str,
    memory_type: MemoryType | None = None,
    category: str | None = None,
    limit: int = 10,
    score_threshold: float | None = None,
    filters: dict[str, Any] | None = None,
    retrieval_strategy: RetrievalStrategy = RetrievalStrategy.SIMILARITY,
    distance_metric: DistanceMetric = DistanceMetric.COSINE,
    max_tokens: int = 4000,
    **kwargs,
) -> list[MemorySearchResult]:
    """Synchronous wrapper for `asearch` that runs the async implementation."""
    return run_coroutine(
        self.asearch(
            config,
            query,
            memory_type=memory_type,
            category=category,
            limit=limit,
            score_threshold=score_threshold,
            filters=filters,
            retrieval_strategy=retrieval_strategy,
            distance_metric=distance_metric,
            max_tokens=max_tokens,
            **kwargs,
        )
    )

setup

setup()

Synchronous setup method for checkpointer.

Returns:

Name	Type	Description
Any	Any	Implementation-defined setup result.

Source code in agentflow/store/base_store.py

def setup(self) -> Any:
    """
    Synchronous setup method for checkpointer.

    Returns:
        Any: Implementation-defined setup result.
    """
    return run_coroutine(self.asetup())

store

store(config, content, memory_type=MemoryType.EPISODIC, category='general', metadata=None, **kwargs)

Synchronous wrapper for astore that runs the async implementation.

Source code in agentflow/store/base_store.py

def store(
    self,
    config: dict[str, Any],
    content: str | Message,
    memory_type: MemoryType = MemoryType.EPISODIC,
    category: str = "general",
    metadata: dict[str, Any] | None = None,
    **kwargs,
) -> str:
    """Synchronous wrapper for `astore` that runs the async implementation."""
    return run_coroutine(
        self.astore(
            config,
            content,
            memory_type=memory_type,
            category=category,
            metadata=metadata,
            **kwargs,
        )
    )

update

update(config, memory_id, content, metadata=None, **kwargs)

Synchronous wrapper for aupdate that runs the async implementation.

Source code in agentflow/store/base_store.py

def update(
    self,
    config: dict[str, Any],
    memory_id: str,
    content: str | Message,
    metadata: dict[str, Any] | None = None,
    **kwargs,
) -> Any:
    """Synchronous wrapper for `aupdate` that runs the async implementation."""
    return run_coroutine(self.aupdate(config, memory_id, content, metadata=metadata, **kwargs))

Functions