通道 (Channels)

langgraph.channels.base

BaseChannel

Bases: Generic[Value, Update, Checkpoint], ABC

所有通道的基类。

方法	描述
copy	返回通道的副本。
checkpoint	返回通道当前状态的可序列化表示。
from_checkpoint	返回一个新的相同通道，可以选择从检查点进行初始化。
get	返回通道的当前值。
is_available	如果通道可用（非空），则返回 True，否则返回 False。
更新	使用给定的更新序列更新通道的值。
consume	通知通道已订阅的任务已运行。默认情况下，无操作。
finish	通知通道 Pregel 运行即将结束。默认情况下，无操作。

ValueType abstractmethod property

ValueType: Any

通道中存储的值的类型。

UpdateType abstractmethod property

UpdateType: Any

通道接收的更新的类型。

copy

copy() -> Self

返回通道的副本。默认情况下，委托给 checkpoint() 和 from_checkpoint()。子类可以重写此方法以实现更高效的实现。

checkpoint

checkpoint() -> Checkpoint | Any

返回通道当前状态的可序列化表示。如果通道为空（从未更新过）或不支持检查点，则引发 EmptyChannelError。

from_checkpoint abstractmethod

from_checkpoint(checkpoint: Checkpoint | Any) -> Self

返回一个新的相同通道，可以选择从检查点进行初始化。如果检查点包含复杂的数据结构，则应将其复制。

get abstractmethod

get() -> Value

返回通道的当前值。

如果通道为空（从未更新过），则引发 EmptyChannelError。

is_available

is_available() -> bool

如果通道可用（非空），则返回 True，否则返回 False。子类应重写此方法，以提供比调用 get() 和捕获 EmptyChannelError 更高效的实现。

update abstractmethod

update(values: Sequence[Update]) -> bool

使用给定的更新序列更新通道的值。序列中更新的顺序是任意的。在每个步骤结束时，Pregel 会对所有通道调用此方法。如果没有更新，则使用空序列调用此方法。如果更新序列无效，则引发 InvalidUpdateError。如果通道已更新，则返回 True，否则返回 False。

consume

consume() -> bool

通知通道已订阅的任务已运行。默认情况下，无操作。通道可以使用此方法修改其状态，以防止值再次被消费。

如果通道已更新，则返回 True，否则返回 False。

finish

finish() -> bool

通知通道 Pregel 运行即将结束。默认情况下，无操作。通道可以使用此方法修改其状态，以防止完成。

如果通道已更新，则返回 True，否则返回 False。

langgraph.channels

Topic

Bases: Generic[Value], BaseChannel[Sequence[Value], Value | list[Value], list[Value]]

一个可配置的发布/订阅（PubSub）主题。

参数	描述
typ	通道中存储的值的类型。 类型： type[Value]
accumulate	是否在多个步骤之间累积值。如果为 False，通道将在每个步骤后被清空。 类型： bool 默认值： False

方法	描述
consume	通知通道已订阅的任务已运行。默认情况下，无操作。
finish	通知通道 Pregel 运行即将结束。默认情况下，无操作。
copy	返回通道的副本。
checkpoint	返回通道当前状态的可序列化表示。
from_checkpoint	返回一个新的相同通道，可以选择从检查点进行初始化。
更新	使用给定的更新序列更新通道的值。
get	返回通道的当前值。
is_available	如果通道可用（非空），则返回 True，否则返回 False。

ValueType property

ValueType: Any

通道中存储的值的类型。

UpdateType property

UpdateType: Any

通道接收的更新的类型。

consume

consume() -> bool

通知通道已订阅的任务已运行。默认情况下，无操作。通道可以使用此方法修改其状态，以防止值再次被消费。

如果通道已更新，则返回 True，否则返回 False。

finish

finish() -> bool

通知通道 Pregel 运行即将结束。默认情况下，无操作。通道可以使用此方法修改其状态，以防止完成。

如果通道已更新，则返回 True，否则返回 False。

copy

copy() -> Self

返回通道的副本。

checkpoint

checkpoint() -> list[Value]

返回通道当前状态的可序列化表示。如果通道为空（从未更新过）或不支持检查点，则引发 EmptyChannelError。

from_checkpoint

from_checkpoint(checkpoint: list[Value]) -> Self

返回一个新的相同通道，可以选择从检查点进行初始化。如果检查点包含复杂的数据结构，则应将其复制。

update

update(values: Sequence[Value | list[Value]]) -> bool

使用给定的更新序列更新通道的值。序列中更新的顺序是任意的。在每个步骤结束时，Pregel 会对所有通道调用此方法。如果没有更新，则使用空序列调用此方法。如果更新序列无效，则引发 InvalidUpdateError。如果通道已更新，则返回 True，否则返回 False。

get

get() -> Sequence[Value]

