Skip to content

Ingestion

varagity.ingest.discovery

Corpus discovery: scan the docs directory and bucket files by parser.

Bucketing (spec §9.1, spec_v2 §8.1): plain-text formats (.txt/.md/.rst) share one extraction path (text_like); .pdf needs Docling with the OCR fallback (pdf); the Office families (incl. macro/template variants), .csv, and OpenDocument share the no-OCR Docling office path (office); .html/.htm/.xhtml likewise (web); bitmap images are OCR-only (image). Extensions outside the ALLOWED_EXTENSIONS whitelist are ignored (logged at DEBUG).

The reserved <DOCS_PATH>/varagity-exports/ folder is skipped whole (ADR-019): it holds backup bundles, never sources, and the rule must hold even if an ALLOWED_EXTENSIONS edit ever admitted a bundle's suffix. The CLI's ingest runs this same discovery, so it skips the folder too.

Buckets dataclass

Discovered corpus files, grouped by extraction path.

Attributes:

Name Type Description
text_like list[Path]

.txt / .md / .rst files (parsed by parsers/text.py).
pdf list[Path]

.pdf files (parsed by parsers/pdf.py).
office list[Path]

Office/OpenDocument/CSV files (the .docx/.pptx/ .xlsx families, .csv, .odt/.ods/.odp), parsed by parsers/office.py.
web list[Path]

.html / .htm / .xhtml files (parsed by parsers/web.py).
image list[Path]

Bitmap images (.png/.jpg/…), OCR'd by parsers/image.py.
Source code in varagity/ingest/discovery.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@dataclass
class Buckets:
    """Discovered corpus files, grouped by extraction path.

    Attributes:
        text_like: ``.txt`` / ``.md`` / ``.rst`` files (parsed by
            ``parsers/text.py``).
        pdf: ``.pdf`` files (parsed by ``parsers/pdf.py``).
        office: Office/OpenDocument/CSV files (the ``.docx``/``.pptx``/
            ``.xlsx`` families, ``.csv``, ``.odt``/``.ods``/``.odp``),
            parsed by ``parsers/office.py``.
        web: ``.html`` / ``.htm`` / ``.xhtml`` files (parsed by
            ``parsers/web.py``).
        image: Bitmap images (``.png``/``.jpg``/…), OCR'd by
            ``parsers/image.py``.
    """

    text_like: list[Path] = field(default_factory=list)
    pdf: list[Path] = field(default_factory=list)
    office: list[Path] = field(default_factory=list)
    web: list[Path] = field(default_factory=list)
    image: list[Path] = field(default_factory=list)

    def by_bucket(self) -> tuple[tuple[str, list[Path]], ...]:
        """List every bucket with its name, in a stable order.

        The single enumeration point renderers iterate, so a future bucket
        added here appears everywhere without caller edits.

        Returns:
            ``(bucket_name, paths)`` pairs, one per bucket.
        """
        return (
            ("text_like", self.text_like),
            ("pdf", self.pdf),
            ("office", self.office),
            ("web", self.web),
            ("image", self.image),
        )

    @property
    def total(self) -> int:
        """Total number of bucketed files.

        Returns:
            The combined size of all buckets.
        """
        return sum(len(paths) for _name, paths in self.by_bucket())

total property

Total number of bucketed files.

Returns:

Type Description
int

The combined size of all buckets.

by_bucket()

List every bucket with its name, in a stable order.

The single enumeration point renderers iterate, so a future bucket added here appears everywhere without caller edits.

Returns:

Type Description
tuple[tuple[str, list[Path]], ...]

(bucket_name, paths) pairs, one per bucket.
Source code in varagity/ingest/discovery.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def by_bucket(self) -> tuple[tuple[str, list[Path]], ...]:
    """List every bucket with its name, in a stable order.

    The single enumeration point renderers iterate, so a future bucket
    added here appears everywhere without caller edits.

    Returns:
        ``(bucket_name, paths)`` pairs, one per bucket.
    """
    return (
        ("text_like", self.text_like),
        ("pdf", self.pdf),
        ("office", self.office),
        ("web", self.web),
        ("image", self.image),
    )

discover_documents(docs_path, verbose=None)

Recursively scan docs_path and bucket ingestable files.

A missing directory is not an error: the app re-scans on every start (spec §9.1), so an empty/unmounted corpus logs a warning and yields empty buckets rather than crashing.

Parameters:

Name Type Description Default
docs_path str

Directory to scan (usually settings.DOCS_PATH).

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
Buckets

The bucketed file paths, each bucket sorted for determinism.

Raises:

Type Description
ValueError

If verbose is invalid.
Source code in varagity/ingest/discovery.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def discover_documents(docs_path: str, verbose: int | None = None) -> Buckets:
    """Recursively scan ``docs_path`` and bucket ingestable files.

    A missing directory is not an error: the app re-scans on every start
    (spec §9.1), so an empty/unmounted corpus logs a warning and yields empty
    buckets rather than crashing.

    Args:
        docs_path: Directory to scan (usually ``settings.DOCS_PATH``).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The bucketed file paths, each bucket sorted for determinism.

    Raises:
        ValueError: If ``verbose`` is invalid.
    """
    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    buckets = Buckets()

    root = Path(docs_path)
    if not root.is_dir():
        logger.warning("docs directory %s does not exist — nothing to ingest", docs_path)
        v_discover(buckets, verbose=verbose)
        return buckets

    allowed = settings.allowed_extension_set
    for path in sorted(root.rglob("*")):
        if is_exports_path(path.relative_to(root)):
            logger.debug("ignoring %s (inside the reserved %s/ folder)", path, EXPORTS_DIRNAME)
            continue
        if not path.is_file():
            continue
        extension = path.suffix.lower()
        if extension not in allowed:
            logger.debug("ignoring %s (extension %r not in ALLOWED_EXTENSIONS)", path, extension)
            continue
        if extension in _TEXT_LIKE_EXTENSIONS:
            buckets.text_like.append(path)
        elif extension in _PDF_EXTENSIONS:
            buckets.pdf.append(path)
        elif extension in _OFFICE_EXTENSIONS:
            buckets.office.append(path)
        elif extension in _WEB_EXTENSIONS:
            buckets.web.append(path)
        elif extension in _IMAGE_EXTENSIONS:
            buckets.image.append(path)
        else:
            logger.warning(
                "%s is allowed by ALLOWED_EXTENSIONS but has no ingestion bucket; skipping", path
            )

    v_discover(buckets, verbose=verbose)
    return buckets

varagity.ingest.parsers.base

Parser protocol and registry (the spec §5.1 registry pattern).

Each parser module defines one implementation decorated with @register("name"); callers resolve implementations with :func:get_parser and never import concrete parsers directly.

Parser

Bases: Protocol

Interface every parser implements.

Source code in varagity/ingest/parsers/base.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class Parser(Protocol):
    """Interface every parser implements."""

    def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
        """Extract text and provenance from a source file.

        Args:
            path: The file to parse.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The extracted document.
        """
        ...

extract(path, verbose=None)

Extract text and provenance from a source file.

Parameters:

Name Type Description Default
path Path

The file to parse.

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
RawDocument

The extracted document.
Source code in varagity/ingest/parsers/base.py 
32
33
34
35
36
37
38
39
40
41
42
43
def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
    """Extract text and provenance from a source file.

    Args:
        path: The file to parse.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The extracted document.
    """
    ...

RawDocument dataclass

A parsed source document, before chunking.

Attributes:

Name Type Description
text str

The full extracted text (normalized by the parser).
source_meta dict[str, Any]

Provenance seeded into every chunk's metadata: source (absolute path), file_name, file_type, and page (None for non-paginated formats).
Source code in varagity/ingest/parsers/base.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
@dataclass
class RawDocument:
    """A parsed source document, before chunking.

    Attributes:
        text: The full extracted text (normalized by the parser).
        source_meta: Provenance seeded into every chunk's metadata:
            ``source`` (absolute path), ``file_name``, ``file_type``, and
            ``page`` (``None`` for non-paginated formats).
    """

    text: str
    source_meta: dict[str, Any]

get_parser(name)

Look up a registered parser by name.

Parameters:

Name Type Description Default
name str

Registry key (e.g. "text", "pdf").

 required

Returns:

Type Description
Parser

The registered parser instance.

Raises:

Type Description
KeyError

