Subgraphs

A Construct can contain other Constructs. When a Construct appears inside another Construct’s nodes list, it becomes a subgraph with its own isolated state. Internal nodes do not leak into the parent.

Two ways to build sub-constructs

Sub-constructs can be built with @node functions or with declarative Node(...) — both produce the same IR and compile identically.

Option A: `@node` functions via `construct_from_functions`

Use construct_from_functions(name, functions, input=X, output=Y) to build a sub-construct from @node-decorated functions. Parameters whose type matches input= are automatically wired to the sub-construct’s input port:

from neograph import node, construct_from_functions, Tool, ToolInteraction

@node(mode="agent", outputs={"result": ExplorationResult, "tool_log": list[ToolInteraction]},
      model="research", prompt="verify/explore", tools=[Tool("search", budget=3)])
def explore(claim: VerifyClaim) -> ExplorationResult: ...

@node(mode="think", outputs=ClaimVerdict, model="judge", prompt="verify/score")
def score(explore_result: ExplorationResult, explore_tool_log: list[ToolInteraction]) -> ClaimVerdict: ...

verify = construct_from_functions(
    "verify", [explore, score],
    input=VerifyClaim, outputs=ClaimVerdict,
)

The explore function’s claim: VerifyClaim parameter matches input=VerifyClaim, so it reads from the sub-construct’s input port. The score function’s parameters explore_result and explore_tool_log reference the upstream explore node’s dict-form output keys — standard @node wiring.

Option B: Declarative `Construct(input=X, output=Y, nodes=[...])`

The declarative form uses Node(...) or Node.scripted(...) directly. This is useful when you need explicit control over node names and wiring, or when building pipelines programmatically at runtime:

from neograph import Construct, Node

enrich = Construct(
    "enrich",
    input=Claims,
    output=ScoredClaims,
    nodes=[lookup, verify, score],
)

Both forms produce the same Construct with input/output boundary ports. The rest of this page applies equally to both.

Declaring a sub-Construct

A sub-Construct must declare both input and output to define what crosses the boundary:

from pydantic import BaseModel
from neograph import Construct, Node

class Claims(BaseModel):
    items: list[str]

class ScoredClaims(BaseModel):
    scored: list[dict]

# Internal nodes use the declarative Node(...) form
lookup = Node("lookup", mode="agent", inputs=Claims, outputs=LookupResults,
              model="reason", prompt="rw/lookup",
              tools=[Tool("search", budget=5)])
verify = Node("verify", mode="think", inputs=LookupResults, outputs=VerifiedClaims,
              model="fast", prompt="rw/verify")
score  = Node("score", mode="think", inputs=VerifiedClaims, outputs=ScoredClaims,
              model="fast", prompt="rw/score")

# Sub-construct with I/O boundary
enrich = Construct(
    "enrich",
    input=Claims,
    output=ScoredClaims,
    nodes=[lookup, verify, score],
)

The input type tells the framework how to extract data from the parent state. At runtime, the framework scans the parent state for a field whose value is an instance of Claims and passes it into the sub-graph. The output type defines what comes back — the sub-graph’s result is stored in parent_state.enrich as a ScoredClaims instance.

Using a sub-Construct in a top-level pipeline

The top-level pipeline uses @node for its own nodes. The sub-construct slots in alongside them:

from neograph import node, construct_from_module, compile, run
import sys

@node(outputs=RawText, prompt='rw/extract', model='fast')
def extract(topic: Annotated[str, FromInput]) -> RawText: ...

@node(outputs=Claims, prompt='rw/decompose', model='reason')
def decompose(extract: RawText) -> Claims: ...

# enrich is the sub-Construct defined above

@node(outputs=FinalReport, prompt='rw/report', model='fast')
def report(enrich: ScoredClaims) -> FinalReport: ...

# construct_from_module discovers @node functions;
# add the sub-construct manually to the node list
pipeline = Construct(
    "ingestion",
    nodes=[extract, decompose, enrich, report],
)

graph = compile(pipeline)
result = run(graph, input={"node_id": "BR-042"})

The compiled graph executes extract -> decompose -> enrich (subgraph) -> report in sequence. The enrich subgraph runs its three internal nodes (lookup -> verify -> score) in isolation, then surfaces the ScoredClaims result back to the parent.

State isolation

The parent state model will contain a field enrich: ScoredClaims | None, but it will not contain lookup, verify, or score fields. Those exist only inside the sub-graph’s state.

This means:

The report node can read enrich (the ScoredClaims output) but cannot see lookup or verify results.
If the sub-Construct has a node named score and the parent also has a node named score, there is no conflict. They write to different state models.
The sub-graph’s internal framework fields (neo_oracle_gen_id, neo_each_item, etc.) are confined to the sub-graph.

Nesting: Construct in Construct

Nesting composes. A sub-Construct can itself contain sub-Constructs:

# Level 2: inner sub-construct
verify_block = Construct(
    "verify-block",
    input=LookupResults,
    output=VerifiedClaims,
    nodes=[cross_check, validate, consolidate],
)

# Level 1: outer sub-construct containing the inner one
enrich = Construct(
    "enrich",
    input=Claims,
    output=ScoredClaims,
    nodes=[lookup, verify_block, score],
)

# Top level
pipeline = Construct(
    "ingestion",
    nodes=[extract, decompose, enrich, report],
)

Each level gets its own isolated state. The inner verify-block is invisible to enrich’s siblings, and enrich’s internals are invisible to pipeline’s nodes.

Modifiers on Constructs

All three modifiers — Oracle, Each, Operator — work on Constructs. When applied, the entire sub-pipeline is treated as the modified unit.

Oracle on a Construct

Run the entire sub-pipeline N times in parallel, then merge the outputs:

enrich = Construct(
    "enrich",
    input=Claims,
    output=ScoredClaims,
    nodes=[lookup, verify, score],
) | Oracle(n=3, merge_fn="combine_scores")

The compiler creates N copies of the sub-graph execution, each receiving the same parent state but running independently. The merge function receives a list of N ScoredClaims results.

Each on a Construct

Run the entire sub-pipeline once per item in a collection:

review_cluster = Construct(
    "review-cluster",
    input=ClusterGroup,
    output=ClusterReview,
    nodes=[analyze, check, summarize],
) | Each(over="clusters.groups", key="label")

If clusters.groups contains 5 items, the sub-pipeline runs 5 times in parallel. The result is a dict:

result["review_cluster"] == {
    "auth": ClusterReview(...),
    "payments": ClusterReview(...),
    "search": ClusterReview(...),
    ...
}

Operator on a Construct

Pause after the entire sub-pipeline completes:

validation = Construct(
    "validation",
    input=Draft,
    output=ValidationResult,
    nodes=[lint, test, report],
) | Operator(when="validation_failed")

The interrupt check runs after all three internal nodes have completed, not after each one individually.

Compile-time validation

The compiler enforces sub-Construct contracts:

A sub-Construct without input raises ConstructError at compile time.
A sub-Construct without output raises ConstructError during state model generation.
An Operator modifier anywhere in the graph without a checkpointer passed to compile() raises ConstructError.