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, durable resume, one-line observability, async execution, an MCP client, 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. 86 compile-time check fixtures (58 should-fail + 28 should-pass) backed by a rustc-style fixture suite. 3,100+ tests, 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.

Batteries included

Everything a production agent needs — typed, wired, and durable:

• MCP client (neograph[mcp]) — connect to Model Context Protocol servers with typed tool results (output_model=), typed resource hydration (Annotated[T, FromResource(uri)]), per-run identity fresh on every request and transport, run-scoped connections that survive interrupt/resume, gated mutations, progress notifications, transport resilience, and keyless test fakes.
• Async-native — one graph, four verbs. The same compiled graph runs under run / arun / stream / astream — no async flag at compile time, no second pipeline. A sync/async driver↔checkpointer mismatch fails loud instead of half-persisting.
• Durable resume. Checkpoint with a thread_id; change a node's output schema and neograph auto-rewinds to re-run only the affected nodes — and fails loud rather than hand back stale results.
• BAML-style prompt rendering. Pydantic models render to a TypeScript-like schema LLMs parse more reliably than JSON Schema; inline ${var} and template-ref prompts; compile_prompt() gives byte-identical prompts inside and outside the graph for eval harnesses.
• One-line observability. observe= auto-attaches Langfuse tracing and flushes on finish; structured logs and named spans on every node.

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

29 runnable examples in examples/, 6 multi-file mini-projects (lead-research, code-review, spec-builder, incident-triage, lead-outreach, rfp-response), 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.