If no parser is registered under name (message lists the available ones).
Source code in varagity/ingest/parsers/base.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def get_parser(name: str) -> Parser:
    """Look up a registered parser by name.

    Args:
        name: Registry key (e.g. ``"text"``, ``"pdf"``).

    Returns:
        The registered parser instance.

    Raises:
        KeyError: If no parser is registered under ``name`` (message lists
            the available ones).
    """
    if name not in PARSER_REGISTRY:
        raise KeyError(f"Unknown parser {name!r}. Available: {list(PARSER_REGISTRY)}")
    return PARSER_REGISTRY[name]

register(name)

Class decorator registering a parser instance under name.

Parameters:

Name Type Description Default
name str

Registry key (matches the discovery bucket, e.g. "text").

 required

Returns:

Type Description
Callable[[T], T]

The decorator, which instantiates and registers the class unchanged.
Source code in varagity/ingest/parsers/base.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def register[T: type[Any]](name: str) -> Callable[[T], T]:
    """Class decorator registering a parser instance under ``name``.

    Args:
        name: Registry key (matches the discovery bucket, e.g. ``"text"``).

    Returns:
        The decorator, which instantiates and registers the class unchanged.
    """

    def deco(cls: T) -> T:
        PARSER_REGISTRY[name] = cls()
        return cls

    return deco

varagity.ingest.parsers.text

Plain-text parser for .txt and .md (spec §9.2).

Reads UTF-8, normalizes newlines, and repairs hyphen-broken words via :func:remove_hyphen_space (carried over from the reference implementation).

TextParser

Parser for the text_like bucket (.txt / .md).

Source code in varagity/ingest/parsers/text.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@register("text")
class TextParser:
    """Parser for the ``text_like`` bucket (``.txt`` / ``.md``)."""

    def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
        """Read and normalize a text file.

        Args:
            path: The ``.txt`` / ``.md`` file to read (UTF-8).
            verbose: Console verbosity (0–2); accepted for interface
                uniformity — this parser renders nothing itself.

        Returns:
            The extracted document; ``page`` is ``None`` (not paginated).

        Raises:
            UnicodeDecodeError: If the file is not valid UTF-8.
            OSError: If the file cannot be read.
        """
        raw = path.read_text(encoding="utf-8")
        text = raw.replace("\r\n", "\n").replace("\r", "\n")
        text = remove_hyphen_space(text)
        return RawDocument(
            text=text,
            source_meta={
                "source": str(path.resolve()),
                "file_name": path.name,
                "file_type": path.suffix.lower().lstrip("."),
                "page": None,
            },
        )

extract(path, verbose=None)

Read and normalize a text file.

Parameters:

Name Type Description Default
path Path

The .txt / .md file to read (UTF-8).

 required
verbose int | None

Console verbosity (0–2); accepted for interface uniformity — this parser renders nothing itself.

 None

Returns:

Type Description
RawDocument

The extracted document; page is None (not paginated).

Raises:

Type Description
UnicodeDecodeError

If the file is not valid UTF-8.
OSError

If the file cannot be read.
Source code in varagity/ingest/parsers/text.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
    """Read and normalize a text file.

    Args:
        path: The ``.txt`` / ``.md`` file to read (UTF-8).
        verbose: Console verbosity (0–2); accepted for interface
            uniformity — this parser renders nothing itself.

    Returns:
        The extracted document; ``page`` is ``None`` (not paginated).

    Raises:
        UnicodeDecodeError: If the file is not valid UTF-8.
        OSError: If the file cannot be read.
    """
    raw = path.read_text(encoding="utf-8")
    text = raw.replace("\r\n", "\n").replace("\r", "\n")
    text = remove_hyphen_space(text)
    return RawDocument(
        text=text,
        source_meta={
            "source": str(path.resolve()),
            "file_name": path.name,
            "file_type": path.suffix.lower().lstrip("."),
            "page": None,
        },
    )

remove_hyphen_space(text)

Rejoin words split by a hyphen at a line break or before a space.

The reference implementation's fix for extraction artifacts like frame-\nwork and frame- work (both → framework).

Parameters:

Name Type Description Default
text str

Text with possible hyphenation artifacts (newlines already normalized to \n).

 required

Returns:

Type Description
str

The text with split words rejoined.
Source code in varagity/ingest/parsers/text.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def remove_hyphen_space(text: str) -> str:
    r"""Rejoin words split by a hyphen at a line break or before a space.

    The reference implementation's fix for extraction artifacts like
    ``frame-\nwork`` and ``frame- work`` (both → ``framework``).

    Args:
        text: Text with possible hyphenation artifacts (newlines already
            normalized to ``\n``).

    Returns:
        The text with split words rejoined.
    """
    text = _HYPHEN_NEWLINE.sub(r"\1\2", text)
    return _HYPHEN_SPACE.sub(r"\1\2", text)

varagity.ingest.parsers.docling_base

Shared Docling conversion core for every rich-document parser (spec_v2 §8.2).

The generic machinery parsers/pdf.py proved in v1 — DocumentConverter conversion → structure-aware markdown export → hyphen repair → page/character provenance → :class:~varagity.ingest.parsers.base.RawDocument assembly — lifted out so the office (.docx/.pptx/.xlsx) and web (.html/.htm) parsers reuse one markdown/table/provenance pipeline.

No OCR lives here. The two-pass OCR fallback is PDF-specific (scanned pages) and stays in pdf.py; the formats parsed through :class:DoclingParser carry digital text, so a single no-OCR conversion suffices and every document is extraction="text".

