Repository: skypilot-org/skypilot
Pull request: DAG API Enhancements: Introducing Downstream Task Parsing and Explicit Flow Definition
<!-- Describe the changes in this PR -->

Resolve #4054 

This PR introduces the `dependencies` feature to SkyPilot, enabling users to specify task dependencies in their YAML configurations. This enhancement allows for more complex workflow definitions and ensures proper execution order of interdependent tasks.

Key changes:
- Extended YAML parsing to support `dependencies` section
- Implemented task graph construction based on dependencies
- Provide much powerful interface for graph construction
- Updated YAML generation to include dependency information

<!-- Describe the tests ran -->
<!-- Unit tests (tests/test_*.py) are part of GitHub CI; below are tests that launch on the cloud. -->

Tested (run the relevant ones):

- [x] Code formatting: `bash format.sh`
- [ ] Any manual or new tests for this PR (please specify below)
- [ ] All smoke tests: `pytest tests/test_smoke.py` 
- [ ] Relevant individual smoke tests: `pytest tests/test_smoke.py::test_fill_in_the_name` 
- [ ] Backward compatibility tests: `conda deactivate; bash -i tests/backward_compatibility_tests.sh`


Problem statement:
[Jobs] API Design for DAG execution
<!-- Describe the bug report / feature request here -->

We should think about the API to specify an DAG of tasks.

Several choices:

1. Add an attribute to `sky.Task` for predecessor/successor
2. Add a dedicated yaml file for node dependencies
3. Merge the node dependency yaml to the pipeline yaml

Hints from issue discussion:
@andylizf seems like i cannot assign you to this issue. I assigned myself first
How about this format?
```yaml
name: pipeline
depends_on:
  train: []
  eval: [train]
  analyze: [eval]
  visualize: [analyze, train]
---
name: train
resources:
  accelerators: V100:8
any_of:
  - use_spot: true
  - use_spot: false
file_mounts:
  /checkpoint:
    name: train-eval
    mode: MOUNT
setup: |
  echo setup for training
run: |
  echo run for training
  echo save checkpoints to /checkpoint
---
name: eval
resources:
  accelerators: T4:1
use_spot: false
file_mounts:
  /checkpoint:
    name: train-eval
    mode: MOUNT
setup: |
  echo setup for eval
run: |
  echo load trained model from /checkpoint
  echo eval model on test set
---
name: analyze
resources:
  cpus: 4
run: |
  echo analyze evaluation results
---
name: visualize
resources:
  cpus: 2
  memory: 16Gi
run: |
  echo create visualizations of results and training process
```
This looks good! Maybe rename `depends_on` to `dependencies`. We should also consider how to determine if the first task is a graph representation or an actual task for the pipeline.

Also, we should prioritize the python API for the user as this YAML representation is not intuitive and limited.

Commit message: add comments for add_edge

