LangGraph 运行时

Pregel 实现了 LangGraph 的运行时，负责管理 LangGraph 应用程序的执行。编译一个 StateGraph 或创建一个 @entrypoint 会产生一个 Pregel 实例，该实例可以通过输入进行调用。本指南将在较高层次上解释该运行时，并提供直接使用 Pregel 实现应用程序的说明。

注意： Pregel 运行时以 Google 的 Pregel 算法命名，该算法描述了一种利用图进行大规模并行计算的高效方法。

概述

在 LangGraph 中，Pregel 将 参与者 和通道结合到一个单一的应用程序中。参与者 从通道读取数据并向通道写入数据。Pregel 将应用程序的执行组织成多个步骤，遵循 Pregel 算法 / 批量同步并行 模型。每个步骤包含三个阶段：

规划：确定在此步骤中要执行哪些 参与者。例如，在第一步中，选择订阅了特殊输入通道的 参与者；在后续步骤中，选择订阅了上一步中已更新通道的 参与者。
执行：并行执行所有选定的 参与者，直到全部完成、其中一个失败或达到超时。在此阶段，通道更新对参与者是不可见的，直到下一步。
更新：使用此步骤中 参与者 写入的值来更新通道。

重复此过程，直到没有 参与者 被选中执行，或者达到最大步骤数。

参与者

参与者 是一个 PregelNode。它订阅通道，从中读取数据，并向其写入数据。它可以被视为 Pregel 算法中的一个 参与者。PregelNode 实现了 LangChain 的 Runnable 接口。

通道

通道用于参与者（PregelNodes）之间的通信。每个通道都有一个值类型、一个更新类型和一个更新函数——该函数接收一系列更新并修改存储的值。通道可用于将数据从一个链发送到另一个链，或在未来的步骤中将数据从一个链发送到自身。

LastValue

LastValue 是默认的通道类型。它存储写入的最后一个值，覆盖任何之前的值。用于输入和输出值，或用于将数据从一个步骤传递到下一个步骤。

from langgraph.channels import LastValue

channel: LastValue[int] = LastValue(int)

Topic

Topic 是一个可配置的 PubSub 通道，适用于在参与者之间发送多个值或跨步骤累积输出。它可以配置为对值进行去重，或累积运行期间写入的所有值。

from langgraph.channels import Topic

# 累积跨步骤写入的所有值
channel: Topic[str] = Topic(str, accumulate=True)

BinaryOperatorAggregate

BinaryOperatorAggregate 存储一个持久值，该值通过将二元运算符应用于当前值和每次新的更新来更新。用于跨步骤计算运行聚合。

import operator
from langgraph.channels import BinaryOperatorAggregate

# 运行总和：每次写入都累加到当前值
total = BinaryOperatorAggregate(int, operator.add)

DeltaChannel

DeltaChannel 需要 langgraph>=1.2 并且目前处于 测试阶段。API 可能在未来的版本中更改。

DeltaChannel 仅存储每一步的增量差异，而不是完整的累积值。这对于频繁写入且随时间累积大量值的通道最为有用——例如，长时间运行线程中的对话消息列表。如果没有增量存储，完整的列表会在每个检查点被重新序列化；使用 DeltaChannel，仅存储每个步骤写入的新消息。

当通道既被频繁写入又随时间增长变大时，可以考虑使用 DeltaChannel。一个很好的信号是：如果你发现某个特定通道的检查点大小随线程长度线性增长，那么 DeltaChannel 可能很适合。

在 Annotated 类型注解中使用 DeltaChannel，方式与使用普通归约器相同：

from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langgraph.channels import DeltaChannel


def my_reducer(state: list[str], writes: Sequence[list[str]]) -> list[str]:
    result = list(state)
    for write in writes:
        result.extend(write)
    return result


class State(TypedDict):
    messages: Annotated[list[str], DeltaChannel(my_reducer)]

批量归约器要求

传递给 DeltaChannel 的 reducer 是一个 批量归约器：它接收当前状态和当前步骤中序列化的所有写入，而不是像标准归约器那样成对调用。这与在 StateGraph 中使用 Annotated 的每键归约器不同，后者每个更新调用一次归约器。

批量归约器 必须是可结合的（批次处理不变性）：

reducer(reducer(state, [xs]), [ys]) == reducer(state, [xs, ys])

如果你的归约器不是可结合的，重构的状态可能会因 LangGraph 如何在步骤间批处理写入而有所不同，从而导致不一致的行为。

归约器在重构时运行，而不是在写入时运行。 与 BinaryOperatorAggregate 不同（其归约器在写入时调用，因此组合值就是序列化到检查点中的内容），DeltaChannel 归约器是在通道值从其持久化写入中重建时调用的。原始的每步写入是被序列化的内容；归约器只在值具体化时被调用——在下一次读取时、在下一步的参与者执行时，或在重放历史时。设计归约器时的实际后果：

