LangGraph's State, Node, and Edge Model — A Practical Walkthrough of Flow Control

Core Concepts

A lower-level orchestration framework than Langchain; used to implement complex workflows and stateful Agent execution, providing runtime capabilities such as streaming, persistence, and human-in-the-loop.

create_agent is built on top of langgraph under the hood.

The graph built from nodes and edges provides more flexible flow orchestration capabilities. The core of LangGraph is State, Node, Edge

• State: A shared data structure during execution, a snapshot at a given moment, the core mechanism for passing data between nodes.
• Node: A concrete execution unit in LangGraph, typically a function. A node receives State as a parameter, executes its logic, and returns a value as the input for the next Node.
• Edge: Used to orchestrate Nodes, determining the execution order and flow between them. Supports both fixed flows and conditional routing.

START -> node_1 -> node_2 -> END

from typing import TypedDict
from langgraph.graph.state import END, START, StateGraph

# 1. Define schema
class StateSchema(TypedDict):
    status: str

# 2. Define node
def order_payment(state: StateSchema):
    state["status"] = "已下单"
    return state

def order_delivery(state: StateSchema):
    state["status"] = "已发货"

# 3. Create state graph
graph_builder = StateGraph(state_schema=StateSchema)

# 4. Define graph nodes
graph_builder.add_node("payment", order_payment)
graph_builder.add_node("delivery", order_delivery)

# 5. Define graph edges
graph_builder.add_edge(START, "payment")
graph_builder.add_edge("payment", "delivery")
graph_builder.add_edge("delivery", END)

# 6. Compile
graph = graph_builder.compile()
result = graph.invoke({"status": "待下单"})

# print(result)
# print(graph.get_graph().draw_ascii())

# Supports drawing mermaid
# graph.get_graph().draw_mermaid()
# graph.get_graph().draw_mermaid_png()

# {'status': '已下单'}
# +-----------+
# | __start__ |
# +-----------+
#       *
#       *
#       *
#  +---------+
#  | payment |
#  +---------+
#       *
#       *
#       *
# +----------+
# | delivery |
# +----------+
#       *
#       *
#       *
#  +---------+
#  | __end__ |
#  +---------+

State definition supports @dataclass TypeDict Pydantic, but because of scenarios requiring Annotated injection, TypeDict has become the mainstream choice.

Different approaches produce different errors when accessing values or failing validation in a Node:

• TypedDict -> KeyError
• dataclass -> TypeError
• Pydantic -> ValidationError

Node

The State in a Node is overwritten by default, taking the output of the previous Node as the input for the next Node. The invoke method passes in the initial State.

The reducer in State adjusts the update mode of State. It is a function (reduction function) injected via Annotated. reducer can use built-in functions or custom ones. When a node operates on a specified key, it follows the reduction function's scheme to operate on the data pointed to by that key.

Commonly used reduction functions include:

from langgraph.graph.message import add_messages
from operator import add, ...

operator is a built-in computation library containing conventional mathematical operations, comparison operations, etc. add_message incrementally adds messages to a list.

from typing import Annotated, TypedDict
from langchain.messages import AnyMessage, HumanMessage
from langgraph.graph.message import add_messages
from langgraph.graph.state import END, START, StateGraph

class StateSchema(TypedDict):
    # operator.add follows + operation, concatenates for lists and strings, increments for numbers
    goods_card: Annotated[list[str], add]
    # 1. If using add_messages, supports tuples or Message objects; final output is converted to Message objects
    # 2. When using, different messages can be assigned IDs for deduplication; later nodes overwrite earlier ones
    messages: Annotated[list[AnyMessage], add_messages]
    # Nodes not operated on will not disappear because they didn't return
    default_val: str

def add_goods_1(state: StateSchema):
    state["goods_card"] = ["T恤"]
    # state["messages"] = [("user", "帮我把这个 T恤 添加到购物车")]
    state["messages"] = [HumanMessage("帮我把这个 短裤 添加到购物车", id=1)]
    return state

def add_goods_2(state: StateSchema):
    # state["messages"] = [("user", "帮我把这个 短裤 添加到购物车")]
    return {
        "goods_card": ["短裤"],
        "messages": [HumanMessage("帮我把这个 短裤 添加到购物车", id=1)],
    }

