BaseRetriever Guide

A minimal guide to how BaseRetriever’s interface works, taught by building a tiny, fully offline retriever. No database, network connection, embeddings, or LLM required.

Before You Start

Have cognee installed (see Installation) — needed only for the BaseRetriever import; no LLM provider or database configuration is required for this example.
No prior OOP knowledge is assumed — the concepts you need (classes, inheritance, abstract methods) are explained at the end of the page.

Code in Action

import asyncio
import json
from pathlib import Path
from typing import Any, Optional

from cognee.modules.retrieval.base_retriever import BaseRetriever


class JsonToyRetriever(BaseRetriever):
    """A minimal offline retriever."""

    def __init__(self, output_path: str | Path = "toy_query.json"):
        self.output_path = Path(output_path)

    async def get_retrieved_objects(
        self,
        query: Optional[str] = None,
        query_batch: Optional[list[str]] = None,
    ) -> dict[str, Any]:
        query = query or ""

        data = {
            "query": query,
            "character_count": len(query),
            "word_count": len(query.split()),
        }

        with self.output_path.open("w", encoding="utf-8") as f:
            json.dump(
                data,
                f,
                indent=2,
                ensure_ascii=False,
                sort_keys=True,
            )

        return data

    async def get_context_from_objects(
        self,
        query: Optional[str] = None,
        query_batch: Optional[list[str]] = None,
        retrieved_objects: Any = None,
    ) -> str:
        return json.dumps(
            retrieved_objects,
            indent=2,
            ensure_ascii=False,
            sort_keys=True,
        )

    async def get_completion_from_context(
        self,
        query: Optional[str] = None,
        query_batch: Optional[list[str]] = None,
        retrieved_objects: Any = None,
        context: Any = None,
    ) -> list[str]:
        return [
            (
                "Toy completion: "
                f"query={retrieved_objects['query']!r}; "
                f"characters={retrieved_objects['character_count']}; "
                f"words={retrieved_objects['word_count']}."
            )
        ]


async def main():
    retriever = JsonToyRetriever()
    query = "How does BaseRetriever work?"

    result = await retriever.get_completion(query)
    print(result[0])

    # Call the same three stages manually, to prove get_completion() just orchestrates them
    objects = await retriever.get_retrieved_objects(query)
    context = await retriever.get_context_from_objects(query, retrieved_objects=objects)
    manual_result = await retriever.get_completion_from_context(
        query, retrieved_objects=objects, context=context
    )

    assert manual_result == result
    print("Manual orchestration matches get_completion():", manual_result[0])


if __name__ == "__main__":
    asyncio.run(main())

What Just Happened

The three methods below — get_retrieved_objects, get_context_from_objects, and get_completion_from_context — are called, in order, by get_completion(), each explained one step at a time below. JsonToyRetriever inherits get_completion() from BaseRetriever rather than defining it.

Step 1: Retrieve Objects

