spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : pipeline.py6# Description : Ingestion pipeline — extract, validate against the AIR schema, route by confidence.7# =============================================================================8"""Ingestion pipeline.910    source text -> Extractor -> schema validation -> routing11                                     |                  |12                                 (Pydantic)     auto-approve  OR  human inbox1314Routing rules (docs/research/llm-structured-extraction.md):15- schema-invalid extraction   -> ALWAYS to the human inbox (with the errors)16- confidence <  threshold     -> human inbox17- confidence >= threshold     -> auto-approved, ready to compile1819The threshold is configurable per home; the default is deliberately20conservative. The LLM's output NEVER goes to the ledger directly — even21auto-approved documents go through the deterministic compiler.22"""23from __future__ import annotations2425import enum26from dataclasses import dataclass, field27from datetime import datetime28from decimal import Decimal29from typing import Any3031from pydantic import ValidationError3233from core.events import AirDocument34from ingestion.extractor import ExtractionResult, Extractor3536DEFAULT_CONFIDENCE_THRESHOLD = Decimal("0.85")373839class Route(str, enum.Enum):40    AUTO_APPROVED = "auto_approved"41    NEEDS_REVIEW = "needs_review"424344@dataclass45class IngestionOutcome:46    route: Route47    extraction: ExtractionResult48    document: AirDocument | None = None          # set when schema-valid49    validation_errors: list[str] = field(default_factory=list)50    reasons: list[str] = field(default_factory=list)515253def _stamp_meta(54    events: list[dict[str, Any]],55    extraction: ExtractionResult,56    ingested_at: datetime | None,57) -> list[dict[str, Any]]:58    """Attach traceability metadata to every extracted event."""59    stamped = []60    for event in events:61        meta = dict(event.get("meta") or {})62        meta["llm"] = {63            "model": extraction.extractor,64            "confidence": str(extraction.confidence),65        }66        if ingested_at is not None:67            meta["timestamps"] = {"ingested": ingested_at.isoformat()}68        stamped.append({**event, "meta": meta})69    return stamped707172def ingest(73    text: str,74    extractor: Extractor,75    threshold: Decimal = DEFAULT_CONFIDENCE_THRESHOLD,76    ingested_at: datetime | None = None,77) -> IngestionOutcome:78    """Run one document through extract -> validate -> route."""79    extraction = extractor.extract(text)80    outcome = IngestionOutcome(route=Route.NEEDS_REVIEW, extraction=extraction)8182    events = _stamp_meta(extraction.events, extraction, ingested_at)83    try:84        outcome.document = AirDocument.model_validate({"events": events})85    except ValidationError as exc:86        outcome.validation_errors = [87            f"{'.'.join(str(p) for p in err['loc'])}: {err['msg']}"88            for err in exc.errors()89        ]90        outcome.reasons.append(91            "extraction does not validate against the AIR schema"92        )93        return outcome  # schema-invalid ALWAYS goes to a human9495    if extraction.confidence < threshold:96        outcome.reasons.append(97            f"confidence {extraction.confidence} below threshold {threshold}"98        )99        return outcome100101    outcome.route = Route.AUTO_APPROVED102    return outcome103