Prefect Flows & Tasks

Learn how to build evaluation workflows with Latent's Prefect integration.

Core Concepts

Flows

Flows are the top-level containers for your evaluation logic. They automatically:

• Load configuration from parameters.yaml and catalog.yaml
• Set up flow-aware logging
• Initialize MLflow experiment tracking
• Handle data loading/saving through the catalog

Tasks

Tasks are the individual units of work within a flow. They automatically:

• Cache results using hybrid strategies (file-based or row-level)
• Retry on failure (3 times by default)
• Create MLflow spans for observability
• Log automatic metrics (duration, row counts)
• Wrap errors with rich context
• Register in the task registry for visualization
• Load/save data from/to the catalog

The @flow Decorator

from latent.prefect import flow
from pydantic import BaseModel

class FlowConfig(BaseModel):
    batch_size: int
    model_name: str

@flow(
    "flow_name",                    # Required: flow name (matches directory name)
    input=["dataset1", "dataset2"], # Optional: input datasets from catalog
    output=["result1", "result2"],  # Optional: output datasets to save
    config_schema=FlowConfig,       # Optional: Pydantic model for typed config
    **flow_kwargs                   # Optional: additional Prefect flow options
)
async def my_flow():
    # Your flow logic
    pass

Basic Flow

@flow("simple_flow")
async def simple_flow():
    """A simple flow with no explicit inputs/outputs."""
    logger.info("Flow started")
    # Your logic here
    return result

Flow with Inputs

@flow(
    "processing_flow",
    input=["raw_data", "config_data"]
)
async def processing_flow(raw_data, config_data):
    """Flow that loads datasets automatically."""
    # raw_data and config_data are loaded from catalog
    processed = process(raw_data, config_data)
    return processed

Cross-Flow Data Loading

@flow(
    "downstream_flow",
    input=["upstream_flow.output_dataset"]
)
async def downstream_flow(upstream_data):
    """Load outputs from another flow."""
    # Loads output_dataset from upstream_flow
    result = analyze(upstream_data)
    return result

The @task Decorator

from latent.prefect import task

@task(
    "task_name",                    # Required: task name
    input="input_dataset",          # Optional: input dataset name
    output="output_dataset",        # Optional: output dataset name
    cache=True,                     # Optional: enable caching (default: True)
    retry=True,                     # Optional: enable retries (default: True)
    span_type="CHAIN",              # Optional: MLflow span type (CHAIN, AGENT, TOOL, etc.)
    cache_expiration=timedelta(days=7),  # Optional: cache duration
    **task_kwargs                   # Optional: additional Prefect task options
)
async def my_task(input_data):
    # Your task logic
    return output_data

Basic Task

@task("process_data")
async def process_data(data):
    """Simple task with no caching."""
    return data.dropna()

Task with Catalog I/O

@task(
    "clean_data",
    input="raw_data",      # Loaded from catalog
    output="cleaned_data"  # Saved to catalog
)
async def clean_data(raw_data: pd.DataFrame) -> pd.DataFrame:
    """Task with automatic data loading/saving."""
    cleaned = raw_data.dropna()
    cleaned = cleaned[cleaned['value'] > 0]
    return cleaned

Task with Custom Caching

from datetime import timedelta

@task(
    "expensive_computation",
    input="data",
    output="results",
    cache_expiration=timedelta(days=30),  # Cache for 30 days
)
async def expensive_computation(data):
    """Task with extended cache duration."""
    # Expensive processing...
    return results

Task without Caching

@task(
    "generate_timestamp",
    cache=False,  # Disable caching for non-deterministic tasks
    retry=False   # Disable retries
)
async def generate_timestamp():
    """Task that always runs (no caching)."""
    return datetime.now().isoformat()

Typed Configuration

Use Pydantic models for type-safe configuration:

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

class MyFlowConfig(BaseModel):
    batch_size: int = Field(gt=0, description="Batch size for processing")
    model_name: str
    learning_rate: float = 0.001
    max_iterations: int = Field(default=1000, le=10000)

