API Reference

NeoGraph exposes three API surfaces that compile to the same IR:

1. @node API — the primary surface for human-written pipelines
2. ForwardConstruct — class-based with Python control flow
3. Programmatic / IR — Node + Construct + | pipe for runtime construction

All three are importable from neograph:

from neograph import (
    # Primary @node API
    node, construct_from_module, construct_from_functions,
    FromInput, FromConfig, merge_fn,
    # ForwardConstruct
    ForwardConstruct,
    # Programmatic IR
    Node, Construct, Oracle, Each, Operator,
    # Tools
    Tool, tool, ToolInteraction,
    # BAML rendering
    describe_type, describe_value,
    # Prompt inspection + standalone compilation
    render_prompt, compile_prompt,
    # Renderers (opt-in alternatives to BAML)
    Renderer, XmlRenderer, DelimitedRenderer, JsonRenderer,
    # Shared
    compile, run,
    ExcludeFromOutput,
    # Errors
    NeographError, ConstructError, CompileError, ConfigurationError,
    CheckpointSchemaError, ExecutionError,
    # Spec loader
    load_spec, register_type, lookup_type,
)

RenderedInput and build_rendered_input are not part of the top-level neograph namespace; import them from neograph.renderers (from neograph.renderers import RenderedInput, build_rendered_input).

---

Symbol Reference

Every public symbol, generated from the library itself (signature + fields) so this table never drifts from the code. Headings are stable anchors — link to any symbol as /reference/api/#<name> (colliding names are disambiguated, e.g. #node-function vs #node-model). Errors live in the Error Hierarchy below.

arun

arun(graph: 'CompiledNeograph', input: 'dict[str, Any] | None' = None, resume: 'dict[str, Any] | None' = None, config: 'RunnableConfig | None' = None, auto_resume: 'bool' = True, observe: 'bool | str | None' = None) -> 'Any'

ask_human

ask_human(payload: 'BaseModel', resume_model: 'type[T] | None' = None) -> 'T | dict'

astream

astream(graph: 'CompiledNeograph', input: 'dict[str, Any] | None' = None, resume: 'dict[str, Any] | None' = None, config: 'RunnableConfig | None' = None, auto_resume: 'bool' = True, stream_mode: 'str | list[str]' = 'values', observe: 'bool | str | None' = None) -> 'Any'

BlobResult

BlobResult(uri: str, mime: str | None = None, text: str | None = None, bytes_b64: str | None = None, size: int | None = None)

Field	Type	Required	Default
uri	str	yes
mime	`str	None`	no
text	`str	None`	no
bytes_b64	`str	None`	no
size	`int	None`	no

compile

compile(construct: 'Construct', checkpointer: 'Any' = None, _context_types: 'dict[str, type] | None' = None, *, llm_factory: 'Any' = None, prompt_compiler: 'Any' = None, renderer: 'Any' = None, cost_callback: 'Any' = None, scripted: 'dict[str, Callable] | None' = None, conditions: 'dict[str, Callable] | None' = None, tool_factories: 'dict[str, Callable] | None' = None, _runtime: 'LlmRuntime | None' = None, _scripted_lookup: 'dict[str, Callable] | None' = None) -> 'CompiledNeograph'

compile_prompt

compile_prompt(template: 'str', input_data: 'Any', *, output_model: 'type[BaseModel] | None' = None, output_schema: 'str | None' = None, di_inputs: 'dict[str, Any] | None' = None, prompt_compiler: 'Any | None' = None, renderer: 'Any | None' = None, node_name: 'str' = '', template_text: 'str | None' = None, loader: 'Any | None' = None) -> 'list[dict]'

configure_image

configure_image(*, max_size_bytes: 'int' = 20971520, allowed_dirs: 'list[str] | None' = None, validate_format: 'bool' = True) -> 'None'

Construct

Construct(name: str, modifier_set: ModifierSet = <factory:ModifierSet>, description: str = '', nodes: list[neograph._ir_protocols.ConstructItem] = <factory:list>, input: type[pydantic.main.BaseModel] | None = None, output: type[pydantic.main.BaseModel] | None = None, llm_config: LlmConfig = <factory:LlmConfig>, renderer: neograph.renderers.Renderer | None = None)

