Custom data loaders in LlamaIndex: a how-to guide

Custom data loaders let you pull documents from any source into LlamaIndex — internal APIs, proprietary databases, legacy file formats, or third-party services without native connectors. This guide walks through building a production-ready loader from scratch, including authentication, incremental sync, and error handling.

Step 1: Understand the base classes

LlamaIndex provides two abstraction layers for data ingestion. BaseReader is the simpler interface — implement load_data() and return a list of Document objects. BasePydanticReader adds Pydantic validation for configuration. Most custom loaders should extend BaseReader unless you need schema validation on init.

from llama_index.core.readers.base import BaseReader
from llama_index.core.schema import Document
from typing import List, Optional

class MyCustomLoader(BaseReader):
    def __init__(self, api_key: str, base_url: str = "https://api.example.com"):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
    
    def load_data(self, **kwargs) -> List[Document]:
        raise NotImplementedError

The load_data method receives arbitrary keyword arguments. Common patterns include query for search-based sources, limit for pagination, and since for incremental loads. Document objects require text and optionally metadata (dict) and id_ (string).

Step 2: Implement authentication and client setup

Real-world sources need auth. Keep credentials out of the loader class itself — inject them at instantiation. Use a session object for connection pooling and retry logic.

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class MyCustomLoader(BaseReader):
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.example.com",
        timeout: float = 30.0,
        max_retries: int = 3,
    ):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.client = httpx.Client(
            base_url=self.base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=timeout,
        )
        self.max_retries = max_retries
    
    @retry(
        wait=wait_exponential(multiplier=1, min=2, max=10),
        stop=stop_after_attempt(3),
    )
    def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
        resp = self.client.request(method, path, **kwargs)
        resp.raise_for_status()
        return resp

The retry decorator handles transient network errors and 5xx responses. Adjust wait_exponential parameters based on your provider’s rate limit behavior.

Step 3: Fetch and transform raw data

Implement the core fetch logic. This example pulls records from a paginated REST endpoint, but the pattern applies to GraphQL, SQL, message queues, or file systems.

from datetime import datetime
from typing import Iterator, Dict, Any

class MyCustomLoader(BaseReader):
    # ... __init__ and _request from Step 2 ...
    
    def _iter_records(self, since: Optional[datetime] = None, limit: Optional[int] = None) -> Iterator[Dict[str, Any]]:
        params = {}
        if since:
            params["updated_after"] = since.isoformat()
        if limit:
            params["limit"] = min(limit, 100)  # API max page size
        
        page = 1
        yielded = 0
        while True:
            params["page"] = page
            resp = self._request("GET", "/v1/records", params=params)
            data = resp.json()
            
            records = data.get("items", [])
            if not records:
                break
            
            for record in records:
                yield record
                yielded += 1
                if limit and yielded >= limit:
                    return
            
            if not data.get("has_more", False):
                break
            page += 1

The iterator pattern keeps memory constant regardless of dataset size. The since parameter enables incremental sync — store the last successful timestamp and pass it on subsequent runs.

Step 4: Map records to Document objects

Transform each raw record into a LlamaIndex Document. Preserve source identifiers in metadata for traceability and deduplication. Include timestamps, URLs, and any fields useful for filtering downstream.

def _record_to_document(self, record: Dict[str, Any]) -> Document:
    # Required: text content for embedding
    text_parts = []
    if record.get("title"):
        text_parts.append(record["title"])
    if record.get("body"):
        text_parts.append(record["body"])
    if record.get("tags"):
        text_parts.append(" ".join(record["tags"]))
    
    text = "\n\n".join(text_parts)
    
    # Metadata for filtering, citation, and deduplication
    metadata = {
        "source_id": record["id"],
        "source_type": "my_custom_api",
        "source_url": f"{self.base_url}/records/{record['id']}",
        "created_at": record.get("created_at"),
        "updated_at": record.get("updated_at"),
        "author": record.get("author", {}).get("name"),
        "tags": record.get("tags", []),
        "status": record.get("status"),
    }
    # Remove None values to keep metadata clean
    metadata = {k: v for k, v in metadata.items() if v is not None}
    
    return Document(
        text=text,
        metadata=metadata,
        id_=record["id"],  # Stable ID enables upsert in vector stores
    )

Stable id_ values are critical. If your source lacks natural keys, generate a hash from immutable fields: hashlib.sha256(f"{record['url']}{record['updated_at']}".encode()).hexdigest()[:16].

Step 5: Wire it together in load_data

Combine the iterator and mapper. Accept standard LlamaIndex conventions: limit, since, and arbitrary passthrough kwargs for source-specific filters.