@flow("my_flow", config_schema=MyFlowConfig)
async def my_flow():
    # params is now a typed Pydantic model
    batch_size: int = params.batch_size  # IDE autocomplete works!
    model: str = params.model_name

    # Validation happens automatically on flow start
    logger.info(f"Running with batch_size={batch_size}")

Benefits of Typed Config

1. IDE Support: Full autocomplete and type checking
2. Validation: Automatic validation on flow start
3. Documentation: Field descriptions in your config
4. Defaults: Built-in default values
5. Constraints: Pydantic validators (gt, le, regex, etc.)

# parameters.yaml
batch_size: 32
model_name: "gpt-4"
learning_rate: 0.001
# max_iterations uses default (1000)

If validation fails, you get a clear error:

ValueError: Configuration validation failed for flow 'my_flow':
  batch_size: ensure this value is greater than 0 (type=value_error.number.not_gt)

Observability & Metrics

Automatic MLflow Spans

Tasks automatically create MLflow spans for observability:

@task("process_data", span_type="CHAIN")
async def process_data(data):
    # Automatically creates an MLflow span with:
    # - span.type = "CHAIN"
    # - Automatic duration logging
    # - Input/output attributes
    return processed_data

Span types follow MLflow conventions: - CHAIN: Sequential processing - AGENT: Agent-based tasks - TOOL: Tool invocations - RETRIEVER: Data retrieval - LLM: Language model calls

Automatic Metrics

Tasks automatically log:

@task("process")
async def process_task(data):
    # Automatically logged:
    # - task.process.duration (in seconds)
    # - task.process.status (success/failure)
    # - Error details (if failed)
    return result

Enhanced Error Context

Task failures are wrapped with rich context:

@task("process", input="data")
async def process_task(data):
    raise ValueError("Processing failed!")

# Error includes:
# - Task name
# - Input arguments
# - Original exception
# - MLflow span attributes

Error output:

TaskExecutionError: Task 'process' failed.
Original error: ValueError: Processing failed!
Context: {'input': 'data', 'row_index': None}

Task Registry & Visualization

Tasks are automatically registered for visualization:

from latent.registry import TaskRegistry

# After running flows, visualize the pipeline
TaskRegistry.print_ascii()

Output:

=== 🌳 Canopy Pipeline Topology ===

📦 Flow: data_pipeline
  ├── ⚙️  load
  │    ⬆️  out: raw_data
  ├── ⚙️  clean
  │    ⬇️  in: raw_data
  │    ⬆️  out: cleaned_data
  ├── ⚙️  process
  │    ⬇️  in: cleaned_data
  │    ⬆️  out: processed_data

Accessing Configuration

Using the params Object

from latent.prefect import params

@task("process")
async def process_task(data):
    # Access parameters from parameters.yaml
    batch_size = params.batch_size
    model = params.model_name

    # Dict-style access
    lr = params["learning_rate"]

    # With default value
    timeout = params.get("timeout", 30)

    # params is a Mapping — merge or splat the whole config
    settings = {"retries": 3, **params}

    return process(data, batch_size, model, lr, settings)

Attribute access is typed, mapping access is data: with a config_schema params.retrieval is the validated sub-model, while params["retrieval"], params.get("retrieval") and {**params} hand back model_dump() data, so nested sections are dicts either way. Reach a nested section's fields through attribute access (params.retrieval.top_k) — params.get("retrieval").top_k is an AttributeError on a dict. Three further ways a schema shifts the mapping view (aliases, unset-but-declared defaults, exclude=True) are listed under Typed parameters.

Cross-Flow Parameter Access

You can access parameters from other flows using dot notation with params.get():

from latent.prefect import params

@task("analyze")
async def analyze_task(data):
    # Access parameters from another flow
    upstream_model = params.get("upstream_flow.model_name")
    upstream_temp = params.get("upstream_flow.temperature", default=0.7)

    # Or that flow's parameters as a whole
    upstream_config = params.flow("upstream_flow")

    # Current flow parameters still work
    batch_size = params.batch_size

    return analyze(data, model=upstream_model, temp=upstream_temp)

