如何将 Pydantic 模型用作图状态¶

前提条件

本指南假设你熟悉以下内容：

状态
节点
Pydantic ：这是一个流行的用于运行时验证的 Python 库。

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 和 v2 BaseModels 而导致错误。
目前，图的 `输出` **不会** 是 pydantic 模型的实例。
运行时验证仅在节点的 **输入** 上进行，而不在输出上进行。
pydantic 的验证错误跟踪不会显示错误出现在哪个节点中。

安装设置¶

首先，我们需要安装所需的软件包。

%%capture --no-stderr
%pip install --quiet -U langgraph

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 构建的大语言模型应用程序 — 点击此处了解更多关于如何开始使用的信息。

输入验证¶

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"})

{'a': 'goodbye'}

使用**无效**输入调用图

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)

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

多节点¶

运行时验证在多节点图中同样有效。在以下示例中，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)

Pydantic 模型的高级用法¶

本节涵盖了在 LangGraph 中使用 Pydantic 模型时的更高级主题。

序列化行为¶

当使用 Pydantic 模型作为状态模式时，理解序列化的工作原理非常重要，特别是在以下情况下： - 将 Pydantic 对象作为输入传递 - 从图中接收输出 - 使用嵌套的 Pydantic 模型

让我们来看看这些行为的实际表现：

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel


class NestedModel(BaseModel):
    value: str


class ComplexState(BaseModel):
    text: str
    count: int
    nested: NestedModel


def process_node(state: ComplexState):
    # Node receives a validated Pydantic object
    print(f"Input state type: {type(state)}")
    print(f"Nested type: {type(state.nested)}")

    # Return a dictionary update
    return {"text": state.text + " processed", "count": state.count + 1}


# Build the graph
builder = StateGraph(ComplexState)
builder.add_node("process", process_node)
builder.add_edge(START, "process")
builder.add_edge("process", END)
graph = builder.compile()

# Create a Pydantic instance for input
input_state = ComplexState(text="hello", count=0, nested=NestedModel(value="test"))
print(f"Input object type: {type(input_state)}")

# Invoke graph with a Pydantic instance
result = graph.invoke(input_state)
print(f"Output type: {type(result)}")
print(f"Output content: {result}")

# Convert back to Pydantic model if needed
output_model = ComplexState(**result)
print(f"Converted back to Pydantic: {type(output_model)}")

运行时类型强制转换¶

Pydantic 会对某些数据类型执行运行时类型强制转换。这可能会很有用，但如果你没有意识到这一点，也可能会导致意外的行为。

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel


class CoercionExample(BaseModel):
    # Pydantic will coerce string numbers to integers
    number: int
    # Pydantic will parse string booleans to bool
    flag: bool


def inspect_node(state: CoercionExample):
    print(f"number: {state.number} (type: {type(state.number)})")
    print(f"flag: {state.flag} (type: {type(state.flag)})")
    return {}


builder = StateGraph(CoercionExample)
builder.add_node("inspect", inspect_node)
builder.add_edge(START, "inspect")
builder.add_edge("inspect", END)
graph = builder.compile()

# Demonstrate coercion with string inputs that will be converted
result = graph.invoke({"number": "42", "flag": "true"})

# This would fail with a validation error
try:
    graph.invoke({"number": "not-a-number", "flag": "true"})
except Exception as e:
    print(f"\nExpected validation error: {e}")

使用消息模型¶

在你的状态架构中使用 LangChain 消息类型时，序列化方面有一些重要的注意事项。当通过网络传输消息对象时，你应该使用 AnyMessage（而不是 BaseMessage）来进行正确的序列化/反序列化：

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
from langchain_core.messages import HumanMessage, AIMessage, AnyMessage
from typing import List


class ChatState(BaseModel):
    messages: List[AnyMessage]
    context: str


def add_message(state: ChatState):
    return {"messages": state.messages + [AIMessage(content="Hello there!")]}


builder = StateGraph(ChatState)
builder.add_node("add_message", add_message)
builder.add_edge(START, "add_message")
builder.add_edge("add_message", END)
graph = builder.compile()

# Create input with a message
initial_state = ChatState(
    messages=[HumanMessage(content="Hi")], context="Customer support chat"
)

result = graph.invoke(initial_state)
print(f"Output: {result}")

# Convert back to Pydantic model to see message types
output_model = ChatState(**result)
for i, msg in enumerate(output_model.messages):
    print(f"Message {i}: {type(msg).__name__} - {msg.content}")

API Reference: HumanMessage | AIMessage | AnyMessage

如何将 Pydantic 模型用作图状态¶

安装设置¶

输入验证¶

多节点¶

多节点¶

Pydantic 模型的高级用法¶

序列化行为¶

运行时类型强制转换¶

使用消息模型¶

Comments