def load_data(
    self,
    limit: Optional[int] = None,
    since: Optional[datetime] = None,
    **kwargs: Any,
) -> List[Document]:
    documents = []
    for record in self._iter_records(since=since, limit=limit):
        try:
            doc = self._record_to_document(record)
            documents.append(doc)
        except Exception as e:
            # Log and continue — don't let one bad record kill the batch
            logger.warning(f"Failed to parse record {record.get('id')}: {e}")
    
    return documents

Add a module-level logger: logger = logging.getLogger(__name__). This lets operators tune verbosity via standard Python logging config.

Step 6: Add incremental sync support

Production pipelines need incremental loads. Implement a get_last_synced_timestamp helper that reads from your persistence layer — a JSON file, database, or key-value store.

import json
from pathlib import Path
from typing import Optional

class IncrementalSyncMixin:
    def __init__(self, state_path: Path, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.state_path = Path(state_path)
        self.state_path.parent.mkdir(parents=True, exist_ok=True)
    
    def get_last_synced(self) -> Optional[datetime]:
        if not self.state_path.exists():
            return None
        try:
            data = json.loads(self.state_path.read_text())
            ts = data.get("last_synced")
            return datetime.fromisoformat(ts) if ts else None
        except (json.JSONDecodeError, ValueError):
            return None
    
    def set_last_synced(self, timestamp: datetime) -> None:
        self.state_path.write_text(json.dumps({"last_synced": timestamp.isoformat()}))

Compose it with your loader:

class MyCustomLoader(IncrementalSyncMixin, BaseReader):
    def __init__(self, api_key: str, state_path: Path, **kwargs):
        BaseReader.__init__(self)
        IncrementalSyncMixin.__init__(self, state_path, **kwargs)
        # ... rest of init ...
    
    def load_data(self, incremental: bool = True, **kwargs) -> List[Document]:
        since = self.get_last_synced() if incremental else None
        docs = super().load_data(since=since, **kwargs)
        if docs and incremental:
            # Use the newest record's updated_at as the new watermark
            latest = max(
                (datetime.fromisoformat(d.metadata["updated_at"]) for d in docs if d.metadata.get("updated_at")),
                default=None,
            )
            if latest:
                self.set_last_synced(latest)
        return docs

Call loader.load_data(incremental=True) in scheduled jobs. The first run fetches everything; subsequent runs only fetch changes.

Step 7: Handle rate limits and provider errors

Respect Retry-After headers and implement circuit-breaker patterns for degraded providers. This is where an inference gateway like n4n.ai helps — it normalizes provider error shapes and handles fallback automatically, but your loader should still be defensive.

from tenacity import retry_if_exception_type

class RateLimitError(Exception):
    def __init__(self, retry_after: Optional[int] = None):
        self.retry_after = retry_after
        super().__init__(f"Rate limited, retry after {retry_after}s" if retry_after else "Rate limited")

class MyCustomLoader(BaseReader):
    # ... existing code ...
    
    @retry(
        wait=wait_exponential(multiplier=1, min=2, max=60),
        stop=stop_after_attempt(5),
        retry=retry_if_exception_type((httpx.HTTPStatusError, RateLimitError)),
    )
    def _request(self, method: str, path: str, **kwargs) -> httpx.Response:
        resp = self.client.request(method, path, **kwargs)
        
        if resp.status_code == 429:
            retry_after = resp.headers.get("Retry-After")
            raise RateLimitError(int(retry_after) if retry_after and retry_after.isdigit() else None)
        
        if 500 <= resp.status_code < 600:
            resp.raise_for_status()  # Triggers retry via HTTPStatusError
        
        resp.raise_for_status()
        return resp

The RateLimitError preserves Retry-After for smarter backoff. In a distributed system, consider a shared token bucket instead of per-process retries.

Step 8: Write integration tests

Test against a mock server, not the real API. pytest-httpx or respx intercept HTTP calls and return fixtures.

# tests/test_custom_loader.py
import pytest
from datetime import datetime, timezone
from pathlib import Path
import respx
import httpx

from my_loaders import MyCustomLoader

FIXTURE_RECORDS = [
    {"id": "rec_1", "title": "First", "body": "Content one", "tags": ["tag1"], "created_at": "2024-01-01T00:00:00Z", "updated_at": "2024-01-01T00:00:00Z", "author": {"name": "Alice"}, "status": "published"},
    {"id": "rec_2", "title": "Second", "body": "Content two", "tags": ["tag2"], "created_at": "2024-01-02T00:00:00Z", "updated_at": "2024-01-02T00:00:00Z", "author": {"name": "Bob"}, "status": "published"},
]

@respx.mock
def test_load_data_basic(tmp_path: Path):
    # Mock the paginated endpoint
    route = respx.get("https://api.example.com/v1/records").mock(
        return_value=httpx.Response(200, json={"items": FIXTURE_RECORDS, "has_more": False})
    )
    
    loader = MyCustomLoader(api_key="test-key", state_path=tmp_path / "state.json")
    docs = loader.load_data(limit=10)
    
    assert route.called
    assert len(docs) == 2
    assert docs[0].id_ == "rec_1"
    assert docs[0].metadata["source_type"] == "my_custom_api"
    assert "Content one" in docs[0].text

@respx.mock
def test_incremental_sync(tmp_path: Path):
    # First call returns all records
    respx.get("https://api.example.com/v1/records").mock(
        side_effect=[
            httpx.Response(200, json={"items": FIXTURE_RECORDS[:1], "has_more": False}),
            httpx.Response(200, json={"items": FIXTURE_RECORDS[1:], "has_more": False}),
        ]
    )
    
    loader = MyCustomLoader(api_key="test-key", state_path=tmp_path / "state.json")
    
    # First run — full sync
    docs1 = loader.load_data(incremental=True)
    assert len(docs1) == 1
    assert docs1[0].id_ == "rec_1"
    
    # Second run — incremental, only new records
    docs2 = loader.load_data(incremental=True)
    assert len(docs2) == 1
    assert docs2[0].id_ == "rec_2"

Run with pytest tests/test_custom_loader.py -v. Verify the state file updates correctly: cat /tmp/test_state.json should show the latest updated_at timestamp.

Step 9: Register as a LlamaIndex hub component (optional)

If you want your loader discoverable via download_loader, package it with a loader.json manifest. This step is optional but enables from llama_index.core import download_loader; MyLoader = download_loader("MyCustomLoader").

my_custom_loader/
├── __init__.py
├── loader.py          # Your MyCustomLoader class
├── loader.json        # Manifest
└── pyproject.toml

loader.json:

{
  "name": "MyCustomLoader",
  "description": "Loads records from Example API with incremental sync support",
  "class_name": "MyCustomLoader",
  "module_name": "my_custom_loader.loader",
  "requirements": ["httpx>=0.25", "tenacity>=8.0"],
  "init_args": {
    "api_key": {"type": "str", "description": "API key for Example API", "required": true},
    "base_url": {"type": "str", "description": "Base URL", "default": "https://api.example.com"},
    "state_path": {"type": "str", "description": "Path to incremental sync state file", "required": true}
  }
}

Publish to PyPI and users can install with pip install my-custom-loader. The hub registry is community-maintained — submit a PR to llama-hub if you want official listing.

Step 10: Verify end-to-end in a pipeline

Plug the loader into an ingestion pipeline with a vector store. This confirms documents have correct shape, metadata, and IDs for upsert.

from llama_index.core import VectorStoreIndex, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore
import chromadb

# Initialize loader
loader = MyCustomLoader(
    api_key="prod-key",
    state_path=Path("/var/lib/llama-index/sync_state.json"),
)

# Load documents (incremental)
documents = loader.load_data(incremental=True, limit=1000)
print(f"Loaded {len(documents)} documents")

# Build index with persistent vector store
chroma_client = chromadb.PersistentClient(path="/var/lib/chroma")
chroma_collection = chroma_client.get_or_create_collection("my_custom_docs")
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context,
    show_progress=True,
)

