Data Catalog (`catalog.yaml`)¶

The catalog.yaml file declares the datasets a flow reads and writes. Instead of hard-coding paths and format logic in your tasks, you name a dataset in the catalog and reference it by name — the framework handles loading, format parsing, saving, schema validation, and MLflow artifact tracking.

flows/
└── my_flow/
    ├── parameters.yaml
    ├── catalog.yaml       # ← this file
    └── flow.py

# flows/my_flow/catalog.yaml
eval_data:                 # ← dataset name (referenced from @task/@flow)
  type: pandas.CSV         # required
  path: data/eval_set.csv  # optional (defaults to "<name>.<ext>")
  schema: myapp.schemas.EvalRow   # optional Pydantic validation

results:
  type: pandas.Parquet
  dataset_type: output

Wiring datasets to tasks and flows¶

Reference a catalog entry by name in input= / output=. The framework loads inputs and injects them as arguments, and saves the return value(s) to the named outputs:

from latent.prefect import flow, task

@task("score", input="eval_data", output="results")
async def score(df):          # df is the loaded eval_data
    scored = await run_judge(df)
    return scored             # saved to the "results" dataset

@flow("my_flow", output=["results"])
async def my_flow():
    return await score()

input / output accept a single name or a list. Multiple outputs map to a returned tuple, position by position.

Anatomy of an entry¶

Key	Required	Description
`type`	✅	Dataset type — selects the format/engine (see the table below)
`path`		File path relative to the workspace. Defaults to `<name>.<ext>` for the type
`schema`		Dotted path to a Pydantic model — validates rows on load (e.g. `myapp.schemas.EvalRow`)
`targets`		Column(s) recorded as MLflow lineage targets
`dataset_type`		`input` (default) or `output` — which area the dataset lives in (see Inputs vs outputs)
type-specific		Reader options like `sheet`, `slug`, `version`, `range` — see Per-type options

All string values support ${VAR} / $VAR environment-variable substitution, same as parameters.yaml.

Dataset types¶

Set the type with the type: key.

Type	Format / Engine	Extension	Loads as
`pandas.CSV`	CSV · pandas	`.csv`	`DataFrame`
`pandas.Parquet`	Parquet · pandas	`.parquet`	`DataFrame`
`pandas.JSON`	JSONL · pandas	`.jsonl`	`DataFrame`
`pandas.Excel`	Excel · pandas	`.xlsx`	`DataFrame` (one sheet)
`pandas.ExcelMultiSheet`	Excel · pandas	`.xlsx`	`dict[str, DataFrame]`
`polars.CSV`	CSV · polars	`.csv`	polars `DataFrame`
`polars.Parquet`	Parquet · polars	`.parquet`	polars `DataFrame`
`polars.Excel` / `polars.ExcelMultiSheet`	Excel · polars	`.xlsx`	`DataFrame` / `dict`
`json.JSON`	JSON · stdlib	`.json`	dict or list
`pickle.Pickle`	Pickle · cloudpickle	`.pkl`	any Python object
`text.Text`	Plain text · stdlib	`.txt`	`str` (or `dict[str, str]` for a directory)
`text.Markdown`	Markdown · stdlib	`.md`	`str` (or `dict[str, str]` for a directory)
`conversation.JSONL`	JSONL · stdlib	`.jsonl`	`list[Conversation]`
`pandas.GoogleSheet`	Live Google Sheet · httpx/Sheets API	—	`DataFrame`
`agent_studio.Dataset`	Remote Agent Studio dataset · httpx	`.jsonl`	`DataFrame`
`agent_studio.Report`	Agent Studio reports API · httpx	`.jsonl`	(output) submits a `StatisticalReport`

Inputs vs outputs¶

A dataset lives in one of two areas under the workspace:

Inputs — data/<flow>/input/ (the default)
Outputs — data/<flow>/output/ (also tracked as MLflow artifacts)

Returning a value to an output= dataset saves it to the output area. A task that needs to read a dataset from the output area — e.g. a previously generated artifact — marks the catalog entry with dataset_type: output:

eval_data:
  type: pandas.CSV          # read from data/my_flow/input/

scored:
  type: pandas.Parquet
  dataset_type: output      # lives in data/my_flow/output/

Cross-flow references¶

To consume another flow's output, reference it by flow_name.dataset_name in input=. The framework loads dataset_name from the source flow's output area — no copy, no manual path:

@task("analyze", input="scoring_flow.scored_data")
async def analyze(scored):
    # `scored` is scoring_flow's output dataset "scored_data"
    ...

@flow("report_flow", input=["scoring_flow.scored_data"])
async def report_flow(scored_data):
    ...

The referenced dataset is a normal entry in the source flow's catalog.yaml (e.g. a pandas.Parquet saved as that flow's output). There is no special output dataset type — the dotted reference is the mechanism.

Schema validation¶

Point schema at a Pydantic model (dotted import path). Each row is validated on load; a mismatch fails fast with a clear error. Validate it ahead of a run with latent check.

eval_data:
  type: pandas.CSV
  schema: myapp.schemas.EvalRow

# myapp/schemas.py
from pydantic import BaseModel

class EvalRow(BaseModel):
    question: str
    answer: str
    ground_truth: str

MLflow lineage targets¶

Use targets to record which column(s) are the prediction/label targets, so MLflow tracks dataset lineage for the run:

predictions:
  type: pandas.Parquet
  dataset_type: output
  targets: prediction

Per-type options¶

Some types accept extra keys that are passed through to the reader:

ExcelGoogle SheetAgent Studio

quarterly:
  type: pandas.Excel
  path: data/q3.xlsx
  sheet: "Summary"     # sheet name or index
# pandas.ExcelMultiSheet ignores `sheet` and returns every sheet as a dict

live_eval:
  type: pandas.GoogleSheet
  # public sheet (CSV export) or service-account auth
  tab: "eval"          # worksheet/tab name
  range: "A:Z"         # A1 range (service-account path)
  revision: latest     # pin or freshen a cached revision

# input — fetch a versioned dataset by slug
eval_set:
  type: agent_studio.Dataset
  slug: support-qa
  version: latest               # integer or "latest"
  # base_url / api_key default to AGENT_STUDIO_URL / AGENT_STUDIO_API_KEY

# output — submit a StatisticalReport back
results_report:
  type: agent_studio.Report
  dataset_slug: support-qa
  dataset_version: latest

See Agent Studio in flows for the end-to-end eval loop.

Stable artifacts¶

Curated, version-controlled artifacts shared across flows live in data/stable/ and are read through catalog.stable — by stem name, no path construction:

from latent.datasets import get_catalog_datasets

catalog = get_catalog_datasets()
glossary = catalog.stable.glossary          # loads data/stable/glossary.json
glossary_file = catalog.stable.glossary_path # -> Path(".../data/stable/glossary.json")

Stable files are matched by stem (the name without extension); an ambiguous or missing stem raises a clear error.

Data Catalog (catalog.yaml)¶