This article introduces LangGraph's core concepts: State, Node, and Edge, and how to build your first state graph. Includes practical code examples showing how to define nodes, add edges, configure entry points, and compile and execute graphs. Ideal for developers building complex multi-step agent workflows.
LangGraph is the core library in the LangChain ecosystem for building stateful, multi-step agent workflows. Unlike traditional Chains, LangGraph models agent behavior as graphs, where nodes represent computational steps, edges represent control flow, and state is passed between nodes.
pip install langgraphState is the shared data structure of the graph, containing current information of the entire workflow:
from typing import TypedDict
from typing_extensions import Annotated
from operator import add
class AgentState(TypedDict):
# Basic field: overwrite update
messages: list[str]
# Field with Reducer: accumulate update
history: Annotated[list[str], add]
Nodes are Python functions that execute specific logic:
def process_node(state: AgentState) -> AgentState:
"""Process node: receives current state, returns updated state"""
current_msg = state["messages"][-1]
return {
"messages": [f"Processed: {current_msg}"]
}
Edges determine which node to execute next:
# Normal edge: from node_a directly to node_b
builder.add_edge("node_a", "node_b")
# Conditional edge: decide routing based on state
def routing_func(state: AgentState) -> str:
if len(state["messages"]) > 5:
return "end_node"
return "continue_node"
builder.add_conditional_edges("router", routing_func)
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 1. Define State
class ChatState(TypedDict):
messages: list[str]
turn_count: int
# 2. Create graph builder
builder = StateGraph(ChatState)
# 3. Define nodes
def should_continue(state: ChatState) -> str:
"""Determine whether to continue the dialogue"""
if state["turn_count"] >= 3:
return "end"
return "continue"
def agent_node(state: ChatState) -> ChatState:
"""Agent core logic"""
last_msg = state["messages"][-1]
response = f"I received: {last_msg} (turn {state['turn_count'] + 1})"
return {
"messages": state["messages"] + [response],
"turn_count": state["turn_count"] + 1
}
# 4. Add nodes and edges
builder.add_node("agent", agent_node)
builder.add_edge(START, "agent")
builder.add_conditional_edges(
"agent",
should_continue,
{"continue": "agent", "end": END}
)
# 5. Compile graph
graph = builder.compile()
# 6. Execute
result = graph.invoke({
"messages": ["Hello"],
"turn_count": 0
})
print(result["messages"])
# ['Hello', 'I received: Hello (turn 1)', 'I received: I received: Hello (turn 1) (turn 2)', ...]
| Feature | LangChain Chain | LangGraph |
|---|---|---|
| Workflow | Linear/fixed | Graph structure/conditional |
| State Management | No built-in state | Built-in State sharing |
| Loop Support | Not supported | Supports arbitrary loops |
| Debug Difficulty | Lower | Requires understanding graph execution |
Q1: What is the purpose of State Reducers?
Q2: How to handle multi-turn conversation memory?
Q3: What are the requirements for conditional edge return values?
所有示例代码在 Python 3.10+ 环境验证通过
代码示例可正常执行,状态图逻辑正确