Field	Type	Required	Default
name	str	yes
modifier_set	ModifierSet	no	<factory:ModifierSet>
description	str	no	''
nodes	list[neograph._ir_protocols.ConstructItem]	no	<factory:list>
input	`type[pydantic.main.BaseModel]	None`	no
output	`type[pydantic.main.BaseModel]	None`	no
llm_config	LlmConfig	no	<factory:LlmConfig>
renderer	`neograph.renderers.Renderer	None`	no

construct_from_functions

construct_from_functions(name: 'str', functions: 'list[Any]', *, llm_config: 'dict[str, Any] | LlmConfig | None' = None, input: 'type[BaseModel] | None' = None, output: 'type[BaseModel] | None' = None) -> 'Construct'

construct_from_module

construct_from_module(mod: 'Any', name: 'str | None' = None, *, llm_config: 'dict[str, Any] | LlmConfig | None' = None, input: 'type[BaseModel] | None' = None, output: 'type[BaseModel] | None' = None) -> 'Construct'

CostCallback

CostCallback(*args, **kwargs)

DefaultPromptCompiler

DefaultPromptCompiler(loader: 'TemplateLoader', *, strict: 'bool' = True, syntax: 'SyntaxSpec' = 'brace', system: 'str | None' = None, schema_var: 'str' = 'json_schema', suffix: 'str' = '.md') -> 'None'

DelimitedRenderer

DelimitedRenderer()

describe_type

describe_type(model: 'type[BaseModel]', *, prefix: 'str' = 'Answer in JSON matching this schema:', hoist_classes: "Literal['auto', 'all'] | list[str]" = 'auto', always_hoist_enums: 'bool' = True, or_splitter: 'str' = ' or ', indent: 'str' = '  ') -> 'str'

describe_value

describe_value(value: 'Any', *, prefix: 'str' = '', indent: 'str' = '  ') -> 'str'

Each

Each(over: str, key: str, on_error: Literal['raise', 'collect'] = 'raise')

Field	Type	Required	Default
over	str	yes
key	str	yes
on_error	Literal['raise', 'collect']	no	'raise'

EachFailure

EachFailure(key: str, error_type: str, message: str)

Field	Type	Required	Default
key	str	yes
error_type	str	yes
message	str	yes

emit_progress

emit_progress(event: 'BaseModel') -> 'None'

ExcludeFromOutput

ExcludeFromOutput()

ForwardConstruct

ForwardConstruct(name: str, modifier_set: ModifierSet = <factory:ModifierSet>, description: str = '', nodes: list[neograph._ir_protocols.ConstructItem] = <factory:list>, input: type[pydantic.main.BaseModel] | None = None, output: type[pydantic.main.BaseModel] | None = None, llm_config: LlmConfig = <factory:LlmConfig>, renderer: neograph.renderers.Renderer | None = None)

FromConfig

FromConfig(*, required: 'bool' = True) -> 'None'

FromInput

FromInput(*, required: 'bool' = True) -> 'None'

FromResource

FromResource(uri: 'str | None' = None, *, ref: 'str | None' = None, mime: 'str | None' = None, parse: 'Callable[[Any, str | None], Any] | None' = None, required: 'bool' = True, max_bytes: 'int | None' = None) -> 'None'

inject_schema

inject_schema(vars: 'dict[str, Any]', output_model: 'type[BaseModel] | None' = None, *, output_schema: 'str | None' = None, key: 'str' = 'json_schema') -> 'dict[str, Any]'

JsonRenderer

JsonRenderer(*, indent: 'int' = 2) -> 'None'

lint

lint(construct: 'Construct', *, config: 'dict[str, Any] | None' = None, known_template_vars: 'set[str] | None' = None, template_resolver: 'Callable[[str], str | None] | None' = None, llm_factory: 'Any' = None, prompt_compiler: 'Any' = None, conditions: 'dict[str, Callable] | None' = None, tool_factories: 'dict[str, Callable] | None' = None) -> 'list[LintIssue]'

LintIssue

