> ## Documentation Index
> Fetch the complete documentation index at: https://fairforge.alquimia.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Retriever

> Loading conversation data into Fair Forge with full-dataset and streaming modes

# 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

```python theme={null}
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]:
        ...
```

<Note>
  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.
</Note>

## Full Dataset Mode

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

### Loading from JSON

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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.

```python theme={null}
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:

```json theme={null}
[
  {
    "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

<AccordionGroup>
  <Accordion title="Use explicit constructor parameters instead of kwargs.get()">
    Declare parameters explicitly for clarity and IDE support:

    ```python theme={null}
    # 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
    ```
  </Accordion>

  <Accordion title="Handle missing fields with Pydantic defaults">
    Use `model_validate` to leverage Pydantic's validation and defaults:

    ```python theme={null}
    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.
  </Accordion>

  <Accordion title="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](/core-concepts/retriever-streaming) page.
  </Accordion>
</AccordionGroup>

## Next Steps

<CardGroup cols={2}>
  <Card title="Streaming Retrievers" icon="arrow-right-arrow-left" href="/core-concepts/retriever-streaming">
    Handle large datasets with stream\_sessions and stream\_batches
  </Card>

  <Card title="Dataset & Batch" icon="table" href="/core-concepts/dataset-batch">
    Understand the data structures used by all retrievers
  </Card>
</CardGroup>