Documents
Home>Documents>Algorithm

Topological Sort in FastAPI for AI Workflow Execution

5 min readMar 5, 2025Mar 6, 2025

Using Topological Sort in FastAPI

This post walks through how to apply the topological sort algorithm — a core component of AI workflow platforms — to a FastAPI backend, step by step.

What Is Topological Sort?

Topological sort is an algorithm that orders the nodes of a directed acyclic graph (DAG) into a linear sequence such that for every edge (u, v), node u appears before node v. It's essential for determining the execution order of tasks that have dependencies.

Basic Implementation: Kahn's Algorithm

The most intuitive approach is BFS-based Kahn's Algorithm.

from collections import deque, defaultdict
from typing import List, Dict

def topological_sort(nodes: List[str], edges: List[tuple]) -> List[str]:
    graph = defaultdict(list)
    in_degree = {node: 0 for node in nodes}

    for src, tgt in edges:
        graph[src].append(tgt)
        in_degree[tgt] += 1

    queue = deque([n for n in nodes if in_degree[n] == 0])
    result = []

    while queue:
        node = queue.popleft()
        result.append(node)
        for neighbor in graph[node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    if len(result) != len(nodes):
        raise ValueError("그래프에 순환이 존재합니다!")

    return result

Integrating with a FastAPI Endpoint

In a real API, the backend receives workflow data from the frontend and computes the execution order.

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

router = APIRouter(prefix="/api/workflow", tags=["workflow"])

class WorkflowRequest(BaseModel):
    nodes: List[dict]
    edges: List[dict]

@router.post("/execute")
async def execute_workflow(req: WorkflowRequest):
    try:
        node_ids = [n["id"] for n in req.nodes]
        edge_tuples = [(e["source"], e["target"]) for e in req.edges]
        order = topological_sort(node_ids, edge_tuples)
        return {"execution_order": order}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

Level-Based Sort for Parallel Execution

Rather than a simple linear ordering, you can extend the algorithm to group nodes by level so that nodes at the same level can run in parallel.

def topological_sort_levels(
    nodes: List[str], edges: List[tuple]
) -> List[List[str]]:
    graph = defaultdict(list)
    in_degree = {node: 0 for node in nodes}

    for src, tgt in edges:
        graph[src].append(tgt)
        in_degree[tgt] += 1

    queue = deque([n for n in nodes if in_degree[n] == 0])
    levels = []

    while queue:
        current_level = list(queue)
        levels.append(current_level)
        next_queue = deque()
        for node in current_level:
            for neighbor in graph[node]:
                in_degree[neighbor] -= 1
                if in_degree[neighbor] == 0:
                    next_queue.append(neighbor)
        queue = next_queue

    total = sum(len(level) for level in levels)
    if total != len(nodes):
        raise ValueError("그래프에 순환이 존재합니다!")

    return levels

Running the Workflow

import asyncio

async def run_workflow(levels: List[List[str]], context):
    for level in levels:
        # nodes at the same level run concurrently
        tasks = [execute_node(node_id, context) for node_id in level]
        await asyncio.gather(*tasks)

Topological sort is useful far beyond AI workflows — build systems, package managers, and many other domains rely on it. Implementing cycle detection alongside level-based parallelism makes it a genuinely powerful tool in production.

Tags
TopologicalSortAlgorithmFastAPIPythonDAG