In [1]:
Copied!
%%capture --no-stderr
%pip install --quiet -U langgraph
%%capture --no-stderr
%pip install --quiet -U langgraph
In [1]:
Copied!
from typing_extensions import TypedDict
class TypedDictState(TypedDict):
foo: str
bar: str
from typing_extensions import TypedDict
class TypedDictState(TypedDict):
foo: str
bar: str
如需更具体的值约束，可以使用诸如 Literal 类型提示等方案。
此处 mood 仅允许为 "happy" 或 "sad" 两种取值。
In [2]:
Copied!
from typing import Literal
class TypedDictState(TypedDict):
name: str
mood: Literal["happy","sad"]
from typing import Literal
class TypedDictState(TypedDict):
name: str
mood: Literal["happy","sad"]
我们可以通过在 StateGraph 中直接传入自定义的状态类（例如这里的 TypedDictState）来在 LangGraph 中使用它。
同时，我们可以将每个状态键（state key）视为图中的一条"通道"（channel）。
正如模块 1 所讨论的，我们会在每个节点中覆盖指定键或"通道"的值。
In [3]:
Copied!
import random
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
def node_1(state):
print("---Node 1---")
return {"name": state['name'] + " is ... "}
def node_2(state):
print("---Node 2---")
return {"mood": "happy"}
def node_3(state):
print("---Node 3---")
return {"mood": "sad"}
def decide_mood(state) -> Literal["node_2", "node_3"]:
# Here, let's just do a 50 / 50 split between nodes 2, 3
if random.random() < 0.5:
# 50% of the time, we return Node 2
return "node_2"
# 50% of the time, we return Node 3
return "node_3"
# Build graph
builder = StateGraph(TypedDictState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
import random
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
def node_1(state):
print("---Node 1---")
return {"name": state['name'] + " is ... "}
def node_2(state):
print("---Node 2---")
return {"mood": "happy"}
def node_3(state):
print("---Node 3---")
return {"mood": "sad"}
def decide_mood(state) -> Literal["node_2", "node_3"]:
# Here, let's just do a 50 / 50 split between nodes 2, 3
if random.random() < 0.5:
# 50% of the time, we return Node 2
return "node_2"
# 50% of the time, we return Node 3
return "node_3"
# Build graph
builder = StateGraph(TypedDictState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
由于我们的状态是一个字典，只需通过传入字典来调用图（graph），即可为状态中的 name 键设置初始值。
In [4]:
Copied!
graph.invoke({"name":"Lance"})
graph.invoke({"name":"Lance"})
---Node 1--- ---Node 2---
Out[4]:
{'name': 'Lance is ... ', 'mood': 'happy'}
In [5]:
Copied!
from dataclasses import dataclass
@dataclass
class DataclassState:
name: str
mood: Literal["happy","sad"]
from dataclasses import dataclass
@dataclass
class DataclassState:
name: str
mood: Literal["happy","sad"]
要访问 dataclass 的键值，我们只需调整 node_1 中使用的下标方式：
- 对于
dataclass状态，我们使用state.name而非之前TypedDict的state["name"]
你会发现一个有趣的现象：在每个节点中，我们仍然返回字典来执行状态更新。
这是可行的，因为 LangGraph 会分别存储状态对象的每个键值。
节点返回的对象只需包含与状态对象匹配的键（属性）即可！
在本例中，dataclass 具有 name 键，因此我们可以像处理 TypedDict 状态时那样，通过从节点传递字典来更新它。
In [6]:
Copied!
def node_1(state):
print("---Node 1---")
return {"name": state.name + " is ... "}
# Build graph
builder = StateGraph(DataclassState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
def node_1(state):
print("---Node 1---")
return {"name": state.name + " is ... "}
# Build graph
builder = StateGraph(DataclassState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
我们通过调用 dataclass 来设置状态中每个键/通道的初始值！
In [7]:
Copied!
graph.invoke(DataclassState(name="Lance",mood="sad"))
graph.invoke(DataclassState(name="Lance",mood="sad"))
---Node 1--- ---Node 3---
Out[7]:
{'name': 'Lance is ... ', 'mood': 'sad'}
Pydantic¶
如前所述，TypedDict 和 dataclasses 虽然提供了类型提示，但不会在运行时强制类型检查。
这意味着你可能会在程序运行时赋予无效值而不会引发错误！
例如，尽管我们的类型提示明确标注了 mood: list[Literal["happy","sad"]]，我们仍然可以将 mood 设置为 mad。
In [8]:
Copied!
dataclass_instance = DataclassState(name="Lance", mood="mad")
dataclass_instance = DataclassState(name="Lance", mood="mad")
Pydantic 是一个基于 Python 类型注解的数据验证与配置管理库。
该库因其强大的验证能力，特别适合在 LangGraph 中定义状态模式。
Pydantic 能够在运行时执行数据验证，检查数据是否符合指定的类型和约束条件。
In [ ]:
Copied!
from pydantic import BaseModel, field_validator, ValidationError
class PydanticState(BaseModel):
name: str
mood: str # "happy" or "sad"
@field_validator('mood')
@classmethod
def validate_mood(cls, value):
# Ensure the mood is either "happy" or "sad"
if value not in ["happy", "sad"]:
raise ValueError("Each mood must be either 'happy' or 'sad'")
return value
try:
state = PydanticState(name="John Doe", mood="mad")
except ValidationError as e:
print("Validation Error:", e)
from pydantic import BaseModel, field_validator, ValidationError
class PydanticState(BaseModel):
name: str
mood: str # "happy" or "sad"
@field_validator('mood')
@classmethod
def validate_mood(cls, value):
# Ensure the mood is either "happy" or "sad"
if value not in ["happy", "sad"]:
raise ValueError("Each mood must be either 'happy' or 'sad'")
return value
try:
state = PydanticState(name="John Doe", mood="mad")
except ValidationError as e:
print("Validation Error:", e)
Validation Error: 1 validation error for PydanticState
mood
Input should be 'happy' or 'sad' [type=literal_error, input_value='mad', input_type=str]
For further information visit https://errors.pydantic.dev/2.8/v/literal_error
我们可以无缝地在图中使用 PydanticState。
In [ ]:
Copied!
# Build graph
builder = StateGraph(PydanticState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
# Build graph
builder = StateGraph(PydanticState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
# Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Add
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
In [ ]:
Copied!
graph.invoke(PydanticState(name="Lance",mood="sad"))
graph.invoke(PydanticState(name="Lance",mood="sad"))
---Node 1--- ---Node 3---
Out[ ]:
{'name': 'Lance is ... ', 'mood': 'sad'}
In [ ]:
Copied!