Page semantics per format (verified against Docling's backends): .pptx maps each slide to a page and .xlsx maps each sheet to a page, so page carries the slide/sheet identity; .docx and .html expose no pagination (document.pages is empty), so page stays None — the same graceful degradation the text parser uses.

Docling imports are deferred to call time: importing this module (which happens on every CLI start via parser self-registration) must not pay for Docling's model machinery.

DoclingParser

Single-pass Docling parser for digital-text formats (no OCR).

The shared base the office and web registrations subclass: one plain DocumentConverter conversion through the same markdown/table/ provenance pipeline as the PDF parser, minus the two-pass OCR fallback (these formats carry digital text — spec_v2 §8.2). The converter is built lazily and cached on the instance (the registry instantiates one parser per process), and the office/web backends are lightweight — no layout-model downloads.

Source code in varagity/ingest/parsers/docling_base.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class DoclingParser:
    """Single-pass Docling parser for digital-text formats (no OCR).

    The shared base the ``office`` and ``web`` registrations subclass: one
    plain ``DocumentConverter`` conversion through the same markdown/table/
    provenance pipeline as the PDF parser, minus the two-pass OCR fallback
    (these formats carry digital text — spec_v2 §8.2). The converter is
    built lazily and cached on the instance (the registry instantiates one
    parser per process), and the office/web backends are lightweight — no
    layout-model downloads.
    """

    def __init__(self) -> None:
        """Initialize the converter cache (nothing heavy happens here)."""
        self._converter: DocumentConverter | None = None

    def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
        """Extract a document's text and provenance via one Docling pass.

        Args:
            path: The file to convert (any format Docling handles without
                OCR; discovery routes only this parser's bucket here).
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The extracted document. ``source_meta`` carries ``page`` (slide
            for ``.pptx``, sheet for ``.xlsx``, ``None`` for formats
            without pagination) and ``extraction="text"`` (never OCR).

        Raises:
            ValueError: If ``verbose`` is invalid.
            docling.exceptions.ConversionError: If conversion fails (a
                malformed file) — the loader counts the file as failed and
                the run continues.
        """
        settings = get_settings()
        check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
        document = self._convert(path)
        text = export_markdown(document)
        return raw_document(path, text, page_char_counts(document), extraction="text")

    def _convert(self, path: Path) -> "DoclingDocument":
        """Run the (cached) Docling converter on one file.

        Args:
            path: The file to convert.

        Returns:
            The converted Docling document.
        """
        if self._converter is None:
            from docling.document_converter import DocumentConverter

            self._converter = DocumentConverter()
        return self._converter.convert(path).document  # raises ConversionError on failure

__init__()

Initialize the converter cache (nothing heavy happens here).

Source code in varagity/ingest/parsers/docling_base.py 
128
129
130
def __init__(self) -> None:
    """Initialize the converter cache (nothing heavy happens here)."""
    self._converter: DocumentConverter | None = None

extract(path, verbose=None)

Extract a document's text and provenance via one Docling pass.

Parameters:

Name Type Description Default
path Path

The file to convert (any format Docling handles without OCR; discovery routes only this parser's bucket here).

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
RawDocument

The extracted document. source_meta carries page (slide
RawDocument

for .pptx, sheet for .xlsx, None for formats
RawDocument

without pagination) and extraction="text" (never OCR).

Raises:

Type Description
ValueError

If verbose is invalid.
ConversionError

If conversion fails (a malformed file) — the loader counts the file as failed and the run continues.
Source code in varagity/ingest/parsers/docling_base.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
    """Extract a document's text and provenance via one Docling pass.

    Args:
        path: The file to convert (any format Docling handles without
            OCR; discovery routes only this parser's bucket here).
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The extracted document. ``source_meta`` carries ``page`` (slide
        for ``.pptx``, sheet for ``.xlsx``, ``None`` for formats
        without pagination) and ``extraction="text"`` (never OCR).

    Raises:
        ValueError: If ``verbose`` is invalid.
        docling.exceptions.ConversionError: If conversion fails (a
            malformed file) — the loader counts the file as failed and
            the run continues.
    """
    settings = get_settings()
    check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    document = self._convert(path)
    text = export_markdown(document)
    return raw_document(path, text, page_char_counts(document), extraction="text")

export_markdown(document)

Export a converted document as normalized markdown.

Docling's structure-aware export (headings, GFM tables) with newlines normalized to \n and hyphen-broken words repaired — the exact text shape the chunkers expect from every parser.

Parameters:

Name Type Description Default
document DoclingDocument

The converted Docling document.

 required

Returns:

Type Description
str

The normalized markdown text.
Source code in varagity/ingest/parsers/docling_base.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def export_markdown(document: "DoclingDocument") -> str:
    r"""Export a converted document as normalized markdown.

    Docling's structure-aware export (headings, GFM tables) with newlines
    normalized to ``\n`` and hyphen-broken words repaired — the exact text
    shape the chunkers expect from every parser.

    Args:
        document: The converted Docling document.

    Returns:
        The normalized markdown text.
    """
    text = document.export_to_markdown().replace("\r\n", "\n").replace("\r", "\n")
    return remove_hyphen_space(text)

page_char_counts(document)

Count extracted non-whitespace characters per page.

Uses Docling item provenance: every content item (including table cells) is attributed to the page(s) it appears on. Pages that no item references keep a zero count — the textless-page signal the PDF OCR trigger reads. Formats without pagination (.docx/.html) yield an empty mapping.

Parameters:

Name Type Description Default
document DoclingDocument

The converted Docling document.

 required

Returns:

Type Description
dict[int, int]

{page_no: char_count} covering every page of the document.
Source code in varagity/ingest/parsers/docling_base.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def page_char_counts(document: "DoclingDocument") -> dict[int, int]:
    """Count extracted non-whitespace characters per page.

    Uses Docling item provenance: every content item (including table
    cells) is attributed to the page(s) it appears on. Pages that no item
    references keep a zero count — the textless-page signal the PDF OCR
    trigger reads. Formats without pagination (``.docx``/``.html``) yield an
    empty mapping.

    Args:
        document: The converted Docling document.

    Returns:
        ``{page_no: char_count}`` covering every page of the document.
    """
    counts: dict[int, int] = dict.fromkeys(document.pages, 0)
    for item, _level in document.iterate_items():
        chars = 0
        text = getattr(item, "text", None)
        if text:
            chars += len(text.strip())
        data = getattr(item, "data", None)  # tables carry text in cells
        table_cells = getattr(data, "table_cells", None) if data is not None else None
        if table_cells:
            chars += sum(len(cell.text.strip()) for cell in table_cells if cell.text)
        if not chars:
            continue
        for prov in getattr(item, "prov", None) or []:
            counts[prov.page_no] = counts.get(prov.page_no, 0) + chars
    return counts

raw_document(path, text, page_counts, *, extraction)

Assemble the parser result with provenance metadata.

Parameters:

Name Type Description Default
path Path

The source file.

 required
text str

The normalized extracted text.

 required
page_counts dict[int, int]

Per-page character counts (for page attribution; empty for non-paginated formats).

 required
extraction str

"text", "ocr", or "ocr_fallback".

 required

Returns:

Name Type Description
The RawDocument

class:~varagity.ingest.parsers.base.RawDocument handed to the
RawDocument

chunker. source_meta carries page — the first page (slide
RawDocument

for .pptx, sheet for .xlsx) that contributed text, None
RawDocument

when the format has no pagination — and extraction.
Source code in varagity/ingest/parsers/docling_base.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def raw_document(
    path: Path, text: str, page_counts: dict[int, int], *, extraction: str
) -> RawDocument:
    """Assemble the parser result with provenance metadata.

    Args:
        path: The source file.
        text: The normalized extracted text.
        page_counts: Per-page character counts (for page attribution;
            empty for non-paginated formats).
        extraction: ``"text"``, ``"ocr"``, or ``"ocr_fallback"``.

    Returns:
        The :class:`~varagity.ingest.parsers.base.RawDocument` handed to the
        chunker. ``source_meta`` carries ``page`` — the first page (slide
        for ``.pptx``, sheet for ``.xlsx``) that contributed text, ``None``
        when the format has no pagination — and ``extraction``.
    """
    pages_with_text = [page for page, count in sorted(page_counts.items()) if count > 0]
    source_meta: dict[str, Any] = {
        "source": str(path.resolve()),
        "file_name": path.name,
        "file_type": path.suffix.lower().lstrip("."),
        "page": pages_with_text[0] if pages_with_text else None,
        "extraction": extraction,
    }
    return RawDocument(text=text, source_meta=source_meta)

varagity.ingest.parsers.pdf

PDF parser: Docling two-pass extraction with an automatic OCR fallback.

Pass 1 converts with OCR disabled — the fast path, since OCR is a multi-× slowdown. If the result has (near-)no text, too many textless pages, or pass 1 raised, pass 2 re-converts with OCR enabled (plan decision #10). Both passes share the docling_base markdown/table/ provenance core (spec_v2 §8.2), so the fallback changes how text is recovered, never its downstream shape — and the office/web parsers share the exact same pipeline.

The OCR engine is pluggable via the OCR_ENGINE setting: a small factory maps engine names to Docling ocr_options (EasyOCR is the benchmark-decided default — ADR-004; Tesseract needs only the system binary). Documents recovered by pass 2 carry extraction="ocr_fallback" provenance on every chunk.

Docling imports are deferred to call time: importing this module (which happens on every CLI start via parser self-registration) must not pay for Docling's model machinery.

Pattern adapted from the docling-ocr-pipeline reference notebook, kept Docling-native (no side pdf2image/pytesseract path) so both passes share one markdown/table/provenance pipeline and the engine swap is pure config.

ExtractionStats dataclass

What pass 1 recovered, as inputs to the fallback trigger.

Attributes:

Name Type Description
non_ws_chars int

Non-whitespace characters in the exported text.
total_pages int

Page count of the source document.
textless_pages int

Pages that contributed no text to the export.
Source code in varagity/ingest/parsers/pdf.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
@dataclass(frozen=True)
class ExtractionStats:
    """What pass 1 recovered, as inputs to the fallback trigger.

    Attributes:
        non_ws_chars: Non-whitespace characters in the exported text.
        total_pages: Page count of the source document.
        textless_pages: Pages that contributed no text to the export.
    """

    non_ws_chars: int
    total_pages: int
    textless_pages: int

PdfParser

Parser for the pdf bucket: Docling with the two-pass OCR fallback.

Converters are built lazily and cached on the instance (the registry instantiates one parser per process), so Docling's models load once per pass type, not once per file.

Source code in varagity/ingest/parsers/pdf.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
@register("pdf")
class PdfParser:
    """Parser for the ``pdf`` bucket: Docling with the two-pass OCR fallback.

    Converters are built lazily and cached on the instance (the registry
    instantiates one parser per process), so Docling's models load once
    per pass type, not once per file.
    """

    def __init__(self) -> None:
        """Initialize the converter caches (nothing heavy happens here)."""
        self._fast_converter: DocumentConverter | None = None
        # Keyed by (engine, languages, force_full_page): tests and
        # long-lived processes may change OCR settings between calls.
        self._ocr_converters: dict[tuple[str, tuple[str, ...], bool], DocumentConverter] = {}

    def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
        """Extract a PDF's text, falling back to OCR when warranted.

        Args:
            path: The ``.pdf`` file to convert.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The extracted document. ``source_meta`` carries ``page`` (the
            first page that contributed text — chunk-level attribution
            would need offset support in the shared chunker, a deferred
            enhancement) and ``extraction`` (``"ocr_fallback"`` when pass
            2 produced the text, else ``"text"``).

        Raises:
            ValueError: If ``verbose`` is invalid.
            docling.exceptions.ConversionError: If conversion fails — for
                pass 1 only when the fallback is disabled; a pass 2
                failure always propagates (the loader counts the file as
                failed and re-attempts it on the next run, rather than
                recording a permanently half-extracted document).
        """
        settings = get_settings()
        verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)

        if settings.PDF_OCR_FORCE_FULL_PAGE:
            # Escape hatch for corrupt-text-layer PDFs: garbage embedded
            # text passes the content triggers, so skip straight to a
            # forced full-page OCR conversion.
            logger.info("%s: PDF_OCR_FORCE_FULL_PAGE is on — full-page OCR conversion", path.name)
            text, page_counts = self._convert(path, ocr=True)
            return self._raw_document(path, text, page_counts, extraction="ocr_fallback")

        try:
            text, page_counts = self._convert(path, ocr=False)
        except Exception:
            if not settings.PDF_OCR_FALLBACK:
                raise
            logger.warning(
                "%s: text-layer conversion raised — retrying with OCR", path.name, exc_info=True
            )
            text, page_counts = self._convert(path, ocr=True)
            return self._raw_document(path, text, page_counts, extraction="ocr_fallback")

        stats = extraction_stats(text, page_counts)
        if settings.PDF_OCR_FALLBACK and needs_ocr_fallback(
            stats,
            min_chars=settings.PDF_OCR_MIN_CHARS,
            textless_page_ratio=settings.PDF_OCR_TEXTLESS_PAGE_RATIO,
        ):
            logger.info(
                "%s: OCR fallback triggered (%d non-whitespace chars, %d/%d textless pages)",
                path.name,
                stats.non_ws_chars,
                stats.textless_pages,
                stats.total_pages,
            )
            text, page_counts = self._convert(path, ocr=True)
            return self._raw_document(path, text, page_counts, extraction="ocr_fallback")

        return self._raw_document(path, text, page_counts, extraction="text")

    def _convert(self, path: Path, *, ocr: bool) -> tuple[str, dict[int, int]]:
        """Run one Docling conversion pass.

        Args:
            path: The PDF to convert.
            ocr: Whether this is the OCR pass (pass 2) or the fast
                text-layer pass (pass 1).

        Returns:
            The normalized markdown text and per-page character counts.
        """
        result = self._converter(ocr=ocr).convert(path)  # raises ConversionError on failure
        document = result.document
        return export_markdown(document), page_char_counts(document)

    def _converter(self, *, ocr: bool) -> "DocumentConverter":
        """Return the cached converter for a pass type, building it lazily.

        Args:
            ocr: Whether OCR is enabled for this converter.

        Returns:
            The (possibly cached) Docling converter.
        """
        from docling.datamodel.base_models import InputFormat
        from docling.datamodel.pipeline_options import PdfPipelineOptions
        from docling.document_converter import DocumentConverter, PdfFormatOption

        settings = get_settings()
        if not ocr:
            if self._fast_converter is None:
                options = PdfPipelineOptions(do_ocr=False)
                self._fast_converter = DocumentConverter(
                    format_options={InputFormat.PDF: PdfFormatOption(pipeline_options=options)}
                )
            return self._fast_converter

        languages = settings.ocr_language_list
        key = (settings.OCR_ENGINE, tuple(languages), settings.PDF_OCR_FORCE_FULL_PAGE)
        if key not in self._ocr_converters:
            options = PdfPipelineOptions(
                do_ocr=True,
                ocr_options=get_ocr_options(
                    settings.OCR_ENGINE,
                    languages,
                    force_full_page=settings.PDF_OCR_FORCE_FULL_PAGE,
                ),
            )
            self._ocr_converters[key] = DocumentConverter(
                format_options={InputFormat.PDF: PdfFormatOption(pipeline_options=options)}
            )
        return self._ocr_converters[key]

    def _raw_document(
        self, path: Path, text: str, page_counts: dict[int, int], *, extraction: str
    ) -> RawDocument:
        """Assemble the parser result with provenance metadata.

        Args:
            path: The source PDF.
            text: The normalized extracted text.
            page_counts: Per-page character counts (for page attribution).
            extraction: ``"text"`` or ``"ocr_fallback"``.

        Returns:
            The :class:`RawDocument` handed to the chunker (assembled by the
            shared :func:`~varagity.ingest.parsers.docling_base.raw_document`).
        """
        return raw_document(path, text, page_counts, extraction=extraction)

__init__()

Initialize the converter caches (nothing heavy happens here).

Source code in varagity/ingest/parsers/pdf.py 
199
200
201
202
203
204
def __init__(self) -> None:
    """Initialize the converter caches (nothing heavy happens here)."""
    self._fast_converter: DocumentConverter | None = None
    # Keyed by (engine, languages, force_full_page): tests and
    # long-lived processes may change OCR settings between calls.
    self._ocr_converters: dict[tuple[str, tuple[str, ...], bool], DocumentConverter] = {}

extract(path, verbose=None)

Extract a PDF's text, falling back to OCR when warranted.

Parameters:

Name Type Description Default
path Path

The .pdf file to convert.

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
RawDocument

The extracted document. source_meta carries page (the
RawDocument

first page that contributed text — chunk-level attribution
RawDocument

would need offset support in the shared chunker, a deferred
RawDocument

enhancement) and extraction ("ocr_fallback" when pass
RawDocument

2 produced the text, else "text").

Raises:

Type Description
ValueError

If verbose is invalid.
ConversionError

If conversion fails — for pass 1 only when the fallback is disabled; a pass 2 failure always propagates (the loader counts the file as failed and re-attempts it on the next run, rather than recording a permanently half-extracted document).
Source code in varagity/ingest/parsers/pdf.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
    """Extract a PDF's text, falling back to OCR when warranted.

    Args:
        path: The ``.pdf`` file to convert.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The extracted document. ``source_meta`` carries ``page`` (the
        first page that contributed text — chunk-level attribution
        would need offset support in the shared chunker, a deferred
        enhancement) and ``extraction`` (``"ocr_fallback"`` when pass
        2 produced the text, else ``"text"``).

    Raises:
        ValueError: If ``verbose`` is invalid.
        docling.exceptions.ConversionError: If conversion fails — for
            pass 1 only when the fallback is disabled; a pass 2
            failure always propagates (the loader counts the file as
            failed and re-attempts it on the next run, rather than
            recording a permanently half-extracted document).
    """
    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)

    if settings.PDF_OCR_FORCE_FULL_PAGE:
        # Escape hatch for corrupt-text-layer PDFs: garbage embedded
        # text passes the content triggers, so skip straight to a
        # forced full-page OCR conversion.
        logger.info("%s: PDF_OCR_FORCE_FULL_PAGE is on — full-page OCR conversion", path.name)
        text, page_counts = self._convert(path, ocr=True)
        return self._raw_document(path, text, page_counts, extraction="ocr_fallback")

    try:
        text, page_counts = self._convert(path, ocr=False)
    except Exception:
        if not settings.PDF_OCR_FALLBACK:
            raise
        logger.warning(
            "%s: text-layer conversion raised — retrying with OCR", path.name, exc_info=True
        )
        text, page_counts = self._convert(path, ocr=True)
        return self._raw_document(path, text, page_counts, extraction="ocr_fallback")

    stats = extraction_stats(text, page_counts)
    if settings.PDF_OCR_FALLBACK and needs_ocr_fallback(
        stats,
        min_chars=settings.PDF_OCR_MIN_CHARS,
        textless_page_ratio=settings.PDF_OCR_TEXTLESS_PAGE_RATIO,
    ):
        logger.info(
            "%s: OCR fallback triggered (%d non-whitespace chars, %d/%d textless pages)",
            path.name,
            stats.non_ws_chars,
            stats.textless_pages,
            stats.total_pages,
        )
        text, page_counts = self._convert(path, ocr=True)
        return self._raw_document(path, text, page_counts, extraction="ocr_fallback")

    return self._raw_document(path, text, page_counts, extraction="text")

extraction_stats(text, page_char_counts)

Summarize an extraction pass for the fallback trigger.

Parameters:

Name Type Description Default
text str

The exported document text.

 required
page_char_counts dict[int, int]

Non-whitespace character count per page number.

 required

Returns:

Type Description
ExtractionStats

The pass's :class:ExtractionStats.
Source code in varagity/ingest/parsers/pdf.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def extraction_stats(text: str, page_char_counts: dict[int, int]) -> ExtractionStats:
    """Summarize an extraction pass for the fallback trigger.

    Args:
        text: The exported document text.
        page_char_counts: Non-whitespace character count per page number.

    Returns:
        The pass's :class:`ExtractionStats`.
    """
    return ExtractionStats(
        non_ws_chars=sum(1 for char in text if not char.isspace()),
        total_pages=len(page_char_counts),
        textless_pages=sum(1 for count in page_char_counts.values() if count == 0),
    )

get_ocr_options(engine, languages, force_full_page=False)

Resolve an OCR_ENGINE name to Docling OCR options.

Parameters:

Name Type Description Default
engine str

Registry key ("easyocr" or "tesseract"; adding an engine means adding a factory above — no caller edits).

 required
languages list[str]

ISO 639-1 language codes, primary first.

 required
force_full_page bool

OCR whole pages even where a text layer exists.

 False

Returns:

Type Description
OcrOptions

Engine-specific Docling ocr_options.

Raises:

Type Description
KeyError

If no factory is registered under engine (message lists the available ones).
Source code in varagity/ingest/parsers/pdf.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_ocr_options(
    engine: str, languages: list[str], force_full_page: bool = False
) -> "OcrOptions":
    """Resolve an ``OCR_ENGINE`` name to Docling OCR options.

    Args:
        engine: Registry key (``"easyocr"`` or ``"tesseract"``; adding an
            engine means adding a factory above — no caller edits).
        languages: ISO 639-1 language codes, primary first.
        force_full_page: OCR whole pages even where a text layer exists.

    Returns:
        Engine-specific Docling ``ocr_options``.

    Raises:
        KeyError: If no factory is registered under ``engine`` (message
            lists the available ones).
    """
    if engine not in OCR_ENGINE_FACTORIES:
        raise KeyError(f"Unknown OCR engine {engine!r}. Available: {list(OCR_ENGINE_FACTORIES)}")
    return OCR_ENGINE_FACTORIES[engine](languages, force_full_page)

needs_ocr_fallback(stats, *, min_chars, textless_page_ratio)

Decide whether pass 1's result warrants the OCR pass (plan decision #10).

Evaluated once per document, as in the reference notebook. Either trigger suffices: a (near-)empty extraction, or a high share of textless pages — the latter catches mixed scanned/digital documents whose digital pages alone pass a document-level length check.

Parameters:

Name Type Description Default
stats ExtractionStats

Pass 1's extraction summary.

 required
min_chars int

Below this many non-whitespace characters the document counts as having no text layer (PDF_OCR_MIN_CHARS).

 required
textless_page_ratio float

Textless-page share at or above which the fallback fires (PDF_OCR_TEXTLESS_PAGE_RATIO).

 required

Returns:

Type Description
bool

True if the document should be re-converted with OCR.
Source code in varagity/ingest/parsers/pdf.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def needs_ocr_fallback(
    stats: ExtractionStats, *, min_chars: int, textless_page_ratio: float
) -> bool:
    """Decide whether pass 1's result warrants the OCR pass (plan decision #10).

    Evaluated once per document, as in the reference notebook. Either
    trigger suffices: a (near-)empty extraction, or a high share of
    textless pages — the latter catches mixed scanned/digital documents
    whose digital pages alone pass a document-level length check.

    Args:
        stats: Pass 1's extraction summary.
        min_chars: Below this many non-whitespace characters the document
            counts as having no text layer (``PDF_OCR_MIN_CHARS``).
        textless_page_ratio: Textless-page share at or above which the
            fallback fires (``PDF_OCR_TEXTLESS_PAGE_RATIO``).

    Returns:
        True if the document should be re-converted with OCR.
    """
    if stats.non_ws_chars < min_chars:
        return True
    return stats.total_pages > 0 and stats.textless_pages / stats.total_pages >= textless_page_ratio

varagity.ingest.parsers.office

Office-document parser: the OOXML families, CSV, and OpenDocument (spec_v2 §8).

A thin registration over the shared Docling core: office formats carry digital text, so a single no-OCR conversion through the same markdown/ table/provenance pipeline as the PDF parser suffices (the two-pass OCR fallback is PDF-only). The bucket covers the .docx/.pptx/.xlsx families (including macro-enabled and template variants — Docling's backends open them like their base formats), single-table .csv, and OpenDocument .odt/.ods/.odp. All backends are lightweight (python-docx / python-pptx / openpyxl / odfdo) — no layout-model downloads.

Provenance: .pptx maps each slide to a Docling page and .xlsx/.ods map each sheet to one, so page carries the slide/sheet identity; .docx, .csv, .odt, and .odp expose no per-item pagination, so page stays None (spec_v2 §8.2).

OfficeParser

Bases: DoclingParser

Parser for the office bucket (OOXML families / CSV / OpenDocument).

Source code in varagity/ingest/parsers/office.py 
23
24
25
@register("office")
class OfficeParser(DoclingParser):
    """Parser for the ``office`` bucket (OOXML families / CSV / OpenDocument)."""

varagity.ingest.parsers.web

Web-document parser: .html / .htm / .xhtml (spec_v2 §8).

A thin registration over the shared Docling core: web pages carry digital text, so a single no-OCR conversion through the same markdown/table/ provenance pipeline as the PDF parser suffices. Docling's HTML backend is lightweight (beautifulsoup4) — no layout-model downloads.

Provenance: HTML has no pagination concept, so page stays None (spec_v2 §8.2) — the same graceful degradation as .txt / .md.

WebParser

Bases: DoclingParser

Parser for the web bucket (.html / .htm / .xhtml).

Source code in varagity/ingest/parsers/web.py 
16
17
18
@register("web")
class WebParser(DoclingParser):
    """Parser for the ``web`` bucket (``.html`` / ``.htm`` / ``.xhtml``)."""

varagity.ingest.parsers.image

Image parser: single-pass Docling conversion with OCR always on.

Bitmap formats carry no text layer, so unlike the PDF parser's two-pass fallback there is nothing to try first: every conversion runs the configured OCR_ENGINE in full-page mode through the same markdown/ table/provenance pipeline as the other Docling-backed parsers (docling_base). Documents carry extraction="ocr" on every chunk — distinct from the PDF parser's "ocr_fallback", because OCR is this format's only extraction path, not a recovery.

Docling imports are deferred to call time: importing this module (which happens on every CLI start via parser self-registration) must not pay for Docling's model machinery.

ImageParser

Parser for the image bucket: Docling's image pipeline, OCR-only.

Converters are built lazily and cached per OCR configuration (the registry instantiates one parser per process; tests and long-lived processes may change OCR settings between calls), so the layout/OCR models load once per configuration, not once per file.

Source code in varagity/ingest/parsers/image.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@register("image")
class ImageParser:
    """Parser for the ``image`` bucket: Docling's image pipeline, OCR-only.

    Converters are built lazily and cached per OCR configuration (the
    registry instantiates one parser per process; tests and long-lived
    processes may change OCR settings between calls), so the layout/OCR
    models load once per configuration, not once per file.
    """

    def __init__(self) -> None:
        """Initialize the converter cache (nothing heavy happens here)."""
        # Keyed by (engine, languages) — full-page OCR is unconditional
        # for images, so it is not part of the key.
        self._converters: dict[tuple[str, tuple[str, ...]], DocumentConverter] = {}

    def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
        """OCR an image into markdown text with provenance.

        Args:
            path: The image file to convert.
            verbose: Console verbosity (0–2); defaults to
                ``settings.DEFAULT_VERBOSE``.

        Returns:
            The extracted document. ``source_meta`` carries ``page``
            (``1`` when any text was recognized — Docling models an image
            as a one-page document) and ``extraction="ocr"``.

        Raises:
            ValueError: If ``verbose`` is invalid.
            docling.exceptions.ConversionError: If conversion fails (a
                corrupt or unreadable image) — the loader counts the file
                as failed and the run continues.
        """
        settings = get_settings()
        check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
        text, page_counts = self._convert(path)
        return raw_document(path, text, page_counts, extraction="ocr")

    def _convert(self, path: Path) -> tuple[str, dict[int, int]]:
        """Run the (cached) OCR converter on one image.

        Args:
            path: The image to convert.

        Returns:
            The normalized markdown text and per-page character counts.
        """
        result = self._converter().convert(path)  # raises ConversionError on failure
        document = result.document
        return export_markdown(document), page_char_counts(document)

    def _converter(self) -> "DocumentConverter":
        """Return the converter for the current OCR settings, building lazily.

        Returns:
            The (possibly cached) Docling converter, configured for
            full-page OCR — an image has no embedded text layer to
            preserve, so there is nothing for partial-page OCR to merge
            with.
        """
        from docling.datamodel.base_models import InputFormat
        from docling.datamodel.pipeline_options import PdfPipelineOptions
        from docling.document_converter import DocumentConverter, ImageFormatOption

        settings = get_settings()
        languages = settings.ocr_language_list
        key = (settings.OCR_ENGINE, tuple(languages))
        if key not in self._converters:
            options = PdfPipelineOptions(
                do_ocr=True,
                ocr_options=get_ocr_options(settings.OCR_ENGINE, languages, force_full_page=True),
            )
            self._converters[key] = DocumentConverter(
                format_options={InputFormat.IMAGE: ImageFormatOption(pipeline_options=options)}
            )
        return self._converters[key]

__init__()

Initialize the converter cache (nothing heavy happens here).

Source code in varagity/ingest/parsers/image.py 
43
44
45
46
47
def __init__(self) -> None:
    """Initialize the converter cache (nothing heavy happens here)."""
    # Keyed by (engine, languages) — full-page OCR is unconditional
    # for images, so it is not part of the key.
    self._converters: dict[tuple[str, tuple[str, ...]], DocumentConverter] = {}

extract(path, verbose=None)

OCR an image into markdown text with provenance.

Parameters:

Name Type Description Default
path Path

The image file to convert.

 required
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None

Returns:

Type Description
RawDocument

The extracted document. source_meta carries page
RawDocument

(1 when any text was recognized — Docling models an image
RawDocument

as a one-page document) and extraction="ocr".

Raises:

Type Description
ValueError

If verbose is invalid.
ConversionError

If conversion fails (a corrupt or unreadable image) — the loader counts the file as failed and the run continues.
Source code in varagity/ingest/parsers/image.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def extract(self, path: Path, verbose: int | None = None) -> RawDocument:
    """OCR an image into markdown text with provenance.

    Args:
        path: The image file to convert.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.

    Returns:
        The extracted document. ``source_meta`` carries ``page``
        (``1`` when any text was recognized — Docling models an image
        as a one-page document) and ``extraction="ocr"``.

    Raises:
        ValueError: If ``verbose`` is invalid.
        docling.exceptions.ConversionError: If conversion fails (a
            corrupt or unreadable image) — the loader counts the file
            as failed and the run continues.
    """
    settings = get_settings()
    check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    text, page_counts = self._convert(path)
    return raw_document(path, text, page_counts, extraction="ocr")

varagity.ingest.loader

Ingestion orchestrator: parse → chunk → contextualize → embed → store (spec §9).

Contextualization (spec §9.4): when settings.CONTEXTUALIZE is on, each chunk gets an LLM-generated situating blurb and contextualized_content = context + "\n\n" + content is what gets embedded; when off, the identity path (context = None, contextualized_content = content) is preserved — the non-contextual eval baseline (plan decision #2). A document's chunks are contextualized sequentially, in order, so llama.cpp can reuse its prompt cache across the shared document preamble.

Dual-write (spec §9.6): every chunk lands in both stores — Elasticsearch (contextual BM25) first, pgvector second — within the same per-file boundary; a store failure after client-level retries fails that file's ingest loudly. The ordering is deliberate: the pgvector documents row is the idempotency marker, so it must commit last — a failure between the two writes leaves no marker, the next run re-attempts the file, and the Elasticsearch bulk (addressed by deterministic chunk_id) overwrites rather than duplicates.

Idempotency: a file whose (doc_id, content_hash) is already recorded is skipped before parsing. Pipeline-setting changes (CONTEXTUALIZE, chunk params) don't change content hashes, so re-processing an unchanged corpus requires reingest=True (the CLI's ingest --reingest), which deletes each discovered document from both stores before ingesting it fresh. A file with no extractable text is never silently dropped: it gets a documents row with n_chunks = 0, a warning, and a dedicated summary count.

Exclusion: a run holds the corpus's advisory lock shared from before its first write until it returns (ADR-019). Ingests overlap each other as they always have; a backup restore — which holds it exclusive — refuses an ingest up front with :class:~varagity.stores.vector_store.CorpusLocked (and is refused by one), whichever process each runs in.

Orchestration seam: each spec §9 stage is a named module function (:func:parse_document, :func:chunk_document, :func:contextualize_chunks, :func:embed_chunks, :func:store_chunks), and the run loop invokes them through an :class:IngestStages bundle. By default the bundle holds the functions themselves; varagity.pipeline.ingest_flow passes task-wrapped equivalents so every stage becomes a tracked, retryable Prefect task run — one orchestration loop, with or without Prefect.

IngestStages dataclass

Call seam through which the run loop invokes each spec §9 stage.

Defaults are the plain stage functions in this module, so constructing the bundle without arguments changes nothing. varagity.pipeline substitutes Prefect @task-wrapped equivalents (same signatures), so the same loop yields tracked, retryable task runs — the orchestration logic exists once.

Attributes:

Name Type Description
discover Callable[..., Buckets]

Corpus discovery (spec §9.1).
parse Callable[..., RawDocument]

Text extraction (spec §9.2).
chunk Callable[..., list[Document]]

Chunking (spec §9.3).
contextualize Callable[..., list[str | None]]

Situating-blurb generation (spec §9.4).
embed Callable[..., list[list[float]]]

Passage embedding (spec §9.5).
store Callable[..., Any]

Dual-store write (spec §9.6).
Source code in varagity/ingest/loader.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
@dataclass(frozen=True)
class IngestStages:
    """Call seam through which the run loop invokes each spec §9 stage.

    Defaults are the plain stage functions in this module, so constructing
    the bundle without arguments changes nothing. ``varagity.pipeline``
    substitutes Prefect ``@task``-wrapped equivalents (same signatures), so
    the same loop yields tracked, retryable task runs — the orchestration
    logic exists once.

    Attributes:
        discover: Corpus discovery (spec §9.1).
        parse: Text extraction (spec §9.2).
        chunk: Chunking (spec §9.3).
        contextualize: Situating-blurb generation (spec §9.4).
        embed: Passage embedding (spec §9.5).
        store: Dual-store write (spec §9.6).
    """

    discover: Callable[..., Buckets] = field(default=discover_documents)
    parse: Callable[..., RawDocument] = field(default=parse_document)
    chunk: Callable[..., list[Document]] = field(default=chunk_document)
    contextualize: Callable[..., list[str | None]] = field(default=contextualize_chunks)
    embed: Callable[..., list[list[float]]] = field(default=embed_chunks)
    store: Callable[..., Any] = field(default=store_chunks)

IngestSummary dataclass

Counters for one ingest run, rendered by the CLI summary table.

Attributes:

Name Type Description
discovered int

Files found in the corpus buckets.
ingested int

Files parsed, chunked, embedded, and stored this run.
skipped int

Unchanged files skipped via the idempotency check.
no_text int

Files with no extractable text (recorded as 0-chunk documents; includes known-empty files seen again).
unsupported int

Files whose bucket has no registered parser (a defensive counter — every v1 bucket has one; it guards future buckets added to discovery before their parser lands).
failed int

Files that raised during ingestion (logged, run continues).
chunks int

Total chunks stored this run.
Source code in varagity/ingest/loader.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@dataclass
class IngestSummary:
    """Counters for one ingest run, rendered by the CLI summary table.

    Attributes:
        discovered: Files found in the corpus buckets.
        ingested: Files parsed, chunked, embedded, and stored this run.
        skipped: Unchanged files skipped via the idempotency check.
        no_text: Files with no extractable text (recorded as 0-chunk
            documents; includes known-empty files seen again).
        unsupported: Files whose bucket has no registered parser (a
            defensive counter — every v1 bucket has one; it guards future
            buckets added to discovery before their parser lands).
        failed: Files that raised during ingestion (logged, run continues).
        chunks: Total chunks stored this run.
    """

    discovered: int = 0
    ingested: int = 0
    skipped: int = 0
    no_text: int = 0
    unsupported: int = 0
    failed: int = 0
    chunks: int = 0

chunk_document(raw, *, verbose)

Split a parsed document with the configured strategy (spec §9.3).

Parameters:

Name Type Description Default
raw RawDocument

The parsed document (text + provenance).

 required
verbose int

Validated console verbosity.

 required

Returns:

Type Description
list[Document]

The chunks, provenance seeded into each chunk's metadata.

Raises:

Type Description
KeyError

If settings.CHUNKING_STRATEGY names an unregistered strategy.
Source code in varagity/ingest/loader.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def chunk_document(raw: RawDocument, *, verbose: int) -> list[Document]:
    """Split a parsed document with the configured strategy (spec §9.3).

    Args:
        raw: The parsed document (text + provenance).
        verbose: Validated console verbosity.

    Returns:
        The chunks, provenance seeded into each chunk's metadata.

    Raises:
        KeyError: If ``settings.CHUNKING_STRATEGY`` names an unregistered
            strategy.
    """
    chunker = get_chunker(get_settings().CHUNKING_STRATEGY)
    return chunker.split(raw.text, source_meta=raw.source_meta, verbose=verbose)

contextualize_chunks(*, document_text, chunk_texts, llm, file_name, progress, verbose)

Generate one situating blurb per chunk, or the identity path (spec §9.4).

The document is fitted into its prompt budget once (:func:~varagity.context.contextual.fit_document_preamble — a served token count per document, not per chunk), then chunks are processed sequentially, in document order: every call shares the byte-identical preamble and llama.cpp reuses its prompt cache. Progress is a per-chunk sub-bar under the run's file bar.

Parameters:

Name Type Description Default
document_text str

The parent document's full extracted text.

 required
chunk_texts list[str]

The document's chunk texts, in order.

 required
llm LLMClient | None

Chat client for contextualization, or None for the identity path (every context None — settings.CONTEXTUALIZE off).

 required
file_name str

Source file name (labels the sub-progress bar).

 required
progress Progress

The run's progress display.

 required
verbose int

Validated console verbosity.

 required

Returns:

Type Description
list[str | None]

One context blurb per chunk, in chunk order (all None when
list[str | None]

llm is None).
Source code in varagity/ingest/loader.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
def contextualize_chunks(
    *,
    document_text: str,
    chunk_texts: list[str],
    llm: LLMClient | None,
    file_name: str,
    progress: Progress,
    verbose: int,
) -> list[str | None]:
    """Generate one situating blurb per chunk, or the identity path (spec §9.4).

    The document is fitted into its prompt budget once
    (:func:`~varagity.context.contextual.fit_document_preamble` — a served
    token count per document, not per chunk), then chunks are processed
    sequentially, in document order: every call shares the byte-identical
    preamble and llama.cpp reuses its prompt cache. Progress is a per-chunk
    sub-bar under the run's file bar.

    Args:
        document_text: The parent document's full extracted text.
        chunk_texts: The document's chunk texts, in order.
        llm: Chat client for contextualization, or ``None`` for the identity
            path (every context ``None`` — ``settings.CONTEXTUALIZE`` off).
        file_name: Source file name (labels the sub-progress bar).
        progress: The run's progress display.
        verbose: Validated console verbosity.

    Returns:
        One context blurb per chunk, in chunk order (all ``None`` when
        ``llm`` is ``None``).
    """
    if llm is None:
        return [None] * len(chunk_texts)
    preamble = fit_document_preamble(document_text, llm=llm)
    contexts: list[str | None] = []
    sub_task: TaskID = progress.add_task(f"  ↳ contextualizing {file_name}", total=len(chunk_texts))
    try:
        for chunk_text in chunk_texts:
            contexts.append(situate_context(preamble, chunk_text, llm=llm, verbose=verbose))
            progress.advance(sub_task)
    finally:
        progress.remove_task(sub_task)
    return contexts

embed_chunks(texts, *, embeddings, verbose)

Embed contextualized chunk texts in e5 passage mode (spec §9.5).

Parameters:

Name Type Description Default
texts list[str]

The contextualized_content of each chunk, in order.

 required
embeddings EmbeddingsClient

The embeddings client.

 required
verbose int

Validated console verbosity.

 required

Returns:

Type Description
list[list[float]]

One embedding vector per text, in order.

Raises:

Type Description
APIError

If embedding still fails after client retries.
Source code in varagity/ingest/loader.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def embed_chunks(
    texts: list[str], *, embeddings: EmbeddingsClient, verbose: int
) -> list[list[float]]:
    """Embed contextualized chunk texts in e5 passage mode (spec §9.5).

    Args:
        texts: The ``contextualized_content`` of each chunk, in order.
        embeddings: The embeddings client.
        verbose: Validated console verbosity.

    Returns:
        One embedding vector per text, in order.

    Raises:
        openai.APIError: If embedding still fails after client retries.
    """
    return embeddings.embed_passages(texts, verbose=verbose)

file_timestamps(path)

Read a file's filesystem timestamps — document provenance, not ingest time.

These land on every chunk's metadata record (file_created_at / file_modified_at), answering "how old is this source?" independently of created_at (which records when the chunk was ingested).

Birth time is best-effort by nature: os.stat exposes st_birthtime on macOS/Windows but not on Linux (CPython doesn't surface statx), so :func:_birthtime_fallback shells out to GNU coreutils there; filesystems without birth times, or a copy/download that reset them, yield None rather than a guess (st_ctime is inode-change time, not creation, and would lie).

Parameters:

Name Type Description Default
path Path

The file to stat.

 required

Returns:

Type Description
datetime | None

A (created, modified) pair of aware UTC datetimes; either is
datetime | None

None when unavailable (both, if the file cannot be stat'd).
Source code in varagity/ingest/loader.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def file_timestamps(path: Path) -> tuple[datetime | None, datetime | None]:
    """Read a file's filesystem timestamps — document provenance, not ingest time.

    These land on every chunk's metadata record (``file_created_at`` /
    ``file_modified_at``), answering "how old is this source?" independently
    of ``created_at`` (which records when the chunk was ingested).

    Birth time is best-effort by nature: ``os.stat`` exposes
    ``st_birthtime`` on macOS/Windows but not on Linux (CPython doesn't
    surface ``statx``), so :func:`_birthtime_fallback` shells out to GNU
    coreutils there; filesystems without birth times, or a copy/download
    that reset them, yield ``None`` rather than a guess (``st_ctime`` is
    inode-change time, not creation, and would lie).

    Args:
        path: The file to stat.

    Returns:
        A ``(created, modified)`` pair of aware UTC datetimes; either is
        ``None`` when unavailable (both, if the file cannot be stat'd).
    """
    try:
        stat = path.stat()
    except OSError:
        return None, None
    modified = datetime.fromtimestamp(stat.st_mtime, tz=UTC)
    birth = getattr(stat, "st_birthtime", None) or _birthtime_fallback(path)
    created = None if birth is None else datetime.fromtimestamp(birth, tz=UTC)
    return created, modified

ingest_corpus(docs_path=None, *, store=None, bm25=None, embeddings=None, llm=None, reingest=False, verbose=None, stages=None, on_file=None)

Ingest every supported document under docs_path into both stores.

Parameters:

Name Type Description Default
docs_path str | None

Corpus directory; defaults to settings.DOCS_PATH.

 None
store ContextualVectorDB | None

Vector store to write to; constructed from settings (and closed on return) when omitted.

 None
bm25 ElasticsearchBM25 | None

BM25 store to write to; constructed from settings (and closed on return) when omitted. Its index is created idempotently at run start.

 None
embeddings EmbeddingsClient | None

Embeddings client; resolved via the model registry when omitted.

 None
llm LLMClient | None

Chat client for contextualization; resolved via the model registry when omitted. Unused when settings.CONTEXTUALIZE is off.

 None
reingest bool

Delete each discovered document's previous ingest (from both stores) and re-process it. Needed after pipeline-setting changes (CONTEXTUALIZE, chunk params): those don't change content hashes, so unchanged files are otherwise skipped.

 False
verbose int | None

Console verbosity (0–2); defaults to settings.DEFAULT_VERBOSE.

 None
stages IngestStages | None

Per-stage call seam; the plain stage functions when omitted. varagity.pipeline.ingest_flow passes task-wrapped stages so each stage is a tracked Prefect task run.

 None
on_file Callable[[Path, str, int], None] | None

Observer called after each file with (path, outcome, chunks_stored), where outcome is "ingested" / "skipped" / "no_text" / "failed" — the seam the API's live ingest-progress stream rides (spec_v2 §4.2). A raising observer is logged and ignored: progress reporting must never fail a run.

 None

Returns:

Type Description
IngestSummary

The run's counters (one file failing is counted and logged, not
IngestSummary

raised — a bad file must not abort a corpus run).

Raises:

Type Description
ValueError

If verbose is invalid.
OperationalError

If the vector store is unreachable.
ConnectionError

If Elasticsearch is unreachable after retries (index creation at run start).
CorpusLocked

If a backup restore holds the corpus — nothing was written.
Source code in varagity/ingest/loader.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
def ingest_corpus(
    docs_path: str | None = None,
    *,
    store: ContextualVectorDB | None = None,
    bm25: ElasticsearchBM25 | None = None,
    embeddings: EmbeddingsClient | None = None,
    llm: LLMClient | None = None,
    reingest: bool = False,
    verbose: int | None = None,
    stages: IngestStages | None = None,
    on_file: Callable[[Path, str, int], None] | None = None,
) -> IngestSummary:
    """Ingest every supported document under ``docs_path`` into both stores.

    Args:
        docs_path: Corpus directory; defaults to ``settings.DOCS_PATH``.
        store: Vector store to write to; constructed from settings (and
            closed on return) when omitted.
        bm25: BM25 store to write to; constructed from settings (and closed
            on return) when omitted. Its index is created idempotently at
            run start.
        embeddings: Embeddings client; resolved via the model registry when
            omitted.
        llm: Chat client for contextualization; resolved via the model
            registry when omitted. Unused when ``settings.CONTEXTUALIZE``
            is off.
        reingest: Delete each discovered document's previous ingest (from
            both stores) and re-process it. Needed after pipeline-setting
            changes (``CONTEXTUALIZE``, chunk params): those don't change
            content hashes, so unchanged files are otherwise skipped.
        verbose: Console verbosity (0–2); defaults to
            ``settings.DEFAULT_VERBOSE``.
        stages: Per-stage call seam; the plain stage functions when omitted.
            ``varagity.pipeline.ingest_flow`` passes task-wrapped stages so
            each stage is a tracked Prefect task run.
        on_file: Observer called after each file with ``(path, outcome,
            chunks_stored)``, where outcome is ``"ingested"`` /
            ``"skipped"`` / ``"no_text"`` / ``"failed"`` — the seam the
            API's live ingest-progress stream rides (spec_v2 §4.2). A
            raising observer is logged and ignored: progress reporting
            must never fail a run.

    Returns:
        The run's counters (one file failing is counted and logged, not
        raised — a bad file must not abort a corpus run).

    Raises:
        ValueError: If ``verbose`` is invalid.
        psycopg.OperationalError: If the vector store is unreachable.
        elastic_transport.ConnectionError: If Elasticsearch is unreachable
            after retries (index creation at run start).
        varagity.stores.vector_store.CorpusLocked: If a backup restore holds
            the corpus — nothing was written.
    """
    settings = get_settings()
    verbose = check_verbose(settings.DEFAULT_VERBOSE if verbose is None else verbose)
    root = Path(docs_path if docs_path is not None else settings.DOCS_PATH)
    stages = stages if stages is not None else IngestStages()

    buckets = stages.discover(str(root), verbose=verbose)
    summary = IngestSummary(discovered=buckets.total)

    # Resolve parsers up front; a bucket without one is
    # counted and skipped loudly, never silently dropped.
    work: list[tuple[Parser, Path]] = []
    for bucket_name, parser_name in _BUCKET_PARSERS:
        paths: list[Path] = getattr(buckets, bucket_name)
        if not paths:
            continue
        try:
            parser = get_parser(parser_name)
        except KeyError:
            logger.warning(
                "no parser registered for %r — skipping %d file(s)",
                parser_name,
                len(paths),
            )
            summary.unsupported += len(paths)
            continue
        work.extend((parser, path) for path in paths)

    owns_store = store is None
    active_store = store if store is not None else ContextualVectorDB()
    owns_bm25 = bm25 is None
    active_bm25 = bm25 if bm25 is not None else ElasticsearchBM25()
    try:
        # Shared, for the whole run, before the first write (the index
        # create): a restore can't empty the tables or rebuild the index
        # under this run, and this run can't write into a restore.
        with active_store.corpus_lock(exclusive=False):
            _ingest_work(
                work,
                root=root,
                store=active_store,
                bm25=active_bm25,
                embeddings=embeddings,
                llm=llm,
                settings=settings,
                reingest=reingest,
                verbose=verbose,
                stages=stages,
                on_file=on_file,
                summary=summary,
            )
    finally:
        if owns_store:
            active_store.close()
        if owns_bm25:
            active_bm25.close()
    return summary

parse_document(parser, path, *, verbose)

Extract a file's text and provenance (spec §9.2).

Parameters:

Name Type Description Default
parser Parser

Parser for the file's discovery bucket.

 required
path Path

The file to parse.

 required
verbose int

Validated console verbosity.

 required

Returns:

Type Description
RawDocument

The extracted document.
Source code in varagity/ingest/loader.py 
174
175
176
177
178
179
180
181
182
183
184
185
def parse_document(parser: Parser, path: Path, *, verbose: int) -> RawDocument:
    """Extract a file's text and provenance (spec §9.2).

    Args:
        parser: Parser for the file's discovery bucket.
        path: The file to parse.
        verbose: Validated console verbosity.

    Returns:
        The extracted document.
    """
    return parser.extract(path, verbose=verbose)

store_chunks(records, vectors, *, store, bm25)

Write one document's chunks to both stores (spec §9.6).

Both writes share one boundary, BM25 first: the pgvector documents row is the idempotency marker and must commit last, so a failure in between leaves the file re-attemptable on the next run (the deterministic chunk_id-addressed BM25 docs then overwrite rather than duplicate). The parent-document row's fields are taken from the records, which all belong to one document.

Parameters:

Name Type Description Default
records list[ChunkRecord]

The document's chunk records (non-empty; the empty- extraction guard runs before this stage).

 required
vectors list[list[float]]

One embedding per record, in order.

 required
store ContextualVectorDB

The vector store.

 required
bm25 ElasticsearchBM25

The BM25 store.

 required

Raises:

Type Description
ValueError

If records is empty.
Source code in varagity/ingest/loader.py 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def store_chunks(
    records: list[ChunkRecord],
    vectors: list[list[float]],
    *,
    store: ContextualVectorDB,
    bm25: ElasticsearchBM25,
) -> None:
    """Write one document's chunks to both stores (spec §9.6).

    Both writes share one boundary, BM25 first: the pgvector ``documents``
    row is the idempotency marker and must commit last, so a failure in
    between leaves the file re-attemptable on the next run (the
    deterministic ``chunk_id``-addressed BM25 docs then overwrite rather
    than duplicate). The parent-document row's fields are taken from the
    records, which all belong to one document.

    Args:
        records: The document's chunk records (non-empty; the empty-
            extraction guard runs before this stage).
        vectors: One embedding per record, in order.
        store: The vector store.
        bm25: The BM25 store.

    Raises:
        ValueError: If ``records`` is empty.
    """
    if not records:
        raise ValueError("store_chunks requires at least one record")
    head = records[0]
    bm25.index_chunks(records)
    store.store_document(
        doc_id=head.doc_id,
        source=head.source,
        file_type=head.file_type,
        content_hash=head.content_hash,
        records=records,
        embeddings=vectors,
    )