Flow Parameters (`parameters.yaml`)¶

Every flow can declare its tunable inputs in a parameters.yaml file. The values are loaded automatically when the flow runs and exposed through the context-aware params accessor — no manual passing, no global state to thread through your task signatures.

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

# flows/my_flow/parameters.yaml
model_name: gpt-4o
batch_size: 32
temperature: 0.0
gates:
  quality: 3.5
  faithfulness: 4.0

from latent.prefect import flow, task, params, logger

@flow("my_flow")
async def my_flow():
    logger.info(f"Scoring with {params.model_name} (batch={params.batch_size})")
    threshold = params.gates["quality"]
    ...

Accessing parameters¶

Inside any @flow- or @task-decorated function, read parameters off params. It is a Mapping, so single keys and the whole config are both reachable:

params.model_name              # attribute access — raises if missing
params["model_name"]           # item access — raises if missing
params.get("model_name")       # returns None if missing
params.get("batch_size", 32)   # returns 32 if missing (default)

dict(params)                   # every parameter, as plain data
{**defaults, **params}         # merge into another dict

Only inside a flow context

params resolves against the current flow. Reading it outside a @flow/@task body raises RuntimeError. Attribute access on a missing key raises AttributeError and item access raises KeyError; use params.get(key, default) when a parameter is optional.

The layering model¶

parameters.yaml is not read in isolation — the final config is merged from several layers. Highest priority wins; nested dicts are deep-merged (so you can override a single nested key without restating the whole block):

Priority	Layer	Source
1 (highest)	CLI flags, caller kwargs, and non-`None` flow-signature defaults	`latent run <flow> --param value`, `my_flow(x=1)`, `def my_flow(x=1)`
2	Flow parameters	`flows/<flow>/parameters.yaml`
3	Global defaults	`flows/global.yaml`
4 (lowest)	Bundled defaults	shipped inside a library flow's package

The three sources in row 1 are one layer, not three: Prefect applies signature defaults before the decorator sees them, so all three arrive as keyword arguments and are indistinguishable. A None default is filtered out and is a no-op — that is how you let parameters.yaml win.

${VAR} / $VAR environment variables are substituted across the YAML layers before CLI flags are merged on top, so a value you pass on the command line reaches params verbatim. The merged result — CLI flags included — is then validated against the flow's Pydantic schema, if it declares one.

Flow-signature defaults outrank parameters.yaml

A non-None default in the flow function's signature is materialised before the flow body runs and merged as a CLI flag, so it wins over parameters.yaml. Write async def my_flow(sample_size: int | None = None) and read the value with params.get("sample_size") when parameters.yaml should be the source of truth.

`global.yaml` — shared defaults¶

Put values common to every flow in flows/global.yaml. Each flow's own parameters.yaml overrides them:

# flows/global.yaml
mlflow_experiment: my-project
model_name: gpt-4o-mini      # default model for all flows

# flows/my_flow/parameters.yaml
model_name: gpt-4o           # this flow overrides the global default
batch_size: 64

At runtime params.model_name == "gpt-4o" and params.mlflow_experiment == "my-project".

Environment variables¶

Any string value may contain ${VAR_NAME} or $VAR_NAME; it is replaced with the environment variable's value at load time. Substitution is recursive (works inside nested dicts and lists). If the variable is unset, the placeholder is left as-is and a warning is logged.

# flows/my_flow/parameters.yaml
api_base: ${OPENAI_API_BASE}
data_root: ${LATENT_WORKSPACE_ROOT}/data
prompt_version: $PROMPT_VERSION

Tip

latent run exports LATENT_WORKSPACE_ROOT before loading parameters, so ${LATENT_WORKSPACE_ROOT} is always resolvable when you run via the CLI.

Typed parameters¶

Pass a Pydantic model as config_schema to validate parameters and get typed, auto-completing attribute access. Validation runs after merging and env substitution; a bad config fails fast with a clear error.

from pydantic import BaseModel
from latent.prefect import flow, params

