typedecide.data.validate

Dataset checks that run without a tokenizer or a model. Every finding code is explained in Read a validation report.

typedecide.data.validate

Telling someone their dataset is broken before they spend a night training on it.

The two checks that earn this module its place are the label balance and the contradiction check. A criterion whose gold is 94% one class trains a model that has learned the prior and nothing else, and it will still score 94%. Two rows with the same evidence, the same question and different answers put a ceiling on accuracy that no amount of training moves, and it is invisible in a metrics table.

DEFAULT_MAX_STATE_TOKENS module-attribute

DEFAULT_MAX_STATE_TOKENS = 1024

IMBALANCE_THRESHOLD module-attribute

IMBALANCE_THRESHOLD = 0.7

SEVERITIES module-attribute

SEVERITIES = ('error', 'warning', 'note')

Finding dataclass

Finding(
    severity: str,
    code: str,
    message: str,
    decisions: tuple[str, ...] = (),
)

One thing worth saying about a dataset. severity is error, warning or note.

severity instance-attribute

severity: str

code instance-attribute

code: str

message instance-attribute

message: str

decisions class-attribute instance-attribute

decisions: tuple[str, ...] = ()

render

render() -> str

Source code in src/typedecide/data/validate.py

def render(self) -> str:
    ids = ""
    if self.decisions:
        shown = ", ".join(self.decisions[:_MAX_LISTED])
        more = (
            f" (+{len(self.decisions) - _MAX_LISTED} more)"
            if len(self.decisions) > _MAX_LISTED
            else ""
        )
        ids = f"\n      rows: {shown}{more}"
    return f"  [{self.severity:<7}] {self.code}: {self.message}{ids}"

DataReport dataclass

DataReport(
    findings: tuple[Finding, ...],
    counts: dict[str, int] = dict(),
    label_balance: dict[str, dict[str, int]] = dict(),
)

What validate found. ok is the one thing a caller must look at.

findings instance-attribute

findings: tuple[Finding, ...]

counts class-attribute instance-attribute

counts: dict[str, int] = field(default_factory=dict)

label_balance class-attribute instance-attribute

label_balance: dict[str, dict[str, int]] = field(
    default_factory=dict
)

ok property

ok: bool

by_severity

by_severity(severity: str) -> tuple[Finding, ...]

Source code in src/typedecide/data/validate.py

def by_severity(self, severity: str) -> tuple[Finding, ...]:
    return tuple(f for f in self.findings if f.severity == severity)

render

render() -> str

Source code in src/typedecide/data/validate.py

def render(self) -> str:
    lines = [
        "Dataset report: " + ("OK" if self.ok else "NOT USABLE AS GIVEN"),
        "  " + ", ".join(f"{k}={v}" for k, v in self.counts.items()),
    ]
    for severity in SEVERITIES:
        group = self.by_severity(severity)
        if group:
            lines.append(f"  -- {severity}s ({len(group)}) --")
            lines.extend(f.render() for f in group)
    if self.label_balance:
        lines.append("  -- label balance --")
        for key in sorted(self.label_balance):
            counts = self.label_balance[key]
            total = sum(counts.values()) or 1
            parts = [
                f"{oid}={n} ({100 * n / total:.0f}%)" for oid, n in sorted(counts.items())
            ]
            lines.append(f"    {key}: " + ", ".join(parts))
    if self.ok and not self.findings:
        lines.append("  Nothing to report.")
    return "\n".join(lines)

estimate_tokens

estimate_tokens(text: str) -> int

A tokenizer-free estimate, so validation never downloads a model.

Deliberately pessimistic: the larger of ~1.3 tokens per whitespace word and one token per four characters. It is a smoke alarm, not a scale.

Source code in src/typedecide/data/validate.py

def estimate_tokens(text: str) -> int:
    """A tokenizer-free estimate, so validation never downloads a model.

    Deliberately pessimistic: the larger of ~1.3 tokens per whitespace word and one
    token per four characters. It is a smoke alarm, not a scale.
    """
    words = len(text.split())
    return max(words + words // 3, (len(text) + 3) // 4)

validate

validate(
    decisions: Sequence[Decision],
    *,
    min_per_class: int = 5,
    max_state_tokens: int = DEFAULT_MAX_STATE_TOKENS
) -> DataReport

Look for the problems that make a training run a waste of an afternoon.

Source code in src/typedecide/data/validate.py

def validate(
    decisions: Sequence[Decision],
    *,
    min_per_class: int = 5,
    max_state_tokens: int = DEFAULT_MAX_STATE_TOKENS,
) -> DataReport:
    """Look for the problems that make a training run a waste of an afternoon."""
    if not decisions:
        return DataReport(
            findings=(
                Finding(
                    "error",
                    "empty",
                    "The dataset has no decisions in it. Fix: check the loader's source path "
                    "and any filter applied after loading.",
                ),
            ),
            counts={"decisions": 0, "labelled": 0, "unlabelled": 0},
            label_balance={},
        )

    findings: list[Finding] = []
    labelled = [d for d in decisions if d.labelled]
    unlabelled = [d for d in decisions if not d.labelled]

    findings.extend(_check_ids(decisions))
    findings.extend(_check_unlabelled(decisions, unlabelled))
    findings.extend(_check_option_sets(decisions))
    findings.extend(_check_contradictions(labelled))
    findings.extend(_check_duplicates(labelled))
    balance = _label_balance(decisions)
    findings.extend(_check_balance(balance))
    findings.extend(_check_rare_classes(balance, min_per_class))
    findings.extend(_check_state_length(decisions, max_state_tokens))
    findings.extend(_check_groups(decisions))

    order = {s: i for i, s in enumerate(SEVERITIES)}
    findings.sort(key=lambda f: (order[f.severity], f.code))

    counts = {
        "decisions": len(decisions),
        "labelled": len(labelled),
        "unlabelled": len(unlabelled),
        "criteria": len({d.criterion.key for d in decisions}),
        "groups": len({d.grouping for d in decisions}),
        "unique_states": len({d.state.strip() for d in decisions}),
        "errors": sum(1 for f in findings if f.severity == "error"),
        "warnings": sum(1 for f in findings if f.severity == "warning"),
    }
    LOGGER.info(
        "validated %d decisions: %d error(s), %d warning(s)",
        len(decisions),
        counts["errors"],
        counts["warnings"],
    )
    return DataReport(findings=tuple(findings), counts=counts, label_balance=balance)