Implementation:Langchain ai Langgraph StateGraph Add Node: Difference between revisions
Auto-imported from implementations/Langchain_ai_Langgraph_StateGraph_Add_Node.md |
Sync from local file |
||
| Line 235: | Line 235: | ||
* [[implements::Langchain_ai_Langgraph_Node_Registration]] | * [[implements::Langchain_ai_Langgraph_Node_Registration]] | ||
* [[Langchain_ai_Langgraph_StateGraph_Init]] | * [[Implementation:Langchain_ai_Langgraph_StateGraph_Init]] | ||
* [[Langchain_ai_Langgraph_StateGraph_Add_Edge]] | * [[Implementation:Langchain_ai_Langgraph_StateGraph_Add_Edge]] | ||
* [[Langchain_ai_Langgraph_StateGraph_Compile]] | * [[Implementation:Langchain_ai_Langgraph_StateGraph_Compile]] | ||
[[Category:Implementations]] | [[Category:Implementations]] | ||
Latest revision as of 10:44, 27 September 2026
| Metadata | Value |
|---|---|
| Type | Implementation (API Doc) |
| Library | langgraph |
| Source File | libs/langgraph/langgraph/graph/state.py
|
| Lines | L569-783 (add_node), L889-934 (add_sequence)
|
| Workflow | Building_a_Stateful_Graph |
Overview
StateGraph.add_node registers a processing function or runnable as a named node in the graph builder. StateGraph.add_sequence is a convenience method that registers multiple nodes and connects them with sequential edges in one call.
Description
add_node accepts either a callable/runnable directly (with the name inferred from __name__ or get_name()) or an explicit string name paired with an action. The method:
- Validates the name is unique, not reserved (
START/END), and does not contain forbidden separator characters. - Coerces the action into a
Runnableviacoerce_to_runnable(). - Inspects the action's type hints to infer input schema and detect
Commandreturn types (for graph visualization). - Stores a
StateNodeSpeccontaining the runnable, metadata, input schema, retry policy, cache policy, destinations, and defer flag. - If a custom
input_schemais provided or inferred, registers it with_add_schema().
add_sequence iterates a list of nodes, calling add_node for each and add_edge between consecutive pairs. It validates that all names are unique.
Both methods return Self to support method chaining.
Usage
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
class State(TypedDict):
x: int
def step_one(state: State) -> dict:
return {"x": state["x"] + 1}
def step_two(state: State) -> dict:
return {"x": state["x"] * 2}
builder = StateGraph(State)
builder.add_node(step_one) # inferred name: "step_one"
builder.add_node("custom_name", step_two) # explicit name
builder.add_edge(START, "step_one")
builder.add_edge("step_one", "custom_name")
Code Reference
Source Location
| Item | Path | Lines |
|---|---|---|
add_node (implementation) |
libs/langgraph/langgraph/graph/state.py |
L569-783 |
add_node (overloads) |
libs/langgraph/langgraph/graph/state.py |
L289-567 |
add_sequence |
libs/langgraph/langgraph/graph/state.py |
L889-934 |
Signature
def add_node(
self,
node: str | StateNode[NodeInputT, ContextT],
action: StateNode[NodeInputT, ContextT] | None = None,
*,
defer: bool = False,
metadata: dict[str, Any] | None = None,
input_schema: type[NodeInputT] | None = None,
retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
cache_policy: CachePolicy | None = None,
destinations: dict[str, str] | tuple[str, ...] | None = None,
) -> Self:
def add_sequence(
self,
nodes: Sequence[
StateNode[NodeInputT, ContextT]
| tuple[str, StateNode[NodeInputT, ContextT]]
],
) -> Self:
Import
from langgraph.graph import StateGraph
I/O Contract
add_node Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
node |
StateNode | required | The function/runnable to register, or a string name (when action is also provided).
|
action |
None | None |
The function/runnable when node is a string name.
|
defer |
bool |
False |
If True, defer execution until the run is about to end.
|
metadata |
None | None |
Arbitrary metadata to attach to the node. |
input_schema |
None | None |
Custom input schema for this node. Defaults to the graph's state schema. |
retry_policy |
Sequence[RetryPolicy] | None | None |
Retry configuration. If a sequence, the first matching policy is applied. |
cache_policy |
None | None |
Caching configuration for this node. |
destinations |
tuple[str, ...] | None | None |
Rendering hint for Command-returning nodes. Does not affect execution. |
Returns: Self -- The StateGraph instance, for method chaining.
Raises:
ValueError-- If the node name already exists, is reserved (START/END), or contains forbidden characters.
add_sequence Parameters
| Parameter | Type | Description |
|---|---|---|
nodes |
tuple[str, StateNode]] | Ordered sequence of nodes or (name, node) tuples.
|
Returns: Self -- The StateGraph instance.
Raises:
ValueError-- If the sequence is empty or contains duplicate names.
Usage Examples
Inferred Node Name
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
class State(TypedDict):
x: int
def my_node(state: State) -> dict:
return {"x": state["x"] + 1}
builder = StateGraph(State)
builder.add_node(my_node) # name inferred as "my_node"
builder.add_edge(START, "my_node")
graph = builder.compile()
graph.invoke({"x": 1})
# {'x': 2}
Explicit Node Name
builder = StateGraph(State)
builder.add_node("my_fair_node", my_node)
builder.add_edge(START, "my_fair_node")
graph = builder.compile()
graph.invoke({"x": 1})
# {'x': 2}
Custom Input Schema
class NodeInput(TypedDict):
x: int
def my_node(state: NodeInput) -> dict:
return {"x": state["x"] + 1}
builder = StateGraph(State)
builder.add_node("my_node", my_node, input_schema=NodeInput)
builder.add_edge(START, "my_node")
graph = builder.compile()
graph.invoke({"x": 1})
# {'x': 2}
Using add_sequence
def step_a(state: State) -> dict:
return {"x": state["x"] + 1}
def step_b(state: State) -> dict:
return {"x": state["x"] * 2}
def step_c(state: State) -> dict:
return {"x": state["x"] - 3}
builder = StateGraph(State)
builder.add_sequence([step_a, step_b, step_c])
builder.add_edge(START, "step_a")
builder.set_finish_point("step_c")
graph = builder.compile()
graph.invoke({"x": 5})
# step_a: 5+1=6, step_b: 6*2=12, step_c: 12-3=9 -> {'x': 9}
Deferred Node
def cleanup(state: State) -> dict:
return {"x": 0}
builder = StateGraph(State)
builder.add_node("main", my_node)
builder.add_node("cleanup", cleanup, defer=True)
builder.add_edge(START, "main")
builder.add_edge("main", "cleanup")
graph = builder.compile()