API reference

import pecca

(
    pecca.connect,
    pecca.profile,
    pecca.train,
    pecca.predict,
    pecca.replace,
    pecca.evaluate,
)
(
    pecca.retrain,
    pecca.audit_pack,
    pecca.register,
    pecca.promote,
    pecca.approve,
    pecca.load,
    pecca.as_tool,
)

Public API. Every function is re-exported from pecca.

connect(source, *, table=None, path=None, columns, input_template=None, **kwargs)

Read a data source into the canonical schema (spec §5.1).

Source code in src/pecca/api.py

def connect(
    source: str | dict[str, Any],
    *,
    table: str | None = None,
    path: str | None = None,
    columns: dict[str, Any],
    input_template: str | None = None,
    **kwargs: Any,
) -> Dataset:
    """Read a data source into the canonical schema (spec §5.1)."""
    cfg: dict[str, Any] = dict(source) if isinstance(source, dict) else {"type": source}
    cfg.setdefault("columns", columns)
    if input_template:
        cfg["input_template"] = input_template
    if table:
        cfg["table"] = table
    if path:
        cfg["path"] = path
    cfg.update(kwargs)
    ds = connectors.create("datasource", cfg)
    df = ds.read()
    return Dataset(df, cfg["columns"], cfg.get("input_template"), source=str(cfg["type"]))

profile(dataset=None, call=None, *, min_rows=500, workspace=None, project=None)

