Integrar tu flujo de trabajo (DVC · MLflow · Dagster)

✅ Estable — DVC, MLflow y Dagster probados en CI sobre la misma loan; lakehouse pendiente.

El motor froga nunca importa ninguna herramienta MLOps. La reproducibilidad del pipeline se delega al seam Reproducer: un adapter por herramienta que implementa la misma interfaz. El campo pipeline.tool en froga.yaml determina qué adapter se activa cuando el motor invoca froga run.

La prueba concreta de este agnosticismo es el escenario loan: el mismo código de evaluación (compliance_eval.py), el mismo tratamiento (train.py), el mismo programa de riesgos (la sección risk: de froga.yaml) — y tres backends distintos, verificados en CI.

---

Categorías MLOps y herramientas soportadas

Cat.	Paradigma	pipeline.tool	Estado
1	Git-native / ficheros versionados	dvc	Estable
2	Experimento → Registry	mlflow	Estable
3	Grafo de assets con linaje	dagster	Estable
4	Lakehouse / tablas	(pendiente)	Futuro

Cat. 4 — lakehouse

La integración con plataformas de tipo lakehouse (Apache Iceberg, Delta Lake) es una línea futura. El seam Reproducer está diseñado para recibirla sin modificar el núcleo.

---

DVC (cat. 1 — git-native)

DVC extiende git para versionar datos y modelos con contenido addressable, y describe el pipeline como un DAG en dvc.yaml. Es la categoría 1 porque la unidad de cambio es un fichero versionado en git.

Configuración en froga.yaml

froga.yaml (extracto — DVC)

pipeline: { tool: dvc, metrics: metrics.json }

DAG multi-stage

El escenario loan define dos stages: featurize prepara el dataset vía Croissant (§2 del Anexo IV) y produce data/features.parquet cacheado; evaluate entrena el modelo y escribe metrics.json y model.pkl.

dvc.yaml — pipeline loan

stages:
  featurize:
    cmd: .venv/bin/python featurize.py
    deps:
      - featurize.py
      - compliance_eval.py
      - data/german_credit.csv
      - data/german_credit.croissant.json
    outs:
      - data/features.parquet
  evaluate:
    cmd: .venv/bin/python evaluate.py
    deps:
      - evaluate.py
      - compliance_eval.py
      - train.py          # el TRATAMIENTO — su cambio marca evaluate como stale
      - data/features.parquet
      - shared_data/policies/assessment_plan.oscal.yaml
    params:
      - seed
    outs:
      - model.pkl:
          cache: true
    metrics:
      - metrics.json:
          cache: false

Staleness selectiva y deriva tipada

dvc repro recomputa únicamente los stages cuyas dependencias han cambiado. Cuando el tratamiento consiste en sustituir train.py (V1 → V2), solo el stage evaluate queda obsoleto; featurize permanece en caché porque el dato no cambió. Este comportamiento es la expresión concreta de la deriva tipada clase B (modelo) del motor: el digest de train.py entra en la fase modelo, y featurize (clase C, datos) no se toca.

froga run invoca dvc repro, lee metrics.json y ancla el digest de dvc.lock en el bundle de evidencia firmado (pipeline_lock_digest). De este modo, el lock del pipeline es parte de la evidencia.

Esquema de metrics.json (contrato común a todos los backends)

metrics.json es un objeto JSON cuyas claves son control_id (los identificadores de los controles del assessment_plan.oscal.yaml). Cada valor admite dos formas, y el motor las acepta ambas en el mismo fichero:

metrics.json — forma escalar (legacy) y forma con power

{
  "unfair-credit-exclusion": {
    "value": 0.046,
    "power": {
      "n": 200,
      "ci_low": 0.003,
      "ci_high": 0.174,
      "ci_level": 0.95,
      "method": "bootstrap",
      "n_boot": 1000,
      "seed": 42
    }
  },
  "global_dice": {
    "value": 0.91,
    "power": {
      "n": 1000,
      "ci_low": 0.0,
      "ci_high": 0.071,
      "ci_level": 0.95,
      "method": "bootstrap",
      "n_boot": 1000,
      "seed": 42
    }
  }
}

• Forma escalar — "control_id": <número>. El valor medido, sin fiabilidad estadística. Compatible con el SDK < 0.6.11; el motor no emite advertencia de power-stats para esa métrica.
• Forma con power — "control_id": { "value": <número>, "power": { … } }. El bloque power (opcional) lleva el intervalo de confianza bootstrap que alimenta la advertencia de subpoderado.

Cuando el bloque power está presente, estos campos son obligatorios (un power al que le falte cualquiera de ellos hace fallar la lectura de metrics.json ruidosamente; no se ignora):

Campo	Tipo	Significado
n	entero	Tamaño de muestra efectivo (filas).
ci_low	número	Extremo inferior del IC percentil del estimador.
ci_high	número	Extremo superior del IC percentil.
ci_level	número	Nivel de confianza (p.ej. 0.95).
method	texto	Método del IC ("bootstrap" o "cluster_bootstrap").
n_boot	entero	Nº de remuestreos del bootstrap (B).
seed	entero	Semilla fija del bootstrap (determinismo byte-a-byte).

