Graph Algorithms

📖 Introduction

Graph algorithms power the infrastructure we use daily—navigation apps finding shortest routes, social networks suggesting connections, and compilers determining build order. When I first implemented Dijkstra's algorithm for a routing project, the elegance of how it systematically finds optimal paths amazed me.

This article covers the essential graph algorithms you'll encounter repeatedly: shortest paths, minimum spanning trees, topological sorting, and the powerful Union-Find data structure.

---

🛤️ Shortest Path Algorithms

BFS for Unweighted Graphs

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

def bfs_shortest_path(
    graph: Dict[str, List[str]], 
    start: str, 
    end: str
) -> Optional[List[str]]:
    """
    Find shortest path in unweighted graph.
    Time: O(V + E)
    """
    if start == end:
        return [start]
    
    visited = {start}
    queue = deque([(start, [start])])
    
    while queue:
        node, path = queue.popleft()
        
        for neighbor in graph[node]:
            if neighbor == end:
                return path + [neighbor]
            
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append((neighbor, path + [neighbor]))
    
    return None

Dijkstra's Algorithm

For graphs with non-negative edge weights.

import heapq
from collections import defaultdict
from typing import Dict, List, Tuple

def dijkstra(
    graph: Dict[str, List[Tuple[str, int]]], 
    start: str
) -> Dict[str, int]:
    """
    Find shortest distances from start to all nodes.
    Time: O((V + E) log V) with binary heap
    """
    distances = {start: 0}
    heap = [(0, start)]
    
    while heap:
        dist, node = heapq.heappop(heap)
        
        # Skip if we've found a better path
        if dist > distances.get(node, float('inf')):
            continue
        
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            
            if new_dist < distances.get(neighbor, float('inf')):
                distances[neighbor] = new_dist
                heapq.heappush(heap, (new_dist, neighbor))
    
    return distances


def dijkstra_with_path(
    graph: Dict[str, List[Tuple[str, int]]], 
    start: str, 
    end: str
) -> Tuple[int, List[str]]:
    """
    Find shortest path and its distance.
    """
    distances = {start: 0}
    previous = {start: None}
    heap = [(0, start)]
    
    while heap:
        dist, node = heapq.heappop(heap)
        
        if node == end:
            # Reconstruct path
            path = []
            current = end
            while current:
                path.append(current)
                current = previous[current]
            return dist, path[::-1]
        
        if dist > distances.get(node, float('inf')):
            continue
        
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            
            if new_dist < distances.get(neighbor, float('inf')):
                distances[neighbor] = new_dist
                previous[neighbor] = node
                heapq.heappush(heap, (new_dist, neighbor))
    
    return float('inf'), []


# Example usage
graph = {
    'A': [('B', 4), ('C', 1)],
    'B': [('D', 1)],
    'C': [('B', 2), ('D', 5)],
    'D': []
}
print(dijkstra(graph, 'A'))  # {'A': 0, 'C': 1, 'B': 3, 'D': 4}

Bellman-Ford Algorithm

Handles negative edge weights, detects negative cycles.

from typing import Dict, List, Tuple, Optional

def bellman_ford(
    vertices: List[str],
    edges: List[Tuple[str, str, int]],
    start: str
) -> Optional[Dict[str, int]]:
    """
    Find shortest paths with possible negative weights.
    Time: O(V * E)
    Returns None if negative cycle exists.
    """
    distances = {v: float('inf') for v in vertices}
    distances[start] = 0
    
    # Relax all edges V-1 times
    for _ in range(len(vertices) - 1):
        for u, v, weight in edges:
            if distances[u] + weight < distances[v]:
                distances[v] = distances[u] + weight
    
    # Check for negative cycles
    for u, v, weight in edges:
        if distances[u] + weight < distances[v]:
            return None  # Negative cycle detected
    
    return distances


# Example
vertices = ['A', 'B', 'C', 'D']
edges = [
    ('A', 'B', 4),
    ('A', 'C', 1),
    ('C', 'B', 2),
    ('B', 'D', 1),
    ('C', 'D', 5)
]
print(bellman_ford(vertices, edges, 'A'))

