Retrievers

BaseRetriever ABC

All retrievers implement BaseRetriever. The retrieve() method returns raw results; query() and query_with_sources() also run the generator.

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

class BaseRetriever(ABC):
    @abstractmethod
    def retrieve(self, query: str, top_k: int = 10) -> list[RetrievalResult]:
        """Return ranked list of relevant chunks — no answer generation."""
        ...

    def query(self, question: str, top_k: int = 10) -> str:
        """Retrieve context + generate a natural-language answer."""
        ...

    def query_with_sources(
        self, question: str, top_k: int = 10
    ) -> dict:
        """
        Returns:
            {
              "answer":  str,
              "sources": list[dict],          # doc_id, page, score
              "chunks":  list[RetrievalResult]
            }
        """
        ...

from cognity_ai.models import RetrievalResult

@dataclass
class RetrievalResult:
    chunk_id: str
    doc_id: str
    text: str
    score: float             # relevance score (higher = more relevant)
    metadata: dict           # page_number, section, source_path, etc.
    retrieval_channel: str   # "vector" | "graph" | "community" | "bridge"

Overview Table

Methodologies

# Must pair chunker + retriever
rag = RAGLibrary(chunker="parent_child", rag_method="parent_child")

Auto-Fallback Logic

cognity-ai automatically degrades gracefully when required backends are unavailable:

Per-Query Method Override

The default rag_method can be overridden for any individual query without changing the pipeline configuration:

from cognity_ai import RAGLibrary

rag = RAGLibrary(rag_method="hybrid_graph")   # default

# Override for a specific query
answer = rag.query("List all companies mentioned", method="graph_only")
summary = rag.query("Give a broad thematic overview", method="microsoft_graphrag")
precise = rag.query("What is the boiling point of water?", method="naive")
detailed = rag.query_with_sources("Who founded Anthropic?", method="hybrid_graph")

# Retrieve without generating an answer
chunks = rag.retrieve("transformer architecture", top_k=5, method="multi_query")
for chunk in chunks:
    print(f"[{chunk.score:.3f}] {chunk.text[:120]}...")

Direct Usage

Instantiate and use a retriever directly without RAGLibrary:

from cognity_ai.retrievers import HybridGraphRetriever, NaiveRetriever
from cognity_ai.stores.vector import ChromaStore
from cognity_ai.stores.graph import Neo4jStore
from cognity_ai.embedders import GeminiEmbedder
from cognity_ai.generators import GeminiGenerator

embedder = GeminiEmbedder(api_key="AIza...")
vector_store = ChromaStore(persist_directory=".chroma")
graph_store = Neo4jStore(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password",
)

# 4-channel hybrid retrieval
retriever = HybridGraphRetriever(
    vector_store=vector_store,
    graph_store=graph_store,
    embedder=embedder,
    generator=GeminiGenerator(api_key="AIza..."),
    top_k_per_channel=10,
    rrf_k=60,
    graph_hops=2,
)

# Retrieve only
results = retriever.retrieve("Who founded the company?")
for r in results:
    print(f"[{r.retrieval_channel}] score={r.score:.3f}: {r.text[:80]}")

# Retrieve + generate answer
answer = retriever.query("What are the main products?")

# Retrieve + answer + sources
result = retriever.query_with_sources("Summarize the Q3 results")
print(result["answer"])
for src in result["sources"]:
    print(f"  {src['doc_id']} p.{src['page']} (score={src['score']:.3f})")

# Naive retriever (no graph needed)
naive = NaiveRetriever(
    vector_store=vector_store,
    embedder=embedder,
    generator=GeminiGenerator(api_key="AIza..."),
)
results = naive.retrieve("climate change impacts", top_k=5)