Agent实战教程：Langgraph的StateGraph以及State怎么用 - 文章 - 开发者社区

LangGraph的StateGraph是一个用于构建状态图的核心组件，用于创建有状态的多步骤工作流程。以下是基本用法： picture.image

StateGraph基本概念

StateGraph允许你定义一个图结构，其中每个节点代表一个操作，边定义了节点之间的连接关系，整个图维护一个共享的状态。

基本用法步骤

1. 定义状态结构

  
from typing import TypedDict  
from langgraph.graph import StateGraph, END  
  
class GraphState(TypedDict):  
    input: str  
    output: str  
    step\_count: int

2. 创建StateGraph实例

  
# 创建StateGraph，指定状态类型  
workflow = StateGraph(GraphState)

3. 定义节点函数

  
def node\_1(state: GraphState) -> GraphState:  
    """第一个节点的处理逻辑"""  
    return {  
        "input": state["input"],  
        "output": f"处理了: {state['input']}",  
        "step\_count": state.get("step\_count", 0) + 1  
    }  
  
def node\_2(state: GraphState) -> GraphState:  
    """第二个节点的处理逻辑"""  
    return {  
        "input": state["input"],  
        "output": state["output"] + " -> 进一步处理",  
        "step\_count": state["step\_count"] + 1  
    }

4. 添加节点

  
workflow.add\_node("step1", node\_1)  
workflow.add\_node("step2", node\_2)

5. 定义边连接

  
# 设置入口点  
workflow.set\_entry\_point("step1")  
  
# 添加边  
workflow.add\_edge("step1", "step2")  
workflow.add\_edge("step2", END)

6. 条件边的使用

你还可以添加条件边来实现分支逻辑：

  
def should\_continue(state: GraphState) -> str:  
    """决定下一步走向的条件函数"""  
    if state["step\_count"] < 3:  
        return "continue"  
    else:  
        return "end"  
  
# 添加条件边  
workflow.add\_conditional\_edges(  
    "step1",  
    should\_continue,  
    {  
        "continue": "step2",  
        "end": END  
    }  
)

7. 编译并运行

  
# 编译图  
app = workflow.compile()  
  
# 运行图  
initial\_state = {  
    "input": "Hello World",  
    "output": "",  
    "step\_count": 0  
}  
  
result = app.invoke(initial\_state)  
print(result)

picture.image

State基本概念

在现代AI应用开发中，如何优雅地管理复杂工作流的状态流转是一个关键挑战。LangGraph的State机制为我们提供了一套强大而灵活的状态管理解决方案。本文将深入探讨LangGraph State的设计理念、核心特性和实际应用。

什么是LangGraph State？

LangGraph State是LangGraph框架中用于定义和管理图状态的核心组件。它不仅定义了数据的结构模式（Schema），还通过reducer函数指定了状态更新的逻辑。State既是所有节点和边的输入模式，也是节点间数据流转的载体。核心设计理念 LangGraph State的设计遵循以下几个核心原则：

类型安全: 通过TypedDict或Pydantic模型确保数据结构的正确性
状态隔离: 支持多层次的状态模式，实现输入、输出和内部状态的分离
灵活性: 节点可以读写任何已声明的状态通道
可扩展性: 支持运行时动态添加新的状态通道

核心关系

**state\_schema**

图的 内部状态模式 ，定义了图内部所有可能的状态字段

**input\_schema**

图的 输入模式 ，是 state\_schema 的子集，定义图接受什么输入

**output\_schema**

图的 输出模式 ，是 state\_schema 的子集，定义图返回什么输出

具体说明

state_schema（必需）

图的 完整内部状态 ，包含所有节点可能读写的字段
所有节点都可以访问和写入这个schema中的任何字段
是图的"全局状态空间"

input_schema（可选）

如果不指定，默认等于 state\_schema
限制图的 输入接口 ，只能传入这些字段
是 state\_schema 的子集或相等

output_schema（可选）

如果不指定，默认等于 state\_schema
限制图的 输出接口 ，只返回这些字段
是 state\_schema 的子集或相等

实际例子

  
# 内部完整状态  
class OverallState(TypedDict):  
    foo: str           # 内部处理字段  
    user\_input: str    # 输入字段  
    graph\_output: str  # 输出字段  
  
# 限制输入只包含user\_input  
class InputState(TypedDict):  
    user\_input: str  
  
# 限制输出只包含graph\_output    
class OutputState(TypedDict):  
    graph\_output: str  
  
# 创建图  
builder = StateGraph(  
    OverallState,           # 内部完整状态  
    input\_schema=InputState,   # 输入限制  
    output\_schema=OutputState  # 输出限制  
)

