Store

The Store is a fundamental abstraction in Agentswarm that decouples state persistence from agent logic. It allows agents to share data (like task results, intermediate variables, or long-term memory) without knowing how or where that data is stored.

The Abstract Concept

The Store class defines a simple key-value interface. By designing your agents to rely on this interface, you make them portable and adaptable to different environments.

• Abstraction: Your agent simply calls store.set("key", value) or store.get("key").
• Flexibility: In a local script, this might just write to a Python dictionary. In a production cloud deployment, the underlying implementation could speak to Redis, a SQL database, or a cloud bucket.

agentswarm.datamodels.Store

Bases: ABC

The Store class defines a simple key/value API to access the store. The implementation can vary from a local dictionary, to a distributed remote store.

Source code in src/agentswarm/datamodels/store.py

class Store(ABC):
    """
    The Store class defines a simple key/value API to access the store.
    The implementation can vary from a local dictionary, to a distributed remote store.
    """

    @abstractmethod
    def get(self, key: str) -> any:
        """
        Obtains the value associated with the given key.
        """
        raise NotImplementedError

    @abstractmethod
    def set(self, key: str, value: any):
        """
        Sets the value associated with the given key.
        """
        raise NotImplementedError

    @abstractmethod
    def has(self, key: str) -> bool:
        """
        Checks if the store has a value associated with the given key.
        """
        raise NotImplementedError

    @abstractmethod
    def items(self) -> dict[str, any]:
        """
        Returns all key-value pairs in the store.
        (Primary used for nice tracing)
        """
        raise NotImplementedError

    @abstractmethod
    def to_dict(self) -> dict:
        """
        Returns the configuration needed to recreate the store.
        WARNING: Never share keys or other secret during this process.
        """
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def recreate(cls, config: dict) -> "Store":
        """
        Recreates a Store instance from a configuration dictionary.
        """
        raise NotImplementedError

get(key) abstractmethod

Obtains the value associated with the given key.

Source code in src/agentswarm/datamodels/store.py

@abstractmethod
def get(self, key: str) -> any:
    """
    Obtains the value associated with the given key.
    """
    raise NotImplementedError

has(key) abstractmethod

Checks if the store has a value associated with the given key.

Source code in src/agentswarm/datamodels/store.py

@abstractmethod
def has(self, key: str) -> bool:
    """
    Checks if the store has a value associated with the given key.
    """
    raise NotImplementedError

items() abstractmethod

Returns all key-value pairs in the store. (Primary used for nice tracing)

Source code in src/agentswarm/datamodels/store.py

@abstractmethod
def items(self) -> dict[str, any]:
    """
    Returns all key-value pairs in the store.
    (Primary used for nice tracing)
    """
    raise NotImplementedError

recreate(config) abstractmethod classmethod

Recreates a Store instance from a configuration dictionary.

Source code in src/agentswarm/datamodels/store.py

@classmethod
@abstractmethod
def recreate(cls, config: dict) -> "Store":
    """
    Recreates a Store instance from a configuration dictionary.
    """
    raise NotImplementedError

set(key, value) abstractmethod

Sets the value associated with the given key.

Source code in src/agentswarm/datamodels/store.py

@abstractmethod
def set(self, key: str, value: any):
    """
    Sets the value associated with the given key.
    """
    raise NotImplementedError

to_dict() abstractmethod

Returns the configuration needed to recreate the store. WARNING: Never share keys or other secret during this process.

Source code in src/agentswarm/datamodels/store.py

@abstractmethod
def to_dict(self) -> dict:
    """
    Returns the configuration needed to recreate the store.
    WARNING: Never share keys or other secret during this process.
    """
    raise NotImplementedError

Custom Implementations

You are encouraged to create your own Store implementations for your specific needs. For example, if you need persistence across reboots, you might implement a FileStore or a RedisStore.

Example: Creating a Redis Store

from agentswarm.datamodels import Store
import redis

class RedisStore(Store):
    def __init__(self, host='localhost', port=6379, db=0):
        self.r = redis.Redis(host=host, port=port, db=db)

    def get(self, key: str) -> any:
        return self.r.get(key)

    def set(self, key: str, value: any):
        self.r.set(key, value)

    def has(self, key: str) -> bool:
        return self.r.exists(key)

    def items(self) -> dict:
        # Implementation to fetch all keys...
        pass

Local Store

The library comes with a ready-to-use LocalStore which implements the interface using an in-memory dictionary. This is perfect for testing, scripts, and single-instance applications where persistence is not required.

agentswarm.datamodels.LocalStore

Bases: Store

The LocalStore class implements a simple key-value store in memory.

Source code in src/agentswarm/datamodels/local_store.py

class LocalStore(Store):
    """
    The LocalStore class implements a simple key-value store in memory.
    """

    def __init__(self):
        self.store = {}

    def get(self, key: str) -> any:
        return self.store[key]

    def set(self, key: str, value: any):
        self.store[key] = value

    def has(self, key: str) -> bool:
        return key in self.store

    def items(self) -> dict[str, any]:
        return self.store.copy()

    def __len__(self) -> int:
        return len(self.store)

    def to_dict(self) -> dict:
        from ..utils.exceptions import RemoteExecutionNotSupportedError

        raise RemoteExecutionNotSupportedError(
            "LocalStore cannot be serialized for remote execution."
        )

    @classmethod
    def recreate(cls, config: dict) -> "LocalStore":
        from ..utils.exceptions import RemoteExecutionNotSupportedError

        raise RemoteExecutionNotSupportedError(
            "LocalStore cannot be recreated from remote configuration."
        )