【LangGraph】langgraph.graph.StateGraph 类的属性和方法

创想小编 · 发表于昨天 17:11

作者：彬彬侠
1. StateGraph 类的概述

StateGraph 是 LangGraph 库的核心类，位于 langgraph.graph 模块中，用于定义和构建基于状态的工作流图。它以 有向图 的形式组织任务，其中：

节点（Nodes）

边（Edges）

状态（State）

StateGraph 的主要作用是：

StateGraph 通常用于构建智能 Agent、对话系统、自动化流程等需要多步骤交互的场景。

2. StateGraph 类的定义和初始化

2.1 类定义

StateGraph 类的完整定义如下：

from langgraph.graph import StateGraph
from typing import Type, Optional, Any, Dict, Union
from langchain_core.runnables import RunnableConfig
classStateGraph:def__init__(
self,
state_schema: Optional[Type[Any]]=None,
config_schema: Optional[Type[Any]]=None)->None:...

复制代码

参数：

2.2 状态和配置的作用

状态（State）

配置（Config）

2.3 初始化示例

以下是一个简单的初始化示例：

from typing import TypedDict
from langgraph.graph import StateGraph
# 定义状态classState(TypedDict):input:str
output:str# 定义配置（可选）classConfigSchema(TypedDict):
user_id:str# 创建 StateGraph
workflow = StateGraph(state_schema=State, config_schema=ConfigSchema)

复制代码

3. StateGraph 的属性

StateGraph 类的属性主要用于存储图的结构和元数据。以下是其核心属性（部分属性是内部使用的，开发者通常通过方法间接访问）：

nodes

edges

conditional_edges

entry_point

finish_points

state_schema

config_schema

注意：这些属性通常是内部管理的，开发者不直接修改它们，而是通过 StateGraph 的方法（如 add_node、add_edge）来操作图结构。

4. StateGraph 的方法

StateGraph 提供了丰富的 API 用于构建和配置工作流图。以下是其主要方法，分为 构建图、设置入口/出口 和 编译图 三类：
4.1 构建图的方法

add_node(name: str, action: Callable) -> None

功能

参数：
用途：定义工作流中的操作单元。
示例：
1. defprocess_input(state: State, config: RunnableConfig)-> State:
2. state["output"]=f"处理: {state['input']}"return state
4. workflow.add_node("process_input", process_input)
复制代码

add_edge(source: Union[str, START, END], target: Union[str, END]) -> None

功能

参数：
用途：定义节点的执行顺序。
示例：
1. from langgraph.graph import START, END
2. workflow.add_edge(START,"process_input")
3. workflow.add_edge("process_input", END)
复制代码

add_conditional_edges(source: str, condition: Callable, edge_mapping: Dict[Any, Union[str, END]]) -> None

功能

参数：
用途：实现动态分支逻辑。
示例：
1. defroute(state: State)->str:return"process_input"iflen(state["input"])>5else"skip"
3. workflow.add_conditional_edges("check_input",
4. route,{"process_input":"process_input","skip":"skip"})
复制代码

4.2 设置入口和出口的方法

set_entry_point(name: str) -> None

功能

参数：
用途：指定工作流开始执行的节点。示例：
1. workflow.set_entry_point("process_input")
复制代码
注意：等效于 add_edge(START, name)。

set_finish_point(name: str) -> None

功能

参数：
用途：指定工作流的结束点。示例：
1. workflow.set_finish_point("finalize_output")
复制代码
注意：等效于 add_edge(name, END)。

4.3 编译图的方法

compile(checkpointer: Optional[BaseCheckpointSaver] = None, interrupt_after: Optional[List[str]] = None, interrupt_before: Optional[List[str]] = None) -> CompiledGraph

功能

参数：
返回值：CompiledGraph 对象，支持 invoke、stream、astream 等方法运行工作流。用途：将图的定义转换为可执行形式。
示例：
1. from langgraph.checkpoint.memory import MemorySaver
2. graph = workflow.compile(checkpointer=MemorySaver())
复制代码

5. 完整示例