# Verify query works
query_engine = index.as_query_engine(similarity_top_k=3)
response = query_engine.query("What are the latest updates?")
print(response)

Run this script manually first. Check Chroma: chroma_client.get_collection("my_custom_docs").count() should match document count. Re-run — incremental sync should fetch zero new docs and query should still work.

Common pitfalls

Missing stable IDs: Without id_ on Document, vector stores create new entries on every ingestion run. Always set id_ from a source primary key or deterministic hash.

Blocking on large payloads: The load_data return type is List[Document], which materializes everything in memory. For millions of records, implement a streaming variant that yields documents and feed them to the index in batches using index.insert(doc) or index.insert_nodes(nodes).

Ignoring provider cache hints: Some APIs return ETag or Last-Modified. Store these alongside your watermark and send If-None-Match/If-Modified-Since headers to avoid downloading unchanged data.

Hardcoding secrets: Never commit API keys. Use environment variables, secret managers, or LlamaIndex’s Settings pattern for configuration.

Next steps

• Add filtering pushdown: pass query parameters to the API instead of filtering in Python
• Implement lazy_load_data returning Iterator[Document] for true streaming
• Add OpenTelemetry tracing around _request and _record_to_document for observability
• Write a BasePydanticReader variant if you need config validation for CLI tools

The loader pattern scales from prototype to production. Start simple, add incremental sync and error handling early, and test against realistic data volumes before deploying to schedule.