Skip to content
Built by Postindustria. We help teams build agentic production systems.

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 / IRNode + 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).


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(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(payload: 'BaseModel', resume_model: 'type[T] | None' = None) -> 'T | dict'
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(uri: str, mime: str | None = None, text: str | None = None, bytes_b64: str | None = None, size: int | None = None)
FieldTypeRequiredDefault
uristryes
mime`strNone`no
text`strNone`no
bytes_b64`strNone`no
size`intNone`no
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(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(*, max_size_bytes: 'int' = 20971520, allowed_dirs: 'list[str] | None' = None, validate_format: 'bool' = True) -> 'None'
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)
FieldTypeRequiredDefault
namestryes
modifier_setModifierSetno<factory:ModifierSet>
descriptionstrno''
nodeslist[neograph._ir_protocols.ConstructItem]no<factory:list>
input`type[pydantic.main.BaseModel]None`no
output`type[pydantic.main.BaseModel]None`no
llm_configLlmConfigno<factory:LlmConfig>
renderer`neograph.renderers.RendererNone`no
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(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(*args, **kwargs)
DefaultPromptCompiler(loader: 'TemplateLoader', *, strict: 'bool' = True, syntax: 'SyntaxSpec' = 'brace', system: 'str | None' = None, schema_var: 'str' = 'json_schema', suffix: 'str' = '.md') -> 'None'
DelimitedRenderer()
describe_type(model: 'Any', *, 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(value: 'Any', *, prefix: 'str' = '', indent: 'str' = ' ') -> 'str'
dump_spec(construct: 'Construct', *, strict: 'bool' = False) -> 'dict[str, Any]'
Each(over: str, key: str, on_error: Literal['raise', 'collect'] = 'raise')
FieldTypeRequiredDefault
overstryes
keystryes
on_errorLiteral['raise', 'collect']no'raise'
EachFailure(key: str, error_type: str, message: str)
FieldTypeRequiredDefault
keystryes
error_typestryes
messagestryes
emit_progress(event: 'BaseModel') -> 'None'
ExcludeFromOutput()
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(*, required: 'bool' = True) -> 'None'
FromInput(*, required: 'bool' = True) -> 'None'
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(vars: 'dict[str, Any]', output_model: 'type[BaseModel] | None' = None, *, output_schema: 'str | None' = None, key: 'str' = 'json_schema') -> 'dict[str, Any]'
input_contract(construct: 'Construct') -> 'list[InputBinding]'
InputBinding(node_name: 'str', param: 'str', kind: 'str', source: 'str', type_name: 'str', required: 'bool', model_name: 'str | None' = None) -> None
FieldTypeRequiredDefault
node_namestryes
paramstryes
kindstryes
sourcestryes
type_namestryes
requiredboolyes
model_name`strNone`no
JsonRenderer(*, indent: 'int' = 2) -> 'None'
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(node_name: 'str', param: 'str', kind: 'str', message: 'str', required: 'bool' = False) -> None
FieldTypeRequiredDefault
node_namestryes
paramstryes
kindstryes
messagestryes
requiredboolnoFalse
LlmFactory(*args, **kwargs)
load_spec(spec: 'str | dict[str, Any]', project: 'str | dict[str, Any] | None' = None) -> 'Construct'
lookup_type(name: 'str') -> 'type[BaseModel]'
Loop(when: str | collections.abc.Callable[[Any], bool], max_iterations: int = 10, on_exhaust: str = 'error')
FieldTypeRequiredDefault
when`strcollections.abc.Callable[[Any], bool]`yes
max_iterationsintno10
on_exhauststrno'error'
merge_fn(fn: 'Callable | None' = None, *, name: 'str | None' = None) -> 'Any'
MergeFallback(*args, **kwargs)
MergePostProcess(*args, **kwargs)
MergePreProcess(*args, **kwargs)
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, map_on_error: "Literal['raise', 'collect']" = 'raise', 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, merge_model: 'str | 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(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)
FieldTypeRequiredDefault
namestryes
modifier_setModifierSetno<factory:ModifierSet>
modeLiteral['think', 'agent', 'act', 'scripted']no'think'
inputsAnynoNone
outputsAnynoNone
model`strNone`no
prompt`strNone`no
llm_configLlmConfigno<factory:LlmConfig>
tools`list[neograph.tool.Toollangchain_core.tools.base.BaseTool]`no
scripted_fn`strNone`no
raw_fn`neograph.node.RawNodeFnNone`no
fan_out_param`strNone`no
renderer`neograph.renderers.RendererNone`no
context`list[str]None`no
skip_when`neograph.node.SkipPredicateNone`no
skip_value`neograph.node.SkipValueFactoryNone`no
gate_tools_when`collections.abc.CallablestrNone`
oracle_gen_type`type[pydantic.main.BaseModel]None`no
Operator(when: str)
FieldTypeRequiredDefault
whenstryes
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)
FieldTypeRequiredDefault
nintno3
models`list[str]None`no
merge_prompt`strNone`no
merge_modelstrno'reason'
merge_fn`strNone`no
merge_pre_process`neograph.modifiers.MergePreProcessNone`no
merge_post_process`neograph.modifiers.MergePostProcessNone`no
merge_fallback`neograph.modifiers.MergeFallbackNone`no
ProducingCall(tool_name: str, args: dict[str, Any] = <factory:dict>, producer_idempotent: bool = False)
FieldTypeRequiredDefault
tool_namestryes
argsdict[str, Any]no<factory:dict>
producer_idempotentboolnoFalse
PromptCompiler(*args, **kwargs)
PromptInput(*args, **kwargs)
RawNodeFn(*args, **kwargs)
register_type(name: 'str', cls: 'type[BaseModel]') -> 'None'
render_input(input_data: 'Any', *, renderer: 'Renderer | None') -> 'Any'
render_inputs(input_data: 'Any', *, renderer: 'Renderer | None' = None) -> 'dict[str, Any]'
render_prompt(node: 'Any', input_data: 'Any', *, config: 'RunnableConfig | None' = None, runtime: 'LlmRuntime | None' = None) -> 'str'
Renderer(*args, **kwargs)
resolve_image(path_or_b64: 'str') -> 'str'
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(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)
FieldTypeRequiredDefault
uristryes
kindstryes
serverstryes
producing_callProducingCallyes
mime`strNone`no
size`intNone`no
fetched_at`strNone`no
ttl_ms`intNone`no
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(*args, **kwargs)
SkipValueFactory(*args, **kwargs)
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(template: 'str', vars: 'dict[str, Any]', *, strict: 'bool' = True, syntax: 'SyntaxSpec' = 'brace') -> 'str'
tool(fn: 'Callable | None' = None, *, name: 'str | None' = None, budget: 'int' = 0, config: 'dict[str, Any] | None' = None) -> 'Any'
Tool(name: str, budget: int = 0, config: dict[str, Any] = <factory:dict>, bound_args: dict[str, str] | None = None, idempotent: bool = False)
FieldTypeRequiredDefault
namestryes
budgetintno0
configdict[str, Any]no<factory:dict>
bound_args`dict[str, str]None`no
idempotentboolnoFalse
ToolInteraction(tool_name: str, args: dict[str, Any] = <factory:dict>, result: str = '', typed_result: Any = None, duration_ms: int = 0)
FieldTypeRequiredDefault
tool_namestryes
argsdict[str, Any]no<factory:dict>
resultstrno''
typed_resultAnynoNone
duration_msintno0
type_display_name(t: 'TypeSpecStatic') -> 'str'
verify_compiled(graph: 'Any') -> 'list[VerifyIssue]'
VerifyIssue(node_name: 'str', kind: 'str', message: 'str') -> None
FieldTypeRequiredDefault
node_namestryes
kindstryes
messagestryes
XmlRenderer(*, include_field_info: "Literal['once', 'always', 'never']" = 'once') -> 'None'

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.

KindSeverityMeaning
act_mode_all_idempotent_toolsWARNmode='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_nodeWARNAn 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.
config_key_unmatchedERRORA key in the config= passed to lint() that matches no DI binding in the construct, so nothing reads it. A key accepted because it is present rather than because a binding names it is how a padded config silences a real unsatisfiable binding: the linter then agrees with a description of the world it was handed, not with the graph.
from_configvariesAnnotated[T, FromConfig] — resolved from config['configurable'], passed directly in config=.
from_config_modelvariesBundled BaseModel via FromConfig — each model field must exist in config.
from_inputvariesAnnotated[T, FromInput] — resolved from config['configurable'], originally from run(input={...}).
from_input_modelvariesBundled BaseModel via FromInput — each model field must exist in config.
from_input_unsatisfiableERRORA FromInput/FromConfig parameter whose value is the Each-fanned item or the Loop carry. No caller can supply it, so the run fails in the DI preflight — and padding a config to silence it makes every branch compute from the padded value. Bind it as a port parameter instead.
llm_kwargs_missingWARNLLM-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_unsafeWARN/ERRORLoop 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_unregisteredERRORLoop when is a string that is not registered in the condition registry.
output_field_unconsumedWARNA field of a node’s typed output that nothing reads: no downstream node takes the model, no template reads it by name, and it is not the graph’s terminal output. It costs tokens on every call and cannot affect the answer.
resource_hydration_kind_unmatchedERRORA 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_input_unreferencedWARNA bound input, DI parameter, or context field that the node’s own template never references. The value reaches the node and the model never sees it. Demand is read from the template text, so a prompt_compiler that composes the message may consume the name without naming it.
template_placeholder_known_vars_onlyWARNPlaceholder only resolvable via known_template_vars, not from actual @node parameter names. Advisory: verify the consumer bridge supplies it at runtime.
template_placeholder_unresolvableERRORPrompt placeholder not found in predicted input keys or known extras.
template_var_requires_async_driverWARNA 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_driverWARNAn 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().

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). A self-loop node’s output field is itself an append-list of every iteration, so result[node] is the full iteration history (last element is the final value).

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(...).

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})

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.


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
├── PromptInputError — a prompt_compiler read a field off already-rendered text
├── 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, PromptInputError, NonIdempotentReplayError, ResourceExpiredError.


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