Stores

BaseVectorStore ABC

• def upsert_chunks(self, chunks: list[SemanticChunk]) -> None abstract
  Insert or update a batch of semantic chunks in the vector store. Uses chunk ID as the upsert key.
• def query_chunks(self, embedding: list[float], top_k: int, filters: dict | None = None) -> list[RetrievalResult] abstract
  Nearest-neighbour search over chunk embeddings. Accepts optional metadata filters (provider-specific syntax).
• def query_by_chunk_ids(self, chunk_ids: list[str]) -> list[RetrievalResult] abstract
  Fetch chunks by their explicit IDs. Used by the bridge channel in HybridGraphRetriever.
• def upsert_community(self, community: CommunityInfo) -> None abstract
  Store a community summary embedding alongside its metadata. Called after detect_communities().
• def query_communities(self, embedding: list[float], top_k: int) -> list[RetrievalResult] abstract
  Semantic search over community summaries. Drives CH3 (community channel) in HybridGraphRetriever.
• def delete_by_doc_id(self, doc_id: str) -> None abstract
  Remove all chunks associated with doc_id. Used during document deprecation and replacement.
• def count(self) -> int abstract
  Return the total number of chunk vectors currently stored.

from abc import ABC, abstractmethod
from cognity_ai.models import SemanticChunk, RetrievalResult, CommunityInfo

class BaseVectorStore(ABC):
    @abstractmethod
    def upsert_chunks(self, chunks: list[SemanticChunk]) -> None: ...
    @abstractmethod
    def query_chunks(
        self, embedding: list[float], top_k: int, filters: dict | None = None
    ) -> list[RetrievalResult]: ...
    @abstractmethod
    def query_by_chunk_ids(self, chunk_ids: list[str]) -> list[RetrievalResult]: ...
    @abstractmethod
    def upsert_community(self, community: CommunityInfo) -> None: ...
    @abstractmethod
    def query_communities(
        self, embedding: list[float], top_k: int
    ) -> list[RetrievalResult]: ...
    @abstractmethod
    def delete_by_doc_id(self, doc_id: str) -> None: ...
    @abstractmethod
    def count(self) -> int: ...

Vector Store Providers

Select a provider by passing its key string or the concrete class. All providers implement BaseVectorStore in full unless noted.

BaseGraphStore ABC

• def upsert_entity(self, entity: Entity) -> None abstract
  Insert or update an entity node. Merges on entity name + type.
• def upsert_relation(self, relation: Relation) -> None abstract
  Insert or update a directed relation edge between two entities.
• def link_chunk_to_entities(self, chunk_id: str, doc_id: str, entity_names: list[str]) -> None abstract
  Create MENTIONED_IN edges from entity nodes to a chunk node. Enables the CH4 bridge channel.
• def retrieve_subgraph(self, entity_names: list[str], hops: int = 2, limit: int = 50) -> list[RetrievalResult] abstract
  Traverse the graph from seed entities up to hops steps, returning scored paths as retrieval results.
• def global_community_search(self, top_n: int = 10) -> list[RetrievalResult] abstract
  Return the top-N community summaries ranked by size and centrality. Used for broad thematic queries.
• def detect_communities(self) -> list[dict] abstract
  Run community detection (Leiden algorithm in Neo4j/GDS). Returns list of community dicts with members and summaries.
• def confirm_source(self, doc_id: str) -> None abstract
  Mark a document source as confirmed, promoting its entities and relations to full confidence.
• def deprecate_source(self, doc_id: str) -> None abstract
  Mark a source as deprecated. Associated entities reduce confidence; relations are flagged for pruning.
• def prune_low_confidence(self, threshold: float) -> int abstract
  Delete all entities and relations with confidence below threshold. Returns the count of deleted nodes/edges.
• def health_report(self) -> dict abstract
  Return a diagnostic dict with counts of nodes, edges, communities, and store connectivity status.

from abc import ABC, abstractmethod
from cognity_ai.models import Entity, Relation, RetrievalResult

class BaseGraphStore(ABC):
    @abstractmethod
    def upsert_entity(self, entity: Entity) -> None: ...
    @abstractmethod
    def upsert_relation(self, relation: Relation) -> None: ...
    @abstractmethod
    def link_chunk_to_entities(
        self, chunk_id: str, doc_id: str, entity_names: list[str]
    ) -> None: ...
    @abstractmethod
    def retrieve_subgraph(
        self, entity_names: list[str], hops: int = 2, limit: int = 50
    ) -> list[RetrievalResult]: ...
    @abstractmethod
    def global_community_search(self, top_n: int = 10) -> list[RetrievalResult]: ...
    @abstractmethod
    def detect_communities(self) -> list[dict]: ...
    @abstractmethod
    def confirm_source(self, doc_id: str) -> None: ...
    @abstractmethod
    def deprecate_source(self, doc_id: str) -> None: ...
    @abstractmethod
    def prune_low_confidence(self, threshold: float) -> int: ...
    @abstractmethod
    def health_report(self) -> dict: ...

Graph Store Providers

All providers implement BaseGraphStore. Community detection capability varies by backend.