Graph

A Graph defines a computation graph from nodes with automatic edge inference.

• Automatic wiring - Edges inferred from matching output/input names

• Build-time validation (strict_types) - Type mismatches caught at construction when strict_types=True

• Hierarchical composition - Graphs nest as nodes via .as_node()

• Partial input semantics - with_entrypoint() and select() narrow inputs to the active subgraph

• Immutable - bind(), select(), with_entrypoint(), and other methods return new instances

from hypergraph import node, Graph

@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

# Edges inferred: double → add_one (via "doubled")
g = Graph([double, add_one])

Edges are inferred automatically: if node A produces output "x" and node B has input "x", an edge A→B is created.

When multiple producers can satisfy the same consumer input p, implicit inference applies producer shadow-elimination:

• Remove edge u -> v (p) iff every valid execution path by which p could flow from u to v passes through another producer of p first.

• If elimination leaves no realizable producer for (v, p), graph construction fails with GraphConfigError and asks you to add ordering, explicit edges, or rename channels.

Constructor

Graph(nodes, *, edges=None, entrypoint=None, name=None, strict_types=False, shared=None)

Create a graph from nodes.

from hypergraph import node, Graph

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

# Basic usage — edges inferred from matching names
g = Graph([process])

# With name (required for nesting)
g = Graph([process], name="processor")

# With type validation
g = Graph([process], strict_types=True)

Args:

• nodes (list[HyperNode]): List of nodes to include in the graph

• edges (list[tuple] | None): Explicit edge declarations. When shared is set, explicit edges are additive (ordering hints on top of auto-inferred edges). Otherwise, disables auto-inference when provided. See Explicit Edges below.

• entrypoint (str | list[str] | tuple[str, ...] | None): Convenience shortcut for with_entrypoint(). Required for cyclic graphs. Accepts a node name or list/tuple of node names. See Entrypoints below.

• name (str | None): Optional graph name. Required if using as_node() for composition.

• strict_types (bool): If True, validate type compatibility between connected nodes at construction time. Default: False.

• shared (list[str] | None): Parameter names that are shared state. Shared params are excluded from auto-wiring — multiple producers are allowed, and nodes read the latest value from run state. The user provides ordering via edges or emit/wait_for. Shared params are required at run() time unless bound. See Shared State below.

Raises:

• GraphConfigError - If duplicate node names exist

• GraphConfigError - If multiple nodes produce the same output name (in auto-inference mode without ordering)

• GraphConfigError - If strict_types=True and types are incompatible or missing

Properties

name: str | None

The graph's name. Used for identification and required for nesting via as_node().

g = Graph([process], name="my_graph")
print(g.name)  # "my_graph"

g2 = Graph([process])
print(g2.name)  # None

strict_types: bool

Whether type validation is enabled for this graph.

g = Graph([producer, consumer], strict_types=True)
print(g.strict_types)  # True

nodes: dict[str, HyperNode]

Map of node name → node object. Returns a copy to prevent mutation.

g = Graph([double, add_one])
print(list(g.nodes.keys()))  # ['double', 'add_one']
print(g.nodes['double'])     # FunctionNode('double')

nx_graph: nx.DiGraph

The underlying NetworkX directed graph. Useful for advanced graph analysis.

g = Graph([double, add_one])
print(g.nx_graph.edges())  # [('double', 'add_one')]
print(g.nx_graph.has_edge('double', 'add_one'))  # True

inputs: InputSpec

Specification of graph input parameters. See InputSpec Reference for details.

g = Graph([double, add_one])
print(g.inputs.required)  # ('x',)
print(g.inputs.optional)  # ()

outputs: tuple[str, ...]

All output names produced by nodes in the graph.

g = Graph([double, add_one])
print(g.outputs)  # ('doubled', 'result')

leaf_outputs: tuple[str, ...]

Outputs from terminal nodes (nodes with no downstream consumers).

g = Graph([double, add_one])
print(g.leaf_outputs)  # ('result',) - only add_one is a leaf

selected: tuple[str, ...] | None

Default output selection set via select(), or None if all outputs are returned.

