cursus.pipeline_catalog.core

Pipeline Catalog Core — factory, builders, and router.

create_pipeline(dag_id=None, dag_path=None, config_path=None, sagemaker_session=None, role=None)

Create a pipeline from a DAG + config.

Parameters:

• dag_id (str | None) – Shared DAG ID from catalog (e.g., “bedrock_pytorch_incremental_edx”)

• dag_path (str | None) – Path to a .dag.json file (alternative to dag_id)

• config_path (str) – Path to pipeline config JSON

• sagemaker_session (Session | None) – SageMaker session

• role (str | None) – IAM role ARN

Returns:

Tuple of (Pipeline, CompilationReport)

Return type:

Tuple[Pipeline, any]

list_available_pipelines()

List all available shared DAG IDs.

build_and_compile(dag_path, config_path, sagemaker_session=None, role=None, project_root=None)

Build and compile a pipeline from DAG + config paths. No class needed.

Parameters:

• dag_path (str) – Path to .dag.json file

• config_path (str) – Path to config JSON

• sagemaker_session (Session | None) – SageMaker session

• role (str | None) – IAM role ARN

• project_root (str | Path | None) – Optional explicit project folder for docker source_dir resolution (the caller hook). When omitted, the compiler infers it from config_path.

Returns:

(Pipeline, CompilationReport)

Return type:

Tuple[Pipeline, any]

build_mods_pipeline(author, version, description, dag_path, config_path, class_name=None)

Generate a @MODSTemplate-decorated pipeline class from declarative config.

The generated class has the standard MODS interface: - __init__(sagemaker_session, execution_role, regional_alias) - generate_pipeline() -> Pipeline

Parameters:

• author (str) – Pipeline author alias

• version (str) – Pipeline version string

• description (str) – Pipeline description

• dag_path (str) – Relative path to DAG JSON (relative to the calling module’s directory)

• config_path (str) – Relative path to config JSON

• class_name (str | None) – Optional class name (default: derived from description)

Returns:

A @MODSTemplate decorated class ready for MODS Lambda

Return type:

Type

recommend_dag(framework=None, features=None, task_type=None, complexity=None, max_results=5)

Recommend DAGs based on requirements. Returns ranked list.

Scoring (0-1):

• Feature overlap: |required ∩ dag_features| / |required| (weight: 0.5)

• Framework match: 1.0 if exact match (weight: 0.25)

• Task type match: 1.0 if substring match (weight: 0.15)

• Complexity match: 1.0 if exact, 0.5 if adjacent (weight: 0.1)

Parameters:

• framework (str | None) – Required framework (pytorch, xgboost, lightgbm, etc.)

• features (List[str] | None) – Required features (e.g., [“training”, “bedrock”, “edx_uploading”])

• task_type (str | None) – Task type keyword (e.g., “incremental”, “end_to_end”)

• complexity (str | None) – Desired complexity (simple, standard, advanced, comprehensive)

• max_results (int) – Maximum number of results to return

Returns:

List of dicts with ‘id’, ‘score’, ‘reasoning’, and all DAG metadata

Return type:

List[Dict[str, Any]]

auto_select_dag(framework=None, features=None, task_type=None, min_score=0.6)

Auto-select the best matching DAG. Returns None if no good match.

Parameters:

• framework (str | None) – Required framework

• features (List[str] | None) – Required features

• task_type (str | None) – Task type keyword

• min_score (float) – Minimum score threshold (0-1)

Returns:

Tuple of (dag_id, PipelineDAG, score) or None

Return type:

Tuple[str, PipelineDAG, float] | None

recommend_for_agent(data_type=None, has_labels=True, needs_llm=False, multi_task=False, incremental=False, data_volume=None, gpu_available=True, framework=None)

Agent-friendly recommendation using semantic constraints.

Returns ranked DAGs with agent_context (when_to_use, prerequisites, config_guidance). Designed for LLM agents to make pipeline selection decisions.

When framework is given it is a HARD filter: only that framework’s DAGs are considered. The filter is applied before scoring/truncation, so a requested framework can never be crowded out of the top-N by higher-scoring other-framework DAGs.

pipeline_catalog_tool(action, dag_id=None, data_type=None, has_labels=True, needs_llm=False, multi_task=False, incremental=False, framework=None, gpu_available=True)

Execute a pipeline catalog tool action. Returns structured response for the agent.

Modules

agent_tool	Agent Tool Interface for Pipeline Catalog
builders	Common Pipeline Builders
pipeline_factory	Pipeline Factory — simplified.
router	DAG Router — recommends or auto-selects the best DAG for a user's requirements.