Skip to main content
Pregel 实现了 LangGraph 的运行时,负责管理 LangGraph 应用的执行。 编译一个 状态图 或创建一个 @entrypoint 会生成一个 Pregel 实例,该实例可以接受输入进行调用。 本指南从高层次解释了运行时,并提供了直接使用Pregel实现应用的说明。
注意: Pregel 运行时以 Google的Pregel算法 命名,该算法描述了一种使用图进行大规模并行计算的高效方法。

概述

在LangGraph中,Pregel将actorschannels合并成一个单一的应用程序。Actors从channels读取数据并将数据写入channels。Pregel将应用程序的执行组织成多个步骤,遵循Pregel Algorithm/Bulk Synchronous Parallel模型。 每个步骤包含三个阶段:
  • 计划:确定在此步骤中要执行哪些演员。例如,在第一步中,选择订阅特殊输入通道的演员;在后续步骤中,选择订阅上一步更新通道的演员
  • 执行:并行执行所有选定的演员,直到所有完成、其中一个失败或达到超时。在此阶段,通道更新对演员来说是不可见的,直到下一个步骤。
  • 更新:使用在此步骤中由演员写入的值更新通道。
重复执行,直到没有 演员 被选中执行,或者达到最大步骤数。

演员

一个 actor 是一个 PregelNode。它订阅频道,从它们中读取数据,并将数据写入它们。它可以被视为 Pregel 算法中的 actorPregelNodes 实现了 LangChain 的 Runnable 接口。

通道

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

示例

虽然大多数用户将通过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(图API)功能API
状态图(Graph API) 是一种高级抽象,简化了 Pregel 应用的创建。它允许您定义节点和边的图。当您编译图时,状态图 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")

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