g = Graph([double, add_one])
print(g.selected)  # None

g2 = g.select("result")
print(g2.selected)  # ('result',)

entrypoints_config: tuple[str, ...] | None

Configured entry point node names set via with_entrypoint(), or None if all nodes are active.

g = Graph([upstream, downstream])
print(g.entrypoints_config)  # None

g2 = g.with_entrypoint("downstream")
print(g2.entrypoints_config)  # ('downstream',)

has_cycles: bool

True if the graph contains cycles.

@node(output_name="x")
def feedback(x: int) -> int:
    return x + 1

g = Graph([feedback])
print(g.has_cycles)  # True - x feeds back to itself

has_async_nodes: bool

True if any node in the graph is async.

@node(output_name="result")
async def fetch(url: str) -> dict:
    return {}

g = Graph([fetch])
print(g.has_async_nodes)  # True

has_interrupts: bool

True if any node in the graph is an interrupt node.

from hypergraph import interrupt

@interrupt(output_name="decision")
def approval(draft: str) -> str | None:
    return None

g = Graph([make_draft, approval, finalize])
print(g.has_interrupts)  # True

interrupt_nodes: list[HyperNode]

Ordered list of interrupt node instances in the graph.

definition_hash: str

Merkle-tree style hash of graph structure. Used for cache invalidation.

g = Graph([double, add_one])
print(len(g.definition_hash))  # 64 (SHA256 hex string)

The hash includes:

• Node names and their definition hashes

• Graph edges (data dependencies)

The hash excludes:

• Bound values (runtime values, not structure)

• Node order in constructor list (normalized by name)

Methods

describe(*, show_types=True) -> str

Return a multiline summary of the active graph scope.

Use describe() when you want one scan-friendly answer to:

• what inputs the graph expects

• which values are already bound

• which outputs the configured graph can produce

• which nodes are active in the current scope

Unlike graph.outputs, the summary respects scope-defining configuration such as with_entrypoint() and select().

By default, type hints are included for both inputs and outputs when available.

@node(output_name="embedding")
def embed(text: str) -> list[float]:
    return [0.1, 0.2, 0.3]

@node(output_name="docs")
def retrieve(embedding: list[float], query: str, top_k: int = 5) -> list[str]:
    return ["doc1", "doc2"]

@node(output_name="answer")
def generate(docs: list[str]) -> str:
    return "done"

g = Graph([embed, retrieve, generate], name="rag")
print(g.describe())

# rag
#   Inputs:
#     required: text (str), query (str)
#     optional: top_k (int)
#   Outputs: embedding (list[float]) → docs (list[str]) → answer (str)
#   Nodes: embed → retrieve → generate

When you slice the graph, the summary follows the active subgraph:

partial = g.with_entrypoint("retrieve")
print(partial.describe())

# rag
#   Inputs:
#     required: embedding (list[float]), query (str)
#     optional: top_k (int)
#   Outputs: docs (list[str]) → answer (str)
#   Nodes: retrieve → generate

Hide type hints when you only want names:

print(g.describe(show_types=False))

bind(**values) -> Graph

Pre-fill input parameters with values. Returns a new Graph (immutable pattern).

g = Graph([double, add_one])
print(g.inputs.required)  # ('x',)

bound = g.bind(x=5)
print(bound.inputs.required)  # ()
print(bound.inputs.bound)     # {'x': 5}

# Original unchanged
print(g.inputs.required)  # ('x',)

Args:

• **values: Named values to bind. Keys must be graph inputs in the current scope.

Returns: New Graph with bindings applied

Raises:

• ValueError - If binding a name not recognized as a graph input in the current scope

Shared State and Non-Copyable Objects

Bound values are intentionally shared across runs, not deep-copied. This makes .bind() ideal for:

• Stateful objects (database connections, vector stores, embedders)

• Non-copyable objects (objects with thread locks, file handles, C extensions)

• Dependency injection (providing shared resources to multiple nodes)

bind() is not an override mechanism for active internal graph outputs. If a value is still produced by an upstream node in the current graph scope, that producer wins at runtime. To provide a formerly-intermediate value yourself, first slice the graph so it becomes an input:

