NeoGraph

Write Python. Get a production graph.

Docs & guides: neograph.pro — full documentation site with tutorials, API reference, and side-by-side LangGraph comparisons.

# uv (recommended)
uv add neograph

# pip
pip install neograph

Define your LLM pipeline as Python functions. The framework infers the topology, validates types at assembly time, and compiles to LangGraph with checkpointing, observability, and tool orchestration. No DSL. No YAML. No add_node / add_edge.

A function is a node. A parameter name is an edge. An if is a branch.

Functions are nodes

from neograph import node, construct_from_module, compile, run

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

@node(outputs=Classified, prompt='rw/classify', model='fast')
def classify(decompose: Claims) -> Classified: ...

@node(outputs=Report)
def report(classify: Classified) -> Report:
    return Report(summary=f"{len(classify.items)} claims processed")

pipeline = construct_from_module(sys.modules[__name__])
graph = compile(pipeline)
result = run(graph, input={'node_id': 'doc-001'})

classify(decompose: Claims) — the parameter name IS the dependency. Rename a function, downstream breaks at import time. Fan-in is just more parameters: def report(claims, scores, verified).

Mode is inferred. prompt= + model= means LLM call (think mode). Neither means the function body runs (scripted mode).

if is a branch

from neograph import ForwardConstruct, Node, compile

class Analysis(ForwardConstruct):
    check   = Node(outputs=CheckResult, prompt='check', model='fast')
    deep    = Node(outputs=Result, prompt='deep-analysis', model='reason')
    shallow = Node(outputs=Result, prompt='quick-scan', model='fast')

    def forward(self, topic):
        checked = self.check(topic)
        if checked.confidence > 0.8:
            return self.shallow(checked)
        else:
            return self.deep(checked)

graph = compile(Analysis())

The if compiles to a conditional edge. for compiles to fan-out. Python is the graph language. Your type checker sees everything. Your debugger works.

Everything else is a keyword

# Fan-out over a collection
@node(outputs=MatchResult, map_over='clusters.groups', map_key='label')
def verify(cluster: ClusterGroup) -> MatchResult: ...

# N-way ensemble with merge
@node(outputs=Claims, prompt='decompose', model='reason',
      ensemble_n=3, merge_fn='merge_claims')
def decompose() -> Claims: ...

# Fan-out + ensemble on the same node (Each x Oracle fusion)
@node(outputs=ClaimGroupingResult, prompt='decompose', model='reason',
      ensemble_n=3, merge_fn='group_claims',
      map_over='chunk_document', map_key='chunk_idx')
def decompose(chunk: ReadContext) -> ClaimGroupingResult: ...

# Human-in-the-loop interrupt
@node(outputs=ValidationResult,
      interrupt_when=lambda s: {'issues': s.check_quality.issues} if not s.check_quality.passed else None)
def check_quality(claims: Claims) -> ValidationResult: ...

# Tool-approval gate — pause before an agent/act tool runs; fail-closed on deny
@node(outputs=WriteResult, mode='act', model='ops', prompt='apply',
      tools=[Tool("write_file", budget=3)],
      gate_tools_when=lambda s: {'action': 'about to write files'})
def apply_changes(plan: Plan) -> WriteResult: ...

# ask_human — pause mid-tool for a typed answer (sugar over LangGraph interrupt())
from neograph import ask_human

@tool
def confirm(path: str) -> str:
    '''Ask a human to approve writing to path.'''
    answer = ask_human(ConfirmRequest(path=path), resume_model=ConfirmReply)
    return "ok" if answer.approved else "skipped"

# Agent with tools — typed tool results preserved
@node(outputs={"result": ExplorationResult, "tool_log": list[ToolInteraction]},
      mode='agent', model='research', prompt='explore',
      tools=[Tool("search", budget=5)],
      context=["catalog"])     # verbatim state injection
def explore(claim: VerifyClaim) -> ExplorationResult: ...

# Non-node parameters: runtime input, config, constants
from typing import Annotated
from neograph import FromInput, FromConfig

@node(outputs=Report)
def summarize(
    claims: Claims,                                   # upstream node
    topic: Annotated[str, FromInput],                 # from run(input={...})
    rate_limiter: Annotated[RateLimiter, FromConfig], # from config
    max_items: int = 10,                              # constant
) -> Report: ...

Catches mistakes before you run

ConstructError: Node 'verify' declares inputs=ClusterGroup but no upstream
  produces a compatible value.
  upstream producers:
    - node 'cluster': Clusters
  hint: did you forget to fan out? try .map(lambda s: s.cluster.groups, key='...')
  at my_pipeline.py:42

Types are validated at assembly time — when you define the pipeline, not when you execute it. 68 compile-time check fixtures (52 should-fail + 16 should-pass) backed by a rustc-style fixture suite. 1362 tests total, including Hypothesis property-based testing. CLI validation: neograph check my_pipeline.py.

Visualize the compiled graph

from neograph import compile, describe_graph

graph = compile(pipeline)
print(describe_graph(graph))   # Mermaid diagram — paste into GitHub, docs, mermaid.live

Set NEOGRAPH_DEV=1 for auto-printed DAG summaries after every compile().

Scales to real systems

Organize by module. Each pipeline is a Python module. Import nodes across modules. construct_from_module finds them all.

Sub-constructs from @node functions. construct_from_functions("verify", [explore, score], input=Claim, output=Result) builds a sub-construct with port param resolution. Mix @node functions and sub-constructs in one construct_from_functions call.

Observe everything. Structured logs on every node. Pass trace providers and shared resources via Annotated[T, FromConfig].

Retry on failure. Output-quality retries (malformed JSON, validation errors) are configured per node via LlmConfig.max_retries. Transient API failures (network, 429, 5xx) belong in your llm_factory via model.with_retry(...). See the retry-semantics page on neograph.pro.

Test at every level. node.run_isolated() for unit tests. compile() + run() for integration. forward() direct-call for debugging.

LLMs can build the graph too

For runtime construction — an LLM emitting a pipeline via tool calls, a config system defining workflows — use the programmatic API with the | pipe syntax:

from neograph import Node, Construct, Oracle, Each, compile, run

decompose = Node("decompose", mode="think", outputs=Claims,
                 prompt="rw/decompose", model="reason") | Oracle(n=3, merge_fn="merge")
verify = Node("verify", mode="agent", outputs=MatchResult,
              prompt="verify", model="fast") | Each(over="decompose.items", key="label")

pipeline = Construct("dynamic", nodes=[decompose, verify])
graph = compile(pipeline)

Three surfaces — @node, ForwardConstruct, Node | Modifier — one compiler.

Documentation

Full documentation is at neograph.pro:

• Quick Start — install, configure, build a pipeline, run it
• The @node API — functions as nodes, modifier kwargs, FromInput/FromConfig, organizing pipelines
• ForwardConstruct — class-based pipelines with Python if/for/try
• Runtime Construction — LLM-driven pipeline assembly, programmatic API
• vs LangGraph — side-by-side for five common patterns
• API Reference

Examples

19 runnable examples in examples/, 3 multi-file mini-projects (lead-research, code-review, spec-builder), and 5 side-by-side LangGraph comparisons in examples/vs_langgraph/. Walkthroughs at neograph.pro.

License

Code: MIT

Documentation content © 2025-2026 Constantine Mirin, mirin.pro. Licensed under CC BY-ND 4.0.