Profile each call in dataset (default: the project's configured datasource).

Source code in src/pecca/api.py

def profile(
    dataset: Dataset | None = None,
    call: str | None = None,
    *,
    min_rows: int = 500,
    workspace: WS = None,
    project: PR = None,
) -> ProfileReport:
    """Profile each call in ``dataset`` (default: the project's configured datasource)."""
    if dataset is None:
        if call is None:
            raise PeccaError(
                "profile() needs a dataset or a call", "profile(dataset) or profile(call=...)"
            )
        dataset = load_dataset(_call(call, workspace, project))
        call = _call(call, workspace, project).name
    return profile_dataset(dataset, call, min_rows)

train(call, dataset=None, *, metric=None, latency_budget_ms=None, target_precision=None, candidates=None, min_rows=None, workspace=None, project=None, progress=None, include_fallbacks=True, include_human_labels=True)

Source code in src/pecca/api.py

def train(
    call: str,
    dataset: Dataset | None = None,
    *,
    metric: str | None = None,
    latency_budget_ms: float | None = None,
    target_precision: float | None = None,
    candidates: list[str] | None = None,
    min_rows: int | None = None,
    workspace: WS = None,
    project: PR = None,
    progress: Callable[[str], None] | None = None,
    include_fallbacks: bool = True,
    include_human_labels: bool = True,
) -> TrainResult:
    c = _call(call, workspace, project)
    pol = c.project.policy(c.name)
    if dataset is None:
        dataset = load_dataset(
            c, include_fallbacks=include_fallbacks, include_human_labels=include_human_labels
        )
    mr = min_rows if min_rows is not None else int(pol.get("min_rows", 500))
    prof: CallProfile = profile_dataset(dataset, c.name, mr).profiles[c.name]
    if not prof.replaceable:
        raise PeccaError(f"{c.path} is not replaceable: {prof.reason}", "see `pecca profile`")
    version = c.next_version()
    with tempfile.TemporaryDirectory() as work:
        out = run_tournament(
            c.name,
            dataset,
            prof,
            version=version,
            workdir=work,
            metric=metric or pol.get("metric"),
            latency_budget_ms=latency_budget_ms
            if latency_budget_ms is not None
            else pol.get("latency_budget_ms"),
            target_precision=target_precision
            if target_precision is not None
            else float(pol.get("target_precision", 0.95)),
            candidates=candidates if candidates is not None else pol.get("candidates"),
            progress=progress,
        )
        mv = out.version
        path = c.registry.save_model(c.key, version, out.artefacts_dir, mv.to_dict())
    st = c.state()
    st.policy = pol
    st.last_trained = iso()
    if st.mode == "live":  # a new model must re-earn live status
        st.mode, st.shadow_since = "shadow", iso()
        c.log_event(st, "retrained_demoted_to_shadow", version=version)
    if st.mode in ("off", "record") or st.current_version is None or st.mode in ("shadow", "live"):
        st.current_version = version
    if not c.has_state():
        st.since = iso()
    c.save_state(st)
    c.notify(
        "trained",
        {"version": version, "message": f"{mv.candidate} {mv.metric_name} {mv.metric:.3f}"},
    )
    return TrainResult(
        version,
        mv.candidate,
        mv.metric_name,
        mv.metric,
        mv.metric_std,
        mv.llm_metric,
        mv.threshold,
        mv.expected_fallback_rate,
        mv.latency_ms,
        mv.leaderboard,
        path,
        mv.lineage,
    )

retrain(call, *, include_fallbacks=True, include_human_labels=True, workspace=None, project=None, **kw)

Source code in src/pecca/api.py

def retrain(
    call: str,
    *,
    include_fallbacks: bool = True,
    include_human_labels: bool = True,
    workspace: WS = None,
    project: PR = None,
    **kw: Any,
) -> TrainResult:
    return train(
        call,
        None,
        include_fallbacks=include_fallbacks,
        include_human_labels=include_human_labels,
        workspace=workspace,
        project=project,
        **kw,
    )

predict(call, inputs, *, workspace=None, project=None)

Source code in src/pecca/api.py

def predict(
    call: str, inputs: list[Any], *, workspace: WS = None, project: PR = None
) -> list[Prediction]:
    return get_predictor(_call(call, workspace, project)).predict(list(inputs))

evaluate(call, since='14d', *, workspace=None, project=None)

Source code in src/pecca/api.py

def evaluate(
    call: str, since: str = "14d", *, workspace: WS = None, project: PR = None
) -> EvalReport:
    return evaluate_call(_call(call, workspace, project), since)

promote(call, mode, *, force=False, workspace=None, project=None)

Source code in src/pecca/api.py

def promote(
    call: str, mode: str, *, force: bool = False, workspace: WS = None, project: PR = None
) -> Decision:
    c = _call(call, workspace, project)
    st = c.state()
    src = st.mode if c.has_state() else "record"
    check_transition(src, mode)
    transition = f"{src}->{mode}"
    if src == mode:
        return Decision(True, transition=transition)
    if mode in ("shadow", "live") and not c.versions():
        raise PeccaError(f"{c.path} has no trained model", f"run `pecca train {c.path}` first")
    dec = Decision(True, transition=transition)
    if transition in _gov.engine.GATED:
        cfg = (c.project.governance.get("transitions") or {}).get(transition) or {}
        if cfg.get("evidence"):
            _audit.generate(c)
        dec = _gov.evaluate(c, transition)
        _ensure_ticket(c, transition, cfg, dec)
    if not dec.allowed:
        if not force:
            return dec
        if os.environ.get("PECCA_ALLOW_FORCE") != "1":
            raise GovernanceError(
                "--force requires PECCA_ALLOW_FORCE=1", "set it explicitly to override gates"
            )
        dec.forced = True
        st = c.state()
        c.log_event(
            st,
            "forced_promotion",
            transition=transition,
            failing_gates=dec.failing_gates,
            pending_approvals=dec.pending_approvals,
            evidence_missing=dec.evidence_missing,
        )
        c.save_state(st)
    st = c.state()
    st.mode, st.since = mode, iso()
    if mode == "shadow" and src in ("record", "off"):
        st.shadow_since = iso()
    if st.current_version is None:
        st.current_version = c.version().version
    c.log_event(st, "mode_change", frm=src, to=mode, forced=dec.forced)
    c.save_state(st)
    c.notify(mode, {"version": st.current_version})
    return dec

approve(call, version, *, approver, ticket=None, note=None, groups=None, workspace=None, project=None)

Source code in src/pecca/api.py

def approve(
    call: str,
    version: str,
    *,
    approver: str,
    ticket: str | None = None,
    note: str | None = None,
    groups: list[str] | None = None,
    workspace: WS = None,
    project: PR = None,
) -> dict[str, Any]:
    c = _call(call, workspace, project)
    c.version(version)  # raises if unknown
    if groups is None:
        try:
            ident = connectors.get("identity", "oidc")().current_principal()
            groups = ident["groups"] if ident.get("user") == approver else []
        except Exception:  # noqa: BLE001
            groups = []
    rec = {
        "version": version,
        "approver": approver,
        "groups": groups,
        "ticket": ticket,
        "note": note,
        "ts": iso(),
    }
    st = c.state()
    st.approvals.append(rec)
    c.log_event(st, "approval", approver=approver, version=version)
    c.save_state(st)
    return rec

audit_pack(call, version=None, *, out='markdown', path=None, workspace=None, project=None)

Source code in src/pecca/api.py

def audit_pack(
    call: str,
    version: str | None = None,
    *,
    out: str = "markdown",
    path: str | None = None,
    workspace: WS = None,
    project: PR = None,
) -> str:
    return _audit.export(_call(call, workspace, project), version, out, path)

load(path)

Load a saved model directory (e.g. downloaded from Hugging Face) for local prediction.

Source code in src/pecca/api.py

def load(path: str) -> Bundle:
    """Load a saved model directory (e.g. downloaded from Hugging Face) for local prediction."""
    return Bundle(path, "local")

as_tool(call, *, project=None, workspace=None, description=None)

A plain typed callable f(text) -> {label, confidence, fallback, version} for agent frameworks.

Source code in src/pecca/api.py

def as_tool(
    call: str, *, project: PR = None, workspace: WS = None, description: str | None = None
) -> Callable[[str], dict[str, Any]]:
    """A plain typed callable ``f(text) -> {label, confidence, fallback, version}`` for agent frameworks."""

    def tool(text: str) -> dict[str, Any]:
        p = predict(call, [text], workspace=workspace, project=project)[0]
        return {
            "label": p.label,
            "confidence": p.confidence,
            "fallback": p.fallback,
            "version": p.version,
        }

    tool.__name__ = call.rsplit("/", 1)[-1]
    tool.__doc__ = description or (
        f"Classify text with the Pecca model for `{call}`. Returns the label, calibrated confidence, "
        "and `fallback=True` when confidence is below the threshold (ask the LLM instead)."
    )
    return tool

Plain dataclasses persisted as JSON via to_dict / from_dict.

CallProfile dataclass

Bases: Serializable

Source code in src/pecca/core/models.py

@dataclass
class CallProfile(Serializable):
    call: str
    n_rows: int
    n_unique_outputs: int
    unique_ratio: float
    task_type: str  # classification | regression | extraction | free_text
    input_type: str  # text | tabular | multi_text
    replaceable: bool
    reason: str | None = None
    labels: list[str] = field(default_factory=list)
    languages: list[str] = field(default_factory=list)
    min_class_count: int | None = None
    n_classes_below_20: int | None = None
    n_human_labels: int = 0

ModelVersion dataclass

Bases: Serializable

Source code in src/pecca/core/models.py

@dataclass
class ModelVersion(Serializable):
    version: str
    candidate: str
    format: str
    task_type: str
    metric_name: str
    metric: float
    metric_std: float
    llm_metric: float | None
    threshold: float | None
    expected_fallback_rate: float | None
    latency_ms: float
    trained_at: str
    target_precision: float = 0.95
    target_met: bool = True
    calibration_kind: str = "none"
    calibration_basis: str = "all_labels"
    holdout_metric: float | None = None
    labels: list[str] = field(default_factory=list)
    leaderboard: list[dict[str, Any]] = field(default_factory=list)
    lineage: dict[str, Any] = field(default_factory=dict)
    confusion_matrix: dict[str, Any] = field(default_factory=dict)
    per_class: dict[str, Any] = field(default_factory=dict)
    calibration_curve: list[list[float]] = field(default_factory=list)
    ece: float | None = None
    environment: dict[str, Any] = field(default_factory=dict)
    git_sha: str | None = None
    input_template: str | None = None
    min_class_count: int | None = None

Prediction dataclass

Bases: Serializable

Source code in src/pecca/core/models.py

@dataclass
class Prediction(Serializable):
    label: str | float
    confidence: float | None
    fallback: bool
    version: str

TrainResult dataclass

Bases: Serializable

Source code in src/pecca/core/models.py

@dataclass
class TrainResult(Serializable):
    version: str
    winner: str
    metric_name: str
    metric: float
    metric_std: float
    llm_metric: float | None
    threshold: float | None
    expected_fallback_rate: float | None
    latency_ms: float
    leaderboard: list[dict[str, Any]]
    model_path: str
    lineage: dict[str, Any]

EvalReport dataclass

Bases: Serializable

Source code in src/pecca/core/models.py

@dataclass
class EvalReport(Serializable):
    agreement_rate: float | None
    fallback_rate: float | None
    n_rows: int
    disagreements: list[dict[str, Any]]
    drift_score: float | None
    per_class_agreement: dict[str, float]
    window: str

Decision dataclass

Bases: Serializable

Source code in src/pecca/core/models.py

@dataclass
class Decision(Serializable):
    allowed: bool
    failing_gates: list[str] = field(default_factory=list)
    pending_approvals: list[str] = field(default_factory=list)
    evidence_missing: list[str] = field(default_factory=list)
    transition: str = ""
    forced: bool = False

CallState dataclass

Bases: Serializable

Source code in src/pecca/core/models.py

@dataclass
class CallState(Serializable):
    mode: str = "record"
    current_version: str | None = None
    since: str | None = None
    policy: dict[str, Any] = field(default_factory=dict)
    approvals: list[dict[str, Any]] = field(default_factory=list)
    tickets: dict[str, str] = field(default_factory=dict)
    last_trained: str | None = None
    last_revalidated: str | None = None
    last_drift_review: str | None = None
    shadow_since: str | None = None
    history: list[dict[str, Any]] = field(default_factory=list)

Bases: ABC

Source code in src/pecca/candidates/base.py

class Candidate(ABC):
    name: ClassVar[str] = ""
    task_types: ClassVar[list[str]] = []
    input_types: ClassVar[tuple[str, ...]] = ("text", "multi_text", "tabular")
    min_rows: ClassVar[int] = 0
    requires_gpu: ClassVar[bool] = False
    estimated_latency_ms: ClassVar[float | None] = None
    format: str = "pickle"  # pickle | onnx | hf  (set after save/load)

    @abstractmethod
    def fit(self, X: list[Any], y: list[Any]) -> None: ...
    @abstractmethod
    def predict_proba(self, X: list[Any]) -> np.ndarray: ...
    @abstractmethod
    def predict(self, X: list[Any]) -> np.ndarray: ...
    @property
    @abstractmethod
    def classes_(self) -> list[str]: ...
    @abstractmethod
    def save(self, path: str) -> None: ...
    @classmethod
    @abstractmethod
    def load(cls, path: str) -> Candidate: ...

    def to_onnx(self, path: str) -> bool:
        """Export to ONNX at ``path``; return False if unsupported."""
        return False

to_onnx(path)

Export to ONNX at path; return False if unsupported.

Source code in src/pecca/candidates/base.py

def to_onnx(self, path: str) -> bool:
    """Export to ONNX at ``path``; return False if unsupported."""
    return False