【LangGraph】什么是 StateGraph

AI小编 · 发表于 10 小时前

作者：彬彬侠
目录

代码示例
高级功能和注意事项与其他 LangGraph 类的关系常见问题和调试技巧总结和学习资源

1. 什么是 StateGraph？

langgraph.graph.StateGraph 是 LangGraph 库中的核心类，用于定义和管理基于状态的有向图工作流。它是 LangGraph 的主要构建块，允许开发者通过节点（Nodes）、边（Edges）和状态（State）构造复杂的工作流，特别适合与大型语言模型（LLM）结合的场景。

定义

用途

模块

StateGraph 的设计目标是提供一个灵活、可扩展的框架，让开发者可以：

2. StateGraph 的核心功能

StateGraph 提供了以下主要功能：

状态管理

节点定义

边管理

图编译

入口和出口

配置支持

扩展性

3. StateGraph 的关键组件

StateGraph 的工作流由以下核心组件组成：

状态（State）

通常使用 typing.TypedDict 或自定义类定义，例如：
1. from typing import TypedDict
3. classState(TypedDict):input:str
4. output:str
复制代码
状态在节点间传递，每个节点可以读取和修改状态。

节点（Nodes）

示例：
1. defmy_node(state: State)-> State:
2. state["output"]=f"处理: {state['input']}"return state
复制代码

边（Edges）

边定义了节点之间的执行顺序：
示例：
1. workflow.add_edge("node1","node2")# 无条件边
复制代码

入口和出口

配置（RunnableConfig）

config ={"configurable":{"user_id":"user123","thread_id":"thread_001"}}

复制代码

4. 如何使用 StateGraph

以下是使用 StateGraph 构建工作流的标准步骤：

导入必要模块：
1. version: python
2. from langgraph.graph import StateGraph, START, END
3. from typing import TypedDict
复制代码
定义状态：
创建一个状态类或字典，定义工作流中需要跟踪的字段。
1. version: python
2. classState(TypedDict):input:str
3. output:str
复制代码
定义节点：
编写节点函数，每个函数处理特定任务并更新状态。
1. version: python
2. defprocess_input(state: State)-> State:
3. state["output"]=f"处理输入: {state['input']}"return state
复制代码
创建 StateGraph 实例：
初始化 StateGraph，传入状态类。
1. version: python
2. workflow = StateGraph(State)
复制代码
添加节点：
使用 add_node 方法将节点添加到图中。
1. version: python
2. workflow.add_node("process_input", process_input)
复制代码
添加边：
使用 add_edge 或 add_conditional_edges 定义节点之间的连接。
1. version: python
2. workflow.add_edge(START,"process_input")
3. workflow.add_edge("process_input", END)
复制代码
编译图：
使用 compile() 方法将图编译为可执行对象。
1. version: python
2. graph = workflow.compile()
复制代码
运行图：
使用 invoke、stream 或 astream 方法运行图，传入初始状态和可选配置。
1. version: python
2. result = graph.invoke({"input":"Hello, LangGraph!","output":""})
复制代码

5. 代码示例

以下是三个不同复杂度的 StateGraph 示例，展示其灵活性。
5.1 简单线性工作流

一个简单的线性工作流，包含两个节点。

from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# 定义状态classState(TypedDict):input:str
output:str# 定义节点defprocess_input(state: State)-> State:
state["output"]=f"处理输入: {state['input']}"return state
deffinalize_output(state: State)-> State:
state["output"]+=" -> 已完成"return state
# 创建图
workflow = StateGraph(State)
workflow.add_node("process_input", process_input)
workflow.add_node("finalize_output", finalize_output)
workflow.add_edge(START,"process_input")
workflow.add_edge("process_input","finalize_output")
workflow.add_edge("finalize_output", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"input":"Hello, LangGraph!","output":""})print(result)

复制代码

输出：

{'input':'Hello, LangGraph!','output':'处理输入: Hello, LangGraph! -> 已完成'}

复制代码

5.2 带条件边的工作流

一个工作流，根据输入长度决定是否处理。

from langgraph.graph import StateGraph, START, END
from typing import TypedDict
# 定义状态classState(TypedDict):input:str
output:str
needs_processing:bool# 定义节点defcheck_input(state: State)-> State:
state["needs_processing"]=len(state["input"])>5return state
defprocess_input(state: State)-> State:
state["output"]=f"处理输入: {state['input']}"return state
defskip_processing(state: State)-> State:
state["output"]="输入太短，无需处理"return state
# 条件函数defroute(state: State)->str:return"process_input"if state["needs_processing"]else"skip_processing"# 创建图
workflow = StateGraph(State)
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})print(result)