LintIssue(node_name: 'str', param: 'str', kind: 'str', message: 'str', required: 'bool' = False) -> None

Field	Type	Required	Default
node_name	str	yes
param	str	yes
kind	str	yes
message	str	yes
required	bool	no	False

LlmFactory

LlmFactory(*args, **kwargs)

load_spec

load_spec(spec: 'str | dict[str, Any]', project: 'str | dict[str, Any] | None' = None) -> 'Construct'

lookup_type

lookup_type(name: 'str') -> 'type[BaseModel]'

Loop

Loop(when: str | collections.abc.Callable[[Any], bool], max_iterations: int = 10, on_exhaust: str = 'error', history: bool = False)

Field	Type	Required	Default
when	`str	collections.abc.Callable[[Any], bool]`	yes
max_iterations	int	no	10
on_exhaust	str	no	'error'
history	bool	no	False

merge_fn

merge_fn(fn: 'Callable | None' = None, *, name: 'str | None' = None) -> 'Any'

MergeFallback

MergeFallback(*args, **kwargs)

MergePostProcess

MergePostProcess(*args, **kwargs)

MergePreProcess

MergePreProcess(*args, **kwargs)

node (function)

node(fn: 'Callable | None' = None, *, mode: "Literal['think', 'agent', 'act', 'scripted', 'raw'] | None" = None, inputs: 'Any' = None, outputs: 'Any' = None, model: 'str | None' = None, prompt: 'str | None' = None, llm_config: 'dict[str, Any] | LlmConfig | None' = None, tools: 'list[Tool] | None' = None, name: 'str | None' = None, map_over: 'str | None' = None, map_key: 'str | None' = None, ensemble_n: 'int | None' = None, models: 'list[str] | None' = None, merge_fn: 'str | None' = None, merge_prompt: 'str | None' = None, merge_pre_process: 'Callable | None' = None, merge_post_process: 'Callable | None' = None, merge_fallback: 'Callable | None' = None, interrupt_when: 'str | Callable | None' = None, renderer: 'Renderer | None' = None, context: 'list[str] | None' = None, skip_when: 'Callable | None' = None, skip_value: 'Callable | None' = None, gate_tools_when: 'Callable | str | None' = None, loop_when: 'str | Callable | None' = None, max_iterations: 'int | None' = None, on_exhaust: "Literal['error', 'last'] | None" = None) -> 'Any'

Node (model)

Node(name: str, modifier_set: ModifierSet = <factory:ModifierSet>, mode: Literal['think', 'agent', 'act', 'scripted'] = 'think', inputs: Any = None, outputs: Any = None, model: str | None = None, prompt: str | None = None, llm_config: LlmConfig = <factory:LlmConfig>, tools: list[neograph.tool.Tool | langchain_core.tools.base.BaseTool] = [], scripted_fn: str | None = None, raw_fn: neograph.node.RawNodeFn | None = None, fan_out_param: str | None = None, renderer: neograph.renderers.Renderer | None = None, context: list[str] | None = None, skip_when: neograph.node.SkipPredicate | None = None, skip_value: neograph.node.SkipValueFactory | None = None, gate_tools_when: collections.abc.Callable | str | None = None, oracle_gen_type: type[pydantic.main.BaseModel] | None = None)

Field	Type	Required	Default
name	str	yes
modifier_set	ModifierSet	no	<factory:ModifierSet>
mode	Literal['think', 'agent', 'act', 'scripted']	no	'think'
inputs	Any	no	None
outputs	Any	no	None
model	`str	None`	no
prompt	`str	None`	no
llm_config	LlmConfig	no	<factory:LlmConfig>
tools	`list[neograph.tool.Tool	langchain_core.tools.base.BaseTool]`	no
scripted_fn	`str	None`	no
raw_fn	`neograph.node.RawNodeFn	None`	no
fan_out_param	`str	None`	no
renderer	`neograph.renderers.Renderer	None`	no
context	`list[str]	None`	no
skip_when	`neograph.node.SkipPredicate	None`	no
skip_value	`neograph.node.SkipValueFactory	None`	no
gate_tools_when	`collections.abc.Callable	str	None`
oracle_gen_type	`type[pydantic.main.BaseModel]	None`	no

