LangGraph开篇-LangGraph 核心元素简介(官网文档解读)
LangGraph 的诞生
在 AI 领域,Transformer 架构的出现,为大模型带来了革命性突破,显著提升其能力的同时,也极大拓展了应用边界。随着大模型技术的持续迭代,以大模型应用开发为核心的框架纷纷涌现。其中,LangChain 凭借丰富便捷的工具与接口脱颖而出,自 2022 年 10 月由 Harrison Chase 开源后,迅速成为开发者构建大模型落地应用的首选。
基于 LangChain,开发者能够高效搭建多样化的大模型应用,包括大模型对话产品、基于 RAG 的知识增强应用,以及基于工作流的 AI Agent 应用等。然而,随着应用场景日趋复杂,LangChain 链式工作流的局限性逐渐显现。面对涉及复杂决策、动态循环和多路径选择的任务时,这种线性结构难以充分激发大模型的思维与判断能力,无法满足复杂 AI Agent 的构建需求。
为突破这一困境,Langchain 团队于 2024 年推出全新框架 LangGraph。当时,大模型应用向更深层次发展,开发者对具备复杂推理和灵活决策能力的 AI Agent 需求愈发迫切。传统工作流框架在处理复杂任务时的弊端日益突出,在此背景下,LangGraph 以创新的图结构重新定义 AI Agent 构建方式,开启了 AI Agent 开发的全新阶段,为复杂 AI 应用场景提供了更优解决方案。
LangChain 到LangGraph 的技术演进
LangChain 基于工作流的构建模式,采用链式调用机制,前一个环节产生的数据依次传递给后续函数处理。这种线性结构在处理简单任务时高效便捷,但面对需要复杂决策、动态循环和多路径选择的任务时,其灵活性和扩展性便捉襟见肘。例如,在需要模拟人类多轮思考、反复验证答案的场景中,工作流结构难以灵活调整执行路径,无法充分展现大模型的推理能力和判断逻辑。
LangGraph 的出现正是为了解决上述问题。它采用 Graph(图)结构对 AI Agent 的工作流进行建模,打破了线性流程的束缚。图结构如同带有循环节点的智能流程图,能够依据实时状态、节点功能和边的连接关系,动态灵活地选择下一步执行动作,为复杂 AI Agent 的构建提供了更强大、更灵活的解决方案。
LangGraph 核心元素
图1 ReAct 思维框架流程图
以非常流行的 AI Agent ReAct 思维框架为例,其流程图由 START、think、execute、END、state 五个关键元素构成。
-
START 作为流程起始节点,触发整个思维链条运转;
-
END 则是流程结束节点,标志任务完成;
-
think 节点承载大模型的思考过程,通过分析问题、规划策略生成初步方案;
-
execute 节点负责将思考结果付诸实践,调用外部工具或执行具体操作;
-
state 状态对象,如同贯穿流程的 “数字血脉”,封装并传递思考与执行过程中产生的所有数据。
这一流程图与 LangGraph 的核心图结构设计高度契合。在 LangGraph 中,可以将 start 和 end 视为特殊节点,其基础架构同样由节点、边和状态三大核心要素构成。
-
节点(Node):功能执行单元:ReAct 框架中的 think 和 execute 节点,对应 LangGraph 中的普通节点。每个节点都绑定一段 Python 函数代码,例如 think 节点的函数可接收当前问题及相关数据,调用大模型进行推理,输出初步分析结果;execute 节点的函数则依据 think 节点的结果,调用 API、数据库等资源完成具体任务。这些节点通过对 State 状态的读取和修改,实现数据处理与状态转换。
-
边(edge):流程导航路径:在 LangGraph 图结构中,边由 Python 函数定义,承担着流程导向的关键职责。类比 ReAct 框架,边会根据当前 State 状态判断何时从 START 节点进入 think 节点开始思考,何时从 think 节点转入 execute 节点执行操作,以及在 execute 节点完成后,如何依据执行结果决定是继续循环思考,还是抵达 END 节点结束流程。这种动态决策机制,让工作流能够灵活应对不同场景需求。
-
状态(state):数据管理中枢:State 状态对象在 LangGraph 中扮演着数据 “总管” 的角色。它不仅存储着 ReAct 框架中 think 节点产生的推理数据、execute 节点的执行结果,还记录着整个流程的中间状态。无论是待处理的用户问题,还是调用工具返回的信息,都被整合在 State 中,为后续节点提供全面的数据支撑,确保整个 AI Agent 工作流的连贯性与智能性。
LangGraph 实现ReAct 架构的代码示例
from typing import TypedDict, Optional, Listfrom Tools.scripts.generate_global_objects import START
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json# 模型初始化
llm = ChatOpenAI()# 定义状态类型 - LangGraph核心概念:State
class ReActState(TypedDict):"""ReAct Agent的状态定义"""question: str # 用户问题thoughts: List[str] # 思考过程action_contents: Optional[str] # 思考结果action_type: str # 思考类型# 通用工具调用方法
def tool_call(tool_call_content: str) -> str:"""通用工具调用方法 - 代表调用各种工具的逻辑Args:tool_call_content: 工具调用的具体内容Returns:工具执行结果"""# 模拟工具调用处理return f"工具处理'{tool_call_content}':返回处理结果"# Node 1: 思考决策节点
def think_node(state: ReActState) -> ReActState:"""思考节点:AI决策下一步操作"""prompt = f"""问题: {state['question']}历史思考: {state['thoughts'][-1] if state['thoughts'] else '开始思考'}请用JSON格式决策下一步:{{"thought": "你的思考过程","action_type": "execute(执行工具)或answer(给出最终答案)","action_content": "具体内容:如果是execute,填写工具调用参数;如果是answer,填写最终答案"}}"""response = llm.invoke(prompt)try:decision = json.loads(response.content.strip())# 更新状态new_state = state.copy()new_state["thoughts"].append(decision.get("thought", ""))# 处理决策action_type = decision.get("action_type", "")action_content = decision.get("action_content", "")# 更新状态字段new_state["action_contents"] = action_contentnew_state["action_type"] = action_typereturn new_stateexcept:new_state = state.copy()new_state["thoughts"].append("思考中...")new_state["action_type"] = "think"return new_state# Node 2: 执行节点 -工具调用
def execute_node(state: ReActState) -> ReActState:"""执行节点:执行工具调用,完成后回到think节点"""new_state = state.copy()# 调用工具方法if state["action_contents"]:result = tool_call(state["action_contents"])new_state["thoughts"].append(f"工具调用结果: {result}")else:new_state["thoughts"].append("无工具调用内容")return new_state# Edge: 路由决策
def route_decision(state: ReActState) -> str:"""决定下一个节点"""# 如果决策类型是answer,结束流程if state["action_type"] == "answer":return END# 如果决策类型是execute,执行工具if state["action_type"] == "execute":return "execute"# 默认继续思考return "think"# 创建LangGraph - 核心框架展示
def create_react_graph() -> StateGraph:"""构建ReAct推理图"""# 1. 初始化StateGraphgraph = StateGraph(ReActState)# 2. 添加节点 (Nodes)graph.add_node("think", think_node) # 决策graph.add_node("execute", execute_node) # 执行# 3. 添加边 (Edges)graph.add_edge(START, "think") # 开始 → 思考graph.add_conditional_edges("think", route_decision) # 思考 → 条件路由graph.add_edge("execute", "think") # 执行 → 思考return graph# 使用示例
if __name__ == "__main__":# 创建图react_graph = create_react_graph()# 输入问题inputStr = "Python是什么时候发明的?"# 初始化状态initial_state: ReActState = {"question": inputStr,"thoughts": [],"action_contents": None,"action_type": ""}# 运行图result = react_graph.invoke(initial_state)# 输出结果print("=" * 50)print("LangGraph ReAct Agent - 简化状态管理")print("=" * 50)print(f"问题: {result['question']}")print(f"最终决策类型: {result['action_type']}")print(f"最终内容: {result['action_contents']}")print(f"\n流程: think → execute → think → ... → END")print("\n思考过程:")for i, thought in enumerate(result['thoughts'], 1):print(f" {i}. {thought}")# 根据action_type显示结果if result['action_type'] == 'answer':print(f"\n✅ 最终答案: {result['action_contents']}")else:print(f"\n🔧 最后操作: {result['action_type']} - {result['action_contents']}")
在这里我使用LangGraph 实现了一个简单的ReAct 思维框架。该框架通过 ReActState 对象统一管理用户问题、思考过程等状态数据;框架内的 think 节点是核心决策节点,负责调用大模型进行决策,生成 execute 调用参数或直接输出 answer 结果。
execute节点
执行后返回think节点形成闭环;think 节点则利用route_decision函数,
按状态类型动态路由节点跳转,最终通过图结构将各组件串联为「思考 - 执行 - 再思考」的可解释推理流程,与 图1中的ReAct 流程图逻辑完全一致。
接下来,我们再根据官网的描述详细介绍下LangGraph 基础框架的三大核心元素。
State(状态)
表示应用程序当前快照的共享数据结构。它可以是任何 Python 类型,但通常是 TypedDict
或 Pydantic BaseModel
。
TypedDict
TypedDict
是 Python 中用于类型提示的工具,在 typing
模块(Python 3.8 及以上)或 typing_extensions
模块中可用。它允许为字典的键和值指定类型,使得代码更具可读性和可维护性,同时也能帮助静态类型检查工具(如 mypy
)进行类型检查。不过,TypedDict
本身只是一个类型定义,不具备运行时的数据验证功能。
from typing_extensions import TypedDictclass UserInfo(TypedDict):name: strage: intemail: struser: UserInfo = {"name": "Alice","age": 30,"email": "alice@example.com"
}
BaseModel
Pydantic 是一个用于数据验证和序列化的 Python 库,BaseModel
是 Pydantic 的核心类之一。通过继承 BaseModel
类,可以定义数据模型,这些模型可以自动进行数据验证、类型转换和序列化操作。在运行时,Pydantic 会检查输入的数据是否符合模型定义,如果不符合则会抛出异常。
-
支持为字段设置默认值,并且可以使用
Field
函数进行复杂的数据验证,如设置字段的最大长度、最小值等。 -
提供了内置的序列化和反序列化方法,如
model_dump_json()
和model_validate_json()
,方便将模型对象转换为 JSON 数据,以及从 JSON 数据创建模型对象。
from pydantic import BaseModel, Field, field_validator
import datetimeclass User(BaseModel):# 用户名,长度在 3 到 20 个字符之间username: str = Field(min_length=3, max_length=20, description="用户名长度必须在 3 到 20 个字符之间")# 年龄,必须为正整数且在 1 到 120 岁之间age: int = Field(gt=0, le=120, description="年龄必须为正整数且在 1 到 120 岁之间")# 注册时间,默认为当前时间registration_time: datetime.datetime = Field(default_factory=datetime.datetime.now, description="注册时间,默认为当前时间")# 用户状态,只能是 "active"、"inactive"、"banned" 中的一个status: str@field_validator("status")@classmethoddef validate_status(cls, value):valid_statuses = ["active", "inactive", "banned"]if value not in valid_statuses:raise ValueError(f"用户状态必须是 {valid_statuses} 中的一个")return value# 示例数据
user_data = {"username": "john_doe","age": 25,"status": "active"
}try:# 创建 User 模型实例,Pydantic 会自动验证输入数据user = User(**user_data)print("验证通过,用户信息如下:")print(user)# 序列化模型实例为 JSON 字符串json_data = user.model_dump_json(indent=2)print("\n序列化后的 JSON 数据:")print(json_data)# 从 JSON 数据反序列化回模型实例deserialized_user = User.model_validate_json(json_data)print("\n反序列化后的用户信息:")print(deserialized_user)except ValueError as e:print(f"验证失败:{e}")
综上所述,TypedDict
适用于简单、轻量级的场景;而 Pydantic 的 BaseModel
更适合复杂、需要严格数据验证的场景。
多模式状态管理
默认情况下,LangGraph 中所有节点共享单一状态模式(即读写相同的状态字段)。但在复杂场景下,我们需要更精细的控制,比如:
-
内部节点通信:某些信息只需在图内部节点间传递,无需暴露给用户(如图的输入 / 输出)。
-
输入 / 输出约束:图的输入可能只需要部分字段,输出也可能只返回特定结果(而非全部内部状态)。
LangGraph 支持通过定义不同的状态模式(如 TypedDict
或 Pydantic BaseModel
)来分离内部状态和公开输入 / 输出。LangGraph 的官网文档给出了多模式状态管理的代码示例:
class InputState(TypedDict):user_input: strclass OutputState(TypedDict):graph_output: strclass OverallState(TypedDict):foo: struser_input: strgraph_output: strclass PrivateState(TypedDict):bar: strdef node_1(state: InputState) -> OverallState:# Write to OverallStatereturn {"foo": state["user_input"] + " name"}def node_2(state: OverallState) -> PrivateState:# Read from OverallState, write to PrivateStatereturn {"bar": state["foo"] + " is"}def node_3(state: PrivateState) -> OutputState:# Read from PrivateState, write to OutputStatereturn {"graph_output": state["bar"] + " Lance"}builder = StateGraph(OverallState,input=InputState,output=OutputState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
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", END)graph = builder.compile()
graph.invoke({"user_input":"My"})
{'graph_output': 'My name is Lance'}
-
自定义状态:
-
内部状态(OverallState):定义图内部所有节点共享的完整状态字段(即节点间可读写的所有字段)
-
输入状态(InputState):定义图的输入约束,即用户调用图时必须提供的字段(通常是
OverallState
的子集)。 -
输出状态(OutputState)定义图的输出约束,即图执行后返回给用户的字段(通常是
OverallState
的子集)。 -
私有状态(PrivateState):定义图内部节点间的私有通信字段,这些字段不参与输入 / 输出,仅用于内部逻辑。
-
-
关键点说明:
-
节点可写入任意内部状态字段。虽然节点 1 的输入是
InputState
(仅含user_input
),但它可以写入OverallState
中的foo
字段。图的状态是所有模式(OverallState
、InputState
、OutputState
)的并集,节点可读写内部状态的任意字段。 -
节点可声明额外的私有状态字段。虽然图初始化时只传入了
OverallState
、Input
、Output
模式,但节点 2 可以写入PrivateState
中的bar
字段。只要状态模式(如PrivateState
)已定义,节点就可以动态添加并读写新的私有字段,无需在图初始化时声明。
-
-
总结
多模式状态管理允许 LangGraph:
-
通过
InputState
和OutputState
约束用户输入 / 输出,隐藏内部实现细节; -
通过
OverallState
和PrivateState
管理节点间通信的内部状态,提高图的灵活性和可维护性。 -
这种设计使复杂的推理流程可以清晰地分离用户交互和内部逻辑,同时保证节点间状态传递的类型安全。
-
Reducer(状态更新器)
Reducer 是 LangGraph 中处理节点对状态(State)更新的关键组件。每个状态字段(Key)都有独立的 Reducer 函数,其核心作用是:
- 若未显式指定 Reducer,则默认使用 覆盖策略(即新值直接替换旧值)
- 定义当节点返回新值时,如何将新值与现有状态合并(而非简单覆盖)。
默认 Reducer(Default Reducer)
from typing_extensions import TypedDictclass State(TypedDict):foo: intbar: list[str]
-
状态初始值:
{"foo": 1, "bar": ["hi"]}
-
节点 1 更新:返回
{"foo": 2}
→ 状态变为{"foo": 2, "bar": ["hi"]}
(foo
被覆盖,bar
不变)。 -
节点 2 更新:返回
{"bar": ["bye"]}
→ 状态变为{"foo": 2, "bar": ["bye"]}
(bar
被覆盖)。 -
核心逻辑:未指定 Reducer 时,节点返回的字段会直接覆盖状态中对应的旧值,未返回的字段保持不变。
为特定字段指定 Reducer(合并策略)
from typing import Annotated
from typing_extensions import TypedDict
from operator import addclass State(TypedDict):foo: intbar: Annotated[list[str], add] # 使用add函数作为bar的Reducer
-
状态初始值:
{"foo": 1, "bar": ["hi"]}
-
节点 1 更新:返回
{"foo": 2}
→ 状态变为{"foo": 2, "bar": ["hi"]}
(foo
被覆盖,bar
不变)。 -
节点 2 更新:返回
{"bar": ["bye"]}
→ 状态变为{"foo": 2, "bar": ["hi", "bye"]}
(bar
被合并,而非覆盖)。 -
核心逻辑:通过
Annotated[type, reducer_func]
为bar
字段指定operator.add
作为 Reducer。当节点返回{"bar": ["bye"]}
时,Reducer 执行 列表相加操作(["hi"] + ["bye"]
),实现状态合并而非覆盖。
消息状态处理函数
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDictclass GraphState(TypedDict):messages: Annotated[list[AnyMessage], add_messages] # 使用add_messages作为Reducer
现代 LLM提供商的聊天模型接口通常接受消息列表作为输入。例如,LangChain 的ChatModel
支持输入Message
对象列表,包括HumanMessage
(用户输入)、AIMessage
(模型响应)等类型。在图状态中存储消息列表有助于保留对话历史,支持多轮交互场景。
若需根据消息 ID 更新现有消息(如人工介入修改历史消息),可以使用预构建的add_messages
函数,该函数会:
-
对新消息执行追加操作;
-
对已有 ID 的消息执行更新操作。
add_messages
函数支持两种消息输入格式的自动反序列化:
-
直接使用 LangChain 消息对象:
{"messages": [HumanMessage(content="Hello")]}
-
使用字典格式(含类型和内容)
{"messages": [{"type": "human", "content": "Hello"}]}
-
反序列化后,可通过点 notation 访问消息属性(如
state["messages"][-1].content
)。
MessagesState:预构建的消息状态类
为简化常用场景,LangGraph 提供了预构建的MessagesState
类:
- 包含一个
messages
字段,类型为list[AnyMessage]
,并默认使用add_messages
作为 Reducer。 - 支持通过继承扩展其他状态字段:
from langgraph.graph import MessagesStateclass CustomState(MessagesState):documents: list[str] # 扩展其他状态字段
Nodes(节点)
用于编码代理逻辑的 Python 函数。它们接收当前值State
作为输入,执行一些计算或副作用,并返回更新后的State
。
节点函数的结构
LangGraph 中的节点本质是 Python 函数(支持同步 / 异步),需满足以下规范:
- 第一个参数:状态(
state
),包含图的当前数据; - 第二个可选参数:配置(
config
),包含可配置参数(如user_id
、thread_id
等)。
from typing_extensions import TypedDict
from langchain_core.runnables import RunnableConfigclass State(TypedDict):input: strresults: str# 带配置参数的节点函数
def my_node(state: State, config: RunnableConfig):print("In node: ", config["configurable"]["user_id"])return {"results": f"Hello, {state['input']}!"}# 无配置参数的节点函数
def my_other_node(state: State):return state
底层实现机制
节点函数会被自动转换为RunnableLambdas
,具备以下增强功能:
- 支持批量处理(
batch
)和异步调用(async
); - 原生集成追踪(
tracing
)和调试(debugging
)功能。
特殊节点START 和 END
START:表示图的入口节点,用于指定流程起点。
-
builder.add_edge(START, "node_a") # 从START连接到node_a
-
builder.set_entry_point("node_a") # 设置入口节点
END:表示图的终端节点,用于标记流程结束。
-
builder.add_edge("node_a", END) # 从node_a连接到END
-
builder.set_finish_point("node_a") # 设置结束节点
节点缓存(Node Caching)机制
-
基于节点输入缓存计算结果,避免重复执行耗时任务;
-
支持配置缓存策略(如缓存键生成规则、过期时间)。
import time
from langgraph.graph import StateGraph, START, END
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicyclass State(TypedDict):x: intresult: intbuilder = StateGraph(State)# 定义耗时节点
def expensive_node(state: State) -> dict[str, int]:time.sleep(2) # 模拟耗时操作return {"result": state["x"] * 2}# 添加节点并设置缓存策略(ttl=3秒)
builder.add_node("expensive_node", expensive_node,cache_policy=CachePolicy(ttl=3)
)
builder.set_entry_point("expensive_node") # 设置入口节点
builder.set_finish_point("expensive_node") # 设置结束节点# 编译图时指定缓存(InMemoryCache)
graph = builder.compile(cache=InMemoryCache())# 第一次调用(无缓存,耗时2秒)
print(graph.invoke({"x": 5})) # 输出: {'result': 10}
# 第二次调用(3秒内,使用缓存)
print(graph.invoke({"x": 5})) # 输出: {'result': 10, '__metadata__': {'cached': True}}
在上面的代码示例中,我们使用cache=InMemoryCache()方法设置了节点缓存;并通过cache_policy=CachePolicy(ttl=3)方法,在添加节点时,设置了3秒过期的缓存策略。
-
ttl
:缓存过期时间(秒),未指定则永不过期。
Edges(边)
根据当前条件确定下一步执行哪个操作的 Python 函数State
。它们可以是条件分支或固定转换。
普通边
固定路由,始终从一个节点到下一个节点。适用于流程明确、无需条件判断的场景(如顺序执行节点)。
graph.add_edge("node_a", "node_b") # 从node_a直接到node_b
条件边
通过路由函数动态决定下一个节点(支持多分支或终止)。适用于需要根据状态(如 LLM 返回结果、工具调用结果)动态选择路径的场景。
核心参数:
-
node_name
:源节点名称; -
routing_function
:路由函数,接收当前状态并返回目标节点名(或列表); -
mapping
(可选):将路由函数的返回值映射到目标节点。
def routing_function(state):if state["condition"]:return "node_b"return "node_c"# 基础用法
graph.add_conditional_edges("node_a", routing_function)# 带映射的用法(根据返回值True/False选择不同节点)
graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"}
)
入口点
明确流程起点(如用户输入直接触发某个处理节点)。
graph.add_edge(START, "node_a") # 图从node_a开始执行
条件入口点
根据自定义逻辑动态选择图的起始节点。适用于需要根据输入类型(如文本 / 图像)或用户身份选择不同处理流程的场景。
graph.add_conditional_edges(START, routing_function) # 启动时执行routing_function决定入口graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"}
)
多出口并行执行
若节点有多个出边(如node_a
同时指向node_b
和node_c
),则这些目标节点会在下一个超级步骤(superstep)中并行执行。
graph.add_edge("node_a", "node_b")
graph.add_edge("node_a", "node_c") # node_b和node_c并行执行
参考文献
LangGraph 官方文档:Overview