graph_builder = StateGraph(state_schema=StateSchema)
graph_builder.add_node("goods_1", add_goods_1)
graph_builder.add_node("goods_2", add_goods_2)

graph_builder.add_edge(START, "goods_1")
graph_builder.add_edge("goods_1", "goods_2")
graph_builder.add_edge("goods_2", END)

result = graph_builder.compile()

# {
#   'goods_card': ['T恤', '短裤'],
#   'messages': [HumanMessage(content='帮我把这个 短裤 添加到购物车', additional_kwargs={}, response_metadata={}, id='1')],
#   'default_val': 'default'
# }
# print(result.invoke({"goods_card": [], "default_val": "default", "messages": []}))

The content injected by Annotated is also obtained and executed internally by the framework through this scheme.

# typing.Annotated[str, <built-in function add>]
print(StateSchema.__annotations__["goods_card"])
# (<built-in function add>,)
print(StateSchema.__annotations__["goods_card"].__metadata__)
# Call the add method
print(StateSchema.__annotations__["goods_card"].__metadata__[0](1, 2))

Overwrite ignores reducer logic

Using Overwrite, you can specify that data in a node does not go through the reducer scheme. From the node that is Overwrited onwards, computation continues.

from operator import add
from typing import Annotated, TypedDict
from langgraph.graph.state import END, START, StateGraph
from langgraph.types import Overwrite

class StateSchema(TypedDict):
    counter: Annotated[int, add]

def add_one(state: StateSchema):
    return {"counter": 1}

def add_many(state: StateSchema):
    # When execution reaches this node, directly overwrite with 10; all previous calculations are discarded
    # If there are subsequent nodes, computation continues from the result returned by this node
    return {"counter": Overwrite(10)}

graph_builder = StateGraph(state_schema=StateSchema)
for node in [add_one, add_many]:
    graph_builder.add_node(node.__name__, node)

graph_builder.add_edge(START, "add_one")
graph_builder.add_edge("add_one", "add_many")
graph_builder.add_edge("add_many", END)

graph = graph_builder.compile()

# {'counter': 10}
print(graph.invoke({"counter": 1}))

Built-in State

Initially, you would define this type to store model messages, along with tuple-to-object conversion functionality. In fact, this is already built-in and can be used directly or extended.

# class StateSchema(TypeDict):
#     messages: Annotated[list[AnyMessage], add_messages]

# MessagesState is the StateSchema type
from langgraph.graph.message import MessagesState
from langgraph.graph.state import END, START, StateGraph
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
load_dotenv()

model = init_chat_model(
    model="openai:kimi-k2.6",
    base_url=os.environ.get("OPENAI_BASE_URL"),
    api_key=os.environ.get("OPENAI_API_KEY"),
    extra_body={"thinking": {"type": "disabled"}},
)

class StateSchema(MessagesState):
    username: str
    plain_output: str

def prompt_node(state: StateSchema):
    return {"messages": [("user", f"你好，我是{state['username']}")]}

def model_node(state: StateSchema):
    result = model.invoke(state["messages"])
    return {"plain_output": result.content, "messages": [result]}

graph_builder = StateGraph(state_schema=StateSchema)

graph_builder.add_node("prompt", prompt_node)
graph_builder.add_node("model", model_node)

graph_builder.add_edge(START, "prompt")
graph_builder.add_edge("prompt", "model")
graph_builder.add_edge("model", END)

graph = graph_builder.compile()

# {
#   'messages': [
#       HumanMessage(content='你好，我是海绵宝宝', additional_kwargs={}, response_metadata={}, id='91c09c1b-3cb7-44d3-a11c-fa707c6d34ac'),
#       AIMessage(content='你好呀，海绵宝宝！🍍\n\n准备好去抓水母了吗', additional_kwargs={'refusal': None}, ....],
#   'username': '海绵宝宝',
#   'plain_output': '你好，海绵宝宝！🍍\n\n很高兴见到你！... 😄'
#  }
# print(graph.invoke({"username": "海绵宝宝"}))

Additionally, AgentState is actually used for create_agent and is generally not used in LangGraph scenarios. The function signature is as follows:

from langchain.agents.middleware.types import AgentState, InputAgentState, OutputAgentState