class MyConfig(BaseModel):
    model_name: str
    batch_size: int = 32
    temperature: float = 0.0

@flow("my_flow", config_schema=MyConfig)
async def my_flow():
    # params now proxies the validated MyConfig instance
    reveal_type = params.batch_size   # int, validated and coerced

With a schema, params.batch_size returns the validated/coerced value, and an unknown or mistyped parameter raises a ValueError at flow start rather than surfacing as a confusing failure deep inside a task.

The schema must cover every top-level key of the merged config¶

Pydantic drops keys a model does not declare, and params.get("key", default) would then answer with your inline default instead of the configured value. So a schema that leaves a configured key undeclared fails at flow start, naming it:

Configuration validation failed for flow 'my_flow': MyConfig does not declare
['concurrency', 'mlflow'], so those configured values would be silently discarded.

Remember the merged config spans global.yaml, bundled defaults and CLI flags — not just your flow's own parameters.yaml. latent scaffold writes mlflow: into global.yaml, and the sampling helpers read sample_size/sample_seed, so subclass FlowConfig to pick up the keys latent reads from layers you do not own:

from latent.prefect import FlowConfig

class MyConfig(FlowConfig):
    model_name: str
    concurrency: int = 5

Only top-level keys are checked. A nested model still silently drops what it does not declare — mlflow: {enabled: true, experiment_name: x} validated against a nested model declaring only enabled loses experiment_name with no error. Declare nested fields exhaustively, or give the nested model extra="allow" too.

To keep undeclared keys instead of naming every one, opt into extra="allow"; they stay readable through params:

from pydantic import BaseModel, ConfigDict

class MyConfig(BaseModel):
    model_config = ConfigDict(extra="allow")

    model_name: str          # typed and validated
    # mlflow, concurrency, … pass through untyped

extra="forbid" and extra="ignore" are honoured as written: the first makes Pydantic itself reject undeclared keys, the second opts into dropping them.

Attribute access is typed, mapping access is data

A schema changes what attribute access returns — params.retrieval is the validated sub-model. The mapping view stays plain data: params["retrieval"], params.get("retrieval") and {**params} hand back model_dump() output, so nested sections are dicts with a schema and without one. A caller that reached a nested section through .get() and then read an attribute off it (params.get("retrieval").top_k) must switch to attribute access (params.retrieval.top_k).

Three further differences a schema introduces in the mapping view, each worth checking before {**defaults, **params}:

the dump is keyed by field name, so Field(alias="samples") moves the key from samples to the field's own name;
the dump is complete, not as-configured: a declared field the YAML never set appears with its schema default, and that default wins the merge over the caller's fallback — {"temperature": 0.9, **params} becomes the schema's 0.0;
a Field(exclude=True) field is absent from the mapping view and from attribute access.

get_config() has no such guarantee — it hands back the model itself, so {**base, **get_config()} stops working the day a schema is added. Merge through params.

Cross-flow parameters¶

Read a parameter from another flow's parameters.yaml with dot notation via params.get(). This loads the other flow's merged config on demand:

# Reuse the judge model configured by the scoring flow
judge_model = params.get("scoring_flow.model_name", default="gpt-4o")

# Or the other flow's parameters as a whole
scoring_config = params.flow("scoring_flow")

Cross-flow access works only through params.get("flow.param") and params.flow("flow") — attribute and item access (params.x) always target the current flow.

A flow that declares no parameters reads as {}, and an undeclared key yields the default. A load that genuinely fails — unreadable or malformed parameters.yaml — raises, so a broken upstream flow cannot masquerade as a missing key.

Overriding from the CLI¶

latent run forwards extra flags into the parameter system, overriding parameters.yaml for that run. Underscores and hyphens are interchangeable:

latent run my_flow --model-name gpt-4o --batch-size 64
# inside the flow: params.model_name == "gpt-4o", params.batch_size == 64

Overrides are matched to the flow's declared parameters and made available through params.get(...). See latent run.

Flow Parameters (parameters.yaml)¶