This is useful when:

• You need to ensure consistency across flows (e.g., same model configuration)
• A downstream flow needs to know upstream configuration
• You want to reference shared settings without duplicating them

How it works:

1. params.get("other_flow.param_name") loads parameters.yaml from flows/other_flow/
2. Global parameters from global.yaml are merged (flow-specific takes precedence)
3. If the flow or parameter doesn't exist, returns the default value
4. If the other flow's parameters.yaml is unreadable or malformed, the read raises — a broken upstream flow is not the same thing as a missing key

# Example: Downstream flow accessing upstream configuration
@flow("downstream_flow")
async def downstream_flow():
    # Get the model that was used in upstream processing
    upstream_model = params.get("upstream_flow.model_name")

    # Use the same model for consistency
    if upstream_model:
        logger.info(f"Using upstream model: {upstream_model}")

    result = await process_task()
    return result

Note

Cross-flow parameter access only works with params.get() and params.flow(), not with attribute access (params.other_flow) or dict access (params["other_flow.param"]).

Using get_config()

from latent.prefect import get_catalog, get_config

@flow("my_flow")
async def my_flow():
    # The current flow's parameters — a dict, or the validated model when the
    # flow declares a config_schema
    config = get_config()

    # The current flow's catalog
    datasets = get_catalog()

Merge through params, not get_config()

get_config()'s return type follows the flow's config_schema: a dict without one, a Pydantic model with one. So {**base, **get_config()} works until the day the flow is typed, then raises TypeError: 'X' object is not a mapping. {**base, **params} is stable across that change.

Logging

Using the logger Object

from latent.prefect import logger

@task("process")
async def process_task(data):
    logger.debug("Detailed debug information")
    logger.info(f"Processing {len(data)} rows")
    logger.warning("This might take a while")
    logger.error("Something went wrong")

    try:
        result = risky_operation()
    except Exception as e:
        logger.exception("Operation failed")
        raise

    return result

Using get_flow_logger()

from latent.prefect import get_flow_logger

@task("process")
async def process_task(data):
    logger = get_flow_logger()
    logger.info("Task started")
    return data

Advanced Patterns

Parallel Task Execution

@flow("parallel_flow")
async def parallel_flow():
    items = load_items()

    # Execute task in parallel for each item, bounded by concurrency.
    # .map() awaits all invocations and returns a list of results in order.
    results = await process_item_task.map(items, concurrency=5)

    return aggregate(results)


@task("process_item")
async def process_item_task(item):
    logger.info(f"Processing item")
    return process(item)

Conditional Task Execution

@flow("conditional_flow")
async def conditional_flow():
    data = load_data()

    if params.enable_preprocessing:
        data = await preprocess_task(data)

    results = await analyze_task(data)

    if params.save_intermediate:
        await save_intermediate_task(data)

    return results

Dynamic Task Configuration

@task("configurable_task")
async def configurable_task(data):
    # Get task-specific config
    task_config = params.get("task_config", {})

    threshold = task_config.get("threshold", 0.5)
    method = task_config.get("method", "default")

    return process(data, threshold=threshold, method=method)

Error Handling

@task("robust_task", retry=True)
async def robust_task(data):
    try:
        result = process(data)
    except ValueError as e:
        logger.warning(f"Validation error: {e}, using defaults")
        result = default_result()
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise

    return result

Subflows

@flow("subflow")
async def subflow(data):
    """A reusable subflow."""
    cleaned = await clean_task(data)
    processed = await process_task(cleaned)
    return processed


@flow("main_flow")
async def main_flow():
    """Main flow that calls subflows."""
    data1 = load_data_1()
    data2 = load_data_2()

    # Call subflow for each dataset
    result1 = await subflow(data1)
    result2 = await subflow(data2)

    return combine(result1, result2)

Hybrid Caching Strategy

