1. Scripted Pipeline

A scripted pipeline is the simplest thing you can build with NeoGraph. Every node is a plain Python function decorated with @node. No API keys, no model configuration, no network calls. Data flows through Pydantic models that enforce types at every boundary.

This walkthrough builds a document processor: extract raw text, split it into claims, classify each claim by category.

What you will learn

Defining Pydantic schemas for typed state
Using the @node decorator to turn functions into graph nodes
Parameter-name dependency inference (the parameter name IS the edge)
Auto-discovering nodes with construct_from_module()
Compiling and running the pipeline

Schemas

Each schema is a frozen Pydantic model. Frozen ensures immutability — once a node produces output, nothing can mutate it downstream.

from pydantic import BaseModel

class RawText(BaseModel, frozen=True):
    text: str

class Claims(BaseModel, frozen=True):
    items: list[str]

class ClassifiedClaims(BaseModel, frozen=True):
    classified: list[dict[str, str]]

Nodes

Each @node-decorated function becomes a graph node. Dependencies are inferred from parameter names: if a parameter is named extract, it wires to the upstream node named extract.

from neograph import node

@node(output=RawText)
def extract() -> RawText:
    """Simulate extracting text from a document source."""
    return RawText(text="The system shall log all access attempts. The system shall validate input.")


@node(output=Claims)
def split(extract: RawText) -> Claims:
    """Split raw text into individual claims by sentence."""
    sentences = [s.strip() for s in extract.text.split(".") if s.strip()]
    return Claims(items=sentences)


@node(output=ClassifiedClaims)
def classify(split: Claims) -> ClassifiedClaims:
    """Classify each claim by category based on keywords."""
    classified = []
    for claim in split.items:
        category = "security" if "access" in claim.lower() or "validate" in claim.lower() else "general"
        classified.append({"claim": claim, "category": category})
    return ClassifiedClaims(classified=classified)

split(extract: RawText) — the parameter name extract IS the dependency. It tells the compiler: “this node consumes the output of the extract node.” Rename the upstream function, and this breaks at import time. No silent wiring bugs.

Building the pipeline

No nodes=[...] list, no ordering, no edge wiring. construct_from_module walks the current module, finds all @node-decorated functions, and topologically sorts them into a Construct:

import sys
from neograph import construct_from_module

pipeline = construct_from_module(sys.modules[__name__], name="doc-processor")

The complete pipeline

"""Scripted pipeline: extract -> split -> classify. No LLM needed.

Run:
    python 01_scripted_pipeline.py
"""

from __future__ import annotations

import sys

from pydantic import BaseModel

from neograph import compile, construct_from_module, node, run


# -- Schemas ----------------------------------------------------------------

class RawText(BaseModel, frozen=True):
    text: str

class Claims(BaseModel, frozen=True):
    items: list[str]

class ClassifiedClaims(BaseModel, frozen=True):
    classified: list[dict[str, str]]


# -- Nodes ------------------------------------------------------------------
# Parameter name = upstream node name. No manual edge wiring.

@node(output=RawText)
def extract() -> RawText:
    """Simulate extracting text from a document source."""
    return RawText(text="The system shall log all access attempts. The system shall validate input.")


@node(output=Claims)
def split(extract: RawText) -> Claims:
    """Split raw text into individual claims by sentence."""
    sentences = [s.strip() for s in extract.text.split(".") if s.strip()]
    return Claims(items=sentences)


@node(output=ClassifiedClaims)
def classify(split: Claims) -> ClassifiedClaims:
    """Classify each claim by category based on keywords."""
    classified = []
    for claim in split.items:
        category = "security" if "access" in claim.lower() or "validate" in claim.lower() else "general"
        classified.append({"claim": claim, "category": category})
    return ClassifiedClaims(classified=classified)


# -- Build pipeline ---------------------------------------------------------

pipeline = construct_from_module(sys.modules[__name__], name="doc-processor")


# -- Run --------------------------------------------------------------------

if __name__ == "__main__":
    graph = compile(pipeline)
    result = run(graph, input={"node_id": "doc-001"})

    print("Claims found:", len(result["classify"].classified))
    for item in result["classify"].classified:
        print(f"  [{item['category']}] {item['claim']}")

Expected output

Claims found: 2
  [security] The system shall log all access attempts
  [security] The system shall validate input

How parameter-name wiring works

The @node decorator inspects each function’s signature. Parameters whose type annotation matches a Pydantic model produced by another node are resolved as dependencies:

extract() — no parameters, so it runs first (root node)
split(extract: RawText) — parameter named extract resolves to the extract node’s output
classify(split: Claims) — parameter named split resolves to the split node’s output

Fan-in is just more parameters: def report(claims: Claims, scores: Scores, verified: Verified) depends on three upstream nodes.

Mode inference

Neither prompt= nor model= is set on any of these nodes. The function body is present. NeoGraph infers mode="scripted" — pure Python, no LLM. If you add prompt= and model=, the body becomes unused and the LLM handles execution.

Key points

No LLM required. Scripted nodes are pure Python. Use them for parsing, formatting, validation, or any deterministic logic.
Type safety at the boundary. Pydantic validates every input and output. If your function returns the wrong shape, you get an error at runtime, not a silent corruption downstream.
No explicit edges. Parameter names are edges. The compiler infers topology from function signatures.
No node list. construct_from_module discovers @node-decorated functions automatically and topologically sorts them.