本教程将介绍如何使用LangGraph库构建和测试状态图。我们将通过一系列示例代码，逐步解释程序的运行逻辑。
1. 基本状态图构建
首先，我们定义一个状态图的基本结构和节点。
定义状态类
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from pydantic import BaseModel# 定义整个图的状态（这是在节点间共享的公共状态）
class OverallState(BaseModel):a: str
定义节点函数
def node(state: OverallState):return {"a": "goodbye"}
构建状态图
# 构建状态图
builder = StateGraph(OverallState)
builder.add_node(node) # 添加节点
builder.add_edge(START, "node") # 从起始节点开始
builder.add_edge("node", END) # 在节点后结束
graph = builder.compile()
测试状态图
# 使用有效输入测试图
graph.invoke({"a": "hello"})
输出结果：
{'a': 'goodbye'}
可视化状态图
from IPython.display import Image, displaytry:display(Image(graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:# 这需要一些额外的依赖项，是可选的print("x")pass
测试无效输入
try:graph.invoke({"a": 123}) # 应该是字符串
except Exception as e:print("An exception was raised because `a` is an integer rather than a string.")print(e)
输出结果：
An exception was raised because `a` is an integer rather than a string.
1 validation error for OverallState
aInput should be a valid string [type=string_type, input_value=123, input_type=int]For further information visit https://errors.pydantic.dev/2.8/v/string_type
2. 处理无效节点
接下来，我们添加一个返回无效状态的节点，并观察其影响。
定义节点函数
def bad_node(state: OverallState):return {"a": 123 # 无效}def ok_node(state: OverallState):return {"a": "goodbye"}
构建状态图
# 构建状态图
builder = StateGraph(OverallState)
builder.add_node(bad_node)
builder.add_node(ok_node)
builder.add_edge(START, "bad_node")
builder.add_edge("bad_node", "ok_node")
builder.add_edge("ok_node", END)
graph = builder.compile()
可视化状态图
from IPython.display import Image, displaytry:display(Image(graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:# 这需要一些额外的依赖项，是可选的pass
测试状态图
# 使用有效输入测试图
try:graph.invoke({"a": "hello"})
except Exception as e:print("An exception was raised because bad_node sets `a` to an integer.")print(e)
输出结果：
An exception was raised because bad_node sets `a` to an integer.
1 validation error for OverallState
aInput should be a valid string [type=string_type, input_value=123, input_type=int]For further information visit https://errors.pydantic.dev/2.8/v/string_type
3. 输入和输出状态的分离
接下来，我们定义输入和输出状态的分离。
定义状态类
from typing_extensions import TypedDict# 定义输入状态的架构
class InputState(TypedDict):question: str# 定义输出状态的架构
class OutputState(TypedDict):answer: str# 定义整体架构，结合输入和输出
class OverallState(InputState, OutputState):pass
定义节点函数
def answer_node(state: InputState):# 示例答案和一个额外的键return {"answer": "bye", "question": state["question"]}
构建状态图
# 使用指定的输入和输出架构构建图
builder = StateGraph(OverallState, input=InputState, output=OutputState)
builder.add_node(answer_node) # 添加答案节点
builder.add_edge(START, "answer_node") # 定义起始边
builder.add_edge("answer_node", END) # 定义结束边
graph1 = builder.compile() # 编译图
可视化状态图
from IPython.display import Image, displaytry:display(Image(graph1.get_graph(xray=True).draw_mermaid_png()))
except Exception:# 这需要一些额外的依赖项，是可选的pass
测试状态图
print(graph1.invoke({"question": "hi"}))
输出结果：
{'answer': 'bye'}
4. 处理私有数据
最后，我们展示如何在节点间传递私有数据。
定义状态类
# 整个图的状态（这是在节点间共享的公共状态）
class OverallState(TypedDict):a: str# node_1的输出包含私有数据，不属于整体状态
class Node1Output(TypedDict):private_data: str
定义节点函数
def node_1(state: OverallState) -> Node1Output:output = {"private_data": "set by node_1"}print(f"Entered node `node_1`:\n\tInput: {state}.\n\tReturned: {output}")return output# node_2的输入仅请求node_1后可用的私有数据
class Node2Input(TypedDict):private_data: strdef node_2(state: Node2Input) -> OverallState:output = {"a": "set by node_2"}print(f"Entered node `node_2`:\n\tInput: {state}.\n\tReturned: {output}")return output# node_3仅能访问整体状态（无法访问node_1的私有数据）
def node_3(state: OverallState) -> OverallState:output = {"a": "set by node_3"}print(f"Entered node `node_3`:\n\tInput: {state}.\n\tReturned: {output}")return output
构建状态图
# 构建状态图
builder = StateGraph(OverallState)
builder.add_node(node_1) # node_1是第一个节点
builder.add_node(node_2) # node_2是第二个节点，接受node_1的私有数据
builder.add_node(node_3) # node_3是第三个节点，无法看到私有数据
builder.add_edge(START, "node_1") # 从node_1开始
builder.add_edge("node_1", "node_2") # 从node_1传递到node_2
builder.add_edge("node_2", "node_3") # 从node_2传递到node_3（仅共享整体状态）
builder.add_edge("node_3", END) # 在node_3后结束
graph = builder.compile()
可视化状态图
from IPython.display import Image, displaytry:display(Image(graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:# 这需要一些额外的依赖项，是可选的pass
测试状态图
# 使用初始状态调用图
response = graph.invoke({"a": "set at start",}
)print()
print(f"Output of graph invocation: {response}")
输出结果：
Entered node `node_1`:Input: {'a': 'set at start'}.Returned: {'private_data': 'set by node_1'}
Entered node `node_2`:Input: {'private_data': 'set by node_1'}.Returned: {'a': 'set by node_2'}
Entered node `node_3`:Input: {'a': 'set by node_2'}.Returned: {'a': 'set by node_3'}Output of graph invocation: {'a': 'set by node_3'}
通过以上步骤，我们展示了如何使用LangGraphState管理，包括处理私有数据和状态验证。希望这个教程对你有所帮助！
参考链接：https://langchain-ai.github.io/langgraph/how-tos/state-model/
https://langchain-ai.github.io/langgraph/how-tos/input_output_schema/
https://langchain-ai.github.io/langgraph/how-tos/pass_private_state/
如果有任何问题，欢迎在评论区提问。
