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 neographDefine 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.
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).
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.
# 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: ...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.
from neograph import compile, describe_graph
graph = compile(pipeline)
print(describe_graph(graph)) # Mermaid diagram — paste into GitHub, docs, mermaid.liveSet NEOGRAPH_DEV=1 for auto-printed DAG summaries after every compile().
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.
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.
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.
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
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.
Code: MIT
Documentation content © 2025-2026 Constantine Mirin, mirin.pro. Licensed under CC BY-ND 4.0.