返回通道的当前值。

如果通道为空（从未更新过），则引发 EmptyChannelError。

is_available

is_available() -> bool

如果通道可用（非空），则返回 True，否则返回 False。子类应重写此方法，以提供比调用 get() 和捕获 EmptyChannelError 更高效的实现。

LastValue

Bases: Generic[Value], BaseChannel[Value, Value, Value]

存储接收到的最后一个值，每一步最多能接收一个值。

方法	描述
consume	通知通道已订阅的任务已运行。默认情况下，无操作。
finish	通知通道 Pregel 运行即将结束。默认情况下，无操作。
copy	返回通道的副本。
from_checkpoint	返回一个新的相同通道，可以选择从检查点进行初始化。
更新	使用给定的更新序列更新通道的值。
get	返回通道的当前值。
is_available	如果通道可用（非空），则返回 True，否则返回 False。
checkpoint	返回通道当前状态的可序列化表示。

ValueType property

ValueType: type[Value]

通道中存储的值的类型。

UpdateType property

UpdateType: type[Value]

通道接收的更新的类型。

consume

consume() -> bool

通知通道已订阅的任务已运行。默认情况下，无操作。通道可以使用此方法修改其状态，以防止值再次被消费。

如果通道已更新，则返回 True，否则返回 False。

finish

finish() -> bool

通知通道 Pregel 运行即将结束。默认情况下，无操作。通道可以使用此方法修改其状态，以防止完成。

如果通道已更新，则返回 True，否则返回 False。

copy

copy() -> Self

返回通道的副本。

from_checkpoint

from_checkpoint(checkpoint: Value) -> Self

返回一个新的相同通道，可以选择从检查点进行初始化。如果检查点包含复杂的数据结构，则应将其复制。

update

update(values: Sequence[Value]) -> bool

使用给定的更新序列更新通道的值。序列中更新的顺序是任意的。在每个步骤结束时，Pregel 会对所有通道调用此方法。如果没有更新，则使用空序列调用此方法。如果更新序列无效，则引发 InvalidUpdateError。如果通道已更新，则返回 True，否则返回 False。

get

get() -> Value

返回通道的当前值。

如果通道为空（从未更新过），则引发 EmptyChannelError。

is_available

is_available() -> bool

如果通道可用（非空），则返回 True，否则返回 False。子类应重写此方法，以提供比调用 get() 和捕获 EmptyChannelError 更高效的实现。

checkpoint

checkpoint() -> Value

返回通道当前状态的可序列化表示。如果通道为空（从未更新过）或不支持检查点，则引发 EmptyChannelError。

EphemeralValue

Bases: Generic[Value], BaseChannel[Value, Value, Value]

存储上一步接收到的值，之后会清除。

方法	描述
consume	通知通道已订阅的任务已运行。默认情况下，无操作。
finish	通知通道 Pregel 运行即将结束。默认情况下，无操作。
copy	返回通道的副本。
from_checkpoint	返回一个新的相同通道，可以选择从检查点进行初始化。
更新	使用给定的更新序列更新通道的值。
get	返回通道的当前值。
is_available	如果通道可用（非空），则返回 True，否则返回 False。
checkpoint	返回通道当前状态的可序列化表示。

ValueType property

ValueType: type[Value]

通道中存储的值的类型。

UpdateType property

UpdateType: type[Value]

通道接收的更新的类型。

consume

consume() -> bool

通知通道已订阅的任务已运行。默认情况下，无操作。通道可以使用此方法修改其状态，以防止值再次被消费。

如果通道已更新，则返回 True，否则返回 False。

finish

finish() -> bool

通知通道 Pregel 运行即将结束。默认情况下，无操作。通道可以使用此方法修改其状态，以防止完成。

如果通道已更新，则返回 True，否则返回 False。

copy

copy() -> Self

返回通道的副本。

from_checkpoint

from_checkpoint(checkpoint: Value) -> Self

返回一个新的相同通道，可以选择从检查点进行初始化。如果检查点包含复杂的数据结构，则应将其复制。

update

update(values: Sequence[Value]) -> bool

使用给定的更新序列更新通道的值。序列中更新的顺序是任意的。在每个步骤结束时，Pregel 会对所有通道调用此方法。如果没有更新，则使用空序列调用此方法。如果更新序列无效，则引发 InvalidUpdateError。如果通道已更新，则返回 True，否则返回 False。

get

get() -> Value

返回通道的当前值。

如果通道为空（从未更新过），则引发 EmptyChannelError。

is_available

is_available() -> bool

如果通道可用（非空），则返回 True，否则返回 False。子类应重写此方法，以提供比调用 get() 和捕获 EmptyChannelError 更高效的实现。

checkpoint

checkpoint() -> Value

