Retriever

The Retriever class is the entry point for loading your conversation data into Fair Forge. Every evaluation requires a custom retriever that implements load_dataset(). Fair Forge supports three iteration strategies that control how data is consumed — from loading everything upfront to yielding individual QA pairs on demand.

Iteration Modes

Choose the mode that fits your dataset size and processing requirements:

Mode	Value	Returns	Use When
`FULL_DATASET`	`"full_dataset"`	`list[Dataset]`	Default. Dataset fits comfortably in memory
`STREAM_SESSIONS`	`"stream_sessions"`	`Iterator[Dataset]`	Many sessions, avoid loading all at once
`STREAM_BATCHES`	`"stream_batches"`	`Iterator[StreamedBatch]`	QA pairs arrive independently (e.g. from a queue)

The default mode is FULL_DATASET. To use streaming, override the iteration_level property.

Interface

from abc import ABC, abstractmethod
from collections.abc import Iterator
from fair_forge.schemas.common import Dataset, IterationLevel, StreamedBatch

class Retriever(ABC):
    def __init__(self, **kwargs):
        self.kwargs = kwargs

    @property
    def iteration_level(self) -> IterationLevel:
        return IterationLevel.FULL_DATASET  # default

    @abstractmethod
    def load_dataset(self) -> list[Dataset] | Iterator[Dataset] | Iterator[StreamedBatch]:
        ...

The return type of load_dataset() must be consistent with iteration_level. Returning an Iterator with the default FULL_DATASET level will raise a ValueError at runtime.

Full Dataset Mode

The simplest and most common mode. Load all sessions into a list and return it.

Loading from JSON

import json
from pathlib import Path
from fair_forge.core.retriever import Retriever
from fair_forge.schemas.common import Dataset

class JSONRetriever(Retriever):
    def __init__(self, file_path: str = "data.json", **kwargs):
        super().__init__(**kwargs)
        self.file_path = Path(file_path)

    def load_dataset(self) -> list[Dataset]:
        with open(self.file_path) as f:
            return [Dataset.model_validate(item) for item in json.load(f)]

# Usage — kwargs are forwarded to the retriever constructor
metrics = Toxicity.run(
    JSONRetriever,
    file_path="conversations.json",
    group_prototypes={...},
)

Loading from a Database

import sqlite3
from fair_forge.core.retriever import Retriever
from fair_forge.schemas.common import Dataset, Batch

class DatabaseRetriever(Retriever):
    def __init__(self, db_path: str, assistant_id: str, **kwargs):
        super().__init__(**kwargs)
        self.db_path = db_path
        self.assistant_id = assistant_id

    def load_dataset(self) -> list[Dataset]:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute(
            "SELECT DISTINCT session_id, context, language FROM conversations WHERE assistant_id = ?",
            (self.assistant_id,),
        )

        datasets = []
        for session_id, context, language in cursor.fetchall():
            cursor.execute(
                "SELECT qa_id, query, assistant, ground_truth FROM conversations WHERE session_id = ? ORDER BY created_at",
                (session_id,),
            )
            batches = [
                Batch(qa_id=row[0], query=row[1], assistant=row[2], ground_truth_assistant=row[3])
                for row in cursor.fetchall()
            ]
            datasets.append(Dataset(
                session_id=session_id,
                assistant_id=self.assistant_id,
                language=language,
                context=context,
                conversation=batches,
            ))

        conn.close()
        return datasets

Loading from an API

import httpx
from fair_forge.core.retriever import Retriever
from fair_forge.schemas.common import Dataset

class APIRetriever(Retriever):
    def __init__(self, api_url: str, api_key: str, **kwargs):
        super().__init__(**kwargs)
        self.api_url = api_url
        self.api_key = api_key

    def load_dataset(self) -> list[Dataset]:
        with httpx.Client() as client:
            response = client.get(
                f"{self.api_url}/conversations",
                headers={"Authorization": f"Bearer {self.api_key}"},
            )
        return [Dataset.model_validate(d) for d in response.json()["datasets"]]

Loading from CSV

import csv
from collections import defaultdict
from fair_forge.core.retriever import Retriever
from fair_forge.schemas.common import Dataset, Batch