复制代码

输出：

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

复制代码

5.3 支持循环和工具调用的复杂工作流

一个智能 Agent 工作流，结合 LLM 和工具调用，支持循环。

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Optional
# 定义工具@tooldefsearch(query:str, config: RunnableConfig)->str:"""模拟从网络上搜获相关的数据进行汇总"""returnf"搜索结果: {query} 的信息"# 定义状态classState(TypedDict):input:str
output:str
needs_search:bool
search_count:int# 定义节点defdecide_action(state: State, config: RunnableConfig)-> State:
llm = ChatOpenAI(model="gpt-4o-mini")
prompt =f"用户问题: {state['input']}\n回答 'yes' 如果需要搜索，'no' 如果可以直接回答。"
response = llm.invoke(prompt, config=config).content
state["needs_search"]= response.lower()=="yes"return state
defsearch_node(state: State, config: RunnableConfig)-> State:
state["search_count"]+=1
result = search.invoke({"query": state["input"]}, config=config)
state["output"]= result
return state
defanswer_node(state: State, config: RunnableConfig)-> State:
llm = ChatOpenAI(model="gpt-4o-mini")
prompt =f"直接回答用户问题: {state['input']}"
response = llm.invoke(prompt, config=config).content
state["output"]= response
return state
# 条件函数defroute(state: State)->str:if state["needs_search"]and state["search_count"]<2:return"search_node"return"answer_node"# 创建图
workflow = StateGraph(State)
workflow.add_node("decide_action", decide_action)
workflow.add_node("search_node", search_node)
workflow.add_node("answer_node", answer_node)
workflow.add_edge(START,"decide_action")
workflow.add_conditional_edges("decide_action", route,{"search_node":"search_node","answer_node":"answer_node"})
workflow.add_conditional_edges("search_node", route,{"search_node":"search_node","answer_node":"answer_node"})
workflow.add_edge("answer_node", END)# 编译和运行
graph = workflow.compile()
result = graph.invoke({"input":"什么是 LangGraph?","output":"","needs_search":False,"search_count":0},
config={"configurable":{"user_id":"user123"}})print(result)

复制代码

输出（假设 LLM 认为需要搜索）：

{'input':'什么是 LangGraph?','output':'搜索结果: 什么是 LangGraph? 的信息','needs_search':True,'search_count':1}

复制代码

6. 高级功能和注意事项

6.1 高级功能

异步支持

示例：
1. asyncdefasync_node(state: State)-> State:
2. state["output"]="异步处理"return state
复制代码

检查点（Checkpoints）

graph = workflow.compile(checkpointer=True)

复制代码

配置模式（Config Schema）

示例：
1. classConfigSchema(TypedDict):
2. user_id: Optional[str]
3. model: Optional[str]
4. workflow = StateGraph(State, config_schema=ConfigSchema)
复制代码

动态节点添加

6.2 注意事项

状态不可变性

示例：
1. defmy_node(state: State)-> State:
2. new_state = state.copy()
3. new_state["output"]="新值"return new_state
复制代码

条件边映射

递归限制

工具调用配置

7. 与其他 LangGraph 类的关系

CompiledGraph

MessageGraph

Pregel

8. 常见问题和调试技巧

8.1 常见问题

问题

原因

解决方案

问题

原因

解决方案

问题

原因

解决方案

8.2 调试技巧

使用 LangSmith

示例：
1. import os
2. os.environ["LANGCHAIN_TRACING_V2"]="true"
3. os.environ["LANGCHAIN_API_KEY"]="your-api-key"
复制代码

打印中间状态

示例：
1. defmy_node(state: State)-> State:print(f"当前状态: {state}")
2. state["output"]="处理完成"return state
复制代码

检查配置

9. 总结和学习资源

9.1 总结

langgraph.graph.StateGraph 是 LangGraph 的核心类，用于构建和管理有状态的有向图工作流。它通过状态、节点和边提供了一个灵活的框架，适合复杂的 LLM 应用场景，如智能 Agent、对话系统和自动化工作流。关键点包括：

通过代码示例，可以看到 StateGraph 如何从简单线性工作流扩展到复杂的工具调用和循环逻辑。结合 LangChain 和 LangSmith，可以进一步增强开发效率和调试能力。
9.2 学习资源

官方文档

LangChain 文档

LangSmith

GitHub 仓库

社区讨论

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

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

【LangGraph】什么是 StateGraph

浏览过的版块

LangChain入门2 RAG详解

关于我们

产品与服务

解决方案

产品与服务