Campos opcionales de power: n_clusters (entero — nº de clústeres si el control declaró input.cluster) y groups (objeto {"<grupo>": <entero>} — tamaño por subgrupo cuando el control trocea por grupo). Quien produce metrics.json es el venturalitica-sdk; este es el contrato que el motor consume, idéntico para DVC, MLflow y Dagster.

---

MLflow (cat. 2 — experimento → registry)

MLflow gestiona el ciclo experimento → registro → promoción. La unidad de cambio es una versión de modelo en el Model Registry; la promoción a @champion es el tratamiento.

Configuración en froga.yaml

froga.yaml (extracto — MLflow)

pipeline: { tool: mlflow, metrics: metrics.json }

Entry point MLflow

El adapter espera un eval/mlflow_entry.py que abra un run de MLflow, ejecute el eval agnóstico, registre el modelo real en el Registry y, si el control bloqueante pasa, lo promueva al alias @champion:

eval/mlflow_entry.py (extracto)

with mlflow.start_run() as run:
    _, model = compliance_eval.run(train.build_model)
    metrics = json.load(open("metrics.json"))
    info = mlflow.sklearn.log_model(model, name="model",
                                    registered_model_name=REGISTERED_MODEL)
    version = info.registered_model_version
    if _val(metrics.get("unfair-credit-exclusion", 1.0)) < 0.092:
        client.set_registered_model_alias(REGISTERED_MODEL, "champion", version)

El store es local (file:./mlruns) y no requiere servidor. MlflowReproducer lee las métricas del run via mlflow runs describe y las entrega al motor.

Separación de concerns

mlflow_entry.py contiene únicamente la integración con MLflow (run, log_model, alias). La medición (venturalitica-sdk) vive en compliance_eval.py y el tratamiento en train.py, idénticos a los del escenario DVC. Leer este fichero no muestra nada de DVC ni de Dagster.

---

Dagster (cat. 3 — grafo de assets con linaje)

Dagster materializa assets con linaje explícito (code_version) y permite declarar checks de calidad nativos (asset_check). La staleness se detecta por code_version, derivado del hash del fichero de tratamiento.

Configuración en froga.yaml

froga.yaml (extracto — Dagster)

pipeline: { tool: dagster, metrics: metrics.json }

Definición del grafo

El escenario define dos assets en cadena y un asset check:

dagster_defs.py (extracto)

def _treatment_code_version() -> str:
    return "train-" + hashlib.sha256(Path("train.py").read_bytes()).hexdigest()[:12]

@asset(code_version=FEATURES_CODE_VERSION)
def credit_features(context):
    rows = len(compliance_eval.load_applications())
    context.add_output_metadata({"rows": rows})
    return rows

@asset(deps=[credit_features], code_version=_treatment_code_version())
def compliance_evaluation(context):
    compliance_eval.run(train.build_model)
    metrics = json.load(open("metrics.json"))
    context.add_output_metadata({k: MetadataValue.float(float(_val(v)))
                                  for k, v in metrics.items()})
    return metrics

@asset_check(asset=compliance_evaluation,
             description="Gate de equidad como check nativo de Dagster")
def unfair_credit_exclusion_gate(context):
    dp = _val(json.load(open("metrics.json")).get("unfair-credit-exclusion", 1.0))
    return AssetCheckResult(passed=dp < 0.092,
                            severity=AssetCheckSeverity.WARN,
                            metadata={"demographic_parity_diff": dp, "threshold": 0.092})

code_version se deriva del hash de train.py. Al sustituir el tratamiento (V1 → V2), el hash cambia y Dagster marca compliance_evaluation como genuinamente stale. froga run invoca dagster asset materialize --select '*' y el asset check actúa como expresión nativa paralela de la barrera de equidad (severidad WARN; el veredicto autoritativo lo emite el motor froga desde el OSCAL).

---

Resumen operativo

Para cada backend, el flujo de comandos es idéntico:

Flujo común (independiente del backend)

uv run froga compile       # programa de riesgos (risk: en froga.yaml) → assessment_plan.oscal.yaml
uv run froga run           # Reproducer → dvc repro | mlflow run | dagster materialize
uv run froga status        # detecta deriva sin recomputar
uv run froga verify        # verifica la firma del bundle
uv run froga reconstruct   # reconstruye el ciclo ISO 23894 por replay de git

Lo único que cambia entre escenarios es el valor de pipeline.tool en froga.yaml y los ficheros de definición del pipeline (dvc.yaml, eval/mlflow_entry.py, dagster_defs.py). El núcleo Rust, los controles, el AssuranceProgram y el eval agnóstico permanecen inalterados.

---

Referencias

• Referencia froga.yaml — campo pipeline y sintaxis completa del manifiesto
• Deriva tipada A/B/C — cómo el tipo de cambio determina qué se recomputa