Great question. In Prefect 3.x there are a few patterns you can use—what’s “best” depends on whether you truly need distinct flow objects in the UI, or just independently scheduled/run “variants” of the same logic. First, a quick clarification: - Do you need each variant to appear as a separate Flow in the UI (unique flow names)? - Or is it enough to have one flow with many independently scheduled/triggered deployments (each with its own params, schedule, tags, concurrency, etc.)? Recommended approach for most cases: one flow, many deployments - Keep a single parameterized flow that encapsulates your model logic. - Create one deployment per model/config. Each deployment is independently runnable, schedulable, taggable, and observable. - This scales well and avoids code duplication. Example:

from prefect import flow, deploy
from prefect.schedules import Cron

@flow(name="train-model")
def train_model(model_name: str, config: dict):
    # your pipeline here
    ...

models = ["xgb", "rf", "nn"]

if __name__ == "__main__":
    deploy(
        *[
            train_model.to_deployment(
                name=f"train-{m}",
                parameters={"model_name": m, "config": {"lr": 0.1}},  # per-model config
                cron="0 3 * * *",
                tags=[m, "ml"],
            )
            for m in models
        ],
        work_pool_name="your-work-pool",
    )

Alternative: one flow, single deployment, multiple schedules - If you want a single deployment to run on different schedules with different parameters:

from prefect import flow
from prefect.schedules import Cron

@flow
def train_model(model_name: str): ...

if __name__ == "__main__":
    train_model.serve(
        schedules=[
            Cron("0 3 * * *", parameters={"model_name": "xgb"}, slug="xgb-nightly"),
            Cron("0 4 * * *", parameters={"model_name": "rf"}, slug="rf-nightly"),
        ]
    )

If you truly need separate Flows (distinct flow names in the UI) - You can generate flows programmatically with a factory and give each a unique name. - Important: define them at module import time and bind them to top-level names so deployments can resolve the entrypoint.

from prefect import flow, deploy

def make_model_flow(model_name: str):
    @flow(name=f"train-{model_name}")
    def _f(config: dict):
        # reuse your underlying logic here
        ...
    return _f

models = ["xgb", "rf", "nn"]

# Create top-level names so Prefect can import them by entrypoint
for m in models:
    globals()[f"train_{m}_flow"] = make_model_flow(m)

if __name__ == "__main__":
    deploy(
        *[
            globals()[f"train_{m}_flow"].to_deployment(
                name=f"train-{m}",
                parameters={"config": {"lr": 0.1}},
                cron="0 3 * * *",
            )
            for m in models
        ],
        work_pool_name="your-work-pool",
    )

Notes and tradeoffs - Multiple deployments of a single flow is the most common and maintainable approach. - Separate flow objects are possible but add overhead and require stable, importable names. - You can control concurrency, tags, schedules, triggers, and parameters per deployment, which often provides the “separation” you need without multiplying flows. Docs and examples: - Deployments concept - Deploy via Python - Create deployments - Create schedules - Workflow design patterns If you can share whether separate flow names in the UI are required vs. separation-by-deployment, I can tailor the code to your exact setup.

Got it — here are several concrete patterns for creating flows dynamically in Prefect 3.x. Pick the one that fits how “separate” you need them to be in the UI and for deployment. 1) Generate separate Flow objects per model (factory pattern) - Creates unique flow functions at import time (so they’re deployable) - Each flow shows up independently in the UI with its own name

from prefect import flow

def make_model_flow(model_name: str):
    @flow(name=f"train-{model_name}", flow_run_name=f"train-{model_name}-{{run_id}}")
    def train(config: dict | None = None):
        # your shared logic here
        print(f"Training {model_name} with {config=}")
        # ...
        return {"model": model_name, "ok": True}

    return train

# Create many flows dynamically and bind them to module globals for importability
MODELS = ["xgb", "rf", "nn"]
for m in MODELS:
    globals()[f"train_{m}_flow"] = make_model_flow(m)

# You can run one directly
if __name__ == "__main__":
    result = train_xgb_flow(config={"lr": 0.1})
    print(result)

2) Wrap a core flow to “bake in” parameters (distinct flows with no parameters) - Handy if you want the same logic but each flow has fixed config

from prefect import flow

@flow(name="train-core")
def train_core(model_name: str, config: dict | None = None):
    print(f"Training {model_name} with {config=}")
    # ...

def make_wrapped_flow(model_name: str, baked_config: dict):
    @flow(name=f"train-{model_name}")
    def train_no_params():
        return train_core(model_name=model_name, config=baked_config)
    return train_no_params

# Dynamically produce parameterless flows per model
specs = {
    "xgb": {"lr": 0.1},
    "rf": {"n_trees": 200},
    "nn": {"layers": [64, 64]},
}

for name, cfg in specs.items():
    globals()[f"train_{name}"] = make_wrapped_flow(name, cfg)

if __name__ == "__main__":
    train_xgb()   # no params; calls core with baked-in config

3) Create dynamic flows on the fly (ad-hoc, notebook-friendly) - Quick creation; fine for running now, but if you plan to deploy, define them at module import time (like in examples 1–2)

from prefect import flow