scoped = graph.with_entrypoint("retrieve").bind(embedding=my_embedding)

class Embedder:
    """Stateful embedder with internal connection pool."""
    def __init__(self):
        self._lock = threading.RLock()  # Non-copyable!
        self._cache = {}

    def embed(self, text: str) -> list[float]:
        with self._lock:
            if text not in self._cache:
                self._cache[text] = compute_embedding(text)
            return self._cache[text]

embedder = Embedder()

@node(output_name="embedding")
def embed_query(query: str, embedder: Embedder) -> list[float]:
    return embedder.embed(query)

# ✅ CORRECT: Use .bind() for shared resources
graph = Graph([embed_query]).bind(embedder=embedder)

Why not use function defaults?

Function signature defaults are deep-copied per run to prevent mutable default mutation:

# ❌ WRONG: Non-copyable object as signature default
@node(output_name="embedding")
def embed_query(query: str, embedder: Embedder = Embedder()) -> list[float]:
    return embedder.embed(query)

graph = Graph([embed_query])
runner.run(graph, {"query": "test"})
# GraphConfigError: Parameter 'embedder' has a default value that cannot be safely copied.
#
# Solution:
#   Use .bind() to provide this value at the graph level instead:
#     graph = Graph([...]).bind(embedder=your_embedder_instance)

The error message explains why copying is needed for signature defaults and suggests using .bind() for shared state.

unbind(*names) -> Graph

Remove specific bindings. Returns a new Graph.

bound = g.bind(x=5, y=10)
print(bound.inputs.bound)  # {'x': 5, 'y': 10}

partial = bound.unbind('x')
print(partial.inputs.bound)  # {'y': 10}

Args:

• *names: Names to unbind

Returns: New Graph with specified bindings removed

select(*names) -> Graph

Set a default output selection. Returns a new Graph (immutable pattern).

This controls which outputs are returned by runner.run() and which outputs are exposed when the graph is used as a nested node via as_node().

select also narrows graph.inputs to only the parameters needed to produce the selected outputs. Nodes that don't contribute to those outputs are excluded from input computation.

from hypergraph import node, Graph

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

@node(output_name="b_val")
def node_b(y: int) -> int:
    return y + 1

g = Graph([node_a, node_b])
print(g.inputs.required)  # ('x', 'y')

# Select narrows inputs to what's actually needed
selected = g.select("a_val")
print(selected.inputs.required)  # ('x',) - y is not needed for a_val

select is scope-defining: it narrows what executes, what is required from the caller, and what is returned to the caller.

g = Graph([embed, retrieve, generate])
print(g.outputs)  # ('embedding', 'docs', 'answer')

# Only return "answer" by default
g_selected = g.select("answer")
result = runner.run(g_selected, {"text": "hello", "query": "what?"})
print(result.values.keys())  # dict_keys(['answer'])

Runtime select= overrides are not supported. Use graph.select(...) to configure output scope before calling run().

Args:

• *names: Output names to include. Must be valid graph outputs.

Returns: New Graph with default selection set

Raises:

• ValueError - If any name is not in graph.outputs

Nested graph behavior

When a graph with select is used as a nested node, only the selected outputs are visible to the parent graph. Unselected outputs cannot be wired to downstream nodes.

inner = Graph([embed, retrieve, generate], name="rag").select("answer")
gn = inner.as_node()
print(gn.outputs)  # ('answer',) — only "answer" is exposed

# Parent graph can only use "answer" from the nested graph
outer = Graph([gn, postprocess])  # postprocess must consume "answer", not "docs"

If the parent graph needs an intermediate output, add it to the selection:

inner = Graph([embed, retrieve, generate], name="rag").select("answer", "docs")

with_entrypoint(*node_names) -> Graph

Set execution entry points. Returns a new Graph (immutable pattern, like bind() and select()).

Entry points define where execution starts. Upstream nodes are excluded from the active subgraph -- their outputs become direct user inputs instead of computed values.

from hypergraph import node, Graph