Operator

Operator(when: str)

Field	Type	Required	Default
when	str	yes

Oracle

Oracle(n: int = 3, models: list[str] | None = None, merge_prompt: str | None = None, merge_model: str = 'reason', merge_fn: str | None = None, merge_pre_process: neograph.modifiers.MergePreProcess | None = None, merge_post_process: neograph.modifiers.MergePostProcess | None = None, merge_fallback: neograph.modifiers.MergeFallback | None = None)

Field	Type	Required	Default
n	int	no	3
models	`list[str]	None`	no
merge_prompt	`str	None`	no
merge_model	str	no	'reason'
merge_fn	`str	None`	no
merge_pre_process	`neograph.modifiers.MergePreProcess	None`	no
merge_post_process	`neograph.modifiers.MergePostProcess	None`	no
merge_fallback	`neograph.modifiers.MergeFallback	None`	no

ProducingCall

ProducingCall(tool_name: str, args: dict[str, Any] = <factory:dict>, producer_idempotent: bool = False)

Field	Type	Required	Default
tool_name	str	yes
args	dict[str, Any]	no	<factory:dict>
producer_idempotent	bool	no	False

PromptCompiler

PromptCompiler(*args, **kwargs)

RawNodeFn

RawNodeFn(*args, **kwargs)

read_blob

register_type

register_type(name: 'str', cls: 'type[BaseModel]') -> 'None'

render_input

render_input(input_data: 'Any', *, renderer: 'Renderer | None') -> 'Any'

render_inputs

render_inputs(input_data: 'Any', *, renderer: 'Renderer | None' = None) -> 'dict[str, Any]'

render_prompt

render_prompt(node: 'Any', input_data: 'Any', *, config: 'RunnableConfig | None' = None, runtime: 'LlmRuntime | None' = None) -> 'str'

Renderer

Renderer(*args, **kwargs)

resolve_image

resolve_image(path_or_b64: 'str') -> 'str'

resource_reader

resource_reader(name: 'str', *, uri_template: 'str', output_model: 'type[BaseModel]', description: 'str', parse: 'Callable[[Any, str | None], BaseModel] | None' = None, budget: 'int' = 0, idempotent: 'bool' = True) -> 'Any'

ResourceRef

ResourceRef(uri: str, kind: str, server: str, producing_call: ProducingCall, mime: str | None = None, size: int | None = None, fetched_at: str | None = None, ttl_ms: int | None = None)

Field	Type	Required	Default
uri	str	yes
kind	str	yes
server	str	yes
producing_call	ProducingCall	yes
mime	`str	None`	no
size	`int	None`	no
fetched_at	`str	None`	no
ttl_ms	`int	None`	no

run

run(graph: 'CompiledNeograph', input: 'dict[str, Any] | None' = None, resume: 'dict[str, Any] | None' = None, config: 'RunnableConfig | None' = None, auto_resume: 'bool' = True, observe: 'bool | str | None' = None) -> 'Any'

SkipPredicate

SkipPredicate(*args, **kwargs)

SkipValueFactory

SkipValueFactory(*args, **kwargs)

stream

stream(graph: 'CompiledNeograph', input: 'dict[str, Any] | None' = None, resume: 'dict[str, Any] | None' = None, config: 'RunnableConfig | None' = None, auto_resume: 'bool' = True, stream_mode: 'str | list[str]' = 'values', observe: 'bool | str | None' = None) -> 'Any'

substitute

substitute(template: 'str', vars: 'dict[str, Any]', *, strict: 'bool' = True, syntax: 'SyntaxSpec' = 'brace') -> 'str'

tool (function)

tool(fn: 'Callable | None' = None, *, name: 'str | None' = None, budget: 'int' = 0, config: 'dict[str, Any] | None' = None) -> 'Any'

Tool (model)

Tool(name: str, budget: int = 0, config: dict[str, Any] = <factory:dict>, idempotent: bool = False)

Field	Type	Required	Default
name	str	yes
budget	int	no	0
config	dict[str, Any]	no	<factory:dict>
idempotent	bool	no	False