使其成为 (state, writes) 的纯函数。 任何副作用、随机性或系统时钟读取（例如 uuid.uuid4()、datetime.now()）每次重构值时都会执行，并在每次重放时产生不同的结果。它们不会被烘焙到持久化的写入中。
不要依赖对传入写入的突变被持久化。 如果你的归约器突变了一个写入对象（例如，为一个没有 ID 的项分配一个稳定的 ID），该突变仅存在于重构的值中。存储的写入仍然具有原始形状，因此下次重构时将再次看到未突变的输入。
将身份标识和其他稳定的元数据附加在上游。 如果下游代码需要通过 ID 跨轮次引用一个项（例如，以便稍后更新或删除它），请在值写入通道之前分配该 ID——而不是在归约器内部。

以下是两种最常见情况的批量归约器：

from typing import Any, Sequence


# 列表：按顺序追加所有写入
def list_reducer(state: list[Any], writes: Sequence[list[Any]]) -> list[Any]:
    result = list(state)
    for write in writes:
        result.extend(write)
    return result


# 字典：合并所有写入，键冲突时最后写入的值获胜
def dict_reducer(
    state: dict[str, Any], writes: Sequence[dict[str, Any]]
) -> dict[str, Any]:
    result = dict(state)
    for write in writes:
        result.update(write)
    return result

两者都是可结合的：一次应用一个批次的结果与一次应用所有批次的结果相同。

使用 snapshot_frequency 限制读取延迟

没有快照时，读取 DeltaChannel 的值需要重放全部写入历史——对于有 N 个步骤的线程，时间复杂度为 O(N)。设置 snapshot_frequency=K 会每 K 个 Pregel 步骤写入一个完整快照，将读取深度限制在最多 K 个步骤：

class State(TypedDict):
    messages: Annotated[
        list[str],
        DeltaChannel(my_reducer, snapshot_frequency=5),
    ]

snapshot_frequency 的值越大，存储开销越小，但读取延迟越高。值越小，延迟限制越严格，但检查点会越大。None（默认值）会完全跳过快照——适用于读取很少或线程较短的情况。

版本兼容性和回滚

不支持回滚到不支持 DeltaChannel 的版本。 langgraph>=1.2 以新格式写入增量通道检查点，旧版本无法读取。一旦某个线程使用了 DeltaChannel，降级 LangGraph 将使这些检查点无法读取，因为旧运行时不理解增量格式，无法重构通道状态。如果需要回滚，请使用 delta-channel-dump 恢复脚本来迁移受影响的线程，或者在降级前将其丢弃。

示例

虽然大多数用户将通过 StateGraph API 或 @entrypoint 装饰器与 Pregel 交互，但也可以直接与 Pregel 交互。下面是一些不同的示例，让你了解 Pregel API 的使用。

单个节点
多个节点
Topic
BinaryOperatorAggregate
循环

from langgraph.channels import EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder

node1 = (
    NodeBuilder().subscribe_only("a")
    .do(lambda x: x + x)
    .write_to("b")
)

app = Pregel(
    nodes={"node1": node1},
    channels={
        "a": EphemeralValue(str),
        "b": EphemeralValue(str),
    },
    input_channels=["a"],
    output_channels=["b"],
)

app.invoke({"a": "foo"})

{'b': 'foofoo'}

from langgraph.channels import LastValue, EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder

node1 = (
    NodeBuilder().subscribe_only("a")
    .do(lambda x: x + x)
    .write_to("b")
)

node2 = (
    NodeBuilder().subscribe_only("b")
    .do(lambda x: x + x)
    .write_to("c")
)


app = Pregel(
    nodes={"node1": node1, "node2": node2},
    channels={
        "a": EphemeralValue(str),
        "b": LastValue(str),
        "c": EphemeralValue(str),
    },
    input_channels=["a"],
    output_channels=["b", "c"],
)

app.invoke({"a": "foo"})

{'b': 'foofoo', 'c': 'foofoofoofoo'}

from langgraph.channels import EphemeralValue, Topic
from langgraph.pregel import Pregel, NodeBuilder

node1 = (
    NodeBuilder().subscribe_only("a")
    .do(lambda x: x + x)
    .write_to("b", "c")
)

node2 = (
    NodeBuilder().subscribe_to("b")
    .do(lambda x: x["b"] + x["b"])
    .write_to("c")
)

app = Pregel(
    nodes={"node1": node1, "node2": node2},
    channels={
        "a": EphemeralValue(str),
        "b": EphemeralValue(str),
        "c": Topic(str, accumulate=True),
    },
    input_channels=["a"],
    output_channels=["c"],
)

app.invoke({"a": "foo"})

{'c': ['foofoo', 'foofoofoofoo']}

此示例演示了如何使用 BinaryOperatorAggregate 通道来实现归约器。

from langgraph.channels import EphemeralValue, BinaryOperatorAggregate
from langgraph.pregel import Pregel, NodeBuilder


node1 = (
    NodeBuilder().subscribe_only("a")
    .do(lambda x: x + x)
    .write_to("b", "c")
)