@node(output_name="embedding")
def embed(text: str) -> list[float]:
    return [0.1, 0.2, 0.3]

@node(output_name="docs")
def retrieve(embedding: list[float], top_k: int = 5) -> list[str]:
    return ["doc1", "doc2"]

@node(output_name="answer")
def generate(docs: list[str], query: str) -> str:
    return f"Answer based on {len(docs)} docs"

g = Graph([embed, retrieve, generate])
print(g.inputs.required)  # ('text', 'query')

# Skip embed, start at retrieve
g2 = g.with_entrypoint("retrieve")
print(g2.inputs.required)  # ('embedding', 'query') - embedding is now a user input

embed is upstream of retrieve, so it is excluded from the active subgraph. Its output embedding is no longer produced by a node -- it becomes a required user input.

Args:

• *node_names: One or more node names to use as entry points.

Returns: New Graph with entry points configured

Raises:

• GraphConfigError - If any name is not a node in the graph

• GraphConfigError - If any name is a gate node (gates control routing, they cannot be entry points)

Chainable

with_entrypoint merges with previous calls. Each call adds to the set:

g2 = g.with_entrypoint("retrieve").with_entrypoint("generate")
print(g2.entrypoints_config)  # ('retrieve', 'generate')

Cycles

Cyclic graphs must be configured with constructor entrypoint:

g = Graph([accumulate, generate, should_continue], entrypoint="accumulate")

with_entrypoint(...) remains useful for DAG slicing and for adding additional entrypoints to an already configured graph.

Composes with select and bind

All four dimensions compose -- with_entrypoint (start), select (end), bind (pre-fill), and function defaults (fallback):

g = Graph([embed, retrieve, generate])

# Start at retrieve, only need "answer", pre-fill top_k
configured = (
    g.with_entrypoint("retrieve")
     .select("answer")
     .bind(top_k=10)
)
print(configured.inputs.required)  # ('embedding', 'query')
print(configured.inputs.optional)  # ('top_k',)

Active-set enforcement

with_entrypoint is not just input narrowing -- it prevents upstream nodes from executing at runtime. Even if you provide all of embed's inputs, it will not run:

g2 = g.with_entrypoint("retrieve")

# embed never runs, even if you provide 'text'
result = runner.run(g2, {"embedding": [0.1, 0.2], "query": "hello"})
# Only retrieve and generate execute

as_node(*, name=None) -> GraphNode

Wrap graph as a node for composition. Returns a new GraphNode.

inner = Graph([double], name="doubler")
gn = inner.as_node()

# Use in outer graph
outer = Graph([gn, add_one])

Args:

• name (str | None): Node name. If not provided, uses graph.name.

Returns: GraphNode wrapping this graph

Raises:

• ValueError - If name is None and graph.name is None

See GraphNode section in Nodes Reference for more details.

Type Validation (strict_types)

When strict_types=True, the Graph validates type compatibility between connected nodes at construction time.

How It Works

For each edge (source_node → target_node):

1. Get the output type from source node

2. Get the input type from target node

3. Check if output type is compatible with input type

4. Raise GraphConfigError if incompatible or missing

Compatible Types

@node(output_name="value")
def producer() -> int:
    return 42

@node(output_name="result")
def consumer(value: int) -> int:
    return value * 2

# Types match - construction succeeds
g = Graph([producer, consumer], strict_types=True)

Type Mismatch Error

@node(output_name="value")
def producer() -> int:
    return 42

@node(output_name="result")
def consumer(value: str) -> str:  # Expects str, gets int
    return value.upper()

Graph([producer, consumer], strict_types=True)
# GraphConfigError: Type mismatch between nodes
#   -> Node 'producer' output 'value' has type: int
#   -> Node 'consumer' input 'value' expects type: str
#
# How to fix:
#   Either change the type annotation on one of the nodes, or add a
#   conversion node between them.

Missing Annotation Error

@node(output_name="value")
def producer():  # Missing return type
    return 42

@node(output_name="result")
def consumer(value: int) -> int:
    return value * 2