class AgentState(TypedDict, Generic[ResponseT]):
    messages: Required[Annotated[list[AnyMessage], add_messages]]
    jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]]
    structured_response: NotRequired[Annotated[ResponseT, OmitFromInput]]

class InputAgentState(TypedDict):
    messages: Required[Annotated[list[AnyMessage | dict[str, Any]], add_messages]]

class OutputAgentState(TypedDict, Generic[ResponseT]):
    messages: Required[Annotated[list[AnyMessage], add_messages]]
    structured_response: NotRequired[ResponseT]

StateGraph Parameters: Input/Output Constraints

When input_schema and output_schema of StateGraph are not passed, the default value is state_schema.

• output_schema is the final output format; no matter how many fields there are, they will be trimmed to fit this type.
• input_schema is the data received by the start node; it will be constrained according to this type during graph.invoke().

Both types are generally subsets of state_schema.

Additionally, during node execution, an intermediate State is allowed. This type is any type other than the main State, commonly used for handling temporary data between nodes. It effectively avoids state interference from reducer.

from langgraph.graph.message import MessagesState
from typing import TypedDict
from langgraph.graph.state import END, START, StateGraph

class StateSchema(MessagesState):
    user_question: str
    answer: str

class InputState(TypedDict):
    user_question: str

class OutputState(TypedDict):
    answer: str

# Intermediate state type
class TempState(OutputState):
    raw_docs: list[dict[str, str | int]]

def retrive_data(state: InputState):
    # Simulate querying a knowledge base
    question = state["user_question"]
    raw_docs = [{"content": "根据该用户购物习惯，喜欢买 T恤", "score": 90}]

    return {
        # Intermediate state data fields
        "raw_docs": raw_docs,
        "messages": [question, ToolMessage("知识库查询完毕", tool_call_id="1")],
    }

def answer(state: TempState):
    return {"answer": state["raw_docs"][0].get("content")}

graph_builder = StateGraph(
    state_schema=StateSchema,
    input_schema=InputState,
    output_schema=OutputState,  # Without this restriction, all fields in StateSchema would be returned
)

graph_builder.add_node("retrive", retrive_data)
graph_builder.add_node("answer", answer)

graph_builder.add_edge(START, "retrive")
graph_builder.add_edge("retrive", "answer")
graph_builder.add_edge("answer", END)

graph = graph_builder.compile()

print(graph.invoke({"user_question": HumanMessage("海绵宝宝喜欢买什么")}))

StateGraph Return Values

Nodes, edges, compilation, etc., are all methods provided by return values, including the previously used add_node | add_edge | compile.

graph_builder = StateGraph(state_schema=StateSchema)

# Common parameters when adding a Node: node is the function name string, action is the function itself
graph_builder.add_node(node='func_name', action=func)

# Adding an Edge specifies the node names added in add_node for orchestration
# START and END serve as the start and end nodes
graph_builder.add_edge(start_key=START, end_key="func_name")
graph_builder.add_edge(start_key="func_name", end_key=END)

# compile is responsible for compilation
graph = graph_builder.compile()

Currently, the methods on the object compiled by compile only involve invoke and a series of drawing methods starting with draw_.

graph_builder's add_sequence method

Receives a list of Nodes and directly draws them as linear edges, without needing additional add_node calls; it handles them internally.

graph_builder.add_edge(START, "question")
# Internally, this will add_node and then add_edge for these two methods
graph_builder.add_sequence([question, answer])
graph_builder.add_edge("answer", END)

graph_builder setting START, END

Use set_entry_point | set_finish_point to set the start and end edges. Under the hood, this is essentially add_edge pre-filled with START | END identifiers.

graph_builder.set_entry_point("question")
graph_builder.add_sequence([question, answer])
graph_builder.set_finish_point("answer")

In the execution flow of a linear graph, END is not mandatory. If there is nothing to execute after the last node, it will also terminate; it does not affect the graph flow and can be treated as an end marker.

graph_builder.set_entry_point("question")
graph_builder.add_sequence([question, answer])

# Without set_finish_point or explicitly declaring END to answer, it will also end
# graph_builder.set_finish_point("answer")
graph_builder.add_edge(start_key="question", end_key="answer")

Edge Execution Branches

Expands from the previous linear structure to a graph structure.

