Skip to main content

Graph

The Graph class defines a computation graph from nodes. It describes what nodes exist and how they connect.

Graph Class

from hypergraph import Graph, node

@node(output_name="doubled")
def double(x: int) -> int:
    return x * 2

@node(output_name="result")
def add_one(doubled: int) -> int:
    return doubled + 1

graph = Graph([double, add_one])

Constructor

nodes
list[HyperNode]
required
List of HyperNode objects (FunctionNode, GraphNode, GateNode, etc.)
edges
list[tuple] | None
Explicit edge declarations. Each edge is a tuple of (source, target) or (source, target, values) where source/target are node names (str) or node objects, and values is a str or list of str specifying which outputs flow on the edge. When provided, auto-inference is disabled
name
str | None
Optional graph name (required for nesting via as_node())
strict_types
bool
default:"False"
If True, validate type compatibility between connected nodes at graph construction time

Properties

name
str | None
Graph name (required for nesting)
nodes
dict[str, HyperNode]
Map of node name → node object
outputs
tuple[str, ...]
All unique output names produced by nodes
leaf_outputs
tuple[str, ...]
Outputs from terminal nodes (no downstream destinations)
inputs
InputSpec
Input specification describing required/optional/entrypoint parameters (cached per instance)
has_cycles
bool
True if graph contains cycles
has_async_nodes
bool
True if any FunctionNode is async
has_interrupts
bool
True if any node is an interrupt node
interrupt_nodes
list
Ordered list of interrupt node instances
strict_types
bool
Whether type validation is enabled
definition_hash
str
Merkle-tree hash of graph structure (cached, includes node hashes and edges)
nx_graph
nx.DiGraph
Underlying NetworkX directed graph
selected
tuple[str, ...] | None
Default output selection, or None if all outputs are returned
entrypoints_config
tuple[str, ...] | None
Configured entry points, or None if all nodes are active
controlled_by
dict[str, list[str]]
Map of node_name → list of controlling gate names

Methods

bind

graph = Graph([...]).bind(x=5, y=10)
Bind default values for inputs or outputs. Returns new Graph (immutable).
**values
Any
required
Keyword arguments mapping parameter names to default values
Graph
New Graph instance with bound values
Accepts any graph input or output name. Bound values act as pre-filled run() values — overridable at run time.
Raises ValueError if key is not a valid graph input or output, or if key is an emit-only output (ordering signal).

unbind

graph = graph.unbind("x", "y")
Remove specific bindings. Returns new Graph.
*keys
str
required
Parameter names to unbind
Graph
New Graph instance without specified bindings

select

graph = Graph([embed, retrieve, generate]).select("answer")
Set default output selection. Returns new Graph (immutable).
*names
str
required
Output names to include. Must be valid graph outputs
Graph
New Graph with default selection set
Controls which outputs are returned by runner.run() and which outputs are exposed when this graph is used as a nested node via as_node(). Also narrows graph.inputs to only parameters needed to produce the selected outputs. A runtime select= passed to runner.run() overrides this default.
Raises ValueError if any name is not a graph output or if names are not unique.

with_entrypoint

graph = Graph([root, process, output]).with_entrypoint("process")
Set execution entry points. Returns new Graph (immutable).
*node_names
str
required
One or more node names to use as entry points
Graph
New Graph with entry points configured
Entry points define where execution enters the graph. Upstream nodes are excluded — their outputs become direct user inputs. Works for both DAGs and cycles:
  • DAG: entry point determines where computation starts
  • Cycle: entry point determines cycle bootstrap requirements
Chainable: graph.with_entrypoint("A").with_entrypoint("B")
Raises GraphConfigError if node doesn’t exist or is a gate node.

add_nodes

graph = graph.add_nodes(new_node_1, new_node_2)
Add nodes to graph. Returns new Graph (immutable).
*nodes
HyperNode
required
Nodes to add
Graph
New Graph with added nodes
Equivalent to rebuilding the graph with the combined node list, then replaying bind/select.
Raises GraphConfigError if graph was constructed with explicit edges. Raises ValueError if existing bindings become invalid after adding nodes.

as_node

inner = Graph([...], name="preprocess")
outer = Graph([inner.as_node(), ...])
Wrap graph as node for composition. Returns new GraphNode.
name
str | None
Optional node name. If not provided, uses graph.name
GraphNode
GraphNode wrapping this graph
Raises ValueError if name is None and graph.name is None.

visualize

graph.visualize(depth=1, theme="dark", show_types=True)
Create an interactive visualization of this graph using React Flow with Kiwi constraint-based layout.
depth
int
default:"0"
How many levels of nested graphs to expand
theme
str
default:"auto"
“dark”, “light”, or “auto” to detect from environment
show_types
bool
default:"False"
Whether to show type annotations on nodes
separate_outputs
bool
default:"False"
Whether to render outputs as separate DATA nodes
filepath
str | None
Path to save HTML file (default: None, display in notebook)
ScrollablePipelineWidget | None
Widget if output is None (displays in notebook), otherwise None (saves to file)

to_mermaid

diagram = graph.to_mermaid(direction="LR")
print(diagram)  # raw Mermaid source
Generate a Mermaid flowchart diagram for this graph.
depth
int
default:"0"
How many levels of nested graphs to expand
show_types
bool
default:"False"
Whether to show type annotations in labels
separate_outputs
bool
default:"False"
Whether to render outputs as separate DATA nodes
direction
str
default:"TD"
Flowchart direction — “TD”, “TB”, “LR”, “RL”, “BT”
colors
dict[str, dict[str, str]] | None
Custom color overrides per node class, e.g. {"function": {"fill": "#fff", "stroke": "#000"}}
MermaidDiagram
Diagram object that auto-renders in notebooks and converts to raw Mermaid source via str() or print()

to_flat_graph

flat = graph.to_flat_graph()
Create a flattened NetworkX graph with all nested nodes.
nx.DiGraph
New DiGraph where all nodes (root + nested) are in one graph with hierarchical node IDs
Returns a new DiGraph where:
  • All nodes (root + nested) are in one graph
  • Node attributes include parent for hierarchy
  • Node IDs are hierarchical to prevent collisions (e.g., “pipeline1/process”)
  • Edges include both root-level and nested edges
  • Graph attributes include input_spec and output_to_sources

iter_nodes

for node in graph.iter_nodes():
    print(node.name)
Iterate over all nodes without copying the internal dict.
Iterable[HyperNode]
Iterator over node objects

debug_viz

debugger = graph.debug_viz()
info = debugger.trace_node("my_node")
Get a debugger for this graph’s visualization.
VizDebugger
VizDebugger instance for tracing nodes/edges and finding issues

Build docs developers (and LLMs) love