跳到主要内容
Pregel 实现了 LangGraph 的运行时，管理 LangGraph 应用程序的执行。 编译 StateGraph 或创建 @entrypoint 会生成一个 Pregel 实例，该实例可以通过输入进行调用。 本指南从高层解释了运行时，并提供了使用 Pregel 直接实现应用程序的说明。
注意： Pregel 运行时以 Google 的 Pregel 算法 命名，该算法描述了一种使用图进行大规模并行计算的有效方法。

​
概览

在 LangGraph 中，Pregel 将 执行器（actors） 和 通道（channels） 组合成一个应用程序。 执行器从通道读取数据并向通道写入数据。Pregel 遵循 Pregel 算法/批量同步并行模型，将应用程序的执行组织成多个步骤。 每个步骤由三个阶段组成：
  • 规划：确定在此步骤中要执行哪些执行器。例如，在第一步中，选择订阅特殊输入通道的执行器；在后续步骤中，选择订阅在前一步中更新的通道的执行器。
  • 执行：并行执行所有选定的执行器，直到所有执行器完成，或其中一个失败，或达到超时。在此阶段，在下一步之前，通道更新对执行器不可见。
  • 更新：用在此步骤中执行器写入的值更新通道。
重复此过程，直到没有执行器被选中执行，或达到最大步数。

​
执行器

一个执行器是一个 PregelNode。它订阅通道，从它们读取数据，并向它们写入数据。它可以被认为是 Pregel 算法中的一个执行器。PregelNodes 实现了 LangChain 的 Runnable 接口。

​
通道

通道用于执行器（PregelNodes）之间的通信。每个通道都有一个值类型、一个更新类型和一个更新函数——它接受一系列更新并修改存储的值。通道可用于将数据从一个链发送到另一个链，或将数据从一个链发送到自身以供将来步骤使用。LangGraph 提供了许多内置通道。
  • LastValue：默认通道，存储发送到通道的最后一个值，适用于输入和输出值，或用于将数据从一个步骤发送到下一个步骤。
  • Topic：一个可配置的 PubSub 主题，适用于在执行器之间发送多个值，或用于累积输出。可以配置为对值进行去重或在多个步骤中累积值。
  • BinaryOperatorAggregate：存储一个持久值，通过将二元运算符应用于当前值和发送到通道的每个更新来更新，适用于计算多个步骤上的聚合；例如，total = BinaryOperatorAggregate(int, operator.add)

​
示例

虽然大多数用户将通过 StateGraph API 或 @entrypoint 装饰器与 Pregel 交互，但也可以直接与 Pregel 交互。 下面是一些不同的示例，让您了解 Pregel API。
  • 单节点
  • 多节点
  • 主题
  • 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'}

​
高级 API

LangGraph 提供了两种用于创建 Pregel 应用程序的高级 API：StateGraph (图 API) 和 函数式 API。
  • StateGraph (图 API)
  • 函数式 API
StateGraph (图 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>}
在 GitHub 上编辑此页面源文件。
以编程方式连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.