START → NODE → END

    START
  ↙       ↘
NODE      NODE  Each NODE may further branch
  ↘       ↙
     END

According to the official documentation, edges are divided into four types:

• Normal Edges: Ordinary edges, i.e., the linear structure add_edge.
• Conditional Edges: Edges built with conditions via add_conditional_edges.
• Entry Point: The entry point, determined by graph.add_edge(START, "node_a").
• Conditional Entry Point: Similarly, the conditional entry point is determined by graph.add_conditional_edges(START, routing_function).

Parallel Branches

• Parallelism at the scheduling level, not program-level; nodes have no dependencies on each other, but the actual execution order still follows the code writing order.
• After execution completes, the unified merged State — if nodes operate on the same State field, reducer must be used appropriately to avoid InvalidUpdateError.

class StateSchema(TypedDict):
    desc: str
    reason: str
    suggest: str
    # Testing InvalidUpdateError: not specifying a reducer will cause an error
    count: Annotated[int, sub]

def node_1(state: StateSchema):
    return {"reason": f"根据你的说法“{state['desc']}”看出来你应该是感冒了", "count": 2}

def node_2(state: StateSchema):
    state["suggest"] = "建议多睡觉"
    state["count"] = 3
    return state

builder = StateGraph(state_schema=StateSchema)
for node in [node_1, node_2, node_3, node_4]:
    builder.add_node(node.__name__, node)

# The core lies in multiple edges
# From start -> node_1 -> end
# From start -> node_2 -> end
builder.add_edge(START, "node_1")
builder.add_edge(START, "node_2")

builder.add_edge("node_1", END)
builder.add_edge("node_2", END)

graph = builder.compile()

# {'desc': '很困，四肢无力', 'reason': '根据你的说法“很困，四肢无力”看出来你应该是感冒了', 'suggest': '建议多睡觉'}
#         +-----------+
#         | __start__ |
#         +-----------+
#           *         *
#         **           **
#        *               *
# +--------+          +--------+
# | node_1 |          | node_2 |
# +--------+          +--------+
#           *         *
#            **     **
#              *   *
#           +---------+
#           | __end__ |
#           +---------+
# print(graph.invoke({"desc": "很困，四肢无力"}))
# print(graph.get_graph().draw_ascii())

Conditional Branches

Using add_conditional_edges, starting from an upstream node, a routing function selects which downstream nodes to jump to. It receives three parameters:

• source: The string name of the starting node.
• path: The routing rule, generally a function, supports returning a sequence or a string.
• path_map: The mapping between the routing function's return value and the actual node names. Can be a dictionary or a list; must be actual node names.

Using the same Nodes from the "Parallel Execution" example above:

def router(state: StateSchema) -> Literal["node_1", "node_2"]:
    if "不提建议" in state["desc"]:
        return "node_1"
    return "node_2"

# No longer adding two STARTs consecutively; instead, add one and specify where to start next via router
builder.add_conditional_edges(source=START, path=router)

builder.add_edge("node_1", END)
builder.add_edge("node_2", END)

graph = builder.compile()

# {'desc': '很困，四肢无力', 'suggest': '建议多睡觉', 'count': -3}
# {'desc': '很困，四肢无力, 不提建议', 'reason': '根据你的说法“很困，四肢无力, 不提建议”看出来你应该是感冒了', 'count': -2}
#         +-----------+
#         | __start__ |
#         +-----------+
#           .         .
#         ..           ..   Between __start__ and node, * changed to .
#        .               .   This further identifies it as a conditional node
# +--------+          +--------+
# | node_1 |          | node_2 |
# +--------+          +--------+
#           *         *
#            **     **
#              *   *
#           +---------+
#           | __end__ |
#           +---------+
print(graph.invoke({"desc": "很困，四肢无力"}))
print(graph.invoke({"desc": "很困，四肢无力, 不提建议"}))
print(graph.get_graph().draw_ascii())

When you don't want the router to directly return node names, you can use the path_map parameter to return meaningful business names for mapping.

def router(state: StateSchema) -> Literal["node_1", "node_2"]:
    if "不提建议" in state["desc"]:
        return "no_suggest"
    return "yes_suggest"