ToolInteraction

ToolInteraction(tool_name: str, args: dict[str, Any] = <factory:dict>, result: str = '', typed_result: Any = None, duration_ms: int = 0)

Field	Type	Required	Default
tool_name	str	yes
args	dict[str, Any]	no	<factory:dict>
result	str	no	''
typed_result	Any	no	None
duration_ms	int	no	0

type_display_name

type_display_name(t: 'TypeSpecStatic') -> 'str'

TypeSpecStatic

verify_compiled

verify_compiled(graph: 'Any') -> 'list[VerifyIssue]'

VerifyIssue

VerifyIssue(node_name: 'str', kind: 'str', message: 'str') -> None

Field	Type	Required	Default
node_name	str	yes
kind	str	yes
message	str	yes

XmlRenderer

XmlRenderer(*, include_field_info: "Literal['once', 'always', 'never']" = 'once') -> 'None'

---

Lint kinds

lint() reports these issue kinds. Severity and meaning are manifest-owned (generated from LINT_KIND_META in lint.py), so this table never drifts from what lint() actually emits.

Kind	Severity	Meaning
act_mode_all_idempotent_tools	WARN	mode='act' (mutations) but all tools are idempotent=True (read-only) — probably a misclassification; use mode='agent' unless a tool is a genuinely idempotent mutation.
ask_human_in_mutating_node	WARN	An act-mode (mutating) tool calls ask_human(); a non-idempotent side effect before the mid-loop pause can double-fire on resume. Make any pre-pause mutation idempotent, or move it after the pause.
from_config	varies	Annotated[T, FromConfig] — resolved from config['configurable'], passed directly in config=.
from_config_model	varies	Bundled BaseModel via FromConfig — each model field must exist in config.
from_input	varies	Annotated[T, FromInput] — resolved from config['configurable'], originally from run(input={...}).
from_input_model	varies	Bundled BaseModel via FromInput — each model field must exist in config.
llm_kwargs_missing	WARN	LLM-mode nodes require llm_factory and prompt_compiler at compile() time. Pass these kwargs to compile() (or configure via configure_llm(), legacy).
loop_condition_none_unsafe	WARN/ERROR	Loop when callable raises when called with None. ERROR for registered string conditions (always crash), WARN for user-supplied callables (may handle None via other means).
loop_condition_unregistered	ERROR	Loop when is a string that is not registered in the condition registry.
resource_hydration_kind_unmatched	ERROR	A node hydrates a manifest kind via FromResource(ref=...) but the construct has no upstream agent/act node that can emit a resource_link, so the manifest is guaranteed empty at runtime.
template_placeholder_known_vars_only	WARN	Placeholder only resolvable via known_template_vars, not from actual @node parameter names. Advisory: verify the consumer bridge supplies it at runtime.
template_placeholder_unresolvable	ERROR	Prompt placeholder not found in predicted input keys or known extras.
template_var_requires_async_driver	WARN	A template var is a FromResource DI param whose fetch is awaited, so it resolves only under the async arun() driver (sync run() fails loud). Drive the graph with arun().
tool_requires_async_driver	WARN	An agent/act node is bound to an async-only tool (e.g. an MCP tool) that cannot run under the sync run() driver. Drive the graph with arun().

---

Guides

Modifiers (pipe syntax)

Apply to a Node or Construct via |:

node | Oracle(n=3, merge_fn="combine")
node | Oracle(models=["reason", "fast"], merge_fn="pick_best")
node | Each(over="upstream.field", key="id")
node | Operator(when="condition_name")
node | Loop(when=lambda d: d is None or d.score < 0.8, max_iterations=5)
construct | Loop(when=lambda d: d is None or d.score < 0.8, max_iterations=10)

The merge_prompt path on Oracle receives upstream context as a dict {"variants": [...], upstream_key: val, ...} instead of a bare variant list; templates use ${variants} for the variant list and ${upstream.field} for upstream data. Each(over, key) fans out over a dotted-path collection, keying results by getattr(item, key). Operator(when) is a human-in-the-loop interrupt whose when is a registered condition name. Loop(when, ...) is a cycle modifier: on a Node it self-loops (output feeds back as input); on a Construct the sub-construct re-runs with its output as the next input. when receives the latest output (which may be None on the first iteration, so the callable must be None-safe, e.g. lambda d: d is None or d.score < 0.8); when history=True, each iteration’s output is collected in a state list field.