async def get_retrieved_objects(
    self,
    query: Optional[str] = None,
    query_batch: Optional[list[str]] = None,
) -> dict[str, Any]:
    query = query or ""

    data = {
        "query": query,
        "character_count": len(query),
        "word_count": len(query.split()),
    }

    with self.output_path.open("w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False, sort_keys=True)

    return data

get_retrieved_objects is the first of the three methods BaseRetriever requires every subclass to implement (get_retrieved_objects, get_context_from_objects, get_completion_from_context). It turns the raw query into a small dictionary and writes it to a JSON file. A production retriever would query a vector database, a graph database, or another storage backend here instead of building a toy dictionary.

Step 2: Build Context

async def get_context_from_objects(
    self,
    query: Optional[str] = None,
    query_batch: Optional[list[str]] = None,
    retrieved_objects: Any = None,
) -> str:
    return json.dumps(retrieved_objects, indent=2, ensure_ascii=False, sort_keys=True)

get_context_from_objects, the second stage, turns retrieved objects into the context that gets handed to the completion step. Here it’s just the same dictionary formatted as JSON text; a real retriever might concatenate document chunks or format graph relationships instead.

Step 3: Produce the Completion

async def get_completion_from_context(
    self,
    query: Optional[str] = None,
    query_batch: Optional[list[str]] = None,
    retrieved_objects: Any = None,
    context: Any = None,
) -> list[str]:
    return [
        (
            "Toy completion: "
            f"query={retrieved_objects['query']!r}; "
            f"characters={retrieved_objects['character_count']}; "
            f"words={retrieved_objects['word_count']}."
        )
    ]

get_completion_from_context, the third stage, would normally call an LLM with the context built in Step 2. This toy version returns a deterministic string instead, so the whole example stays offline and reproducible.

Step 4: Run the Retriever

async def main():
    retriever = JsonToyRetriever()
    query = "How does BaseRetriever work?"

    result = await retriever.get_completion(query)
    print(result[0])

get_completion() is not implemented by JsonToyRetriever — it’s inherited from BaseRetriever. get_completion() calls the three stages above in order, passing each stage’s return value into the next, and returns the final result: the base class defines the workflow, while the subclass only defines the behavior of each step. Running this prints:

Toy completion: query='How does BaseRetriever work?'; characters=28; words=4.

and creates a toy_query.json file alongside it.

Step 5: Prove the Orchestration Manually

objects = await retriever.get_retrieved_objects(query)
context = await retriever.get_context_from_objects(query, retrieved_objects=objects)
manual_result = await retriever.get_completion_from_context(
    query, retrieved_objects=objects, context=context
)

assert manual_result == result
print("Manual orchestration matches get_completion():", manual_result[0])

To prove that’s really what’s happening, calling the three stages manually — get_retrieved_objects, then get_context_from_objects, then get_completion_from_context, passing each result into the next — produces the exact same output as get_completion(). The assert above never fails: it’s the same three calls either way, just written out instead of hidden inside the inherited method. Running this prints an extra line:

Manual orchestration matches get_completion(): Toy completion: query='How does BaseRetriever work?'; characters=28; words=4.

OOP Concepts You Need

Classes, Inheritance, and Abstract Methods

A class is a blueprint describing what an object can do; an instance is one actual object created from it.One class can inherit from another, gaining its methods for free, and a subclass can override an inherited method with its own implementation:

class Animal:
    def speak(self):
        return "..."

class Dog(Animal):
    def speak(self):
        return "woof"

Sometimes a base class needs to require a method without providing it — an abstract method: declared on the base class, but with no implementation there, forcing every subclass to supply its own. Python’s abc module (Abstract Base Classes) provides this via the @abstractmethod decorator:

from abc import ABC, abstractmethod

class Animal(ABC):
    @abstractmethod
    def speak(self):
        pass

If a subclass forgets to implement speak, Python raises an error before the object can even be created. See Python’s abc module docs for the full mechanism.BaseRetriever works the same way: it declares get_retrieved_objects, get_context_from_objects, and get_completion_from_context as abstract methods, so calling BaseRetriever() directly raises a TypeError unless all three methods are implemented.

Search

See how built-in retrievers map to search types, and how to register a custom one

Search Basics

Run your first real Cognee search once you’re ready to move past this toy example

Getting Started

Core Concepts

Setup Configuration

Guides

Examples

CLI

Rust SDK

TypeScript SDK

OSS

Before You Start

Code in Action

What Just Happened

Step 1: Retrieve Objects

Step 2: Build Context

Step 3: Produce the Completion

Step 4: Run the Retriever

Step 5: Prove the Orchestration Manually

OOP Concepts You Need

Search

Search Basics

​Before You Start

​Code in Action

​What Just Happened

​Step 1: Retrieve Objects

​Step 2: Build Context

​Step 3: Produce the Completion

​Step 4: Run the Retriever

​Step 5: Prove the Orchestration Manually

​OOP Concepts You Need

Search

Search Basics

Before You Start

Code in Action

What Just Happened

Step 1: Retrieve Objects

Step 2: Build Context

Step 3: Produce the Completion

Step 4: Run the Retriever

Step 5: Prove the Orchestration Manually

OOP Concepts You Need