Latent uses two caching strategies depending on the task type:

1. File-Based Caching (Batch)

For tasks with explicit input datasets, uses file hashes:

@task("cached_task", input="raw_data", output="processed_data")
async def cached_task(raw_data):
    # Cached based on:
    # 1. The hash of raw_data
    # 2. The task code signature
    # 3. Cross-flow references (if any)

    # If raw_data hasn't changed,
    # the cached result is returned immediately
    return expensive_processing(raw_data)

Benefits: - Based on actual data content - Works across flows (cross-flow dependencies) - Invalidates automatically when data changes

2. Row-Level Caching (Granular)

For tasks without explicit inputs (e.g., mapped over rows):

@task("process_row")
async def process_row(row_data: dict) -> dict:
    # Cached based on stable hash of row_data
    # Each unique row_data gets its own cache entry
    return expensive_computation(row_data)

@flow("my_flow")
async def my_flow():
    rows = load_data()

    # Each row is cached independently. .map() awaits all invocations
    # and returns a list of results in order.
    results = await process_row.map(rows, concurrency=10)

    return results

Benefits: - Fine-grained caching (per-row or per-argument) - Efficient for mapped tasks - Handles DataFrames, dicts, Pydantic models - Stable hashing ensures consistency

Cache Key Generation

File-Based (with input):

MD5(hash_1 + hash_2 + ... + task_signature)

Row-Level (without input):

MD5(task_name + stable_hash(arg1) + stable_hash(arg2) + ...)

How Caching Works

1. Cache Key Generation: Based on strategy (file or row-level)
2. Cache Lookup: Prefect checks if a cached result exists
3. Cache Hit: Return cached result (no execution)
4. Cache Miss: Execute task and cache result
5. Cache Expiration: Default 7 days, configurable

Disabling Caching

# For non-deterministic tasks
@task("random_task", cache=False)
async def random_task():
    return random.random()

# For always-fresh data
@task("fetch_latest", cache=False)
async def fetch_latest():
    return api.get_latest()

Custom Cache Duration

from datetime import timedelta

# Cache for 30 days
@task("expensive_task", input="data", cache_expiration=timedelta(days=30))
async def expensive_task(data):
    return very_expensive_computation(data)

# Cache for 1 hour
@task("api_task", cache_expiration=timedelta(hours=1))
async def api_task():
    return fetch_from_api()

Resuming a Long Run

Task-level caching resumes at task boundaries. A task that walks 10,000 rows through a live agent has no boundary to resume from: if it dies at row 7,500, the retry starts at row 0.

@task("process_documents")
async def process_documents(documents: list[dict]) -> list[dict]:
    results = []
    for doc in documents:  # 10,000 documents
        result = await llm_client.complete(doc["prompt"])
        results.append(result)

    return results

# If this fails at document 7,500, the retry reprocesses all 10,000.

The Solution: resumable

resumable turns the loop into a keyed work list. Each result is appended to a JSONL file the moment it is produced; a rerun subtracts the recorded keys and hands back only what is left.

from latent.prefect import resumable, task

@task("process_documents")
async def process_documents(documents: list[dict]) -> list[dict]:
    with resumable("process_documents", documents, key=lambda d: d["id"]) as run:
        for doc in run.pending:  # only what this run still owes
            run.record(doc, await llm_client.complete(doc["prompt"]))

    return run.results()  # every result, in input order

# Rerun after a failure at document 7,500:
# "Resuming 'process_documents': 7500/10000 recorded, 2500 remaining"

.map(concurrency=N) still owns the batching half. resumable owns the durability half only — it runs nothing, so keep whatever execution model you already have:

with resumable("score_rows", rows, key=lambda r: r["id"]) as run:
    pending = run.pending
    scores = await asyncio.gather(*[score(row) for row in pending])
    for row, score in zip(pending, scores):
        run.record(row, score)

How It Works

