Implementation:Diagram of thought Diagram of thought DAG Traversal Algorithms

Overview

This page describes concrete algorithms for traversing and analyzing DoT reasoning DAGs, including topological sort, critical path extraction, and branching analysis.

Description

Graph traversal algorithms applied to the DoT reasoning DAG. After the typed records (@node, @edge, @status) have been parsed and the DAG has been reconstructed into an adjacency list with node metadata, these algorithms extract analytical insights about the structure, depth, and quality of the model's reasoning process.

Usage

Use these algorithms after DAG reconstruction, to extract analytical insights from a completed DoT reasoning trace. The algorithms require only the adjacency list and node metadata dictionary produced by the reconstruction step.

Code Reference

Source

• README.md:L126-130 -- Semantic view describing validated subobjects, colimit synthesis, and graph invariance
• README.md:L22 -- Auditable and verifiable traces enabling post-hoc analysis of reasoning paths

Signature

The analysis is composed of several functions:

def topological_sort(adjacency: dict[int, list[int]]) -> list[int]
def longest_path(adjacency: dict[int, list[int]], start: int, targets: list[int]) -> list[int]
def find_branch_points(adjacency: dict[int, list[int]]) -> list[int]
def analyze_dot_dag(adjacency: dict[int, list[int]], metadata: dict[int, dict]) -> dict

Import

Standard Python only. No external dependencies are required. Optionally, networkx can be used for graph utilities:

# Standard library (no imports needed beyond built-ins)
from collections import deque

# Optional: networkx for convenience
# import networkx as nx

I/O Contract

Direction	Description
Inputs	Validated DAG (from reconstruction step): an adjacency list dict[int, list[int]] mapping each node ID to its list of successor node IDs Node metadata dict[int, dict] mapping each node ID to its attributes (including "role" and "status" fields)
Outputs	Topological order: list[int] -- nodes in dependency-respecting order Critical path: list[int] -- longest path from the problem node to a summarizer node Branch points: list[int] -- node IDs with multiple outgoing edges Validated nodes: set[int] -- nodes marked validated by the critic Invalidated nodes: set[int] -- nodes marked invalidated by the critic Statistics: total node count and validation ratio

Usage Examples

Helper Functions

Topological sort using Kahn's algorithm (BFS-based):

from collections import deque

def topological_sort(adjacency):
    """Compute a topological ordering of the DAG using Kahn's algorithm."""
    in_degree = {}
    for node in adjacency:
        in_degree.setdefault(node, 0)
        for neighbor in adjacency[node]:
            in_degree[neighbor] = in_degree.get(neighbor, 0) + 1

    queue = deque(node for node, deg in in_degree.items() if deg == 0)
    order = []

    while queue:
        node = queue.popleft()
        order.append(node)
        for neighbor in adjacency.get(node, []):
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    return order

Longest path computation using dynamic programming over the topological order:

def longest_path(adjacency, start, targets):
    """Find the longest path from start to any node in targets.

    Uses DP over topological order. Returns the path as a list of node IDs.
    """
    topo = topological_sort(adjacency)
    dist = {node: float('-inf') for node in adjacency}
    dist[start] = 0
    parent = {node: None for node in adjacency}

    for node in topo:
        if dist[node] == float('-inf'):
            continue
        for neighbor in adjacency.get(node, []):
            if dist[node] + 1 > dist[neighbor]:
                dist[neighbor] = dist[node] + 1
                parent[neighbor] = node

    # Find the target with the longest distance
    best_target = max(targets, key=lambda t: dist.get(t, float('-inf')))
    if dist.get(best_target, float('-inf')) == float('-inf'):
        return []

    # Reconstruct path
    path = []
    current = best_target
    while current is not None:
        path.append(current)
        current = parent[current]
    path.reverse()
    return path

Complete Analysis

The main analysis function that combines all traversal algorithms:

def analyze_dot_dag(adjacency, metadata):
    """Perform complete path analysis on a DoT reasoning DAG.

    Args:
        adjacency: dict mapping node ID -> list of successor node IDs
        metadata: dict mapping node ID -> {"role": str, "status": str, ...}

    Returns:
        dict with topological_order, critical_path, branch_points,
        validated_nodes, invalidated_nodes, total_nodes, validation_ratio
    """
    # Topological sort (DAG guarantees this exists)
    topo_order = topological_sort(adjacency)

    # Critical path: longest path from node 1 to any summarizer
    summarizer_ids = [nid for nid, m in metadata.items() if m["role"] == "summarizer"]
    critical_path = longest_path(adjacency, start=1, targets=summarizer_ids)

    # Branch points: nodes with multiple outgoing edges
    branch_points = [nid for nid, neighbors in adjacency.items() if len(neighbors) > 1]

    # Validated vs invalidated subgraphs
    validated = {nid for nid, m in metadata.items() if m.get("status") == "validated"}
    invalidated = {nid for nid, m in metadata.items() if m.get("status") == "invalidated"}

    return {
        "topological_order": topo_order,
        "critical_path": critical_path,
        "branch_points": branch_points,
        "validated_nodes": validated,
        "invalidated_nodes": invalidated,
        "total_nodes": len(metadata),
        "validation_ratio": len(validated) / max(1, len(validated) + len(invalidated))
    }

Example Invocation

Given a reconstructed DAG from a character-counting task:

adjacency = {
    1: [2],        # problem -> proposer
    2: [3, 4],     # proposer -> critic, proposer (branch point)
    3: [],         # critic (validated node 2)
    4: [5],        # proposer (refined) -> critic
    5: [6],        # critic (validated node 4) -> summarizer
    6: []          # summarizer
}

metadata = {
    1: {"role": "problem",    "status": None},
    2: {"role": "proposer",   "status": "validated"},
    3: {"role": "critic",     "status": None},
    4: {"role": "proposer",   "status": "validated"},
    5: {"role": "critic",     "status": None},
    6: {"role": "summarizer", "status": None}
}

result = analyze_dot_dag(adjacency, metadata)
# result["critical_path"]     -> [1, 2, 4, 5, 6]
# result["branch_points"]     -> [2]
# result["validated_nodes"]   -> {2, 4}
# result["invalidated_nodes"] -> set()
# result["validation_ratio"]  -> 1.0

Related Pages

• Principle:Diagram_of_thought_Diagram_of_thought_Reasoning_Path_Analysis -- Theoretical foundation for Reasoning Path Analysis