ZerdaZerda

Advanced Usage

On this page

Building a custom loader

Subclass BaseFileLoader (for file-backed sources) or BaseDocumentLoader (for anything else — APIs, databases, in-memory data) and implement load().

Python · 14 lines
from typing import List
from zerda.document_loaders.base_loader import BaseFileLoader, LoadedDocument

class YAMLLoader(BaseFileLoader):
    SUPPORTED_EXTENSIONS = [".yaml", ".yml"]

    def load(self) -> List[LoadedDocument]:
        import yaml
        with open(self.file_path, encoding=self.encoding) as f:
            data = yaml.safe_load(f)

        content = "\n".join(f"{k}: {v}" for k, v in data.items())
        meta = self._build_file_metadata(record_count=len(data))
        return [LoadedDocument(page_content=content, metadata=meta)]

Because it subclasses BaseFileLoader, this loader automatically gets:

  • File-existence and extension validation at construction (_validate_file())
  • _build_file_metadata() for consistent base metadata
  • load_and_split() for free
  • A default lazy_load() (override it yourself for true streaming if the format benefits)

Wiring a custom loader into directory scanning

EXTENSION_MAP and DirectoryLoader._get_loader_for_file() / AutoLoader._create_file_loader() are not designed for external registration — they're hardcoded if/elif chains. To integrate a custom loader with directory scanning, subclass DirectoryLoader:

Python · 18 lines
from zerda.document_loaders import DirectoryLoader
from zerda.document_loaders.config_loader import EXTENSION_MAP, LoaderType

class ExtendedDirectoryLoader(DirectoryLoader):
    def _get_loader_for_file(self, file_path):
        if file_path.suffix.lower() in (".yaml", ".yml"):
            return YAMLLoader(str(file_path))
        return super()._get_loader_for_file(file_path)

    def _collect_files(self):
        # also match .yaml/.yml, which aren't in EXTENSION_MAP
        files = super()._collect_files()
        extra = [
            f for f in self.directory.rglob("**/*")
            if f.is_file() and f.suffix.lower() in (".yaml", ".yml")
            and not self._is_excluded(f)
        ]
        return sorted(set(files + extra))

Building a full ingestion pipeline: load → split → dedupe → embed

Python · 26 lines
from zerda.document_loaders import DirectoryLoader

class SimpleSplitter:
    def __init__(self, chunk_size=800, overlap=100):
        self.chunk_size = chunk_size
        self.overlap = overlap

    def split_text(self, text: str):
        chunks = []
        step = self.chunk_size - self.overlap
        for i in range(0, len(text), step):
            chunk = text[i:i + self.chunk_size]
            if chunk.strip():
                chunks.append(chunk)
        return chunks

def ingest(directory: str, seen_ids: set):
    loader = DirectoryLoader(directory, silent_errors=True)
    splitter = SimpleSplitter()

    for raw_doc in loader.lazy_load():
        for chunk in loader.load_and_split(text_splitter=splitter):
            if chunk.doc_id in seen_ids:
                continue  # deterministic doc_id enables cheap deduplication
            seen_ids.add(chunk.doc_id)
            yield chunk
Python · 3 lines
def ingest(directory: str):
    loader = DirectoryLoader(directory, silent_errors=True)
    return loader.load_and_split(text_splitter=SimpleSplitter())

Mixing loaders explicitly for fine-grained control

When AutoLoader's one-config-fits-all dispatch isn't precise enough (e.g. you need different CSVLoaderConfigs for different files), bypass it and call loaders directly per source:

Python · 11 lines
from zerda.document_loaders import CSVLoader, CSVLoaderConfig, PDFLoader, PDFLoaderConfig

sources = [
    ("customers.csv", CSVLoader, CSVLoaderConfig(content_columns=["name", "notes"])),
    ("invoices.csv", CSVLoader, CSVLoaderConfig(content_columns=["invoice_text"])),
    ("contract.pdf", PDFLoader, PDFLoaderConfig(password="secret")),
]

all_docs = []
for path, loader_cls, config in sources:
    all_docs.extend(loader_cls(path, config=config).load())

Combining GraphLoader and TableLoader with the standard pipeline

Since neither is wired into AutoLoader, combine them manually with everything else:

Python · 6 lines
from zerda.document_loaders import AutoLoader, GraphLoader, TableLoader

docs = []
docs += AutoLoader.load("./text_corpus/")          # DirectoryLoader under the hood
docs += GraphLoader("entities.graphml").load()
docs += TableLoader("financials.xlsx", output_format="markdown").load()

Parallelizing across loader types with ThreadPoolExecutor

document_loaders itself doesn't provide a unified "load everything in parallel across mixed formats" API beyond DirectoryLoader (single format-agnostic directory) and MultiURLLoader (URLs only). For mixed heterogeneous sources, build your own pool:

Python · 13 lines
from concurrent.futures import ThreadPoolExecutor, as_completed
from zerda.document_loaders import AutoLoader

sources = ["report.pdf", "data.csv", "https://example.com/page", "./docs/"]

all_docs = []
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(AutoLoader.load, s): s for s in sources}
    for future in as_completed(futures):
        try:
            all_docs.extend(future.result())
        except Exception as e:
            print(f"Failed to load {futures[future]}: {e}")

Related pages