1. pending: The items whose key has no recorded result yet, in input order
2. record(item, result): Appends one JSON line to .latent/checkpoints/<flow>.<name>.jsonl and flushes it, so anything recorded before a crash survives it
3. Completion: When every item is recorded, the checkpoint file is deleted — a finished batch never serves its results to the next run
4. Interruption: The file is kept, and a warning names it and the progress inside it — including between a task's automatic retries, which re-enter the block and pick up where the failed attempt stopped

Unlike @checkpoint, resumable is active in every environment. It is safe in production precisely because it only survives a run that died: a clean run leaves nothing behind to go stale.

Rules and Failure Modes

• The key is the contract. Two runs must agree on it, and anything that invalidates a result belongs in it. A key that ignores a changed field resumes onto stale results. Duplicate keys are rejected.
• Results must be JSON-serializable — dump Pydantic models before recording. A result that is not raises at the record() call that caused it, not at the end.
• A torn final line costs one item, not the file. A process killed mid-write leaves half a line; that line is dropped, its item reruns, and the rest of the checkpoint is used.
• Entries for items no longer in the work list are ignored, and logged as a warning — the input changed since the checkpoint was written.
• results() refuses to guess. If any item was never recorded it raises instead of returning a short list.
• One writer per checkpoint name. Threads and coroutines within a process are safe; two processes on the same name are not.

Checkpointing in Development

@checkpoint is a different tool: an argument-hash disk memoizer for a fast edit-rerun cycle, active only in development mode.

from latent.prefect import task, checkpoint

@task("process_documents")
async def process_documents(documents: list[dict]) -> list[dict]:

    @checkpoint  # Results persist to disk (dev only)
    async def expensive_llm_call(doc: dict) -> dict:
        response = await llm_client.complete(doc["prompt"])
        return {"id": doc["id"], "response": response}

    return [await expensive_llm_call(doc) for doc in documents]

Environment	Checkpoint Behavior
dev / development	Enabled - memoizes to disk
production (default)	Disabled - no-op, runs fresh every time

# Enable checkpointing (development mode)
export LATENT_ENVIRONMENT=dev

# Or in config/latent.toml
environment = "dev"

Rationale: the cache is keyed on call arguments and outlives the run, so in production it would serve yesterday's answers as today's results. Use resumable for production resumability.

Which One

Use resumable when:

• ✅ A task iterates over many expensive items and a crash must not restart it
• ✅ The run is a production run (or any run you cannot afford to redo)
• ✅ Each item has a stable id

Use @checkpoint when:

• ✅ You are iterating on flow code locally and want the expensive calls skipped
• ✅ The call is keyed adequately by its arguments

Use task .map() when:

• ✅ Items can be processed as separate task runs
• ✅ You want Prefect's built-in concurrency control
• ✅ You need visibility into individual item status in Prefect UI

Use task-level caching (cache=True) when:

• ✅ Entire task result should be cached (not individual items)
• ✅ Task is deterministic and data-driven

Multiple Checkpointed Functions

You can checkpoint multiple functions independently:

@task("complex_pipeline")
async def complex_pipeline(items: list[dict]) -> list[dict]:

    @checkpoint
    async def step_1_extract(item: dict) -> dict:
        return await extraction_model.run(item)

    @checkpoint
    async def step_2_transform(extracted: dict) -> dict:
        return await transformation_model.run(extracted)

    @checkpoint
    async def step_3_validate(transformed: dict) -> dict:
        return await validation_model.run(transformed)

    results = []
    for item in items:
        # Each step independently cached
        extracted = await step_1_extract(item)
        transformed = await step_2_transform(extracted)
        validated = await step_3_validate(transformed)
        results.append(validated)

    return results

Clearing Checkpoints

Manually clear checkpoint cache when needed:

from latent.prefect import clear_checkpoints

# After successful completion
clear_checkpoints()

Or delete the cache directory:

rm -rf .latent/cache/

On-Disk Location

.latent/
  cache/                          # @checkpoint (dev only)
    cache.db                      # SQLite managed by diskcache
  checkpoints/                    # resumable()
    my_flow.process_documents.jsonl   # one line per completed item

