cai-causal-graph

name: cai-causal-graph description: | Define, manipulate, and serialize causal graph structures (DAG, CPDAG, MAG, PAG) using cai-causal-graph. Provides CausalGraph, Node, Edge, Skeleton classes with variable type support (CONTINUOUS, BINARY, MULTICLASS, ORDINAL). Use when user imports from cai_causal_graph, works with causal graphs, needs graph conversion (NetworkX, adjacency matrix, dict), or mentions DAG/CPDAG/PAG/MAG representations. Do NOT trigger for graph neural networks or knowledge graphs unrelated to causality. license: Apache-2.0 metadata: author: causalens version: "0.5.14" sdk_schema_version: "0.1.0" sdk_dependencies: "cai-causal-graph>=0.5.0"

CAI Causal Graph

Python library for defining, manipulating, and serializing causal graph structures (DAG, CPDAG, MAG, PAG).

Quick Reference

All public classes are re-exported from cai_causal_graph directly:

from cai_causal_graph import CausalGraph, TimeSeriesCausalGraph, EdgeType, NodeVariableType, Skeleton

Graph Types

Choose the graph representation based on what the edges mean:

CausalGraph enforces acyclicity only over directed edges. It does not enforce which combination of edge types is "valid" for a given graph type — that is the caller's responsibility.

Edge Semantics

EdgeType values and their string representations:

Critical edge semantics differences by graph type:

Node Variable Types

NodeVariableType values:

variable_type defaults to UNSPECIFIED and is mutable post-creation:

from cai_causal_graph import CausalGraph, NodeVariableType

cg = CausalGraph()
cg.add_node('x', variable_type=NodeVariableType.CONTINUOUS)
node = cg.get_node('x')
node.variable_type = NodeVariableType.BINARY  # mutable

UNSPECIFIED is dangerous — downstream tools may silently treat it as continuous.

Core Usage

Building a graph

from cai_causal_graph import CausalGraph, EdgeType, NodeVariableType

cg = CausalGraph()

# Nodes are auto-created when adding edges
cg.add_edge('x', 'y')                                          # default: DIRECTED_EDGE
cg.add_edge('z', 'y', edge_type=EdgeType.DIRECTED_EDGE)
cg.add_edge('a', 'b', edge_type=EdgeType.UNDIRECTED_EDGE)      # CPDAG undirected

# Or add nodes explicitly first
cg.add_node('w', variable_type=NodeVariableType.BINARY)
cg.add_edges_from([('w', 'x'), ('w', 'z')])

print(cg.is_dag())           # True if all edges are directed and acyclic
print(cg.get_node_names())
print(cg.get_edge_pairs())

Inspecting nodes and edges

node = cg.get_node('x')
print(node.identifier, node.variable_type)
print(node.is_source_node(), node.is_sink_node())
print(node.get_inbound_edges(), node.get_outbound_edges())

edge = cg.get_edge('x', 'y')
print(edge.source.identifier, edge.destination.identifier)
print(edge.get_edge_type())   # EdgeType.DIRECTED_EDGE
print(edge.descriptor)        # '(x -> y)'

Serialization

# dict round-trip (`from_dict` deepcopies metadata; `to_dict` does not deep-copy nested meta)
d = cg.to_dict()
cg2 = CausalGraph.from_dict(d)

# NetworkX (directed edges only -> DiGraph; undirected only -> Graph; mixed raises GraphConversionError)
nx_graph = cg.to_networkx()
cg3 = CausalGraph.from_networkx(nx_graph)

# Numpy adjacency matrix (directed/undirected only; bidirected/unknown raises TypeError)
adj, node_names = cg.to_numpy()
cg4 = CausalGraph.from_adjacency_matrix(adj, node_names=node_names)

Skeleton

skeleton = cg.skeleton          # property; returns Skeleton instance
print(skeleton.nodes)           # nodes without edge direction info
print(skeleton.edges)           # all edges as UNDIRECTED_EDGE
adj = skeleton.adjacency_matrix # symmetric binary matrix

Time series graph

from cai_causal_graph import TimeSeriesCausalGraph, EdgeType

ts = TimeSeriesCausalGraph()
ts.add_edge('X1 lag(n=1)', 'X1', edge_type=EdgeType.DIRECTED_EDGE)
ts.add_edge('X2 lag(n=1)', 'X2', edge_type=EdgeType.DIRECTED_EDGE)

node = ts.get_node('X1 lag(n=1)')
print(node.variable_name)   # 'X1'
print(node.time_lag)        # -1

summary = ts.get_summary_graph()    # CausalGraph collapsing lags
minimal = ts.get_minimal_graph()    # TimeSeriesCausalGraph with minimum lag structure
lag_0_nodes = ts.get_nodes_at_lag(0)

Domain knowledge constraints

from cai_causal_graph.type_definitions import EdgeConstraint

# EdgeConstraint is used by discovery algorithms, not enforced by CausalGraph itself
EdgeConstraint.HARD_DIRECTED_EDGE    # force edge direction
EdgeConstraint.HARD_UNDIRECTED_EDGE  # force undirected
EdgeConstraint.SOFT_DIRECTED_EDGE    # prefer direction
EdgeConstraint.FORBIDDEN_EDGE        # forbid the edge

Assumptions and Limitations

• Self-loops are not allowed — adding source == destination raises CyclicConnectionError.
• Nodes are auto-created when adding edges if they do not already exist.
• add_edge raises CyclicConnectionError if directed edges would form a cycle; the graph is rolled back.
• to_numpy() / from_adjacency_matrix() only support DIRECTED_EDGE and UNDIRECTED_EDGE — bidirected or unknown edge types raise TypeError.
• from_adjacency_matrix() expects a numpy.ndarray input. Passing plain Python lists can raise AttributeError before adjacency validation.
• to_networkx() only produces DiGraph (all directed) or Graph (all undirected) — mixed edge types raise GraphConversionError.
• Node and edge metadata is shallow-copied when passed to constructors; from_dict deepcopies.
• replace_node() copies edge meta shallowly from originals but does not preserve edge object identity.
• networkx version is bounded to >=3.0.0, <3.3.0 due to a bug in get_all_simple_paths in 3.3.

Gotchas

• -- (undirected) in a CPDAG means orientation is unknown and can be resolved either way. The same symbol in a MAG implies selection bias — a completely different semantic.
• Deleting a node invalidates graph-backed node queries — methods like get_inbound_edges() and get_outbound_edges() raise NodeDoesNotExistError after deletion.
• node.variable_type is mutable; changing it after creation is intentional and supported.
• Edge.__eq__ treats UNDIRECTED_EDGE, BIDIRECTED_EDGE, and UNKNOWN_EDGE as directionless — (a -- b) and (b -- a) are equal.
• CausalGraph.is_dag() returns False immediately if any non-directed edge is present (does not check only directed subgraph).
• Skeleton.edges returns all edges as UNDIRECTED_EDGE regardless of original type.

Error Types

All errors are nested inside CausalGraphErrors:

from cai_causal_graph.exceptions import CausalGraphErrors

try:
    cg.get_node('missing')
except CausalGraphErrors.NodeDoesNotExistError:
    ...