Graph([producer, consumer], strict_types=True)
# GraphConfigError: Missing type annotation in strict_types mode
#   -> Node 'producer' output 'value' has no type annotation
#
# How to fix:
#   Add type annotation: def producer(...) -> ReturnType

Union Type Compatibility

A more specific type satisfies a broader union type:

@node(output_name="value")
def producer() -> int:
    return 42

@node(output_name="result")
def consumer(value: int | str) -> str:  # Accepts int OR str
    return str(value)

# Works! int is compatible with int | str
g = Graph([producer, consumer], strict_types=True)

When to Use strict_types

• Development: Enable it to catch wiring mistakes early

• Production: Enable it for safety in critical pipelines

• Prototyping: Disable it (default) for quick experiments

# Quick prototype
g = Graph([node1, node2])  # strict_types=False by default

# Production code
g = Graph([node1, node2], strict_types=True)

GraphConfigError

Raised when graph configuration is invalid.

Common Causes

Duplicate node names:

@node(output_name="x")
def process(a: int) -> int: return a

Graph([process, process])
# GraphConfigError: Duplicate node name: 'process'

Multiple nodes produce same output:

@node(output_name="result")
def a(x: int) -> int: return x

@node(output_name="result")
def b(x: int) -> int: return x

Graph([a, b])
# GraphConfigError: Multiple nodes produce 'result'

Three ways to resolve this:

1. emit/wait_for — Prove ordering between the producers. See Shared Outputs in a Cycle.

2. Explicit edges — Declare the full topology manually via edges=[...]. See Explicit Edges below.

3. Mutually exclusive gate branches — Place producers in different branches of an @ifelse or @route gate.

Invalid identifiers:

process.with_name("123-invalid")
# Node names must be valid Python identifiers

Inconsistent defaults:

@node(output_name="x")
def a(value: int = 10) -> int: return value

@node(output_name="y")
def b(value: int) -> int: return value  # No default!

Graph([a, b])
# GraphConfigError: Inconsistent defaults for 'value'

Explicit Edges

By default, Graph infers edges from matching output/input names. Pass edges to disable auto-inference and declare the graph topology manually.

from hypergraph import Graph, node

@node(output_name="messages")
def add_query(messages: list, query: str) -> list:
    return [*messages, {"role": "user", "content": query}]

@node(output_name="response")
def generate(messages: list) -> str:
    return llm.chat(messages)

@node(output_name="messages")
def add_response(messages: list, response: str) -> list:
    return [*messages, {"role": "assistant", "content": response}]

chat = Graph(
    [add_query, generate, add_response],
    edges=[
        (add_query, generate),         # messages
        (generate, add_response),      # response
        (add_response, add_query),     # messages (cycle)
    ],
)

Both add_query and add_response produce messages. With auto-inference this would raise GraphConfigError. Explicit edges make the topology unambiguous.

Edge Format

Each edge is a tuple of (source, target) or (source, target, values):

Format	Behavior
(source, target)	Values auto-detected from intersection of source.outputs and target.inputs
(source, target, "name")	Explicit single value
(source, target, ["a", "b"])	Explicit multiple values

Source and target can be node objects or string names:

# These are equivalent
Graph([a, b], edges=[(a, b)])
Graph([a, b], edges=[("a", "b")])

When a 2-tuple has no overlapping output/input names, it becomes an ordering-only edge — a structural dependency with no data flow.

Restrictions

• add_nodes() raises GraphConfigError on a graph with explicit edges. Create a new Graph with the complete node and edge lists instead.

• Auto-inference and explicit edges don't mix. When edges is provided, only declared edges exist.

• Gate control edges and emit/wait_for ordering edges are still auto-wired in explicit mode. Don't declare edges from gate nodes to their targets — the explicit edge would override the auto-wired control edge type, affecting visualization.

Shared State

When multiple nodes read and write the same value (e.g., messages in a chat loop), auto-wiring can't determine which producer feeds which consumer. Instead of requiring nested graphs or fully explicit edges, declare the param as shared:

from hypergraph import Graph, node, route, END

