Workflow orchestration

Workflow orchestration manages complex multi-step processes with task dependencies, parallel execution where possible, and error handling. Tasks are modeled as a DAG (Directed Acyclic Graph) where each task declares its dependencies and the orchestrator determines execution order.

A DAG (Directed Acyclic Graph) is a flowchart with no loops. Each step points forward to the next step, but you can never go backwards. This guarantees your workflow will always finish instead of getting stuck in an infinite loop.

Workflow dag

patterns/24_workflow_orchestration.py

python

class WorkflowTask:
    def __init__(self, name, dependencies=None):
        self.name = name
        self.dependencies = dependencies or []
        self.status = "pending"  # pending, running, completed, failed

class WorkflowOrchestrator:
    def __init__(self):
        self.tasks = {}

    def calculate_execution_order(self):
        """Topological sort for dependency-based ordering."""
        visited = set()
        order = []
        def dfs(task_name):
            if task_name in visited:
                return
            visited.add(task_name)
            for dep in self.tasks[task_name].dependencies:
                dfs(dep)
            order.append(task_name)
        for name in self.tasks:
            dfs(name)
        return order

    def execute_workflow(self):
        order = self.calculate_execution_order()
        context = {}
        for task_name in order:
            task = self.tasks[task_name]
            task.status = "running"
            result = self.execute_task(task, context)
            context[task_name] = result
            task.status = "completed"
        return context

Topological sort determines execution order based on dependencies.

Quiz: Quiz

Loading practice…

Ordering exercise: Order the workflow orchestration steps

Loading practice…

Flashcards: Flashcards

Loading practice…

You can now orchestrate complex workflows with dependency management. Next, we learn to build modular, reusable workflow components with subgraphs.

Workflow orchestration

Workflow dag

patterns/24_workflow_orchestration.py

python

class WorkflowTask:
    def __init__(self, name, dependencies=None):
        self.name = name
        self.dependencies = dependencies or []
        self.status = "pending"  # pending, running, completed, failed

class WorkflowOrchestrator:
    def __init__(self):
        self.tasks = {}

    def calculate_execution_order(self):
        """Topological sort for dependency-based ordering."""
        visited = set()
        order = []
        def dfs(task_name):
            if task_name in visited:
                return
            visited.add(task_name)
            for dep in self.tasks[task_name].dependencies:
                dfs(dep)
            order.append(task_name)
        for name in self.tasks:
            dfs(name)
        return order

    def execute_workflow(self):
        order = self.calculate_execution_order()
        context = {}
        for task_name in order:
            task = self.tasks[task_name]
            task.status = "running"
            result = self.execute_task(task, context)
            context[task_name] = result
            task.status = "completed"
        return context

Topological sort determines execution order based on dependencies.

Quiz: Quiz

Loading practice…

Ordering exercise: Order the workflow orchestration steps

Loading practice…

Flashcards: Flashcards

Loading practice…

You can now orchestrate complex workflows with dependency management. Next, we learn to build modular, reusable workflow components with subgraphs.