LangGraph 的运行时 (Pregel)¶

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

编译 StateGraph 或创建 entrypoint 会生成一个 Pregel 实例，该实例可以接受输入进行调用。

本指南从高层次解释了运行时，并提供了直接使用 Pregel 实现应用程序的说明。

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

概述¶

在 LangGraph 中，Pregel 将 参与者 和通道组合到一个应用程序中。参与者 从通道读取数据，并将数据写入通道。Pregel 按照 Pregel 算法/批量同步并行 模型将应用程序的执行组织成多个步骤。

每个步骤包含三个阶段

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

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

参与者¶

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

通道¶

通道用于在参与者 (PregelNodes) 之间通信。每个通道都有一个值类型、一个更新类型和一个更新函数——该函数接受一系列更新并修改存储的值。通道可用于将数据从一个链发送到另一个链，或在未来步骤中将数据从一个链发送给自己。LangGraph 提供了许多内置通道

基本通道：LastValue 和 Topic¶

LastValue：默认通道，存储发送到通道的最后一个值，适用于输入和输出值，或用于将数据从一个步骤发送到下一个步骤。
Topic：一个可配置的 PubSub 主题，适用于在参与者之间发送多个值，或用于累积输出。可以配置为去重值或在多个步骤中累积值。

高级通道：Context 和 BinaryOperatorAggregate¶

Context：暴露上下文管理器的值，管理其生命周期。适用于访问需要设置和/或清理的外部资源；例如，client = Context(httpx.Client)。
BinaryOperatorAggregate：存储一个持久值，通过将二元运算符应用于当前值和发送到通道的每个更新来更新，适用于计算多个步骤中的聚合；例如，total = BinaryOperatorAggregate(int, operator.add)

示例¶

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

以下是几个不同的示例，让您了解 Pregel API。

单节点多节点主题BinaryOperatorAggregate循环

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

node1 = (
    Channel.subscribe_to("a")
    | (lambda x: x + x)
    | Channel.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, Channel 

node1 = (
    Channel.subscribe_to("a")
    | (lambda x: x + x)
    | Channel.write_to("b")
)

node2 = (
    Channel.subscribe_to("b")
    | (lambda x: x + x)
    | Channel.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, Channel 

node1 = (
    Channel.subscribe_to("a")
    | (lambda x: x + x)
    | {
        "b": Channel.write_to("b"),
        "c": Channel.write_to("c")
    }
)

node2 = (
    Channel.subscribe_to("b")
    | (lambda x: x + x)
    | {
        "c": Channel.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 通道实现一个 reducer。

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


node1 = (
    Channel.subscribe_to("a")
    | (lambda x: x + x)
    | {
        "b": Channel.write_to("b"),
        "c": Channel.write_to("c")
    }
)

node2 = (
    Channel.subscribe_to("b")
    | (lambda x: x + x)
    | {
        "c": Channel.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"})

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

from langgraph.channels import EphemeralValue
from langgraph.pregel import Pregel, Channel, ChannelWrite, ChannelWriteEntry

example_node = (
    Channel.subscribe_to("value")
    | (lambda x: x + x if len(x) < 10 else None)
    | ChannelWrite(writes=[ChannelWriteEntry(channel="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（图 API）和函数式 API。

StateGraph（图 API）函数式 API

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

from typing import TypedDict, Optional

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

class Essay(TypedDict):
    topic: str
    content: Optional[str]
    score: Optional[float]

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

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

from typing import TypedDict, Optional

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

class Essay(TypedDict):
    topic: str
    content: Optional[str]
    score: Optional[float]


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

LangGraph 的运行时 (Pregel)¶

概述¶

参与者¶

通道¶

基本通道：LastValue 和 Topic¶

高级通道：Context 和 BinaryOperatorAggregate¶

示例¶

高级 API¶

评论