def create_and_run(name: str, payload: dict):
    @flow(name=name)
    def _f():
        print(f"Running {name} with {payload=}")
        return payload

    return _f()

if __name__ == "__main__":
    create_and_run("one-off-flow", {"hello": "world"})

4) Programmatically deploy multiple dynamically-created flows - After creating flows via a factory, use

to_deployment

+

deploy

to register many at once

from prefect import deploy

# Assuming example (1) ran and created train_xgb_flow, train_rf_flow, train_nn_flow
ALL_FLOWS = [train_xgb_flow, train_rf_flow, train_nn_flow]

if __name__ == "__main__":
    deploy(
        *[
            f.to_deployment(
                name=f"{f.name}-deployment",
                parameters={"config": {"lr": 0.1}},     # per-flow or per-model params
                cron="0 3 * * *",                      # or interval/rrule/schedules
                tags=[f.name],
            )
            for f in ALL_FLOWS
        ],
        work_pool_name="your-work-pool",
    )

5) Long-running “serve” with many dynamic deployments (local process) - Useful if you want a single process polling and running them

from prefect import serve

if __name__ == "__main__":
    deployments = [
        train_xgb_flow.to_deployment(name="xgb-dpl", cron="0 3 * * *"),
        train_rf_flow.to_deployment(name="rf-dpl", cron="0 4 * * *"),
        train_nn_flow.to_deployment(name="nn-dpl", cron="0 5 * * *"),
    ]
    serve(*deployments)

Tips and gotchas - If you plan to deploy, ensure dynamic flows are created at import time and bound to module-level names (so Prefect can import them via entrypoint). - Give each flow a unique

name=

to keep them distinct in the UI.

- If you have hundreds, consider whether many deployments of a single flow might be simpler operationally, but dynamic flows as shown above are supported. Docs for deeper reference: - Concepts: Deployments - Deploy via Python: Deploy via Python - Create deployments: Create deployments - Create schedules: Create schedules If you share how many flows you expect and how you want to separate them (names, schedules, tags, infra), I can tailor one of these patterns precisely to your use case.

Thanks for sharing the traceback — that error happens because the flow functions produced by your factory are nested (their qualname includes “<locals>”), and Prefect Cloud tries to import them by entrypoint. Python can’t

getattr(..., "<locals>")

, so import fails. To make dynamic flows deployable, they must be defined at module top-level (not inside another function), so their qualname is just the function name. Here are two working patterns. Option A: Generate top-level flow functions (codegen) - Define a reusable core flow once. - Generate real top-level functions for each variant using

exec

so they are importable by name.

# file: flows_dynamic.py
from prefect import flow, deploy
import re

@flow(name="train-core")
def train_core(model_name: str, config: dict | None = None):
    print(f"Training {model_name} with {config=}")
    # shared logic here
    return {"model": model_name, "ok": True}

def sanitize_identifier(s: str) -> str:
    s = re.sub(r"\W|^(?=\d)", "_", s)
    return s

MODELS = ["xgb", "rf", "nn"]  # can be loaded from a DB/config

for model in MODELS:
    fn_name = f"train_{sanitize_identifier(model)}_flow"
    src = f"""
from prefect import flow
@flow(name='train-{model}')
def {fn_name}(config: dict | None = None):
    return train_core('{model}', config)
"""
    # define top-level function in this module’s globals
    exec(src, globals())

# Optional: programmatically deploy all generated flows
if __name__ == "__main__":
    flows = [obj for name, obj in globals().items() if name.startswith("train_") and name.endswith("_flow")]
    deploy(
        *[
            f.to_deployment(
                name=f"{f.name}-deployment",
                cron="0 3 * * *",
                parameters={"config": {"lr": 0.1}},
                tags=[f.name],
            )
            for f in flows
        ],
        work_pool_name="your-work-pool",
    )

Why this works: - Each generated flow is a real top-level function (e.g.,

flows_dynamic:train_xgb_flow

) with no “<locals>” in its qualname, so Prefect Cloud can import it by entrypoint. Option B: Prefer deployments over many flows (no import issues) - If you don’t truly need distinct Flow objects in the UI, keep one flow and create many deployments. This is usually simpler and avoids factory/codegen entirely.

from prefect import flow, deploy

@flow(name="train-model")
def train_model(model_name: str, config: dict | None = None):
    print(f"Training {model_name} with {config=}")
    return {"model": model_name}

MODELS = ["xgb", "rf", "nn"]

if __name__ == "__main__":
    deploy(
        *[
            train_model.to_deployment(
                name=f"train-{m}",
                parameters={"model_name": m, "config": {"lr": 0.1}},
                cron="0 3 * * *",
                tags=[m, "ml"],
            )
            for m in MODELS
        ],
        work_pool_name="your-work-pool",
    )

Notes - Any approach that defines flows inside another function will produce

...<locals>...

qualnames and fail to import in Cloud. Binding to

globals()

alone is not enough; the qualname still reflects nesting. - Using

serve()

is fine for local, long-running processes, but Cloud work pools still need an importable entrypoint. Helpful docs: - Deploy via Python - Create deployments If you share how you’re packaging (local files vs. Git) and how you’d like to name these flows, I can tailor the codegen snippet to your setup (including sanitizing names, grouping tags, and schedules).