When output != input on a Construct Loop, the sub-construct re-reads original inputs from parent state on each iteration instead of feeding output back — this enables produce+validate patterns. The | syntax returns a new Node/Construct with the modifier appended, so you can chain: node | Oracle(...) | Operator(...).

Facade verbs

compile() returns a CompiledNeograph facade over the underlying LangGraph graph. Beyond run()/arun()/stream()/astream(), the facade delegates a closed set of methods for inspection, streaming, and state editing — sync: invoke, stream, get_state, get_state_history, get_graph, update_state; async: ainvoke, astream, astream_events, aget_state, aget_state_history, aupdate_state. The raw LangGraph graph is available as .graph for anything outside this set.

graph = compile(pipeline, checkpointer=saver)

# Event-level streaming (token/step observability):
async for event in graph.astream_events({"topic": "x"}, config=config, version="v2"):
    ...

# Inspect checkpointed state without re-running:
snapshot = await graph.aget_state(config)

# Edit state before a HITL resume:
graph.update_state(config, {"human_approved": True})

Runtime wiring (compile() dict kwargs)

The programmatic / runtime-constructed surface references scripted functions, conditions, and tool factories by string name. Supply the implementations as dicts to compile() — there are no global registration functions:

• scripted={"name": fn} — Resolves Node.scripted(fn="name") (and @node(merge_fn="name")).
• conditions={"name": fn} — Resolves Operator(when="name") and string-form interrupt_when="name" / loop_when="name".
• tool_factories={"name": factory} — Resolves declarative tool lookup; factory(config, tool_config) returns a tool.

graph = compile(
    pipeline,
    scripted={"extract_fn": extract_fn, "merge_claims": merge_claims},
    conditions={"needs_review": needs_review},
    tool_factories={"search_code": lambda config, tool_config: SearchTool()},
)

These are only needed for the programmatic API. @node, @tool, and @merge_fn handle wiring automatically. See Pipeline Spec Format for the YAML/JSON spec surface loaded by load_spec.

---

Error Hierarchy

Parentage follows one rule, by lifecycle phase: a failure raised while the graph is executing subclasses ExecutionError; an assembly/validation failure is a ConstructError; a compile() failure is a CompileError; a bad or missing configuration (including a resume-time precondition mismatch) is a ConfigurationError. This means a single except ExecutionError around run() catches every runtime failure — including resource-expiry and non-idempotent-replay.

NeographError (base)
  ├── ConstructError (ValueError)  — assembly-time validation failures
  ├── CompileError                 — compilation failures
  ├── ConfigurationError           — missing LLM config, unregistered tools
  │     └── CheckpointSchemaError  — checkpoint state schema mismatch on resume
  └── ExecutionError               — runtime failures during a run
        ├── PromptVarMissing          — strict prompt placeholder had no value
        ├── StateMissingError         — required state read missed
        ├── NodeOutputError           — a node ran and produced None vs its declared output
        ├── NonIdempotentReplayError  — refused to replay a non-idempotent producing tool
        └── ResourceExpiredError      — a manifest ResourceRef could not be hydrated

CheckpointSchemaError — raised when resuming from a checkpoint whose state schema fingerprint differs from the current graph. Has an invalidated_nodes: set[str] attribute listing which state fields changed. Only raised when auto_resume=False; when auto_resume=True (the default), run() automatically rewinds instead. It subclasses ConfigurationError, not ExecutionError: it fires at resume, before any node re-executes, as a precondition check on the persisted checkpoint against the current graph — a configuration mismatch you resolve with a new thread_id or a schema migration, not a retry of node logic.

All errors are importable: from neograph import ConstructError, CompileError, ConfigurationError, CheckpointSchemaError, ExecutionError, PromptVarMissing, NonIdempotentReplayError, ResourceExpiredError.

---

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