@node(output_name="messages")
def add_user_message(messages: list, user_input: str) -> list:
    return [*messages, {"role": "user", "content": user_input}]

@node(output_name="response")
def generate(messages: list) -> str:
    return llm.chat(messages)

@node(output_name="messages")
def add_response(messages: list, response: str) -> list:
    return [*messages, {"role": "assistant", "content": response}]

@route(targets=["add_user_message", END])
def should_continue(messages: list) -> str:
    return "add_user_message" if len(messages) < 10 else END

graph = Graph(
    [add_user_message, generate, add_response, should_continue],
    shared=["messages"],
    entrypoint="add_user_message",
    edges=[
        (add_user_message, generate),
        (add_response, should_continue),
    ],
)

How it works

1. No data edges are inferred for messages — it's excluded from auto-wiring

2. Multiple producers (add_user_message, add_response) are allowed without conflict

3. Non-shared params (user_input, response) are still auto-wired normally

4. Ordering must be provided via edges or emit/wait_for — the explicit edges above become ordering-only edges (since messages is shared, no data flows through them)

5. Initial value is required at run() time: runner.run(graph, {"messages": [], ...})

Connectivity validation

If auto-wiring with shared params leaves the graph disconnected, construction fails with a helpful error showing which node groups need connecting:

Graph is disconnected after auto-wiring with shared=['messages'].

These groups of nodes have no edges connecting them:

  [add_response, generate]
    generate -> add_response (response)

  [add_user_message, should_continue]
    should_continue -> add_user_message (control)

How to fix:
  Add edges=[(node_a, node_b), ...] or emit/wait_for to connect them.

In the visualization

Shared params are omitted from the interactive input lane and from INPUT/INPUT_GROUP rendering, and are rendered as a %% shared state: messages comment in Mermaid output.

visualize(*, depth=0, theme="auto", show_types=False, separate_outputs=False, show_inputs=True, show_bounded_inputs=False, filepath=None)

Render an interactive visualization of the graph.

graph.visualize()                          # Display in notebook
graph.visualize(depth=1, show_types=True)  # Expand nested graphs, show types
graph.visualize(filepath="graph.html")     # Save standalone HTML

Args:

• depth (int): How many levels of nested graphs to expand. Default: 0 (all collapsed).

• theme (str): "dark", "light", or "auto" (detects from notebook environment). Default: "auto".

• show_types (bool): Display type annotations on nodes. Default: False.

• separate_outputs (bool): Render outputs as separate DATA nodes instead of direct edges. Default: False.

• show_inputs (bool): Show INPUT/INPUT_GROUP nodes. Default: True.

• show_bounded_inputs (bool): Include bound INPUT/INPUT_GROUP nodes when show_inputs=True. If show_inputs=False, the input lane stays hidden. Default: False.

• filepath (str | None): Save to HTML file instead of displaying inline. Default: None.

Returns: ScrollablePipelineWidget if filepath=None, otherwise None (saves to file).

See Visualize Graphs for the full usage guide.

Complete Example

from hypergraph import node, Graph

# Define nodes
@node(output_name="embedding")
def embed(text: str) -> list[float]:
    return [0.1, 0.2, 0.3]  # Simplified

@node(output_name="docs")
def retrieve(embedding: list[float], top_k: int = 5) -> list[str]:
    return ["doc1", "doc2"]  # Simplified

@node(output_name="answer")
def generate(docs: list[str], query: str) -> str:
    return f"Answer based on {len(docs)} docs"

# Build graph with type validation
g = Graph([embed, retrieve, generate], strict_types=True)

# Inspect graph
print(g.inputs.required)  # ('text', 'query')
print(g.inputs.optional)  # ('top_k',)
print(g.outputs)          # ('embedding', 'docs', 'answer')

# Bind default query
bound = g.bind(query="What is hypergraph?")
print(bound.inputs.required)  # ('text',)

# Create nested graph
g_named = Graph([embed, retrieve, generate], name="rag", strict_types=True)
rag_node = g_named.as_node()
print(rag_node.name)    # 'rag'
print(rag_node.inputs)  # ('text', 'top_k', 'query')