LangGraph 运行时

Pregel 实现了 LangGraph 的运行时，管理 LangGraph 应用的执行。

编译一个 StateGraph 或创建一个 @entrypoint 会生成一个 Pregel 实例，可以用输入来调用它。

本指南从高层介绍运行时，并提供直接使用 Pregel 实现应用的说明。

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

概述

在 LangGraph 中，Pregel 将actor和通道组合成单个应用。Actor 从通道读取数据并向通道写入数据。Pregel 将应用的执行组织为多个步骤，遵循 Pregel 算法/整体同步并行模型。

每个步骤包含三个阶段：

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

重复执行，直到没有选定要执行的 actor，或达到最大步骤数。

Actors

actor 是一个 PregelNode。它订阅通道，从中读取数据，并向其中写入数据。可以将它看作 Pregel 算法中的一个 actor。PregelNodes 实现了 LangChain 的 Runnable 接口。

频道

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

LastValue

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

from langgraph.channels import LastValue

channel: LastValue[int] = LastValue(int)

Topic

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

from langgraph.channels import Topic

# Accumulate all values written across steps
channel: Topic[str] = Topic(str, accumulate=True)

BinaryOperatorAggregate

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

import operator
from langgraph.channels import BinaryOperatorAggregate

# Running total: each write adds to the current value
total = BinaryOperatorAggregate(int, operator.add)

DeltaChannel

WARNING

DeltaChannel 要求 langgraph>=1.2，目前处于 beta 阶段。该 API 在未来的版本中可能会发生变化。

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

TIP

当一个通道既被频繁写入又随时间变得很大时，考虑使用 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 一起使用的按键归约器不同，后者的归约器在每次更新时调用一次。

WARNING

批量归约器必须满足结合律（批处理不变性）：

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

如果你的归约器不满足结合律，那么重建的状态可能会因 LangGraph 跨步骤对写入进行批处理的方式不同而不同，从而产生不一致的行为。

WARNING

归约器在重建时运行，而不是在写入时运行。 与 BinaryOperatorAggregate 不同——前者的归约器在写入时被调用，因此组合后的值会被序列化到检查点中——DeltaChannel 的归约器是在通道值从持久化的写入中重建时被调用的。被序列化的是每一步的原始写入；归约器只在值被实例化时才会被调用——在下一次读取时、在下一步骤的 actor 中，或在重放历史时。

设计归约器时的实际后果：

• 让它成为 (state, writes) 的纯函数。 任何副作用、随机性或挂钟时间读取（例如 uuid.uuid4()、datetime.now()）都会在每次重建值时执行，并在每次重放时产生不同的结果。它们不会被固化到持久化的写入中。
• 不要依赖对传入写入的变更会被持久化。 如果你的归约器修改了某个写入对象（例如，为一个到达时没有 ID 的条目分配一个稳定的 ID），该变更只存在于重建后的值中。所存储的写入仍保持原始形态，因此下一次重建时会再次看到未变更的输入。
• 在上游附加身份和其他稳定元数据。 如果下游代码需要在多轮之间通过 ID 引用某个条目（例如，以后更新或删除它），请在值被写入通道之前分配该 ID——而不是在归约器内部。

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

from typing import Any, Sequence

# List: append all writes in order
def list_reducer(state: list[Any], writes: Sequence[list[Any]]) -> list[Any]:
    result = list(state)
    for write in writes:
        result.extend(write)
    return result

# Dict: merge all writes, last write wins on key conflicts
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（默认值）完全跳过快照——适用于读取很少或线程很短的情况。

版本兼容性与回滚

WARNING

不建议将已持久化的通道从 DeltaChannel 更改为非 delta 通道。检查点对这些通道类型的编码方式不同，因此更改现有线程的通道类型可能导致状态重建不完整或不正确。请在线程的整个生命周期内保持通道定义稳定。在更改通道类型之前，请将受影响的线程迁移到新的表示形式，或者丢弃它们并启动新线程。

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

示例

虽然大多数用户会通过 StateGraph API 或 @entrypoint 装饰器与 Pregel 交互，但也可以直接与 Pregel 交互。

下面是几个不同的示例，让你感受一下 Pregel API。

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'}

高层 API

LangGraph 提供两个用于创建 Pregel 应用的高层 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")

# Compile the graph.
# This will return a Pregel instance.
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>}

---