如何将 Pydantic 模型用作图状态¶
一个 StateGraph 在初始化时接受一个 state_schema
参数,该参数指定了图中的节点可以访问和更新的状态的“形状”。
在我们的示例中,我们通常使用 Python 原生的 TypedDict
作为 state_schema
(或者在 MessageGraph 的情况下,一个 列表),但 state_schema
可以是任何 类型。
在本指南中,我们将看到如何使用 Pydantic BaseModel 作为 state_schema
,以在 输入 上添加运行时验证。
已知限制
-
本笔记本使用 Pydantic v2
BaseModel
,需要langchain-core >= 0.3
。使用langchain-core < 0.3
将会导致错误,因为 Pydantic v1 和 v2BaseModels
混合使用。 - 目前,图的 `output` 将**不会**是 Pydantic 模型的一个实例。
- 运行时验证仅发生在进入节点的 **输入** 上,而不是在输出上。
- Pydantic 的验证错误跟踪不会显示错误发生在哪个节点。
环境搭建¶
首先我们需要安装所需的包
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
为LangGraph开发设置 LangSmith
注册 LangSmith,可以快速发现并解决您的 LangGraph 项目中的问题,并提高项目性能。LangSmith 让您可以使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序——更多关于如何开始的信息,请参阅 这里。
输入验证¶
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from pydantic import BaseModel
# The overall state of the graph (this is the public state shared across nodes)
class OverallState(BaseModel):
a: str
def node(state: OverallState):
return {"a": "goodbye"}
# Build the state graph
builder = StateGraph(OverallState)
builder.add_node(node) # node_1 is the first node
builder.add_edge(START, "node") # Start the graph with node_1
builder.add_edge("node", END) # End the graph after node_1
graph = builder.compile()
# Test the graph with a valid input
graph.invoke({"a": "hello"})
API Reference: StateGraph | START | END
使用**无效**的输入调用图
try:
graph.invoke({"a": 123}) # Should be a string
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
a
Input should be a valid string [type=string_type, input_value=123, input_type=int]
For further information visit https://errors.pydantic.dev/2.9/v/string_type
多节点¶
运行时验证同样适用于多节点图。在下面的例子中,bad_node
将 a
更新为一个整数。
因为运行时验证发生在**输入**上,所以当调用 ok_node
时会出现验证错误(而不是在 bad_node
返回一个与模式不一致的状态更新时出现)。
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from pydantic import BaseModel
# The overall state of the graph (this is the public state shared across nodes)
class OverallState(BaseModel):
a: str
def bad_node(state: OverallState):
return {
"a": 123 # Invalid
}
def ok_node(state: OverallState):
return {"a": "goodbye"}
# Build the state graph
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()
# Test the graph with a valid input
try:
graph.invoke({"a": "hello"})
except Exception as e:
print("An exception was raised because bad_node sets `a` to an integer.")
print(e)
API Reference: StateGraph | START | END
An exception was raised because bad_node sets `a` to an integer.
1 validation error for OverallState
a
Input should be a valid string [type=string_type, input_value=123, input_type=int]
For further information visit https://errors.pydantic.dev/2.9/v/string_type