Building a Parallel Workflow Execution Engine

Implementing a Parallel Workflow Execution Engine

This post covers how we dramatically improved workflow execution speed by using topological sorting to determine execution order, then running same-level nodes in parallel.

The Problem: Sequential Execution Doesn't Scale

Initially, all nodes ran sequentially. With three LLM nodes at the same level, each taking 5 seconds, the total came to 15 seconds. Running them in parallel brings that down to 5 seconds.

Parallel Execution with asyncio.gather

import asyncio
from typing import List, Dict, Any

class ParallelWorkflowExecutor:
    def __init__(self, nodes: dict, edges: list):
        self.nodes = nodes
        self.edges = edges
        self.context = ExecutionContext()
        self.topo = WorkflowExecutor(list(nodes.values()), edges)

    async def execute(self) -> Dict[str, Any]:
        levels = self.topo.get_execution_order()

        for level_idx, level in enumerate(levels):
            tasks = []
            for node_id in level:
                node = create_node(NodeConfig(**self.nodes[node_id]))
                inputs = self.context.get_input(node_id, self.edges)
                tasks.append(self._execute_node(node, inputs))

            # Run all nodes at the same level concurrently
            results = await asyncio.gather(*tasks, return_exceptions=True)

            for node_id, result in zip(level, results):
                if isinstance(result, Exception):
                    raise WorkflowExecutionError(
                        f"노드 {node_id} 실행 실패: {result}"
                    )
                self.context.set_result(node_id, result)

        return self.context.results

    async def _execute_node(
        self, node: BaseNode, inputs: Dict
    ) -> Dict[str, Any]:
        try:
            return await asyncio.wait_for(
                node.execute(inputs),
                timeout=300,  # 5-minute timeout
            )
        except asyncio.TimeoutError:
            raise WorkflowExecutionError(
                f"노드 {node.node_id} 실행 시간 초과 (300초)"
            )

Limiting Concurrency with a Semaphore

Because LLM APIs enforce rate limits, we needed a way to cap the number of concurrent executions.

class RateLimitedExecutor(ParallelWorkflowExecutor):
    def __init__(self, nodes, edges, max_concurrent: int = 5):
        super().__init__(nodes, edges)
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def _execute_node(self, node, inputs):
        async with self.semaphore:
            return await super()._execute_node(node, inputs)

Performance Comparison

Measured on a real workflow (8 nodes, 3 levels):

Overall, we achieved a 60–75% reduction in execution time.

Error Handling Strategy

In parallel execution, how you handle a single node failure matters. We implemented three strategies:

class ErrorStrategy:
    FAIL_FAST = "fail_fast"       # Abort everything on any failure
    CONTINUE = "continue"          # Skip only the failed node
    RETRY = "retry"                # Retry up to 3 times

FAIL_FAST is the default, but users can change it in the workflow settings. The RETRY strategy applies exponential backoff, which handles API rate limit issues gracefully.

The parallel execution engine is one of the most impactful features we've built for the XGen platform.