AgenticGraph

nae.AgenticGraph

AgenticGraph(state: type | None = None, start_node: Node | None = None, end_nodes: Node | set | list | tuple | None = None, name: str | None = None, checkpointer: Any = None, cache: Any = None, validate: bool = True, inputs: list | set | None = None, prompt_cache: bool = False, prompt_cache_key: str | None = None)

Bases: StateGraph

Compile nodes wired with the > DSL into a runnable LangGraph graph.

Hand it the start_node and end_nodes of a graph you built with the > operator and it materializes and compiles a native LangGraph StateGraph. The DSL it understands:

• a > b > c — sequential chain (each node appends to the shared state).
• a > fanout(b, c) > d (or a > [b, c] > d) — parallel fan-out; the join node d is deferred and runs once after both branches finish.
• worker.map("items") > collector — map-reduce: worker runs once per item of state["items"] via LangGraph Send, replies join into collector.
• decision["choice"] > handler — conditional routing off a DecisionNode/ConditionNode.
• subgraph composition — an AgenticGraph may itself be a node in a larger AgenticGraph (graph_0 > graph_1).

Build-time behavior: the schema is inferred (undeclared keys a node reads/writes are auto-registered) so you rarely pass state=, and a forward dataflow validator FAILS the build (GraphValidationError) if a node reads a key nothing produces upstream. Run it with invoke / stream / ainvoke / astream.

Examples:

from nae import AgentNode, AgenticGraph

outline = AgentNode(llm=llm, node_prompt="Outline the answer.")
draft = AgentNode(llm=llm, node_prompt="Write a draft from the outline.")
polish = AgentNode(llm=llm, node_prompt="Polish the draft.")

outline > draft > polish

graph = AgenticGraph(start_node=outline, end_nodes={polish})
graph.invoke("Explain what a monad is.")

Initialize the AgenticGraph.

Parameters:

Name	Type	Description	Default
state	type | None	The state schema (a TypedDict). OPTIONAL — defaults to the standard AgenticState (messages/log/token). Any scalar key a node writes, and any non-optional key a node reads that no node writes (a graph INPUT seeded at invoke), is auto-registered as a PLAIN (last-writer-wins) channel when the schema doesn't already declare it — so you rarely need a hand-written schema. Pass an explicit state= (with reducers) when you want to type/reduce a key precisely.	None
start_node	Node | None	The initial node in the graph	None
end_nodes	Node | set | list | tuple | None	Set of nodes that represent terminal states	None
checkpointer	Any	Optional LangGraph checkpointer (e.g. InMemorySaver). Required for HumanNode interrupt/resume. When set, every invoke must pass a config with a thread_id.	None
cache	Any	Optional LangGraph cache backend. Defaults to an InMemoryCache when any node sets cache_ttl; pass explicitly to override.	None
inputs	list | set | None	Keys you pass at invoke() that are PLAIN (non-reduced) AND written by some node — declare them here so the validator treats them as provided at entry, not produced. (Reduced channels and keys no node writes are already inferred as inputs; you only need this for a plain key you seed at invoke that a node also writes.)	None
validate	bool	When True (default), statically check that every key each node reads is produced upstream/in the schema and every key it writes is declared, FAILING the build (GraphValidationError) on a hard error. Set False to skip (escape hatch). See validate().	True
prompt_cache	bool	Opt-in (default False). When True AND a node's model is OpenAI, every model call passes a prompt_cache_key so OpenAI routes the shared SystemMessage/history prefix to one cache. No effect on non-OpenAI models.	False
prompt_cache_key	str | None	The cache key to use verbatim for the whole run. When omitted, one key is generated per graph build and shared by every node, so all sibling/sequential calls land on the same cache.	None

__call__

__call__(state: AgenticState, config: dict | None = None, durability=None) -> dict

Process the current state through the graph.

Parameters:

Name	Type	Description	Default
state	AgenticState	Initial state, or a Command (e.g. Command(resume=...)) to resume an interrupted run.	required
config	dict | None	Optional LangGraph config, e.g. {"configurable": {"thread_id": "abc"}} (required when a checkpointer is set).	None

Returns:

Type	Description
dict	Updated state after processing through the graph

__gt__

__gt__(other: Node | list | Branch) -> Node | Branch

Wire this graph forward to one child, or fan out to several.

Parameters:

Name	Type	Description	Default
other	Node | list | Branch	A single node (sequential edge), or a list of nodes (parallel fan-out — each becomes a child).	required

Returns:

Type	Description
Node | Branch	The single child (chain building), or a Branch wrapping the
Node | Branch	fan-out heads whose own > wires the fan-in/join.

ainvoke async

ainvoke(input=_UNSET, /, *, message=None, config=None, durability=None, **extra)

Async variant of invoke. Sync nodes run in LangGraph's threadpool.

See invoke for the full input/config contract.

astream async

astream(input=_UNSET, /, *, message=None, config=None, durability=None, **extra)

Async streaming variant of stream.

See invoke for the full input/config contract.

build_graph

build_graph() -> None

Build the graph structure starting from the start node.

compile

compile(checkpointer=None, *, cache=None, store=None, interrupt_before=None, interrupt_after=None, debug=False)

Compile the built graph into a runnable LangGraph CompiledStateGraph.

Thin override of StateGraph.compile that keeps checkpointer as the first positional argument for convenience; all other options (cache, store, interrupt_before, interrupt_after, debug) pass through unchanged. AgenticGraph.__init__ calls this for you, so you rarely invoke it directly — the compiled result is stored on self.compiled_graph and used by invoke / stream / ainvoke / astream.

invoke

invoke(input: Any = _UNSET, /, *, message: str | list | None = None, config: dict | None = None, durability=None, **extra: Any) -> dict

Process an input through the graph.

Parameters:

Name	Type	Description	Default
input	Any	A plain str (wrapped to a HumanMessage on messages), a list of messages (placed on messages), a full state dict (used as-is, back-compat), or a Command (e.g. Command(resume=...)) to resume an interrupted run. For the str/list/dict forms, boilerplate channels are auto-seeded (at minimum log: []).	_UNSET
message	str | list | None	Keyword form of the str/list shorthand, e.g. invoke(message="hi"). Mutually exclusive with a positional input (passing both raises ValueError).	None
config	dict | None	Optional LangGraph config, e.g. {"configurable": {"thread_id": "abc"}} (required when a checkpointer is set).	None
**extra	Any	Additional state keys to seed alongside the shorthand input, e.g. invoke("hi", topic="product").	{}

Returns:

Type	Description
dict	Updated state after processing through the graph

stream

stream(input=_UNSET, /, *, message=None, config=None, durability=None, **extra)

Stream graph execution, yielding updates as nodes complete.

Normalizes input/message like invoke (str/list/dict/Command + **extra state keys), then passes through to the compiled graph's stream. See invoke for the full input/config contract.

summary

summary(file=sys.stdout) -> None

Print a Keras-model.summary()-style table of every node's reads, writes, and available (must) keys (may`mustkeys suffixed?`).

validate

validate(strict: bool = False, *, interrupt=None) -> ValidationReport

Statically check state availability; returns a ValidationReport.

Raises nothing itself (the build path raises on errors). With strict=True the returned report has its warnings promoted to errors.