Classifiers

Add a classifier that says what a document, a page or a chunk is, asked only when a request enables it.

A classifier says what a text is — its type, the unit that produced it, the industry it belongs to — as facets of ranked labels. It is asked after every reader has produced its pages, and it is handed text rather than bytes. There are three of them, one per unit: DocumentClassifier sees the whole document, PageClassifier is asked once per page, and ChunkClassifier once per chunk. The protocol is identical in all three — only the unit and the provider hook change — so this page is written for the document one and the delta for the others is at the end. Unlike a language detector they run only when a request names them, because a classifier costs a call and may leave the box. indx-classifier-words, indx-classifier-zeroshot and indx-classifier-llm ship this way, through the same entry-point group your distribution uses. Package and install it as the overview describes.

from collections.abc import Mapping
from typing import Any

from indx_interfaces import (
    CapabilityDescriptor,
    CapabilityId,
    ClassifierId,
    Device,
    DocumentClassifier,
    LabelScore,
)


class PurchaseOrderClassifier:
    # What a request names in `classification.document_ids`. IDs are one
    # namespace across all five classifier and extractor ports, so no two may
    # share this string.
    id = ClassifierId("acme-document-type")
    device = Device.CPU

    def classify(self, text: str) -> Mapping[str, tuple[LabelScore, ...]]:
        # `text` is the full text of your unit. Bound it yourself if you need it
        # bounded. A facet left out is "no opinion": the next enabled classifier
        # is asked for it.
        if "purchase order" not in text.lower():
            return {}
        return {"document_type": (LabelScore(label="purchase_order", confidence=0.8),)}


class Provider:
    def descriptors(self) -> tuple[CapabilityDescriptor, ...]:
        # This distribution annotates what other capabilities read; it reads nothing.
        return ()

    def create(self, capability_id: CapabilityId) -> Any:
        raise ValueError(f"this provider declares no capabilities, so not {capability_id}")

    def document_classifiers(self) -> tuple[DocumentClassifier, ...]:
        return (PurchaseOrderClassifier(),)

DocumentClassifierProvider is optional, like LanguageDetectorProvider. A classifier carries an ID because a request enables it by name, and the installed IDs are advertised on the capability snapshot — outside its content hash, so installing yours moves no plan. A caller enables it with {"classification": {"document_ids": ["acme-document-type"]}} on encode, or --classifier acme-document-type on the CLI; the classifiers named run in that order, and the first with an opinion on a facet wins it.

You receive the full text of your unit, and bounding it is yours. There is no request field and no deployment default that truncates it any more: sampling left the contract, and an implementation that needs a bound — a token window, a cost ceiling — applies its own from its own settings model, using the dependency-free helper in indx-interfaces. A classifier that reads everything it is handed pays for a 300-page filing what it does not pay for a three-page one, and that is now a decision you make rather than one made for you.

For a page or a chunk instead

Change two things. Return the same Mapping[str, tuple[LabelScore, ...]] from the same classify(text), and declare it through page_classifiers() or chunk_classifiers() instead of document_classifiers(); a request then names your ID in classification.page_ids or classification.chunk_ids. Your answer is written to each page block’s metadata, or — for a chunk classifier — to the document block’s chunk_classification, keyed by chunk block ID, because chunk blocks carry no metadata of their own.

Two things to expect. A chunk classifier requires chunk granularity: naming one in a request that did not ask for CHUNK is a 422 before the source is fetched, since there would be nothing to classify. And a per-unit classifier is called once per page or once per chunk rather than once per document, so a model that costs a call now costs several hundred — put per-call setup behind the first call, not in it.

Raising means “mine, and broken” — a verdict about the classifier, not the source — so the executor logs it and asks the next one instead of failing an encode the run already paid to read. Three attributes are read off your class with defaults: device (Device.CPU when absent; declare Device.EXTERNAL for a hosted model, and a request carrying data_residency is refused before the source is fetched rather than silently classified elsewhere), cost_usd, set after classify() and added to the run’s usage, and facets, a tuple[str, ...] of the facet names you can answer. Declare it and you are not called at all when an earlier classifier has already taken every one of them, which is the difference between being asked and being billed; leave it off and you are always asked.

The executor writes the merged answer onto the document block’s metadata under classification. That key is reserved: a caller supplying it on EncodeRequest is refused rather than overwritten.

Checklist

Leave a facet out rather than guess at it; order labels highest first.
Advertise nothing when you cannot run — without your extra, or without a model configured — rather than a classifier that raises.
Keep module scope cheap; the engine import belongs inside classify(), and building the model behind the first call.

Classifiers

For a page or a chunk instead

Checklist

Reference

Developer