Pregel由Node和Channel这两个核心部件组成，Channel不仅维护了整个图的状态，还是驱动Node执行的 “原力” 。在前面演示的一系列实例中，我们已经使用了三种Channel类型，包括频繁使用的LastValue，能够将所有添加的值保留下来的BinaryOperatorAggregate，以及帮助我们轻松解决多Node依赖问题的NamedBarrierValue。为什么Channel有这么多类型，我们应该在何种场景中使用何种类型的Channel，要回答这个问题，就无法回避（BSP：Bulk Synchronous Parallel）这一个算法模式，毫不夸张地说，整个Pregel就是围绕BSP进行设计的。
1. BSP
我们知道一个Agent利用作为决策者的LLM和一系列作为执行者的Tool协助完成指定的任务。LLM（这里主要指基于文本生成的GPT）由于其逐Token的生成机制，加上各种工具可能涉及到大数据的处理、IO读写、跨网络的交互等耗时操作，所以要确保Agent能够提供可接受的性能保障，并行计算是唯一的解决方案。考虑到对并行计算的 “钢需” ，那么Pregel为什么选择Actor模型来实现也就很容易理解了，因为高并发正式Actor模型最擅长的领域。BSP中的BP（Bulk Parallel）可以理解为Pregel多个Node在同一时段的并行执行，那么S（Synchronous）有作何理解呢？
在很多典型的Actor模型里，Actor大多是有状态的，但是作为Pregel的Actor，其Node对象是完全无状态的。开始执行之前，他从注册的Channel读取数据，执行结果以最终输出到相应的Channel。我们知道高并发、低延时和数据一致性是三个永远不可能同时兼顾的指标，如果要兼顾其中两个，就不得不牺牲第三个。大部分系统的架构与设计都是在对三者作一个平衡，Pregel亦是如此。
由于多个并行执行的Node会涉及针对同一份数据（Channel）的读写，如果要保证数据的完整性，就不得不施加相应的同步机制，比如各种不同类型的锁。不论多轻量级的锁，它们都是高并发最强的杀手，那么能否实现无锁的解决方案呢？BSP就是答案，而这其中又涉及一个名为Superstep（Super Step，由于这一个概念会频繁提及，后续内容直接将Superstep视为一个专有名词来指代这一个概念）的概念，我们可以将其视为整个图有条不紊向前推进的一个单步，是Pregel执行引擎的一个脉冲，是所有Node执行的“节拍”，每个Superstep具有一个自增的序号。
在每个Superstep结束之后，每个Channel的状态被固定下来，引擎根据Node针对Channel的订阅和Channel在当前Superstep内的更新情况确定在下一Superstep需要执行的Node，并为它们创建创建相应的任务。这些任务在新的Superstep开始是并发执行，而且它们只能访问前一超步固化下来的值。这一策略确保输入的一致性，保证所有并发执行的Node所见的状态都是一样的；另一方面Channel在Superstep内保持只读状态就完全不用考虑并发脏读的问题了。
并发执行的Node不允许对任何Channel实施写入，它只能将写入诉求提交给执行引擎，后者按照提交的顺序存储下来。让所有任务成功执行后，Superstep进入一个同步屏障，执行引擎将对收集到的Channel写入请求按照顺序统一应用，使所有Channel最终能准确地体现Superstep结束后的状态。然后再确定下一Superstep的任务，如此反复，周而复始。
第一个Superstep的序号是从-1开始计数的，这个所谓的“创世超步”开始于针对Pregel兑现的常规调用，引擎利用这一Superstep将原始输入写入对应的Channel，并计算需要在Superstep 0中执行的任务，所以Node的首次执行是在Superstep 0中执行的。
由于Agent可以涉及一个需要长时间执行的处理流程，而且这个过程还可能出现中断，此中断可能是由于系统故障，也可能是需要认为介入导致，所以Agent的状态不可能常驻内容。而且Agent还提供了 “时间旅行” 的功能，这一个功能使我们从过去的某个时间点开启一个新的分支来执行流程，这相当于开启一个 “平行世界” 。所有的这一切都离不开针对状态的持久化，这是整个Pregel最复杂的部分。
我们将在后续内容详细介绍Pregel针对状态持久化的设计与实现，这里我们做一个概括性介绍。当Superstep进入同步屏障之后，持久化发生在 “解析待执行任务” 之后，此时引擎会针对当前Channel的状态创建一份名为Checkpoint快照并利用注册的Checkpointer记录下来。
每个被记录的Checkpoint不仅具有一个唯一标识，还登记了当前的Superstep序号，还有一个能够精确描述当前Agent对应的图在整个执行图中的位置（多Agent应用中，一个Agent可作为 “子Agent” 被调用，对应的图就是整张执行图的 “子图” ）和调用顺序的 “命名空间” 。完整的标识体系的唯一左右就是，在Agent从某个点恢复执行的时候，相应的Checkpointer能够准确地提取对应的Checkpoint来 “恢复现场” 。下图基本反映了Pregel针对单一Superstep的执行流程。
但是很多文档中你看到他们将Superstep划分为“计划”、“执行”、“更新”和“持久化”，看起来合理，但是和真正的执行流程是有出入的。计划阶段得到其实不是当前Superstep待执行的任务，而是下一Superstep待执行的任务，因为这有这样采用解释为什么针对Superstep -1的Checkpoint会包含任务列表，这些任务分明是在Superstep 0中才会被执行。实际上Checkpoint中的任务列表反映的都是下一Superstep待执行的操作，所以唯独最后一个Checkpoint才不会存在任务列表。表示状态快照的StateSnapshot将返回任务名称列表的字段命名为next，我想也应该是处于这个原因。
2. BaseChannel
所有的Channel类型都派生于如下这个名为BaseChannel的抽象类，它的泛型类型参数Value和Update分别表示存储类型、更新时提供的值类型，我们可以利用属性ValueType和UpdateType获取前两种类型。抽象方法get和update提供针对这两个类型的读写,update方法返回的布尔值表示是否具有实质性的更新，引擎由此判断Channel的值是否发生改变。另一个泛型类型参数Checkpoint与持久化有关，引擎在恢复执行时会从持久化的快照中提取对应的数据并将其转换成此对象，最终调用from_checkpoint方法恢复Channel的状态，在进行持久化时，引擎会调用checkpoint方法将自身的状态转换成该类型，并提交给Checkpointer实施持久化。- Value = TypeVar("Value")Update = TypeVar("Update")Checkpoint = TypeVar("Checkpoint")class BaseChannel(Generic[Value, Update, Checkpoint], ABC): __slots__ = ("key", "typ") def __init__(self, typ: Any, key: str = "") -> None @property @abstractmethod def ValueType(self) -> Any @property @abstractmethoddef UpdateType(self) -> Any @abstractmethoddef get(self) -> Value @abstractmethoddef update(self, values: Sequence[Update]) -> bool @abstractmethoddef from_checkpoint(self, checkpoint: Checkpoint | Any) -> Selfdef checkpoint(self) -> Checkpoint | Any def copy(self) -> Self def consume(self) -> bool def finish(self) -> bool def is_available(self) -> bool
Channel的值并非在每个时间点都是可用的，比如对于某些Channel，在Superstep N中写入的值只能在Superstep N+1中可用，Channel的可用性由is_available方法决定。finish是一个“钩子（Hook）”方法，引擎会在每个Superstep完成时调用此方法，对应的Channel可用利用此方法做一些类似于“可用性控制”的操作。指的一体的是，只有在当前Superstep中被更新的且被Node订阅的Channel，它们的finish方法才会被调用。
3. Channel
Pregel提供了若干预定义的Channel类型，它们的读写行为以及可用性上都有差异，我们需要根据数据或者状态的特性选择适合的Channel类型。
3.1 LastValue
LastValue只存储单值，并且采用后来居上的策略，存储最后一个提交更新的值。从如下提供的具体实现可用看出，LastValue的值类型、更新类型和Checkpoint类型三者合一，所以只通过字段value维护唯一的值。get、checkpoint和update方法实现了针对该字段的读写，如果没有对value字段显式赋值，is_available方法会返回False，get方法也会抛出异常。copy和from_checkpoint方法返回的都是一个新的LastValue对象，提供的值会为其value字段赋值。- class LastValue(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("value",) value: Value | Any def __init__(self, typ: Any, key: str = "") -> None: super().__init__(typ, key) self.value = MISSING def __eq__(self, value: object) -> bool: return isinstance(value, LastValue) @property def ValueType(self) -> type[Value]: return self.typ @property def UpdateType(self) -> type[Value]: return self.typ def copy(self) -> Self: empty = self.__class__(self.typ, self.key) empty.value = self.value return empty def from_checkpoint(self, checkpoint: Value) -> Self: empty = self.__class__(self.typ, self.key) if checkpoint is not MISSING: empty.value = checkpoint return empty def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: return False if len(values) != 1: msg = create_error_message( message=f"At key '{self.key}': Can receive only one value per step. Use an Annotated key to handle multiple values.", error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE, ) raise InvalidUpdateError(msg) self.value = values[-1] return True def get(self) -> Value: if self.value is MISSING: raise EmptyChannelError() return self.value def is_available(self) -> bool: return self.value is not MISSING def checkpoint(self) -> Value: return self.valueMISSING = object()
- from langgraph.channels import LastValuefrom langgraph.pregel import Pregel, NodeBuilderfrom langgraph.errors import InvalidUpdateErrordef build(node_name:str)->NodeBuilder: return (NodeBuilder() .subscribe_to("start") .do(lambda _:node_name) .write_to("output"))app = Pregel( nodes={name: build(name) for name in ["foo", "bar", "baz"]}, channels={ "start": LastValue(None), "output": LastValue(str), }, input_channels=["start"], output_channels=["output"],)try: app.invoke(input={"start": None}) assert False, "Expected InvalidUpdateError"except Exception as e: assert isinstance(e, InvalidUpdateError)
AnyValue类型的Channel同样是用于存储单值，所以它的实现于LastValue在很多地方都类似。两者之间最大的不同在于，AnyValue支持同一个Superstep的多次更新，但是将所有的更新视为等效，所以它的update方法会使用最后一个更新提供的值。- class AnyValue(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("typ", "value") value: Value | Any def __init__(self, typ: Any, key: str = "") -> None: super().__init__(typ, key) self.value = MISSING def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: if self.value is MISSING: return False else: self.value = MISSING return True self.value = values[-1] return True
- from langgraph.channels import LastValue,AnyValuefrom langgraph.pregel import Pregel, NodeBuilderimport timefrom typing import Anyfrom functools import partialchanges: list[Any] = []class ExtendedAnyValue(AnyValue): def update(self, values: list[Any]) -> bool: global changes changes = values return super().update(values) def handle(node_name:str, args:dict[str,Any])->str: if node_name != "foo": time.sleep(1) # Simulate some processing delay return node_namedef build(node_name:str)->NodeBuilder: return (NodeBuilder() .subscribe_to("start") .do(partial(handle, node_name)) .write_to("output"))app = Pregel( nodes={name: build(name) for name in ["foo","bar", "baz"]}, channels={ "start": LastValue(None), "output": ExtendedAnyValue(str), }, input_channels=["start"], output_channels=["output"], )result = app.invoke(input={"start": None})print("collected changes:", changes)print("channel value:", result["output"])
- collected changes: ['bar', 'baz', 'foo']channel value: foo
LastValueAfterFinish类型的Channel弥补了LastValue在同一Superstep中不能多次更新的不足，它的update方法采用了与AnyValue类似的逻辑，总是会提取最后提交的更新。不过我们在前面已经说过，我们不应该对更新顺序作任何假设，所以它是选择第一个还是最后一个其实都没有什么不同。- class LastValueAfterFinish( Generic[Value], BaseChannel[Value, Value, tuple[Value, bool]]): value: Value | Any finished: bool def __init__(self, typ: Any, key: str = "") -> None: super().__init__(typ, key) self.value = MISSING self.finished = False def update(self, values: Sequence[Value | Any]) -> bool: if len(values) == 0: return False self.finished = False self.value = values[-1] return True def consume(self) -> bool: if self.finished: self.finished = False self.value = MISSING return True return False def finish(self) -> bool: if not self.finished and self.value is not MISSING: self.finished = True return True else: return False def get(self) -> Value: if self.value is MISSING or not self.finished: raise EmptyChannelError() return self.value def is_available(self) -> bool: return self.value is not MISSING and self.finished
赋予了LastValueAfterFinish “跨步延迟” 的可见性，我们可以从如下这个演示程序看出它和LastValue的不同之处。如代码片段所示，Pregel唯一的Node订阅了“foo”和“bar”这两个Channel，其类型分别为LastValue和LastValueAfterFinish。在Superstep -1提供的两个输入foo和bar，前者在Superstep -1可见，所以会在Superstep 0触发节点执行，此时后者是不可用的状态。等到Superstep 0结束，finish方法被调用之后，bar变得可用，其更新触发节点与Superstep 1中再次执行。- from langgraph.channels import LastValue,LastValueAfterFinishfrom langgraph.pregel import Pregel, NodeBuilderfrom langchain_core.runnables import RunnableConfigfrom typing import Anyinputs:list = []def handle(args:dict[str,Any], config: RunnableConfig)->None: step = config.get("metadata", {}).get("langgraph_step") inputs.append((step, args.get("foo"), args.get("bar")))node = (NodeBuilder() .subscribe_to("foo","bar", read=True) .do(handle))app = Pregel( nodes={"body": node}, channels={ "foo": LastValue(str), "bar": LastValueAfterFinish(str), }, input_channels=["foo", "bar"], output_channels=[],)app.invoke(input={"foo": "123", "bar": "456",})assert len(inputs) == 2assert inputs[0] == (0, "123", None)assert inputs[1] == (1, "123", "456")
- from langgraph.channels import LastValue,LastValueAfterFinishfrom langgraph.pregel import Pregel, NodeBuildernode = (NodeBuilder() .subscribe_only("input") .do(lambda args:args) .write_to("output"))app = Pregel( nodes={"body": node}, channels={ "input": LastValueAfterFinish(str), "output": LastValue(str), }, input_channels=["input"], output_channels=["output"],)result = app.invoke(input={"input": "foobar"})assert result == None
3.4 EphemeralValue
EphemeralValue是一种专门用于短寿命数据传递的Channel类型，因为它的核心特性就是 “更新数据邻步有效” 。EphemeralValue在默认情况下也像LastValue一样不支持通过Superstep中的多个更新，但是我们可用在构造函数中将guard参数设置为False来关闭这个限制。如果允许单个Superstep内的多次更新，它也只会选择最后一次更新提供的值。- class EphemeralValue(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("value", "guard") value: Value | Any guard: bool def __init__(self, typ: Any, guard: bool = True) -> None: super().__init__(typ) self.guard = guard self.value = MISSING def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: if self.value is not MISSING: self.value = MISSING return True else: return False if len(values) != 1 and self.guard: raise InvalidUpdateError( f"At key '{self.key}': EphemeralValue(guard=True) can receive only one value per step. Use guard=False if you want to store any one of multiple values." ) self.value = values[-1] return True
- from langgraph.channels import LastValue,EphemeralValuefrom langgraph.pregel import Pregel, NodeBuilderfrom langchain_core.runnables import RunnableConfigfrom typing import Anyinputs:list = []def handle(args:dict[str,Any], config: RunnableConfig)->None: step = config.get("metadata", {}).get("langgraph_step") inputs.append((step, args.get("foo"), args.get("bar")))node1 = (NodeBuilder() .subscribe_to("node1", read=False) .read_from("foo", "bar") .do(handle) .write_to(node2= None))node2 = (NodeBuilder() .subscribe_to("node2", read=False) .read_from("foo", "bar") .do(handle))app = Pregel( nodes={"node1": node1, "node2": node2}, channels={ "foo": LastValue(str), "bar": EphemeralValue(str), "node1": LastValue(None), "node2": LastValue(None), }, input_channels=["node1","foo", "bar"], output_channels=[],)app.invoke(input={"node1": None,"foo": "123", "bar": "456",})assert len(inputs) == 2assert inputs[0] == (0, "123", "456")assert inputs[1] == (1, "123", None)
UntrackedValue是一种特殊的Channel类型，其核心特性在于它的 “非持久性” 和 “不可追踪性” 。写入UntrackedValue的值常驻内存，并会参与持久化。这一特性体现在的checkpoint和from_checkpoint方法上，前者为持久化提供一个空值，后者直接放弃从持久化存储加载的值。与EphemeralValue一样，UntrackedValue也通过构造函数的guard参数决定是否允许同一Superstep的多次写入。- class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]): def checkpoint(self) -> Value | Any: return MISSING def from_checkpoint(self, checkpoint: Value) -> Self: empty = self.__class__(self.typ, self.guard) empty.key = self.key return empty def update(self, values: Sequence[Value]) -> bool: if len(values) == 0: return False if len(values) != 1 and self.guard: raise InvalidUpdateError( f"At key '{self.key}': UntrackedValue(guard=True) can receive only one value per step. Use guard=False if you want to store any one of multiple values." ) self.value = values[-1] return True
- 安全性与隐私：某些敏感信息（如临时访问令牌、API Keys 或用户私密凭证）不应写入磁盘或持久化数据库的快照中。
- 性能优化：如果某些数据量极大（如大型图像 Buffer、大型文件流或复杂的对象实例），且这些数据在系统崩溃后可以重新生成或不再需要，排除它们可以显著减小 Checkpoint 的体积，提升存储性能。
- 对象非序列化兼容：有些 Python 对象（如打开的文件句柄、数据库连接、正在运行的线程或特定的第三方库对象）是无法被序列化的。将它们放入UntrackedValueChannel可以避免图在保存状态时报错。
UntrackedValueChannel不参与持久化的特性可用通过如下这个程序来验证。如代码片段所示，我们分别定义了一个组输入Channel（foo和bar）和输出Channel（baz和qux），每组中一个是常规的LastValue，另一个则是UntrackedValueChannel。我们为Pregel对象指定了Checkpointer, 并在调用时利用RunnableConfig指定了Thread ID，那么调用的历史将会被持久化下来。- from langgraph.channels import LastValue,UntrackedValuefrom langgraph.pregel import Pregel, NodeBuilderfrom langchain_core.runnables import RunnableConfigfrom langgraph.checkpoint.memory import InMemorySavernode = (NodeBuilder() .subscribe_to("foo","bar") .do(lambda args:args) .write_to(baz=lambda r: r["foo"], qux=lambda r: r["bar"]))app = Pregel( nodes={"body": node}, channels={ "foo": LastValue(str), "bar": UntrackedValue(str), "baz": LastValue(str), "qux": UntrackedValue(str), }, input_channels=["foo", "bar"], output_channels=["baz", "qux"], checkpointer= InMemorySaver())config:RunnableConfig = {"configurable":{"thread_id":"123"}}app.invoke(input={"start": None,"foo": "123", "bar": "456",}, config=config)for state in app.get_state_history(config=config):print(f"step {state.metadata['step']}: {state.values}")
- step 0: {'foo': '123', 'baz': '123'}step -1: {'foo': '123'}
BinaryOperatorAggregate是功能最强大的Channel类型。首先，它支持同一Superstep中针对它的多次写入；其次，我们可用利用它来存储不同类型的数据，可以是单值，也可以是列表、集合胡总和字典，因为最终存储的值是由我们提供的一个二元操作符决定的。这个操作符体现为一个Callable[[Value, Value], Value] 对象，两个输入参数分别表示现有的值和新写入的值，这个可执行对象的返回值最终写入Channel的值。- class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]): __slots__ = ("value", "operator") def __init__(self, typ: type[Value], operator: Callable[[Value, Value], Value]): super().__init__(typ) self.operator = operator … def update(self, values: Sequence[Value]) -> bool: if not values: return False if self.value is MISSING: self.value = values[0] values = values[1:] seen_overwrite: bool = False for value in values: is_overwrite, overwrite_value = _get_overwrite(value) if is_overwrite: if seen_overwrite: msg = create_error_message( message="Can receive only one Overwrite value per super-step.", error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE, ) raise InvalidUpdateError(msg) self.value = overwrite_value seen_overwrite = True continue if not seen_overwrite: self.value = self.operator(self.value, value) return True
- from langgraph.channels import LastValue,BinaryOperatorAggregatefrom langgraph.pregel import Pregel, NodeBuilderimport operatordef build_node(node_name:str)->NodeBuilder: return (NodeBuilder() .subscribe_to("start", read=False) .do(lambda _: [node_name]) .write_to("result"))app = Pregel( nodes={name: build_node(name) for name in ["foo", "bar", "baz"]}, channels={ "start": LastValue(None), "result": BinaryOperatorAggregate(list, operator.add), }, input_channels=["start"], output_channels=["result"],)result = app.invoke(input={"start": None})assert all(x in result["result"] for x in ["foo", "bar", "baz"])
- from langgraph.channels import LastValue,BinaryOperatorAggregatefrom langgraph.pregel import Pregel, NodeBuilderfrom typing import Anydef build_node(node_name:str)->NodeBuilder: return (NodeBuilder() .subscribe_to("start", read=False) .do(lambda _: node_name) .write_to("result"))def append(a:list,b:Any)->list: if isinstance(b,list): return a + b else: a.append(b) return aapp = Pregel( nodes={name: build_node(name) for name in ["foo", "bar", "baz"]}, channels={ "start": LastValue(None), "result": BinaryOperatorAggregate(list, append), }, input_channels=["start"], output_channels=["result"],)result = app.invoke(input={"start": None})assert all(x in result["result"] for x in ["foo", "bar", "baz"])
- @dataclass(slots=True)class Overwrite: value: Any
- from langgraph.channels import LastValue,BinaryOperatorAggregatefrom langgraph.pregel import Pregel, NodeBuilderfrom langgraph.types import Overwritefoo = (NodeBuilder() .subscribe_to("foo", read=False) .write_to(output = ["foo"], bar=None))bar = (NodeBuilder() .subscribe_to("bar", read=False) .do(lambda _: Overwrite(value=["bar"])) .write_to("output"))app = Pregel( nodes={"foo": foo, "bar": bar}, channels={ "foo": LastValue(None), "bar": LastValue(None), "output": BinaryOperatorAggregate(list, lambda a,b: a + b), }, input_channels=["foo"], output_channels=["output"],)result = app.invoke(input={"foo": None}, interrupt_after= "foo")assert result == {"output": ["foo"]}result = app.invoke(input={"foo": None})assert result == {"output": ["bar"]}
- from langgraph.channels import LastValue,BinaryOperatorAggregatefrom langgraph.pregel import Pregel, NodeBuilderfoo = (NodeBuilder() .subscribe_to("foo", read=False) .write_to(output = ["foo"], bar=None))bar = (NodeBuilder() .subscribe_to("bar", read=False) .do(lambda _: {"__overwrite__": ["bar"]}) .write_to("output"))app = Pregel( nodes={"foo": foo, "bar": bar}, channels={ "foo": LastValue(None), "bar": LastValue(None), "output": BinaryOperatorAggregate(list, lambda a,b: a + b), }, input_channels=["foo"], output_channels=["output"],)result = app.invoke(input={"foo": None}, interrupt_after= "foo")assert result == {"output": ["foo"]}result = app.invoke(input={"foo": None})assert result == {"output": ["bar"]}
我们在前面的演示实例中利用NamedBarrierValue这个特殊的Channel解决了“多Node依赖”的问题，我们从中也能大致知道它的工作原理：它内部维护两个集合，一个是初始化指定的名称集合，另一个写入填充的集合，等两个集合一致的时候该Channel才变得可用。这两个集合对应NamedBarrierValue如下所示的names和seen集合。当update在应用更新的时候，如果提供的名称没有在names集合中，会直接抛出InvalidUpdateError异常。如果已经在seen集合中，它会直接忽略并返回False，否则才会将其添加到seen集合中，并返回True。- class NamedBarrierValue(Generic[Value], BaseChannel[Value, Value, set[Value]]): names: set[Value] seen: set[Value] def __init__(self, typ: type[Value], names: set[Value]) -> None: super().__init__(typ) self.names = names self.seen: set[str] = set() def checkpoint(self) -> set[Value]: return self.seen def from_checkpoint(self, checkpoint: set[Value]) -> Self: empty = self.__class__(self.typ, self.names) empty.key = self.key if checkpoint is not MISSING: empty.seen = checkpoint return empty def update(self, values: Sequence[Value]) -> bool: updated = False for value in values: if value in self.names: if value not in self.seen: self.seen.add(value) updated = True else: raise InvalidUpdateError( f"At key '{self.key}': Value {value} not in {self.names}" ) return updated def get(self) -> Value: if self.seen != self.names: raise EmptyChannelError() return None def is_available(self) -> bool: return self.seen == self.names def consume(self) -> bool: if self.seen == self.names: self.seen = set() return True return False
3.8 NamedBarrierValueAfterFinish
NamedBarrierValueAfterFinish在NamedBarrierValue基础上添加了基于 “AfterFinish” 的可用性限制。即使names和seen集合导致一致状态，其生效的时机会延后致下一个Superstep。NamedBarrierValueAfterFinish会在内部维护一个finished状态，当seen被填充满时针对finish方法的调用会将此状态设置为True。在达到可用条件后针对consume方法的调用会将此状态设置为False。实现的逻辑基本上与和LastValueAfterFinish一致。
3.9 Topic
Topic类型的Channel就相当于一个消息队列，它会保留所有提供的数据。在默认情况下，在完成基于Superstep的更新应用之前，这个对象会被清空。如果在初始化的时候利用accumulate参数开启了 “累积” 模式，数据在跨越Superstep时会被保留。这个逻辑体现在如下所示的update方法上。- class Topic( Generic[Value], BaseChannel[Sequence[Value], Value | list[Value], list[Value]]): __slots__ = ("values", "accumulate") def __init__(self, typ: type[Value], accumulate: bool = False) -> None: super().__init__(typ) self.accumulate = accumulate self.values = list[Value]() def update(self, values: Sequence[Value | list[Value]]) -> bool: updated = False if not self.accumulate: updated = bool(self.values) self.values = list[Value]() if flat_values := tuple(_flatten(values)): updated = True self.values.extend(flat_values) return updated
- from langgraph.channels import LastValue,NamedBarrierValue, Topicfrom langgraph.pregel import Pregel, NodeBuildernode1 = (NodeBuilder() .subscribe_to("start", read=False) .write_to(trigger="node1", foo="node1", bar="node1"))node2 = (NodeBuilder() .subscribe_to("start", read=False) .write_to(trigger="node2", foo="node2", bar="node2"))node3 = (NodeBuilder() .subscribe_to("trigger", read=False) .write_to(foo="node3", bar="node3"))node4 = (NodeBuilder() .subscribe_to("trigger", read=False) .write_to(foo="node4", bar="node4"))app = Pregel( nodes={"node1": node1, "node2": node2, "node3": node3, "node4": node4}, channels={ "start": LastValue(None), "trigger": NamedBarrierValue(list,names={"node1", "node2"}), "foo": Topic(list), "bar": Topic(list, accumulate=True), }, input_channels=["start"], output_channels=["foo", "bar"],) result = app.invoke(input={"start": None})assert set(result["foo"]) == {"node3", "node4"}assert set(result["bar"]) == {"node1", "node2", "node3", "node4"}
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！ |
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
照妖镜
