Implementation:Diagram of thought Diagram of thought Adjacency List Builder

Overview

Concrete pattern for building a validated DAG data structure from parsed DoT typed records using adjacency lists with node metadata.

Description

The Adjacency List Builder constructs an adjacency list representation of the reasoning DAG, validates the acyclicity constraint, and attaches role and validation status metadata to each node. Given the parsed @node, @edge, and @status records extracted from the model's output, this pattern produces a fully populated graph data structure that is confirmed acyclic and ready for reasoning path analysis, visualization, or formal verification.

The adjacency list maps each node ID to a list of its outgoing edges (each annotated with a destination and edge kind). A parallel metadata dictionary maps each node ID to its role and validation status. The acyclicity invariant is enforced by asserting src < dst for every edge during construction.

Usage

Apply the Adjacency List Builder after parsing typed records (from the regex extraction step) and before analysis or visualization. This is the bridge between raw parsed records and a queryable graph structure.

Code Reference

Source

README.md:L70 -- Acyclicity constraint: must have i < n on all @edge records.
README.md:L116-124 -- Structural view: "This protocol guarantees the graph is acyclic (since @edge sources must have smaller IDs than destinations)."

Signature

The DAG builder function accepts parsed nodes, edges, and statuses, and returns a validated graph:

def build_dag(
    nodes: List[NodeRecord],
    edges: List[EdgeRecord],
    statuses: List[StatusRecord],
) -> Tuple[Dict[int, List[Dict]], Dict[int, Dict]]:
    """Build a validated DAG from parsed DoT typed records.

    Returns:
        adjacency: mapping from node ID to list of outgoing edges
        metadata: mapping from node ID to role and status attributes
    Raises:
        AssertionError: if any edge violates the acyclicity constraint (src >= dst)
    """
    ...

Import

from typing import Dict, List, Optional, Tuple

For the NetworkX variant:

import networkx as nx

I/O Contract

Inputs

Parsed nodes: A list of node records, each containing an integer id and a string role (one of problem, proposer, critic, summarizer). Extracted from @node typed records.
Parsed edges: A list of edge records, each containing integer src and dst fields and a string kind (one of use, critique, refine). Extracted from @edge typed records.
Parsed statuses: A list of status records, each containing an integer target and a string mark (one of validated, invalidated). Extracted from @status typed records.

Outputs

Validated DAG: An adjacency list mapping each node ID to its outgoing edges (with destination and kind), plus a metadata dictionary mapping each node ID to its role and validation status. The graph is confirmed acyclic by the src < dst assertion on every edge.

Usage Examples

Pure Python Implementation

A library-agnostic implementation using only built-in data structures:

def build_dag(nodes, edges, statuses):
    """Build a validated DAG as an adjacency list with node metadata.

    Args:
        nodes: parsed @node records (each has .id and .role)
        edges: parsed @edge records (each has .src, .dst, .kind)
        statuses: parsed @status records (each has .target, .mark)

    Returns:
        adjacency: dict mapping node ID -> list of outgoing edge dicts
        metadata: dict mapping node ID -> {"role": str, "status": str|None}
    """
    adjacency = {n.id: [] for n in nodes}
    metadata = {n.id: {"role": n.role, "status": None} for n in nodes}

    # Attach validation status marks to nodes
    for s in statuses:
        if s.target in metadata:
            metadata[s.target]["status"] = s.mark

    # Build adjacency list with acyclicity enforcement
    for e in edges:
        assert e.src < e.dst, f"Acyclicity violation: {e.src} >= {e.dst}"
        adjacency[e.src].append({"dst": e.dst, "kind": e.kind})

    return adjacency, metadata

NetworkX Implementation

An implementation using the NetworkX graph library, which provides additional graph algorithms (shortest path, topological sort, etc.) out of the box:

import networkx as nx


def build_dag_nx(nodes, edges, statuses):
    """Build a validated DAG as a NetworkX DiGraph.

    Args:
        nodes: parsed @node records (each has .id and .role)
        edges: parsed @edge records (each has .src, .dst, .kind)
        statuses: parsed @status records (each has .target, .mark)

    Returns:
        G: a networkx.DiGraph confirmed to be acyclic
    """
    G = nx.DiGraph()

    for n in nodes:
        G.add_node(n.id, role=n.role, status=None)

    # Attach validation status marks to nodes
    for s in statuses:
        if s.target in G:
            G.nodes[s.target]["status"] = s.mark

    # Build edges with acyclicity enforcement
    for e in edges:
        assert e.src < e.dst, f"Acyclicity violation: {e.src} >= {e.dst}"
        G.add_edge(e.src, e.dst, kind=e.kind)

    # Belt-and-suspenders: verify DAG property via NetworkX
    assert nx.is_directed_acyclic_graph(G), "Graph contains a cycle"

    return G

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment