Skip to content

datarobot_genai.dragent.workflow_paths

workflow_paths

Locate workflow.yaml for DRAgent runtimes and deployments.

discover_workflow_yaml 

discover_workflow_yaml() -> Path | None

Return workflow.yaml when it can be located on disk.

Resolution order:

  1. DRAGENT_CONFIG_FILE when it points at an existing file
  2. $CODE_DIR/workflow.yaml (DataRobot custom-model deployments set CODE_DIR)
  3. workflow.yaml in the current working directory or any parent directory
Source code in datarobot_genai/dragent/workflow_paths.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def discover_workflow_yaml() -> Path | None:
    """Return ``workflow.yaml`` when it can be located on disk.

    Resolution order:

    1. ``DRAGENT_CONFIG_FILE`` when it points at an existing file
    2. ``$CODE_DIR/workflow.yaml`` (DataRobot custom-model deployments set ``CODE_DIR``)
    3. ``workflow.yaml`` in the current working directory or any parent directory
    """
    config_file = os.environ.get(DRAGENT_CONFIG_FILE_ENV)
    if config_file:
        for candidate in _workflow_candidates_from_hint(config_file):
            if (found := _existing_workflow(candidate)) is not None:
                return found

    code_dir = os.environ.get(CODE_DIR_ENV)
    if code_dir and (found := _existing_workflow(Path(code_dir) / WORKFLOW_FILENAME)) is not None:
        return found

    for directory in [Path.cwd(), *Path.cwd().parents]:
        if (found := _existing_workflow(directory / WORKFLOW_FILENAME)) is not None:
            return found

    return None

publish_dragent_config_file_env 

publish_dragent_config_file_env(workflow_path: Path | str | None = None) -> Path | None

Set DRAGENT_CONFIG_FILE to an absolute workflow.yaml path when found.

Source code in datarobot_genai/dragent/workflow_paths.py 
72
73
74
75
76
77
78
79
80
81
82
83
def publish_dragent_config_file_env(
    workflow_path: Path | str | None = None,
) -> Path | None:
    """Set ``DRAGENT_CONFIG_FILE`` to an absolute ``workflow.yaml`` path when found."""
    resolved = (
        _existing_workflow(Path(workflow_path))
        if workflow_path is not None
        else discover_workflow_yaml()
    )
    if resolved is not None:
        os.environ[DRAGENT_CONFIG_FILE_ENV] = str(resolved)
    return resolved