关系图示

  
输入 → [InputState] → [OverallState] → [OutputState] → 输出  
      (输入过滤)      (内部完整状态)    (输出过滤)

这种设计的好处：

封装性 : 内部处理细节不会暴露给外部
接口清晰 : 明确定义输入输出契约
灵活性 : 内部可以有更多临时状态字段用于节点间通信

下面是一些例子

基础单模式使用

最简单的情况下，所有节点共享同一个状态模式：

  
from typing import TypedDict  
from langgraph.graph import StateGraph, START, END  
  
class SimpleState(TypedDict):  
    input: str  
    output: str  
    step\_count: int  
  
def process\_node(state: SimpleState) -> SimpleState:  
    return {  
        "input": state["input"],  
        "output": f"处理结果: {state['input']}",  
        "step\_count": state.get("step\_count", 0) + 1  
    }  
  
# 创建简单的状态图  
workflow = StateGraph(SimpleState)  
workflow.add\_node("processor", process\_node)  
workflow.add\_edge(START, "processor")  
workflow.add\_edge("processor", END)  
  
app = workflow.compile()  
result = app.invoke({"input": "Hello", "output": "", "step\_count": 0})

多模式架构：输入输出分离

在实际应用中，我们常常需要区分输入、输出和内部状态：

  
from typing import TypedDict  
from langgraph.graph import StateGraph, START, END  
  
  
class InputState(TypedDict):  
    """用户输入接口"""  
    user\_input: str  
  
  
class OutputState(TypedDict):  
    """最终输出接口"""  
    graph\_output: str  
    step\_count:int  
  
class OverallState(TypedDict):  
    """内部完整状态"""  
    user\_input: str  
    intermediate\_result: str  
    processed\_data: str  
    graph\_output: str  
    step\_count: int  
  
  
class PrivateState(TypedDict):  
    """节点间私有通信"""  
    private\_data: str  
    internal\_flag: bool  
  
  
def input\_node(state: InputState) -> OverallState:  
    """输入处理节点"""  
    return {  
        "user\_input": state["user\_input"],  
        "intermediate\_result": f"开始处理: {state['user\_input']}",  
        "step\_count": 1  
    }  
  
  
def processing\_node(state: OverallState) -> PrivateState:  
    """数据处理节点"""  
    processed = state["intermediate\_result"].upper()  
    return {  
        "private\_data": f"私有处理: {processed}",  
        "internal\_flag": len(state["user\_input"]) > 5  
    }  
  
  
def intermediate\_node(state: PrivateState) -> OverallState:  
    """中间处理节点"""  
    additional\_info = " (长文本)" if state["internal\_flag"] else " (短文本)"  
    return {  
        "processed\_data": state["private\_data"] + additional\_info  
    }  
  
  
def output\_node(state: OverallState) -> OutputState:  
    """输出生成节点"""  
    final\_result = f"最终结果: {state['processed\_data']} | 步骤: {state.get('step\_count', 0)}"  
    return {  
        "graph\_output": final\_result  
    }  
  
  
# 创建多模式状态图  
builder = StateGraph(  
    OverallState,  
    input=InputState,  
    output=OutputState  
)  
  
builder.add\_node("input\_processor", input\_node)  
builder.add\_node("data\_processor", processing\_node)  
builder.add\_node("intermediate\_processor", intermediate\_node)  
builder.add\_node("output\_processor", output\_node)  
  
builder.add\_edge(START, "input\_processor")  
builder.add\_edge("input\_processor", "data\_processor")  
builder.add\_edge("data\_processor", "intermediate\_processor")  
builder.add\_edge("intermediate\_processor", "output\_processor")  
builder.add\_edge("output\_processor", END)  
  
graph = builder.compile()  
  
# 测试输入  
input\_data = {"user\_input": "Hello LangGraph","extra\_input":"额外输入"}  
  
print("输入:", input\_data)  
  
 # 执行图  
result = graph.invoke(input\_data)  
  
print("输出:", result)  
  
  
png\_data=graph.get\_graph().draw\_mermaid\_png()  
with open("state.png", "wb") as f:  
    f.write(png\_data)

picture.image 输出如下

  
输入: {'user\_input': 'Hello LangGraph', 'extra\_input': '额外输入'}  
输出: {'graph\_output': '最终结果: 私有处理: 开始处理: HELLO LANGGRAPH (长文本) | 步骤: 1', 'step\_count': 1}

picture.image

添加微信，备注” LLM “进入大模型技术交流群

picture.image

如果你觉得这篇文章对你有帮助，别忘了点个赞、送个喜欢

/ 作者：致Great

/ 作者：欢迎转载，标注来源即可