node2 = (
    NodeBuilder().subscribe_only("b")
    .do(lambda x: x + x)
    .write_to("c")
)

def reducer(current, update):
    if current:
        return current + " | " + update
    else:
        return update

app = Pregel(
    nodes={"node1": node1, "node2": node2},
    channels={
        "a": EphemeralValue(str),
        "b": EphemeralValue(str),
        "c": BinaryOperatorAggregate(str, operator=reducer),
    },
    input_channels=["a"],
    output_channels=["c"],
)

app.invoke({"a": "foo"})

{ 'c': 'foofoo | foofoofoofoo' }

此示例演示了如何在图中引入循环，通过让一个链写入它订阅的通道。执行将继续，直到向通道写入 None 值。

from langgraph.channels import EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry

example_node = (
    NodeBuilder().subscribe_only("value")
    .do(lambda x: x + x if len(x) < 10 else None)
    .write_to(ChannelWriteEntry("value", skip_none=True))
)

app = Pregel(
    nodes={"example_node": example_node},
    channels={
        "value": EphemeralValue(str),
    },
    input_channels=["value"],
    output_channels=["value"],
)

app.invoke({"value": "a"})

{'value': 'aaaaaaaaaaaaaaaa'}

高级 API

LangGraph 提供了两个用于创建 Pregel 应用程序的高级 API：StateGraph (Graph API) 和函数式 API。

StateGraph (Graph API)
函数式 API

StateGraph (Graph API) 是一个更高级别的抽象，简化了 Pregel 应用程序的创建。它允许你定义一个节点和边的图。当你编译该图时，StateGraph API 会自动为你创建 Pregel 应用程序。

from typing import TypedDict

from langgraph.constants import START
from langgraph.graph import StateGraph

class Essay(TypedDict):
    topic: str
    content: str | None
    score: float | None

def write_essay(essay: Essay):
    return {
        "content": f"Essay about {essay['topic']}",
    }

def score_essay(essay: Essay):
    return {
        "score": 10
    }

builder = StateGraph(Essay)
builder.add_node(write_essay)
builder.add_node(score_essay)
builder.add_edge(START, "write_essay")
builder.add_edge("write_essay", "score_essay")

# 编译图。
# 这将返回一个 Pregel 实例。
graph = builder.compile()

编译后的 Pregel 实例将与一个节点和通道列表相关联。你可以通过打印来检查节点和通道。

print(graph.nodes)

你会看到类似这样的内容：

{'__start__': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1810>,
 'write_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba14d0>,
 'score_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1710>}

print(graph.channels)

你应该会看到类似这样的内容

{'topic': <langgraph.channels.last_value.LastValue at 0x7d05e3294d80>,
 'content': <langgraph.channels.last_value.LastValue at 0x7d05e3295040>,
 'score': <langgraph.channels.last_value.LastValue at 0x7d05e3295980>,
 '__start__': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e3297e00>,
 'write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e32960c0>,
 'score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8ab80>,
 'branch:__start__:__self__:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e32941c0>,
 'branch:__start__:__self__:score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d88800>,
 'branch:write_essay:__self__:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e3295ec0>,
 'branch:write_essay:__self__:score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8ac00>,
 'branch:score_essay:__self__:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d89700>,
 'branch:score_essay:__self__:score_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8b400>,
 'start:write_essay': <langgraph.channels.ephemeral_value.EphemeralValue at 0x7d05e2d8b280>}

在函数式 API中，你可以使用 @entrypoint 来创建一个 Pregel 应用程序。entrypoint 装饰器允许你定义一个接受输入并返回输出的函数。

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint

class Essay(TypedDict):
    topic: str
    content: str | None
    score: float | None


checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def write_essay(essay: Essay):
    return {
        "content": f"Essay about {essay['topic']}",
    }

print("Nodes: ")
print(write_essay.nodes)
print("Channels: ")
print(write_essay.channels)

Nodes:
{'write_essay': <langgraph.pregel.read.PregelNode object at 0x7d05e2f9aad0>}
Channels:
{'__start__': <langgraph.channels.ephemeral_value.EphemeralValue object at 0x7d05e2c906c0>, '__end__': <langgraph.channels.last_value.LastValue object at 0x7d05e2c90c40>, '__previous__': <langgraph.channels.last_value.LastValue object at 0x7d05e1007280>}

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Capabilities

Production

Frontend

LangGraph APIs

概述

参与者

通道

LastValue

Topic

BinaryOperatorAggregate

DeltaChannel

批量归约器要求

使用 snapshot_frequency 限制读取延迟

版本兼容性和回滚

示例

高级 API

​概述

​参与者

​通道

​LastValue

​Topic

​BinaryOperatorAggregate

​DeltaChannel

​批量归约器要求

​使用 snapshot_frequency 限制读取延迟

​版本兼容性和回滚

​示例

​高级 API

概述

参与者

通道

LastValue

Topic

BinaryOperatorAggregate

DeltaChannel

批量归约器要求

使用 snapshot_frequency 限制读取延迟

版本兼容性和回滚

示例

高级 API