builder.add_conditional_edges(
    source=START,
    path=router,
    # Map via path_map to keep the router function cleaner
    path_map={"no_suggest": "node_1", "yes_suggest": "node_2"}
)

router supports returning multiple nodes via a list. Similarly, all nodes must be added to edge.

In this scenario, it's best to specify path_map to aid mermaid relationship drawing.

def router(state: StateSchema) -> Literal["node_1", "node_2"]:
    if "不提建议" in state["desc"]:
        return ["node_1", "node_3"]
    return ["node_2", "node_4"]

builder.add_conditional_edges(
    source=START,
    path=router,
    path_map={
        "node_1": "node_1",
        "node_2": "node_2",
        "node_3": "node_3",
        "node_4": "node_4",
    },
    # Dictionary or list
    # path_map=["node_1", "node_2", "node_3", "node_4"],
)

builder.add_edge("node_1", END)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
builder.add_edge("node_4", END)

defer Delayed Execution

This is a parameter of add_node, defaulting to False. If True, the node will execute last, commonly used for logging, auditing, summarization, validation, etc.

class StateSchema(TypedDict):
    status: str
    mount: Annotated[int, sub]
    log: str

def order_payment(state: StateSchema):
    return {"status": "已下单"}

def order_status():
    return {"mount": 20}

def order_failed(state: StateSchema):
    return {"status": "下单失败", "mount": state["mount"]}

def order_log(state: StateSchema):
    return {"log": f"该商品状态 {state['status']}, 花费 {state['mount']}"}

builder = StateGraph(state_schema=StateSchema)

# Add nodes
builder.add_node(node="payment", action=order_payment)
builder.add_node(node="status", action=order_status)
builder.add_node(node="failed", action=order_failed)
# This node will execute after all other nodes have finished
builder.add_node(node="log", action=order_log, defer=True)

def router(state: StateSchema):
    if state["mount"] < 20:
        return ["failed", "log"]
    else:
        return ["payment", "status", "log"]

# Add conditional edges
builder.add_conditional_edges(
    source=START, path=router, path_map=["payment", "status", "failed", "log"]
)
# Add edges, all edges point to END
builder.add_edge("payment", END)
builder.add_edge("status", END)
builder.add_edge("failed", END)
builder.add_edge("log", END)

graph = builder.compile()

print(graph.invoke({"status": "待下单", "mount": 0}))
print(graph.get_graph().draw_mermaid())

Dynamic Parallel Branches: Send

Unlike orchestrated parallelism, Send is used within a router function to simultaneously dispatch multiple tasks, determining which downstream tasks to trigger during node execution.

It's not dynamically registering Nodes, but dynamically deciding which Nodes to run.

Send receives two parameters: the node name and arguments.

from typing import Literal, TypedDict

from langgraph.graph.state import START, StateGraph
from langgraph.types import Send

class StateSchema(TypedDict):
    content: str
    tw: str
    us: str

class TranslateSchema(StateSchema):
    translate: Literal["tw", "us"]

def translate_node(state: TranslateSchema):
    print(state["content"], state["translate"])  # Simulate content translation logic
    return {"tw": "妳好"} if state["translate"] == "tw" else {"us": "hello"}

def router(state: StateSchema):
    return [
        # Parallel calls to translate_node to complete different translation tasks
        Send(
            node="translate_node",
            arg={"translate": lang, "content": state["content"]},
        )
        for lang in ["us", "tw"]
    ]

builder = StateGraph(state_schema=StateSchema)

# Nodes used by Send must also be added before they can be Sent
builder.add_node("translate_node", translate_node)
builder.add_conditional_edges(START, path=router)
graph = builder.compile()

# {'content': '你好', 'tw': '妳好', 'us': 'hello'}
print(graph.invoke({"content": "你好"}))
print(graph.get_graph().draw_mermaid())

Although translate_node is executed n times, it is merged into the same node.

Dynamic Conditional Branches: Command

Similarly, during the execution of a router, it determines which node should be used next. It receives four parameters:

• goto: The target node to jump to after the node completes.
• update: Updates the graph state, equivalent to updating the node's return value.
• resume: Resumes an interrupted execution, usable for human-in-the-loop stages.
• graph: When subgraphs exist, specifies which layer of the graph the jump occurs in, e.g., jumping from a subgraph to a parent graph.

