Base KnowledgeStore

Base Knowledge Store.

BaseKnowledgeStore

Bases: BaseModel, ABC

Base Knowledge Store Class.

This class represent the base knowledge store component of a RAG system.

Attributes:

Name	Type	Description
name	str	The name of knowledge store.

Source code in src/fed_rag/base/knowledge_store.py

class BaseKnowledgeStore(BaseModel, ABC):
    """Base Knowledge Store Class.

    This class represent the base knowledge store component of a RAG system.

    Attributes:
        name: The name of knowledge store.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)
    name: str = Field(
        description="Name of Knowledge Store used for caching and loading.",
        default=DEFAULT_KNOWLEDGE_STORE_NAME,
    )

    @abstractmethod
    def load_node(self, node: "KnowledgeNode") -> None:
        """Load a "KnowledgeNode" into the KnowledgeStore.

        Args:
            node (KnowledgeNode): The node to load to the knowledge store.
        """

    @abstractmethod
    def load_nodes(self, nodes: list["KnowledgeNode"]) -> None:
        """Load multiple "KnowledgeNode"s in batch.

        Args:
            nodes (list[KnowledgeNode]): The nodes to load.
        """

    @abstractmethod
    def retrieve(
        self, query_emb: list[float], top_k: int
    ) -> list[tuple[float, "KnowledgeNode"]]:
        """Retrieve top-k nodes from KnowledgeStore against a provided user query.

        Args:
            query_emb (list[float]): the query represented as an encoded vector.
            top_k (int): the number of knowledge nodes to retrieve.

        Returns:
            A list of tuples where the first element represents the similarity score
            of the node to the query, and the second element is the node itself.
        """

    @abstractmethod
    def batch_retrieve(
        self, query_embs: list[list[float]], top_k: int
    ) -> list[list[tuple[float, "KnowledgeNode"]]]:
        """Batch retrieve top-k nodes from KnowledgeStore against provided user queries.

        Args:
            query_embs (list[list[float]]): the list of encoded queries.
            top_k (int): the number of knowledge nodes to retrieve.

        Returns:
            A list of list of tuples where the first element represents the similarity score
            of the node to the query, and the second element is the node itself.
        """

    @abstractmethod
    def delete_node(self, node_id: str) -> bool:
        """Remove a node from the KnowledgeStore by ID, returning success status.

        Args:
            node_id (str): The id of the node to delete.

        Returns:
            bool: Whether or not the node was successfully deleted.
        """

    @abstractmethod
    def clear(self) -> None:
        """Clear all nodes from the KnowledgeStore."""

    @property
    @abstractmethod
    def count(self) -> int:
        """Return the number of nodes in the store."""

    @abstractmethod
    def persist(self) -> None:
        """Save the KnowledgeStore nodes to a permanent storage."""

    @abstractmethod
    def load(self) -> None:
        """Load the KnowledgeStore nodes from a permanent storage using `name`."""

count abstractmethod property

count

Return the number of nodes in the store.

load_node abstractmethod

load_node(node)

Load a "KnowledgeNode" into the KnowledgeStore.

Parameters:

Name	Type	Description	Default
node	KnowledgeNode	The node to load to the knowledge store.	required

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def load_node(self, node: "KnowledgeNode") -> None:
    """Load a "KnowledgeNode" into the KnowledgeStore.

    Args:
        node (KnowledgeNode): The node to load to the knowledge store.
    """

load_nodes abstractmethod

load_nodes(nodes)

Load multiple "KnowledgeNode"s in batch.

Parameters:

Name	Type	Description	Default
nodes	list[KnowledgeNode]	The nodes to load.	required

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def load_nodes(self, nodes: list["KnowledgeNode"]) -> None:
    """Load multiple "KnowledgeNode"s in batch.

    Args:
        nodes (list[KnowledgeNode]): The nodes to load.
    """

retrieve abstractmethod

retrieve(query_emb, top_k)

Retrieve top-k nodes from KnowledgeStore against a provided user query.

Parameters:

Name	Type	Description	Default
query_emb	list[float]	the query represented as an encoded vector.	required
top_k	int	the number of knowledge nodes to retrieve.	required

Returns:

Type	Description
list[tuple[float, KnowledgeNode]]	A list of tuples where the first element represents the similarity score
list[tuple[float, KnowledgeNode]]	of the node to the query, and the second element is the node itself.

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def retrieve(
    self, query_emb: list[float], top_k: int
) -> list[tuple[float, "KnowledgeNode"]]:
    """Retrieve top-k nodes from KnowledgeStore against a provided user query.

    Args:
        query_emb (list[float]): the query represented as an encoded vector.
        top_k (int): the number of knowledge nodes to retrieve.

    Returns:
        A list of tuples where the first element represents the similarity score
        of the node to the query, and the second element is the node itself.
    """

batch_retrieve abstractmethod

batch_retrieve(query_embs, top_k)

Batch retrieve top-k nodes from KnowledgeStore against provided user queries.

Parameters:

Name	Type	Description	Default
query_embs	list[list[float]]	the list of encoded queries.	required
top_k	int	the number of knowledge nodes to retrieve.	required

Returns:

Type	Description
list[list[tuple[float, KnowledgeNode]]]	A list of list of tuples where the first element represents the similarity score
list[list[tuple[float, KnowledgeNode]]]	of the node to the query, and the second element is the node itself.

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def batch_retrieve(
    self, query_embs: list[list[float]], top_k: int
) -> list[list[tuple[float, "KnowledgeNode"]]]:
    """Batch retrieve top-k nodes from KnowledgeStore against provided user queries.

    Args:
        query_embs (list[list[float]]): the list of encoded queries.
        top_k (int): the number of knowledge nodes to retrieve.

    Returns:
        A list of list of tuples where the first element represents the similarity score
        of the node to the query, and the second element is the node itself.
    """

delete_node abstractmethod

delete_node(node_id)

Remove a node from the KnowledgeStore by ID, returning success status.

Parameters:

Name	Type	Description	Default
node_id	str	The id of the node to delete.	required

Returns:

Name	Type	Description
bool	bool	Whether or not the node was successfully deleted.

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def delete_node(self, node_id: str) -> bool:
    """Remove a node from the KnowledgeStore by ID, returning success status.

    Args:
        node_id (str): The id of the node to delete.

    Returns:
        bool: Whether or not the node was successfully deleted.
    """

clear abstractmethod

clear()

Clear all nodes from the KnowledgeStore.

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def clear(self) -> None:
    """Clear all nodes from the KnowledgeStore."""

persist abstractmethod

persist()

Save the KnowledgeStore nodes to a permanent storage.

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def persist(self) -> None:
    """Save the KnowledgeStore nodes to a permanent storage."""

load abstractmethod

load()

Load the KnowledgeStore nodes from a permanent storage using name.

Source code in src/fed_rag/base/knowledge_store.py

@abstractmethod
def load(self) -> None:
    """Load the KnowledgeStore nodes from a permanent storage using `name`."""