Floyd-Warshall Algorithm

All-pairs shortest paths.

def floyd_warshall(
    graph: List[List[int]]
) -> List[List[int]]:
    """
    Find shortest paths between all pairs.
    Time: O(V³), Space: O(V²)
    graph[i][j] = weight or inf if no edge
    """
    n = len(graph)
    dist = [row[:] for row in graph]  # Copy
    
    for k in range(n):
        for i in range(n):
            for j in range(n):
                if dist[i][k] + dist[k][j] < dist[i][j]:
                    dist[i][j] = dist[i][k] + dist[k][j]
    
    return dist


# Example
INF = float('inf')
graph = [
    [0, 3, INF, 5],
    [2, 0, INF, 4],
    [INF, 1, 0, INF],
    [INF, INF, 2, 0]
]
result = floyd_warshall(graph)

Algorithm Comparison

Algorithm	Time	Space	Negative Weights	Use Case
BFS	O(V+E)	O(V)	N/A (unweighted)	Unweighted shortest path
Dijkstra	O((V+E)logV)	O(V)	No	Single-source, positive weights
Bellman-Ford	O(VE)	O(V)	Yes	Negative weights, detect cycles
Floyd-Warshall	O(V³)	O(V²)	Yes	All-pairs shortest paths

---

🌲 Minimum Spanning Tree

A spanning tree that connects all vertices with minimum total edge weight.

Prim's Algorithm

Grow MST from a starting vertex.

import heapq
from collections import defaultdict
from typing import Dict, List, Tuple

def prim(
    graph: Dict[str, List[Tuple[str, int]]]
) -> List[Tuple[str, str, int]]:
    """
    Find MST using Prim's algorithm.
    Time: O(E log V) with binary heap
    """
    if not graph:
        return []
    
    start = next(iter(graph))
    visited = {start}
    edges = [(weight, start, neighbor) for neighbor, weight in graph[start]]
    heapq.heapify(edges)
    
    mst = []
    
    while edges and len(visited) < len(graph):
        weight, u, v = heapq.heappop(edges)
        
        if v in visited:
            continue
        
        visited.add(v)
        mst.append((u, v, weight))
        
        for neighbor, w in graph[v]:
            if neighbor not in visited:
                heapq.heappush(edges, (w, v, neighbor))
    
    return mst


# Example
graph = {
    'A': [('B', 4), ('C', 1)],
    'B': [('A', 4), ('C', 2), ('D', 1)],
    'C': [('A', 1), ('B', 2), ('D', 5)],
    'D': [('B', 1), ('C', 5)]
}
print(prim(graph))  # [('A', 'C', 1), ('C', 'B', 2), ('B', 'D', 1)]

Kruskal's Algorithm

Sort edges, add if doesn't create cycle (using Union-Find).

from typing import List, Tuple

def kruskal(
    vertices: List[str],
    edges: List[Tuple[str, str, int]]
) -> List[Tuple[str, str, int]]:
    """
    Find MST using Kruskal's algorithm.
    Time: O(E log E) for sorting
    """
    # Union-Find with path compression and union by rank
    parent = {v: v for v in vertices}
    rank = {v: 0 for v in vertices}
    
    def find(x):
        if parent[x] != x:
            parent[x] = find(parent[x])
        return parent[x]
    
    def union(x, y):
        px, py = find(x), find(y)
        if px == py:
            return False
        if rank[px] < rank[py]:
            px, py = py, px
        parent[py] = px
        if rank[px] == rank[py]:
            rank[px] += 1
        return True
    
    # Sort edges by weight
    edges = sorted(edges, key=lambda x: x[2])
    
    mst = []
    for u, v, weight in edges:
        if union(u, v):
            mst.append((u, v, weight))
            if len(mst) == len(vertices) - 1:
                break
    
    return mst

---

🔗 Union-Find (Disjoint Set Union)