A resumable file is named after the flow and the checkpoint, so two flows scaffolded from one template do not resume from each other. It exists only between a crash and the rerun that finishes the batch; delete one by hand to force a full reprocess.

Note: Add .latent/ to your .gitignore (not committed to version control).

Concurrency and Parallel Execution

The @checkpoint decorator is safe for concurrent use. The underlying cache uses proper locking to handle concurrent access. Since latent-py is async-only, run checkpointed calls concurrently with asyncio.gather:

import asyncio

@task("parallel_processing")
async def parallel_processing(items: list[dict]) -> list[dict]:

    @checkpoint
    async def process_item(item: dict) -> dict:
        return await expensive_operation(item)

    # Run all items concurrently; .gather preserves input order
    results = await asyncio.gather(*[process_item(item) for item in items])

    return list(results)

How it works:

• The checkpoint cache is initialized with thread-safe locking (double-checked locking pattern)
• Only one cache instance is created, shared across all coroutines
• The underlying diskcache library handles concurrent read/write operations safely

Best practices for parallel checkpointing:

1. Define checkpointed functions at the top level of your task - not inside loops or nested functions
2. Use unique function names - each checkpointed function should have a distinct name
3. Avoid redefining checkpointed functions - dynamically creating @checkpoint decorated functions in parallel contexts can cause cache key collisions

# ✅ Good: Define checkpointed function once at task level
@task("process_batch")
async def process_batch(items: list[dict]) -> list[dict]:

    @checkpoint
    async def call_api(item: dict) -> dict:
        return await api.process(item)

    # Use the same function for all items
    return [await call_api(item) for item in items]


# ❌ Avoid: Creating checkpointed functions inside parallel workers
@task("process_batch")
async def process_batch(items: list[dict]) -> list[dict]:
    results = []
    for item in items:
        # Creating @checkpoint functions in a loop is inefficient
        @checkpoint
        async def process_this_item(x: dict) -> dict:
            return await api.process(x)

        results.append(await process_this_item(item))
    return results

Troubleshooting parallel execution issues:

If you experience hangs or deadlocks with checkpointing during parallel execution:

1. Disable checkpointing temporarily: Set environment = "production" in config/latent.toml
2. Check concurrency settings: Reduce concurrency if experiencing SQLite lock contention
3. Clear the cache: Run rm -rf .latent/cache/ to remove potentially corrupted cache data

Best Practices

1. Small, Focused Tasks

# ✅ Good: Single responsibility
@task("clean_data")
async def clean_data(data):
    return data.dropna()

@task("transform_data")
async def transform_data(data):
    return data.apply(transform_fn)

# ❌ Bad: Too much in one task
@task("process_everything")
async def process_everything(data):
    data = data.dropna()
    data = data.apply(transform_fn)
    data = data.merge(other_data)
    # ... many more operations

2. Use Type Hints

# ✅ Good: Clear types
@task("process")
async def process_task(data: pd.DataFrame) -> pd.DataFrame:
    return data.dropna()

# ❌ Bad: No type hints
@task("process")
async def process_task(data):
    return data.dropna()

3. Descriptive Logging

# ✅ Good: Informative logs
@task("process")
async def process_task(data):
    logger.info(f"Processing {len(data)} rows with batch_size={params.batch_size}")
    result = process(data)
    logger.info(f"Processed {len(result)} rows, dropped {len(data) - len(result)}")
    return result

# ❌ Bad: No context
@task("process")
async def process_task(data):
    logger.info("Processing")
    return process(data)

4. Handle Errors Gracefully

# ✅ Good: Explicit error handling
@task("api_call")
async def api_call_task():
    try:
        result = api.fetch()
    except TimeoutError:
        logger.warning("API timeout, using cached data")
        result = load_cached()
    except Exception as e:
        logger.error(f"API call failed: {e}")
        raise
    return result

See Also

• MLflow Tracking - Track experiments within flows
• API Reference - Complete API documentation