from operator import add
from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

class LottoSchema(TypedDict):
    amount: Annotated[int, add]
    action: str

def zero_amount(state):
    return {"amount": -10}

def ten_amount(state):
    return {"amount": 10}

def twenty_amount(state):
    return {"amount": 20}

def start_node(
    state: LottoSchema,
) -> Command[Literal["zero_amount", "ten_amount", "twenty_amount", END]]:  # type: ignore
    # Declaring the return type helps the graph generate better
    if state["action"] == "喜相逢":
        return Command(goto="zero_amount")
    if state["action"] == "好运十倍":
        return Command(goto="ten_amount")
    if state["action"] == "百发百中":
        return Command(goto="twenty_amount")
    return Command(goto=END)

builder = StateGraph(state_schema=LottoSchema)
for node in [zero_amount, ten_amount, twenty_amount, start_node]:
    # Add nodes
    builder.add_node(node.__name__, node)
    # Add edges
    if node.__name__ != "start_node":
        builder.add_edge(node.__name__, END)
    else:
        builder.add_edge(START, "start_node")

graph = builder.compile()
print(graph.invoke({"action": "喜相逢"}))
print(graph.get_graph().draw_mermaid())

Node Execution Order and Execution Units

In the second config parameter of a Node, langgraph_step can be used to group execution units. This can be understood as an internal batch processing mode. If the same langgraph_step operates on the same State["Field"], a reducer strategy must be configured.

It also allows a deeper understanding of Node execution order and the number of times a Node executes in multi-node parallel scenarios.

# Test code
from typing import TypedDict
from langchain_core.runnables import RunnableConfig # config type
from langgraph.graph import END, START, StateGraph
from loguru import logger

class EmptyState(TypedDict): pass

def node_1(state, config: RunnableConfig):
    logger.info("node_1 executing step {}", config["metadata"]["langgraph_step"])
    return state

def node_2(state, config: RunnableConfig):
    logger.info("node_2 executing step {}", config["metadata"]["langgraph_step"])
    return state

def node_3(state, config: RunnableConfig):
    logger.info("node_3 executing step {}", config["metadata"]["langgraph_step"])
    return state

def node_4(state, config: RunnableConfig):
    logger.info("node_4 executing step {}", config["metadata"]["langgraph_step"])
    return state

builder = StateGraph(state_schema=EmptyState)
for node in [node_1, node_2, node_3, node_4]:
    builder.add_node(node.__name__, node)

Parallel execution: all Nodes are in one execution unit and all execute once. If operating on State["field"] within the same execution unit, you must consider using reducer to manage it.

builder.add_edge(START, "node_1")
builder.add_edge(START, "node_2")
builder.add_edge(START, "node_3")
builder.add_edge(START, "node_4")
builder.add_edge("node_1", END)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
builder.add_edge("node_4", END)

graph = builder.compile()
# 2026-08-27 17:46:32.236 | INFO     | __main__:node_1:371 - node_1 executing step 1
# 2026-08-27 17:46:32.236 | INFO     | __main__:node_2:376 - node_2 executing step 1
# 2026-08-27 17:46:32.236 | INFO     | __main__:node_3:381 - node_3 executing step 1
# 2026-08-27 17:46:32.236 | INFO     | __main__:node_4:386 - node_4 executing step 1
graph.invoke({})  # Trigger logs
print(graph.get_graph().draw_mermaid())

Standard linear structure: each Node is a separate execution unit.

builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", "node_4")
builder.add_edge("node_4", END)

graph = builder.compile()
# 2026-08-27 17:51:58.484 | INFO     | __main__:node_1:371 - node_1 executing step 1
# 2026-08-27 17:51:58.484 | INFO     | __main__:node_2:376 - node_2 executing step 2
# 2026-08-27 17:51:58.484 | INFO     | __main__:node_3:381 - node_3 executing step 3
# 2026-08-27 17:51:58.484 | INFO     | __main__:node_4:386 - node_4 executing step 4
graph.invoke({})  # Trigger logs
print(graph.get_graph().draw_mermaid())

A parallel structure with a common non-END end node will cause that non-END end node to execute multiple times.

# Two parallel starts both end with node_4
# Ultimately causing node_4 to run multiple times in different execution units in the logs

builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_3")
builder.add_edge("node_3", "node_4")

builder.add_edge(START, "node_2")
builder.add_edge("node_2", "node_4")

graph = builder.compile()
# 2026-08-27 17:55:22.918 | INFO     | __main__:node_1:371 - node_1 executing step 1
# 2026-08-27 17:55:22.919 | INFO     | __main__:node_2:376 - node_2 executing step 1
# 2026-08-27 17:55:22.919 | INFO     | __main__:node_3:381 - node_3 executing step 2
# 2026-08-27 17:55:22.919 | INFO     | __main__:node_4:386 - node_4 executing step 2
# 2026-08-27 17:55:22.920 | INFO     | __main__:node_4:386 - node_4 executing step 3
graph.invoke({})  # Trigger logs
print(graph.get_graph().draw_mermaid())

Multiple executions of nodes in a parallel structure can be resolved by passing a list to the first parameter of add_edge, and the final graph structure will not be changed.

builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_3")

builder.add_edge(START, "node_2")
# Remove the add_edge for node_3 -> node_4 and node_2 -> node_4
# Directly pass ["node_2", "node3"] to indicate that 4 runs only after 2 and 3 finish
builder.add_edge(["node_2", "node_3"], "node_4")

graph = builder.compile()
# 2026-08-27 18:11:49.822 | INFO     | __main__:node_1:371 - node_1 executing step 1
# 2026-08-27 18:11:49.822 | INFO     | __main__:node_2:376 - node_2 executing step 1
# 2026-08-27 18:11:49.823 | INFO     | __main__:node_3:381 - node_3 executing step 2
# 2026-08-27 18:11:49.823 | INFO     | __main__:node_4:386 - node_4 executing step 3
graph.invoke({})  # Trigger logs
print(graph.get_graph().draw_mermaid())

Edge Loop Structures

An Agent itself is a ReAct reasoning + acting architecture. The model decides whether to call a tool or MCP based on the question, calls it, then thinks again based on the result to decide if another call is needed, thinks again... a loop.

• add_conditional_edges: The loop logic mainly resides in the router function, i.e., during the process of describing edge relationships.
• Command: Dynamically implemented, with logic inside the actual nodes.

add_conditional_edges Loop Structure

Essentially, the router function controls whether the model enters a loop between tools.

import random
from typing import Literal
from langchain.messages import AIMessage, HumanMessage, ToolMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import MessagesState

class StateSchema(MessagesState):
    question: str
    output: str

def model_node(state: StateSchema):
    # model.invoke simulation
    # High probability of returning tool_calls
    tool_calls = (
        [{"id": "123", "name": "tool_name", "args": {"a": "1"}}]
        if random.randint(0, 9) < 8
        else []
    )
    return {"messages": [AIMessage("完成问题的回答", tool_calls=tool_calls)]}

def tool_node(state: StateSchema):
    return {"messages": [ToolMessage("调用工具完成", tool_call_id="123")]}

def output_node(state: StateSchema):
    return {"output": state["messages"][-1].content}

def router(state: StateSchema) -> Literal["tool_node", "output_node"]:
    # If there are tool_calls, go to tool_node
    if state["messages"][-1].tool_calls:
        return "tool_node"
    return "output_node"

builder = StateGraph(state_schema=StateSchema)
for node in [model_node, tool_node, output_node]:
    builder.add_node(node.__name__, node)

builder.add_edge(START, "model_node")
# After model_node ends, decide whether to call tool_node based on whether tool_calls exist in model_node
# When there are no tool_calls, it goes to output_node, i.e., the loop ends
builder.add_conditional_edges("model_node", router)
# Also draw the edge from tool_node to model_node,
# After tool_node finishes, go to model_node
builder.add_edge("tool_node", "model_node")
builder.add_edge("output_node", END)

graph = builder.compile()
print(graph.invoke({"messages": [HumanMessage("你好")]}))
print(graph.get_graph().draw_mermaid())

Command Loop Structure

Thanks to the usage scenarios of Command, the judgment logic can be moved into the actual Node to achieve a loop structure; only the node implementation needs to be improved.