Efficiently track connected components.

class UnionFind:
    """
    Union-Find with path compression and union by rank.
    Operations are nearly O(1) amortized.
    """
    
    def __init__(self, n: int):
        self.parent = list(range(n))
        self.rank = [0] * n
        self.count = n  # Number of components
    
    def find(self, x: int) -> int:
        """Find root with path compression."""
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])
        return self.parent[x]
    
    def union(self, x: int, y: int) -> bool:
        """
        Union two sets. Returns True if they were different.
        """
        px, py = self.find(x), self.find(y)
        
        if px == py:
            return False
        
        # Union by rank
        if self.rank[px] < self.rank[py]:
            px, py = py, px
        
        self.parent[py] = px
        
        if self.rank[px] == self.rank[py]:
            self.rank[px] += 1
        
        self.count -= 1
        return True
    
    def connected(self, x: int, y: int) -> bool:
        """Check if two elements are in same set."""
        return self.find(x) == self.find(y)
    
    def get_count(self) -> int:
        """Get number of disjoint sets."""
        return self.count

Union-Find Applications

Number of Connected Components

def count_components(n: int, edges: List[List[int]]) -> int:
    """Count connected components in undirected graph."""
    uf = UnionFind(n)
    
    for u, v in edges:
        uf.union(u, v)
    
    return uf.get_count()

Redundant Connection

def find_redundant_connection(edges: List[List[int]]) -> List[int]:
    """
    Find edge that creates a cycle.
    """
    n = len(edges)
    uf = UnionFind(n + 1)
    
    for u, v in edges:
        if not uf.union(u, v):
            return [u, v]
    
    return []

Accounts Merge

from collections import defaultdict

def accounts_merge(accounts: List[List[str]]) -> List[List[str]]:
    """
    Merge accounts with common emails.
    """
    email_to_id = {}
    email_to_name = {}
    
    # Assign IDs to emails
    for account in accounts:
        name = account[0]
        for email in account[1:]:
            if email not in email_to_id:
                email_to_id[email] = len(email_to_id)
            email_to_name[email] = name
    
    # Union emails in same account
    uf = UnionFind(len(email_to_id))
    
    for account in accounts:
        first_email = account[1]
        first_id = email_to_id[first_email]
        
        for email in account[2:]:
            uf.union(first_id, email_to_id[email])
    
    # Group emails by root
    groups = defaultdict(list)
    for email, id in email_to_id.items():
        root = uf.find(id)
        groups[root].append(email)
    
    # Build result
    result = []
    for root, emails in groups.items():
        name = email_to_name[emails[0]]
        result.append([name] + sorted(emails))
    
    return result

---

📊 Topological Sort

Linear ordering of vertices in a DAG where for every edge u→v, u comes before v.

Kahn's Algorithm (BFS)

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