返回通道当前状态的可序列化表示。如果通道为空（从未更新过）或不支持检查点，则引发 EmptyChannelError。

BinaryOperatorAggregate

Bases: Generic[Value], BaseChannel[Value, Value, Value]

存储对当前值和每个新值应用二元运算符的结果。

import operator

total = Channels.BinaryOperatorAggregate(int, operator.add)

方法	描述
consume	通知通道已订阅的任务已运行。默认情况下，无操作。
finish	通知通道 Pregel 运行即将结束。默认情况下，无操作。
copy	返回通道的副本。
from_checkpoint	返回一个新的相同通道，可以选择从检查点进行初始化。
更新	使用给定的更新序列更新通道的值。
get	返回通道的当前值。
is_available	如果通道可用（非空），则返回 True，否则返回 False。
checkpoint	返回通道当前状态的可序列化表示。

ValueType property

ValueType: type[Value]

通道中存储的值的类型。

UpdateType property

UpdateType: type[Value]

通道接收的更新的类型。

consume

consume() -> bool

通知通道已订阅的任务已运行。默认情况下，无操作。通道可以使用此方法修改其状态，以防止值再次被消费。

如果通道已更新，则返回 True，否则返回 False。

finish

finish() -> bool

通知通道 Pregel 运行即将结束。默认情况下，无操作。通道可以使用此方法修改其状态，以防止完成。

如果通道已更新，则返回 True，否则返回 False。

copy

copy() -> Self

返回通道的副本。

from_checkpoint

from_checkpoint(checkpoint: Value) -> Self

返回一个新的相同通道，可以选择从检查点进行初始化。如果检查点包含复杂的数据结构，则应将其复制。

update

update(values: Sequence[Value]) -> bool

使用给定的更新序列更新通道的值。序列中更新的顺序是任意的。在每个步骤结束时，Pregel 会对所有通道调用此方法。如果没有更新，则使用空序列调用此方法。如果更新序列无效，则引发 InvalidUpdateError。如果通道已更新，则返回 True，否则返回 False。

get

get() -> Value

返回通道的当前值。

如果通道为空（从未更新过），则引发 EmptyChannelError。

is_available

is_available() -> bool

如果通道可用（非空），则返回 True，否则返回 False。子类应重写此方法，以提供比调用 get() 和捕获 EmptyChannelError 更高效的实现。

checkpoint

checkpoint() -> Value

返回通道当前状态的可序列化表示。如果通道为空（从未更新过）或不支持检查点，则引发 EmptyChannelError。

AnyValue

Bases: Generic[Value], BaseChannel[Value, Value, Value]

存储接收到的最后一个值，假设如果接收到多个值，它们都是相等的。

方法	描述
consume	通知通道已订阅的任务已运行。默认情况下，无操作。
finish	通知通道 Pregel 运行即将结束。默认情况下，无操作。
copy	返回通道的副本。
from_checkpoint	返回一个新的相同通道，可以选择从检查点进行初始化。
更新	使用给定的更新序列更新通道的值。
get	返回通道的当前值。
is_available	如果通道可用（非空），则返回 True，否则返回 False。
checkpoint	返回通道当前状态的可序列化表示。

ValueType property

ValueType: type[Value]

通道中存储的值的类型。

UpdateType property

UpdateType: type[Value]

通道接收的更新的类型。

consume

consume() -> bool

通知通道已订阅的任务已运行。默认情况下，无操作。通道可以使用此方法修改其状态，以防止值再次被消费。

如果通道已更新，则返回 True，否则返回 False。

finish

finish() -> bool

通知通道 Pregel 运行即将结束。默认情况下，无操作。通道可以使用此方法修改其状态，以防止完成。

如果通道已更新，则返回 True，否则返回 False。

copy

copy() -> Self

返回通道的副本。

from_checkpoint

from_checkpoint(checkpoint: Value) -> Self

返回一个新的相同通道，可以选择从检查点进行初始化。如果检查点包含复杂的数据结构，则应将其复制。

update

update(values: Sequence[Value]) -> bool

使用给定的更新序列更新通道的值。序列中更新的顺序是任意的。在每个步骤结束时，Pregel 会对所有通道调用此方法。如果没有更新，则使用空序列调用此方法。如果更新序列无效，则引发 InvalidUpdateError。如果通道已更新，则返回 True，否则返回 False。

get

get() -> Value

返回通道的当前值。

如果通道为空（从未更新过），则引发 EmptyChannelError。

is_available

is_available() -> bool

如果通道可用（非空），则返回 True，否则返回 False。子类应重写此方法，以提供比调用 get() 和捕获 EmptyChannelError 更高效的实现。

checkpoint

checkpoint() -> Value

返回通道当前状态的可序列化表示。如果通道为空（从未更新过）或不支持检查点，则引发 EmptyChannelError。