Patch to review:
diff --git a/examples/dag/diamond.yml b/examples/dag/diamond.yml
new file mode 100644
index 00000000000..c6791138658
--- /dev/null
+++ b/examples/dag/diamond.yml
@@ -0,0 +1,77 @@
+name: pipeline
+dependencies:
+  infer_A:
+    - data-processing
+  infer_B:
+    - data-processing
+  eval_A_B:
+    - infer_A
+    - infer_B
+---
+name: data-processing
+
+resources:
+  cpus: 2+
+  use_spot: true
+  any_of:
+    - cloud: aws
+    - cloud: gcp
+
+setup: |
+  echo setup for data-processing
+  sleep 20
+
+run: |
+  echo data-processing starts
+  sleep 30
+  echo data-processing ends
+
+---
+name: infer_A
+
+resources:
+  cpus: 2+
+  any_of:
+    - cloud: aws
+    - cloud: gcp
+
+setup: |
+  echo setup for infer_A
+  sleep 20
+
+run: |
+  echo infer_A starts
+  sleep 20
+  echo infer_A ends
+
+---
+
+name: infer_B
+
+resources:
+  cpus: 2+
+
+setup: |
+  echo setup for infer_B
+  sleep 20
+
+run: |
+  echo infer_B starts
+  sleep 20
+  echo infer_B ends
+
+---
+
+name: eval_A_B
+
+resources:
+  cpus: 2+
+
+setup: |
+  echo setup for eval
+  sleep 20
+
+run: |
+  echo eval starts
+  sleep 20
+  echo eval ends
diff --git a/sky/cli.py b/sky/cli.py
index 093db23adbf..f401dd7e0a0 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -761,7 +761,7 @@ def _make_task_or_dag_from_entrypoint_with_overrides(
     if is_yaml:
         assert entrypoint is not None
         usage_lib.messages.usage.update_user_task_yaml(entrypoint)
-        dag = dag_utils.load_chain_dag_from_yaml(entrypoint, env_overrides=env)
+        dag = dag_utils.load_dag_from_yaml(entrypoint, env_overrides=env)
         if len(dag.tasks) > 1:
             # When the dag has more than 1 task. It is unclear how to
             # override the params for the dag. So we just ignore the
@@ -778,7 +778,7 @@ def _make_task_or_dag_from_entrypoint_with_overrides(
     else:
         task = sky.Task(name='sky-cmd', run=entrypoint)
         task.set_resources({sky.Resources()})
-        # env update has been done for DAG in load_chain_dag_from_yaml for YAML.
+        # env update has been done for DAG in load_dag_from_yaml for YAML.
         task.update_envs(env)
 
     # Override.
diff --git a/sky/dag.py b/sky/dag.py
index 4af5adc76b5..9497d88c465 100644
--- a/sky/dag.py
+++ b/sky/dag.py
@@ -2,74 +2,239 @@
 import pprint
 import threading
 import typing
-from typing import List, Optional
+from typing import Dict, List, Optional, Set, Union
+
+import networkx as nx
+
+from sky.serve import serve_utils
 
 if typing.TYPE_CHECKING:
     from sky import task
 
+TaskOrName = Union['task.Task', str]
+
 
 class Dag:
     """Dag: a user application, represented as a DAG of Tasks.
 
+    This class allows users to define and manage directed acyclic graphs
+    (DAGs) of tasks, representing complex workflows.
+
     Examples:
         >>> import sky
-        >>> with sky.Dag() as dag:
-        >>>     task = sky.Task(...)
+        >>> with sky.Dag(name='my_pipeline') as dag:
+        >>>     task1 = sky.Task(name='task1', ...)
+        >>>     task2 = sky.Task(name='task2', ...)
+        >>>     task1 >> task2
     """
 
-    def __init__(self) -> None:
+    def __init__(self, name: Optional[str] = None) -> None:
+        """Initialize a new DAG.
+
+        Args:
+            name: Optional name for the DAG.
+        """
+
+        self.name = name
         self.tasks: List['task.Task'] = []
-        import networkx as nx  # pylint: disable=import-outside-toplevel
+        self._task_name_lookup: Dict[str, 'task.Task'] = {}
+        self.dependencies: Dict['task.Task', Set['task.Task']] = {}
 
         self.graph = nx.DiGraph()
-        self.name: Optional[str] = None
+
+    def _get_task(self, task_or_name: TaskOrName) -> 'task.Task':
+        """Get a task object from a task or its name.
+
+        Args:
+            task_or_name: Either a Task object or the name of a task.
+
+        Returns:
+            The Task object.
+
+        Raises:
+            ValueError: If the task name is not found in the DAG.
+        """
+        if not isinstance(task_or_name, str):
+            return task_or_name
+        name = task_or_name
+        if name not in self._task_name_lookup:
+            raise ValueError(f'Task {name} not found in DAG')
+        return self._task_name_lookup[name]
 
     def add(self, task: 'task.Task') -> None:
+        """Add a task to the DAG.
+
+        Args:
+            task: The Task object to add.
+
+        Raises:
+            ValueError: If the task already exists in the DAG or if its name
+            is already used.
+        """
+        if task.name is None:
+            task.name = serve_utils.generate_task_name(task)
+        if task in self.tasks:
+            raise ValueError(f'Task {task.name} already exists in the DAG.')
+        if task.name in self._task_name_lookup:
+            raise ValueError(
+                f'Task name "{task.name}" is already used in the DAG.')
         self.graph.add_node(task)
         self.tasks.append(task)
+        self._task_name_lookup[task.name] = task
+
+    def remove(self, task: Union['task.Task', str]) -> None:
+        """Remove a task from the DAG.
+
+        Args:
+            task: The Task object or name of the task to remove.
 
-    def remove(self, task: 'task.Task') -> None:
+        Raises:
+            ValueError: If the task is still being depended on by other tasks.
+        """
+        task = self._get_task(task)
+
+        dependents = list(self.graph.successors(task))
+        for dependent in dependents:
+            self.remove_dependency(dependent, task)
+
+        # TODO(andy): Stuck by optimizer's wrong way to remove dummy sources
+        # and sink nodes.
+        # if dependents:
+        #     dependent_names = ', '.join([dep.name for dep in dependents])
+        #     raise ValueError(f'Task {task.name} is still being depended on '
+        #                      f'by tasks {dependent_names!r}. Try to remove '
+        #                      'the dependencies first.')
+
+        self.dependencies.pop(task, None)
         self.tasks.remove(task)
         self.graph.remove_node(task)
+        assert task.name is not None
+        self._task_name_lookup.pop(task.name, None)
+
+    def add_dependency(self, dependent: TaskOrName,
+                       dependency: TaskOrName) -> None:
+        """Add a single dependency for a task.
 
+        This method adds a new dependency without removing existing ones.
+
+        Args:
+            dependent: The task that depends on another.
+            dependency: The task that the dependent task depends on.
+
+        Raises:
+            ValueError: If a task tries to depend on itself or if the
+            dependency task is not in the DAG.
+        """
+        dependent = self._get_task(dependent)
+        dependency = self._get_task(dependency)
+
+        if dependent == dependency:
+            raise ValueError(f'Task {dependent.name} cannot depend on itself.')
+        if dependency not in self.tasks:
+            raise ValueError(
+                f'Dependency task {dependency.name} is not in the DAG.')
+
+        self.graph.add_edge(dependency, dependent)
+        if dependent not in self.dependencies:
+            self.dependencies[dependent] = set()
+        self.dependencies[dependent].add(dependency)
+
+    # Backward compatibility for old DAG API.
+    # TODO(andy): Remove this after 2 minor releases, which is 0.6.3.
+    # Be careful the parameter order is different from add_dependency.
     def add_edge(self, op1: 'task.Task', op2: 'task.Task') -> None:
-        assert op1 in self.graph.nodes
-        assert op2 in self.graph.nodes
-        self.graph.add_edge(op1, op2)
+        return self.add_dependency(op2, op1)
+
+    def set_dependencies(
+            self, dependent: TaskOrName,
+            dependencies: Union[List[TaskOrName], TaskOrName]) -> None:
+        """Set dependencies for a task, replacing any existing dependencies.
+
+        Args:
+            dependent: The task to set dependencies for.
+            dependencies: The task(s) that the dependent task depends on.
+        """
+        dependent = self._get_task(dependent)
+        self.remove_all_dependencies(dependent)
+        if not isinstance(dependencies, list):
+            dependencies = [dependencies]
+        for dependency in dependencies:
+            self.add_dependency(dependent, dependency)
+
+    def remove_dependency(self, dependent: TaskOrName,
+                          dependency: TaskOrName) -> None:
+        """Remove a specific dependency for a task.
+
+        Args:
+            dependent: The task to remove a dependency from.
+            dependency: The dependency to remove.
+        """
+        dependent = self._get_task(dependent)
+        dependency = self._get_task(dependency)
+
+        if dependent in self.dependencies and dependency in self.dependencies[
+                dependent]:
+            self.dependencies[dependent].remove(dependency)
+            self.graph.remove_edge(dependency, dependent)
+
+    def remove_all_dependencies(self, task: TaskOrName) -> None:
+        """Remove all dependencies for a given task.
+
+        Args:
+            task: The task to remove all dependencies from.
+        """
+        task = self._get_task(task)
+        if task in self.dependencies:
+            for dependency in list(self.dependencies[task]):
+                self.graph.remove_edge(dependency, task)
+            self.dependencies[task].clear()
+
+    def get_dependencies(self, task: TaskOrName) -> Set['task.Task']:
+        """Get all dependencies for a given task.
+
+        Args:
+            task: The task to get dependencies for.
+
+        Returns:
+            A set of tasks that the given task depends on.
+        """
+        task = self._get_task(task)
+        return self.dependencies.get(task, set())
 
     def __len__(self) -> int:
+        """Return the number of tasks in the DAG."""
         return len(self.tasks)
 
     def __enter__(self) -> 'Dag':
+        """Enter the runtime context related to this object."""
         push_dag(self)
         return self
 
     def __exit__(self, exc_type, exc_value, traceback) -> None:
+        """Exit the runtime context related to this object."""
         pop_dag()
 
     def __repr__(self) -> str:
+        """Return a string representation of the DAG."""
         pformat = pprint.pformat(self.tasks)
         return f'DAG:\n{pformat}'
 
     def get_graph(self):
+        """Return the networkx graph representing the DAG."""
         return self.graph
 
     def is_chain(self) -> bool:
-        # NOTE: this method assumes that the graph has no cycle.
-        is_chain = True
-        visited_zero_out_degree = False
-        for node in self.graph.nodes:
-            out_degree = self.graph.out_degree(node)
-            if out_degree > 1:
-                is_chain = False
-                break
-            elif out_degree == 0:
-                if visited_zero_out_degree:
-                    is_chain = False
-                    break
-                else:
-                    visited_zero_out_degree = True
-        return is_chain
+        """Check if the DAG is a linear chain of tasks.
+
+        Returns:
+            True if the DAG is a linear chain, False otherwise.
+        """
+        nodes = list(self.graph.nodes)
+        out_degrees = [self.graph.out_degree(node) for node in nodes]
+
+        return (len(nodes) <= 1 or
+                (all(degree <= 1 for degree in out_degrees) and
+                 sum(degree == 0 for degree in out_degrees) == 1))
 
 
 class _DagContext(threading.local):
@@ -78,11 +243,13 @@ class _DagContext(threading.local):
     _previous_dags: List[Dag] = []
 
     def push_dag(self, dag: Dag):
+        """Push a DAG onto the stack."""
         if self._current_dag is not None:
             self._previous_dags.append(self._current_dag)
         self._current_dag = dag
 
     def pop_dag(self) -> Optional[Dag]:
+        """Pop the current DAG from the stack."""
         old_dag = self._current_dag
         if self._previous_dags:
             self._current_dag = self._previous_dags.pop()
@@ -91,6 +258,7 @@ def pop_dag(self) -> Optional[Dag]:
         return old_dag
 
     def get_current_dag(self) -> Optional[Dag]:
+        """Get the current DAG."""
         return self._current_dag
 
 
diff --git a/sky/jobs/controller.py b/sky/jobs/controller.py
index f3cd81576e2..aeb7cdabf16 100644
--- a/sky/jobs/controller.py
+++ b/sky/jobs/controller.py
@@ -37,7 +37,7 @@
 
 
 def _get_dag_and_name(dag_yaml: str) -> Tuple['sky.Dag', str]:
-    dag = dag_utils.load_chain_dag_from_yaml(dag_yaml)
+    dag = dag_utils.load_dag_from_yaml(dag_yaml)
     dag_name = dag.name
     assert dag_name is not None, dag
     return dag, dag_name
diff --git a/sky/jobs/core.py b/sky/jobs/core.py
index c4f59f65eca..be93b105e36 100644
--- a/sky/jobs/core.py
+++ b/sky/jobs/core.py
@@ -83,7 +83,7 @@ def launch(
 
     with tempfile.NamedTemporaryFile(prefix=f'managed-dag-{dag.name}-',
                                      mode='w') as f:
-        dag_utils.dump_chain_dag_to_yaml(dag, f.name)
+        dag_utils.dump_dag_to_yaml(dag, f.name)
         controller = controller_utils.Controllers.JOBS_CONTROLLER
         controller_name = controller.value.cluster_name
         prefix = managed_job_constants.JOBS_TASK_YAML_PREFIX
diff --git a/sky/optimizer.py b/sky/optimizer.py
index 4326329579d..825e4826c83 100644
--- a/sky/optimizer.py
+++ b/sky/optimizer.py
@@ -188,6 +188,7 @@ def _remove_dummy_source_sink_nodes(dag: 'dag_lib.Dag'):
         if len(source) == len(sink) == 0:
             return
         assert len(source) == len(sink) == 1, dag.tasks
+
         dag.remove(source[0])
         dag.remove(sink[0])
 
diff --git a/sky/serve/serve_utils.py b/sky/serve/serve_utils.py
index 0ecf34135a7..ef7fd4f5e11 100644
--- a/sky/serve/serve_utils.py
+++ b/sky/serve/serve_utils.py
@@ -37,6 +37,7 @@
 if typing.TYPE_CHECKING:
     import fastapi
 
+    from sky import task as task_lib
     from sky.serve import replica_managers
 
 SKY_SERVE_CONTROLLER_NAME: str = (
@@ -191,6 +192,13 @@ def generate_remote_service_dir_name(service_name: str) -> str:
     return os.path.join(constants.SKYSERVE_METADATA_DIR, service_name)
 
 
+def generate_task_name(_task: 'task_lib.Task') -> str:
+    timestamp = int(time.time())
+    unique_suffix = uuid.uuid4().hex[:6]
+    name = f'task_{timestamp}_{unique_suffix}'
+    return name
+
+
 def generate_remote_tmp_task_yaml_file_name(service_name: str) -> str:
     dir_name = generate_remote_service_dir_name(service_name)
     # Don't expand here since it is used for remote machine.
diff --git a/sky/utils/dag_utils.py b/sky/utils/dag_utils.py
index e6b491c3168..484b3600017 100644
--- a/sky/utils/dag_utils.py
+++ b/sky/utils/dag_utils.py
@@ -64,53 +64,108 @@ def convert_entrypoint_to_dag(entrypoint: Any) -> 'dag_lib.Dag':
     return converted_dag
 
 
-def load_chain_dag_from_yaml(
+def load_dag_from_yaml(
     path: str,
     env_overrides: Optional[List[Tuple[str, str]]] = None,
 ) -> dag_lib.Dag:
-    """Loads a chain DAG from a YAML file.
+    """Loads a DAG from a YAML file.
+
+    Supports various formats:
+    1. Single task without separators
+    2. Multiple tasks with separators (linear dependency by default)
+    3. DAG with explicit 'dependencies' field
 
     Has special handling for an initial section in YAML that contains only the
     'name' field, which is the DAG name.
 
     'env_overrides' is a list of (key, value) pairs that will be used to update
-    the task's 'envs' section. If it is a chain dag, the envs will be updated
-    for all tasks in the chain.
+    the task's 'envs' section. For multi-task DAGs, the envs will be updated
+    for all tasks.
 
     Returns:
-      A chain Dag with 1 or more tasks (an empty entrypoint would create a
+      A Dag with 1 or more tasks (an empty entrypoint will create a
       trivial task).
+
+    Raises:
+      ValueError: If the YAML structure is invalid or inconsistent.
     """
     configs = common_utils.read_yaml_all(path)
-    dag_name = None
-    if set(configs[0].keys()) == {'name'}:
-        dag_name = configs[0]['name']
-        configs = configs[1:]
-    elif len(configs) == 1:
-        dag_name = configs[0].get('name')
-
-    if len(configs) == 0:
-        # YAML has only `name: xxx`. Still instantiate a task.
-        configs = [{'name': dag_name}]
-
-    current_task = None
-    with dag_lib.Dag() as dag:
-        for task_config in configs:
-            if task_config is None:
-                continue
-            task = task_lib.Task.from_yaml_config(task_config, env_overrides)
-            if current_task is not None:
-                current_task >> task  # pylint: disable=pointless-statement
-            current_task = task
-    dag.name = dag_name
+    dag = dag_lib.Dag()
+
+    if not configs:
+        raise ValueError('Empty YAML file.')
+
+    has_header = set(configs[0].keys()) <= {'name', 'dependencies'}
+    if has_header:
+        header = configs.pop(0)
+        dag.name = header.get('name')
+        dependencies = header.get('dependencies', {})
+    else:
+        multi_tasks = len(configs) > 1
+        if multi_tasks:
+            raise ValueError('Multiple task definitions without a valid'
+                             'header.')
+        # Single task without header
+        dag.name = configs[0].get('name')
+        dependencies = {}
+
+    # Handle YAML with only 'name'
+    if not configs:
+        configs = [{'name': dag.name}]
+
+    tasks = []
+
+    # Create tasks
+    for config in configs:
+        if not isinstance(config, dict):
+            raise ValueError(f'Invalid task configuration: {config}')
+
+        task = task_lib.Task.from_yaml_config(config, env_overrides)
+        tasks.append(task)
+        dag.add(task)
+
+    # Handle dependencies
+    if dependencies:
+        # Explicit dependencies
+        for dependent, deps in dependencies.items():
+            deps = [deps] if not isinstance(deps, list) else deps
+            dag.set_dependencies(dependent, deps)
+    elif len(tasks) > 1:
+        # Implicit linear dependency
+        for i in range(1, len(tasks)):
+            dag.add_dependency(tasks[i], tasks[i - 1])
+
+    if not tasks:
+        raise ValueError('No tasks defined in the YAML file')
+
     return dag
 
 
-def dump_chain_dag_to_yaml(dag: dag_lib.Dag, path: str) -> None:
-    assert dag.is_chain(), dag
-    configs = [{'name': dag.name}]
-    for task in dag.tasks:
-        configs.append(task.to_yaml_config())
+def dump_dag_to_yaml(dag: dag_lib.Dag, path: str) -> None:
+    """Dumps a DAG to a YAML file.
+
+    This function supports both chain DAGs and DAGs with more complex dependency
+    structures.
+
+    Args:
+        dag: The DAG object to be dumped.
+        path: The file path where the YAML will be written.
+    """
+    header: Dict[str, Any] = {'name': dag.name}
+
+    if not dag.is_chain():
+        dependencies = {}
+        for task in dag.tasks:
+            deps = dag.get_dependencies(task)
+            if deps:
+                dependencies[task.name] = [dep.name for dep in deps]
+        if dependencies:
+            header['dependencies'] = dependencies
+
+    task_configs = [task.to_yaml_config() for task in dag.tasks]
+
+    configs = [header] + task_configs
+
     common_utils.dump_yaml(path, configs)
 
 


Code review comments:
path: examples/dag/diamond.yml
text: @user1:
I realized this would change the direction of the edge, e.g. for `dependencies` we need to specify its predecessor but not its successor, which is inconsistent with our python api (`>>`). Could we use sth like `downstream` and change the direction here?

@author:
Quick questions:
1. Should we swap the parameter order in `add_dependency`?
2. Are you suggesting replacing 'dependencies' to 'downstream' in the YAML header?

Note: Wouldn't using `__lshift__` instead be more intuitive? Like `task1 << task2` means `task1` is dependent on `task2`.

@user1:
answer to both questions is yes. We could discuss more on naming, though I think we should fix the direction. The reason to use `>>` is from airflow's UX design, and using UX that is counterwise with the main player in the domain might not be the best choice ;)

@author:
Thanks for the feedback. I've already renamed all relevant parts to use the downstream/edge style consistently.

diff_hunk: @@ -64,53 +64,108 @@ def convert_entrypoint_to_dag(entrypoint: Any) -> 'dag_lib.Dag':
     return converted_dag
 
 
-def load_chain_dag_from_yaml(
+def load_dag_from_yaml(
     path: str,
     env_overrides: Optional[List[Tuple[str, str]]] = None,
 ) -> dag_lib.Dag:
-    """Loads a chain DAG from a YAML file.
+    """Loads a DAG from a YAML file.
+
+    Supports various formats:
+    1. Single task without separators
+    2. Multiple tasks with separators (linear dependency by default)
+    3. DAG with explicit 'dependencies' field
 
     Has special handling for an initial section in YAML that contains only the
     'name' field, which is the DAG name.
 
     'env_overrides' is a list of (key, value) pairs that will be used to update
-    the task's 'envs' section. If it is a chain dag, the envs will be updated
-    for all tasks in the chain.
+    the task's 'envs' section. For multi-task DAGs, the envs will be updated
+    for all tasks.
 
     Returns:
-      A chain Dag with 1 or more tasks (an empty entrypoint would create a
+      A Dag with 1 or more tasks (an empty entrypoint will create a
       trivial task).
+
+    Raises:
+      ValueError: If the YAML structure is invalid or inconsistent.
     """
     configs = common_utils.read_yaml_all(path)
-    dag_name = None
-    if set(configs[0].keys()) == {'name'}:
-        dag_name = configs[0]['name']
-        configs = configs[1:]
-    elif len(configs) == 1:
-        dag_name = configs[0].get('name')
-
-    if len(configs) == 0:
-        # YAML has only `name: xxx`. Still instantiate a task.
-        configs = [{'name': dag_name}]
-
-    current_task = None
-    with dag_lib.Dag() as dag:
-        for task_config in configs:
-            if task_config is None:
-                continue
-            task = task_lib.Task.from_yaml_config(task_config, env_overrides)
-            if current_task is not None:
-                current_task >> task  # pylint: disable=pointless-statement
-            current_task = task
-    dag.name = dag_name
+    dag = dag_lib.Dag()
+
+    if not configs:
+        raise ValueError('Empty YAML file.')
+
+    has_header = set(configs[0].keys()) <= {'name', 'dependencies'}
+    if has_header:
+        header = configs.pop(0)
+        dag.name = header.get('name')
+        dependencies = header.get('dependencies', {})
+    else:
+        multi_tasks = len(configs) > 1
+        if multi_tasks:
+            raise ValueError('Multiple task definitions without a valid'
+                             'header.')
+        # Single task without header
+        dag.name = configs[0].get('name')
+        dependencies = {}
+
+    # Handle YAML with only 'name'
+    if not configs:
+        configs = [{'name': dag.name}]
+
+    tasks = []
+
+    # Create tasks
+    for config in configs:
+        if not isinstance(config, dict):
+            raise ValueError(f'Invalid task configuration: {config}')
+
+        task = task_lib.Task.from_yaml_config(config, env_overrides)
+        tasks.append(task)
+        dag.add(task)
+
+    # Handle dependencies
+    if dependencies:
+        # Explicit dependencies
+        for dependent, deps in dependencies.items():
+            deps = [deps] if not isinstance(deps, list) else deps
+            dag.set_dependencies(dependent, deps)
+    elif len(tasks) > 1:
+        # Implicit linear dependency
+        for i in range(1, len(tasks)):
+            dag.add_dependency(tasks[i], tasks[i - 1])
+
+    if not tasks:
+        raise ValueError('No tasks defined in the YAML file')
path: sky/utils/dag_utils.py
text: @user1:
Lets add ux_utils.print_exception_no_traceback to all those errors.