class CSVRetriever(Retriever):
    def __init__(self, file_path: str, **kwargs):
        super().__init__(**kwargs)
        self.file_path = file_path

    def load_dataset(self) -> list[Dataset]:
        sessions: dict[str, list] = defaultdict(list)
        with open(self.file_path, newline="") as f:
            for row in csv.DictReader(f):
                sessions[row["session_id"]].append(row)

        datasets = []
        for session_id, rows in sessions.items():
            batches = [
                Batch(
                    qa_id=row["qa_id"],
                    query=row["query"],
                    assistant=row["assistant"],
                    ground_truth_assistant=row.get("ground_truth", ""),
                )
                for row in rows
            ]
            datasets.append(Dataset(
                session_id=session_id,
                assistant_id=rows[0].get("assistant_id", "unknown"),
                language=rows[0].get("language", "english"),
                context=rows[0].get("context", ""),
                conversation=batches,
            ))

        return datasets

Multi-Assistant Retriever (BestOf)

BestOf expects one Dataset per assistant, all answering the same questions. The retriever loads them — BestOf handles the tournament logic automatically.

import json
from pathlib import Path
from fair_forge.core.retriever import Retriever
from fair_forge.schemas.common import Dataset

class MultiAssistantRetriever(Retriever):
    """Each Dataset in the file represents one assistant answering the same questions."""

    def __init__(self, file_path: str = "dataset_bestof.json", **kwargs):
        super().__init__(**kwargs)
        self.file_path = Path(file_path)

    def load_dataset(self) -> list[Dataset]:
        with open(self.file_path) as f:
            return [Dataset.model_validate(item) for item in json.load(f)]

# Usage — BestOf only needs model and criteria; the retriever provides the assistants
metrics = BestOf.run(
    MultiAssistantRetriever,
    model=judge_model,
    use_structured_output=True,
    criteria="Overall response quality, helpfulness, and clarity",
)

The JSON file must contain one entry per assistant, all sharing the same qa_id values so BestOf can pair their responses:

[
  {
    "session_id": "eval-session",
    "assistant_id": "assistant_alpha",
    "language": "english",
    "context": "",
    "conversation": [
      {"qa_id": "q1", "query": "What are the benefits of renewable energy?", "assistant": "...", "ground_truth_assistant": ""}
    ]
  },
  {
    "session_id": "eval-session",
    "assistant_id": "assistant_beta",
    "language": "english",
    "context": "",
    "conversation": [
      {"qa_id": "q1", "query": "What are the benefits of renewable energy?", "assistant": "...", "ground_truth_assistant": ""}
    ]
  }
]

Best Practices

Use explicit constructor parameters instead of kwargs.get()

Declare parameters explicitly for clarity and IDE support:

# Preferred
class MyRetriever(Retriever):
    def __init__(self, file_path: str, assistant_id: str = "default", **kwargs):
        super().__init__(**kwargs)
        self.file_path = file_path
        self.assistant_id = assistant_id

Handle missing fields with Pydantic defaults

Use model_validate to leverage Pydantic’s validation and defaults:

def load_dataset(self) -> list[Dataset]:
    with open(self.file_path) as f:
        return [Dataset.model_validate(item) for item in json.load(f)]

Fields with defaults (language="english") are filled in automatically.

Use streaming for large datasets

If your dataset has thousands of sessions or QA pairs, prefer streaming to avoid loading everything into memory. See the Streaming page.

Getting Started

Core Concepts

Metrics

Generators

Runners

Prompt Optimizer

Explainability

Retriever

Retriever

Iteration Modes

Interface

Full Dataset Mode

Loading from JSON

Loading from a Database

Loading from an API

Loading from CSV

Multi-Assistant Retriever (BestOf)

Best Practices

Next Steps

Streaming Retrievers

Dataset & Batch

​Retriever

​Iteration Modes

​Interface

​Full Dataset Mode

​Loading from JSON

​Loading from a Database

​Loading from an API

​Loading from CSV

​Multi-Assistant Retriever (BestOf)

​Best Practices

​Next Steps

Streaming Retrievers

Dataset & Batch

Retriever

Iteration Modes

Interface

Full Dataset Mode

Loading from JSON

Loading from a Database

Loading from an API

Loading from CSV

Multi-Assistant Retriever (BestOf)

Best Practices

Next Steps