def topological_sort_kahn(
    vertices: List[str],
    edges: List[Tuple[str, str]]
) -> Optional[List[str]]:
    """
    Topological sort using BFS.
    Returns None if cycle exists.
    Time: O(V + E)
    """
    graph = defaultdict(list)
    in_degree = {v: 0 for v in vertices}
    
    for u, v in edges:
        graph[u].append(v)
        in_degree[v] += 1
    
    # Start with nodes of in-degree 0
    queue = deque([v for v in vertices if in_degree[v] == 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(vertices):
        return None  # Cycle detected
    
    return result

DFS-Based Topological Sort

from collections import defaultdict
from typing import List, Optional

def topological_sort_dfs(
    vertices: List[str],
    edges: List[Tuple[str, str]]
) -> Optional[List[str]]:
    """
    Topological sort using DFS.
    """
    graph = defaultdict(list)
    for u, v in edges:
        graph[u].append(v)
    
    # 0: unvisited, 1: visiting, 2: visited
    state = {v: 0 for v in vertices}
    result = []
    
    def dfs(node):
        if state[node] == 1:  # Cycle
            return False
        if state[node] == 2:  # Already processed
            return True
        
        state[node] = 1  # Visiting
        
        for neighbor in graph[node]:
            if not dfs(neighbor):
                return False
        
        state[node] = 2  # Visited
        result.append(node)
        return True
    
    for vertex in vertices:
        if state[vertex] == 0:
            if not dfs(vertex):
                return None
    
    return result[::-1]

Course Schedule II

def find_order(
    num_courses: int, 
    prerequisites: List[List[int]]
) -> List[int]:
    """
    Find valid course order.
    """
    graph = defaultdict(list)
    in_degree = [0] * num_courses
    
    for course, prereq in prerequisites:
        graph[prereq].append(course)
        in_degree[course] += 1
    
    queue = deque([c for c in range(num_courses) if in_degree[c] == 0])
    result = []
    
    while queue:
        course = queue.popleft()
        result.append(course)
        
        for next_course in graph[course]:
            in_degree[next_course] -= 1
            if in_degree[next_course] == 0:
                queue.append(next_course)
    
    return result if len(result) == num_courses else []

---

🎯 Advanced Graph Problems

Cheapest Flights Within K Stops

import heapq
from collections import defaultdict

def find_cheapest_price(
    n: int, 
    flights: List[List[int]], 
    src: int, 
    dst: int, 
    k: int
) -> int:
    """
    Find cheapest price with at most k stops.
    Modified Dijkstra with stops tracking.
    """
    graph = defaultdict(list)
    for u, v, price in flights:
        graph[u].append((v, price))
    
    # (cost, node, stops)
    heap = [(0, src, 0)]
    best = {}  # (node, stops) -> min cost
    
    while heap:
        cost, node, stops = heapq.heappop(heap)
        
        if node == dst:
            return cost
        
        if stops > k:
            continue
        
        if (node, stops) in best and best[(node, stops)] <= cost:
            continue
        
        best[(node, stops)] = cost
        
        for neighbor, price in graph[node]:
            heapq.heappush(heap, (cost + price, neighbor, stops + 1))
    
    return -1

Alien Dictionary

def alien_order(words: List[str]) -> str:
    """
    Derive order of letters from sorted alien words.
    """
    graph = defaultdict(set)
    in_degree = {c: 0 for word in words for c in word}
    
    # Build graph from adjacent word pairs
    for i in range(len(words) - 1):
        w1, w2 = words[i], words[i + 1]
        
        # Check for invalid case: prefix comes after
        if len(w1) > len(w2) and w1.startswith(w2):
            return ""
        
        for c1, c2 in zip(w1, w2):
            if c1 != c2:
                if c2 not in graph[c1]:
                    graph[c1].add(c2)
                    in_degree[c2] += 1
                break
    
    # Topological sort
    queue = deque([c for c in in_degree if in_degree[c] == 0])
    result = []
    
    while queue:
        c = queue.popleft()
        result.append(c)
        
        for neighbor in graph[c]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    
    if len(result) != len(in_degree):
        return ""
    
    return ''.join(result)

Critical Connections (Bridges)

def critical_connections(
    n: int, 
    connections: List[List[int]]
) -> List[List[int]]:
    """
    Find bridges in graph (edges that disconnect graph if removed).
    Tarjan's algorithm.
    """
    graph = defaultdict(list)
    for u, v in connections:
        graph[u].append(v)
        graph[v].append(u)
    
    disc = [0] * n  # Discovery time
    low = [0] * n   # Lowest reachable time
    bridges = []
    time = [0]
    
    def dfs(node, parent):
        time[0] += 1
        disc[node] = low[node] = time[0]
        
        for neighbor in graph[node]:
            if disc[neighbor] == 0:  # Unvisited
                dfs(neighbor, node)
                low[node] = min(low[node], low[neighbor])
                
                # Bridge condition
                if low[neighbor] > disc[node]:
                    bridges.append([node, neighbor])
            elif neighbor != parent:
                low[node] = min(low[node], disc[neighbor])
    
    dfs(0, -1)
    return bridges

---

📝 Practice Exercises

Exercise 1: Path with Maximum Probability

def max_probability(
    n: int, 
    edges: List[List[int]], 
    success_prob: List[float], 
    start: int, 
    end: int
) -> float:
    """
    Find path with maximum success probability.
    edges[i] = [a, b], success_prob[i] = probability of edge i.
    """
    pass

Solution

import heapq
from collections import defaultdict

def max_probability(
    n: int, 
    edges: List[List[int]], 
    success_prob: List[float], 
    start: int, 
    end: int
) -> float:
    graph = defaultdict(list)
    for i, (a, b) in enumerate(edges):
        prob = success_prob[i]
        graph[a].append((b, prob))
        graph[b].append((a, prob))
    
    # Max probability -> negate for min heap
    probs = [0.0] * n
    probs[start] = 1.0
    heap = [(-1.0, start)]
    
    while heap:
        prob, node = heapq.heappop(heap)
        prob = -prob
        
        if node == end:
            return prob
        
        if prob < probs[node]:
            continue
        
        for neighbor, edge_prob in graph[node]:
            new_prob = prob * edge_prob
            if new_prob > probs[neighbor]:
                probs[neighbor] = new_prob
                heapq.heappush(heap, (-new_prob, neighbor))
    
    return 0.0

Exercise 2: Reconstruct Itinerary

def find_itinerary(tickets: List[List[str]]) -> List[str]:
    """
    Reconstruct itinerary starting from "JFK".
    Use all tickets exactly once.
    Return lexicographically smallest itinerary.
    
    tickets = [["MU","LHR"],["JFK","MU"],["LHR","JFK"]]
    Output: ["JFK","MU","LHR","JFK"]
    """
    pass

Solution

from collections import defaultdict

def find_itinerary(tickets: List[List[str]]) -> List[str]:
    graph = defaultdict(list)
    for src, dst in sorted(tickets, reverse=True):
        graph[src].append(dst)
    
    result = []
    
    def dfs(airport):
        while graph[airport]:
            dfs(graph[airport].pop())
        result.append(airport)
    
    dfs("JFK")
    return result[::-1]

Exercise 3: Evaluate Division

def calc_equation(
    equations: List[List[str]], 
    values: List[float], 
    queries: List[List[str]]
) -> List[float]:
    """
    equations = [["a","b"],["b","c"]], values = [2.0,3.0]
    means a/b = 2.0, b/c = 3.0
    queries = [["a","c"]] -> [6.0]
    """
    pass

Solution

from collections import defaultdict

def calc_equation(
    equations: List[List[str]], 
    values: List[float], 
    queries: List[List[str]]
) -> List[float]:
    graph = defaultdict(dict)
    
    for (a, b), val in zip(equations, values):
        graph[a][b] = val
        graph[b][a] = 1.0 / val
    
    def dfs(src, dst, visited):
        if src not in graph or dst not in graph:
            return -1.0
        if dst in graph[src]:
            return graph[src][dst]
        
        visited.add(src)
        
        for neighbor, val in graph[src].items():
            if neighbor not in visited:
                result = dfs(neighbor, dst, visited)
                if result != -1.0:
                    return val * result
        
        return -1.0
    
    return [dfs(a, b, set()) for a, b in queries]

---

🔑 Key Takeaways

1. BFS for unweighted shortest paths

2. Dijkstra for non-negative weighted graphs

3. Bellman-Ford for negative weights and cycle detection

4. Union-Find for dynamic connectivity problems

5. Topological sort for dependency ordering

6. Know when to use which: problem constraints guide algorithm choice

---

🚀 What's Next?

In the final article, we'll consolidate everything with problem-solving patterns:

• Pattern recognition strategies

• Interview approach

• Common patterns catalog

• Time management tips

Next: Article 15 - Problem-Solving Patterns

---

📚 References

• CLRS Chapters 22-26: Graph Algorithms

• "Algorithm Design" - Kleinberg & Tardos

• Competitive programming resources