# Note the return value annotation, it affects the final graph drawing
def model_node(state: StateSchema) -> Command[Literal["tool_node", "output_node"]]:
    tool_calls = (
        [{"id": "123", "name": "tool_name", "args": {"a": "1"}}]
        if random.randint(0, 9) < 8
        else []
    )
    # Adjust to Command jump based on random logic
    if tool_calls:
        return Command(
            goto="tool_node",
            update={"messages": [AIMessage("完成问题的回答", tool_calls=tool_calls)]},
        )
    return Command(goto="output_node", update={"messages": [AIMessage("完成问题的回答")]})

def tool_node(state: StateSchema):
    # Tool node unconditionally jumps to model node
    return Command(
        goto="model_node",
        update={"messages": [ToolMessage("调用工具完成", tool_call_id="123")]},
    )

# In the drawing phase, remove router but still need to keep tool_node -> model_node
builder.add_edge(START, "model_node")
# builder.add_conditional_edges("model_node", router)
builder.add_edge("tool_node", "model_node")
builder.add_edge("output_node", END)

Loop Limits

In practice, this is configured beforehand via the upper limit of config["metadata"]["langgraph_step"] in the node, to prevent infinite loops in looping nodes. The type is from langchain_core.runnables import RunnableConfig.

Passed via the config parameter of graph.invoke(), the default value is 25. Exceeding it raises GraphRecursionError.

In State, a remaining_steps parameter can also be received to get a countdown to the exception, which can be used to decide whether to end the node's execution.

The upper limit refers to the step limit for the entire graph execution, not for a specific node.

from typing import Literal, TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.errors import GraphRecursionError # The exception for exceeding
from langgraph.graph import END, StateGraph
from langgraph.managed import RemainingSteps # Countdown steps
from langgraph.types import Command
from loguru import logger

class StateSchema(TypedDict):
    remaining_steps: RemainingSteps

def node_1(state: StateSchema, config: RunnableConfig) -> Command[Literal["node_1"]]:
    # Get cur_step, remaining_step
    cur_step = config["metadata"]["langgraph_step"]
    remaining_step = state["remaining_steps"]

    logger.info("node_1 current step {}, remaining steps {}", cur_step, remaining_step)
    # Self-loop logic: this scheme terminates when there are almost no steps left
    if remaining_step < 2:
        return Command(goto=END)
    return Command(goto="node_1")

builder = StateGraph(state_schema=StateSchema)
builder.add_node("node_1", node_1)
builder.set_entry_point("node_1")
builder.set_finish_point("node_1")
graph = builder.compile()

# Even without internal handling logic, the error can be caught here
try:
    graph.invoke({}, config={"recursion_limit": 10})
except GraphRecursionError as e:
    logger.error("Exceeded loop limit {}", e)

print(graph.get_graph().draw_mermaid())

# Jumping to END prematurely will cause an exception
# 2026-08-28 11:45:04.966 | INFO     | __main__:node_1:18 - node_1 current step 1, remaining steps 9
# 2026-08-28 11:45:04.966 | INFO     | __main__:node_1:18 - node_1 current step 2, remaining steps 8
# 2026-08-28 11:45:04.966 | INFO     | __main__:node_1:18 - node_1 current step 3, remaining steps 7
# 2026-08-28 11:45:04.967 | INFO     | __main__:node_1:18 - node_1 current step 4, remaining steps 6
# 2026-08-28 11:45:04.967 | INFO     | __main__:node_1:18 - node_1 current step 5, remaining steps 5
# 2026-08-28 11:45:04.967 | INFO     | __main__:node_1:18 - node_1 current step 6, remaining steps 4
# 2026-08-28 11:45:04.967 | INFO     | __main__:node_1:18 - node_1 current step 7, remaining steps 3
# 2026-08-28 11:45:04.967 | INFO     | __main__:node_1:18 - node_1 current step 8, remaining steps 2
# 2026-08-28 11:45:04.967 | INFO     | __main__:node_1:18 - node_1 current step 9, remaining steps 1
# 2026-08-28 11:45:04.967 | INFO     | __main__:node_1:18 - node_1 current step 10, remaining steps 0
# 2026-08-28 11:45:04.967 | ERROR    | __main__:<module>:34 - Exceeded loop limit Recursion limit of 10 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
# For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/errors/GRAPH_RECURSION_LIMIT