以下是一个完整的示例，展示如何使用 StateGraph 构建一个具有条件分支的工作流：

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_core.runnables import RunnableConfig
# 定义状态classState(TypedDict):input:str
output:str
needs_processing:bool# 定义配置（可选）classConfigSchema(TypedDict):
user_id:str# 节点函数defcheck_input(state: State, config: RunnableConfig)-> State:
state["needs_processing"]=len(state["input"])>5
user_id = config.get("configurable",{}).get("user_id","default_user")
state["output"]=f"检查输入 by {user_id}"return state
defprocess_input(state: State, config: RunnableConfig)-> State:
state["output"]=f"处理输入: {state['input']}"return state
defskip_processing(state: State, config: RunnableConfig)-> State:
state["output"]="输入太短，无需处理"return state
# 条件函数defroute(state: State)->str:return"process_input"if state["needs_processing"]else"skip_processing"# 创建 StateGraph
workflow = StateGraph(state_schema=State, config_schema=ConfigSchema)# 添加节点
workflow.add_node("check_input", check_input)
workflow.add_node("process_input", process_input)
workflow.add_node("skip_processing", skip_processing)# 添加边
workflow.add_edge(START,"check_input")
workflow.add_conditional_edges("check_input",
route,{"process_input":"process_input","skip_processing":"skip_processing"})
workflow.add_edge("process_input", END)
workflow.add_edge("skip_processing", END)# 编译图
graph = workflow.compile()# 运行工作流
result = graph.invoke({"input":"Hi","output":"","needs_processing":False},
config={"configurable":{"user_id":"user123"}})print(result)

复制代码

输出（输入 “Hi”）：

{'input':'Hi','output':'输入太短，无需处理','needs_processing':False}

复制代码

输出（输入 “Hello, LangGraph!”）：

{'input':'Hello, LangGraph!','output':'处理输入: Hello, LangGraph!','needs_processing':True}

复制代码

这个示例展示了：

6. CompiledGraph 的关键方法

编译后的 CompiledGraph 对象（通过 workflow.compile() 获得）支持以下主要方法，用于执行工作流：

invoke(input: Dict[str, Any], config: Optional[RunnableConfig] = None) -> Dict[str, Any]

result = graph.invoke({"input":"test"}, config={"configurable":{"user_id":"user123"}})

复制代码

stream(input: Dict[str, Any], config: Optional[RunnableConfig] = None) -> Iterator[Dict[str, Any]]

for state in graph.stream({"input":"test"}):print(state)

复制代码

astream(input: Dict[str, Any], config: Optional[RunnableConfig] = None) -> AsyncIterator[Dict[str, Any]]

asyncfor state in graph.astream({"input":"test"}):print(state)

复制代码

astream_events(input: Dict[str, Any], config: Optional[RunnableConfig] = None, version: str = "v1") -> AsyncIterator[Dict[str, Any]]

asyncfor event in graph.astream_events({"input":"test"}, version="v1"):print(event)

复制代码

7. 高级功能和注意事项

7.1 检查点（Checkpointing）

示例：
1. from langgraph.checkpoint.memory import MemorySaver
2. graph = workflow.compile(checkpointer=MemorySaver())
复制代码

7.2 中断（Interrupts）

graph = workflow.compile(interrupt_after=["process_input"])

复制代码

7.3 异步支持

示例：
1. asyncdefasync_node(state: State, config: RunnableConfig)-> State:
2. state["output"]=await some_async_function(state["input"])return state
复制代码

7.4 注意事项

状态一致性

条件边逻辑

版本兼容性

8. 与其他类的关系

CompiledGraph

RunnableConfig

BaseCheckpointSaver

9. 总结

langgraph.graph.StateGraph 是 LangGraph 的核心类，用于定义基于状态的工作流图。其主要功能包括：

其属性（如 nodes、edges）管理图的内部结构，方法（如 add_node、compile）提供灵活的 API。结合状态管理和检查点功能，StateGraph 适合构建复杂的 LLM 驱动应用。
参考文献：

原文地址：https://blog.csdn.net/u013172930/article/details/147963138

25位大厂高管转战生成式AI创业！吸金猛，不

【LangGraph】langgraph.graph.StateGraph 类的属性和方法

LangGraph 调用工具过程详解

关于我们

产品与服务

解决方案

产品与服务