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

​
概览

在 LangGraph 中，Pregel 将执行器和通道组合成一个应用程序。执行器从通道读取数据并向通道写入数据。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。
  • 单个节点
  • 多个节点
  • 主题
  • 二元运算符聚合
  • 循环
import { EphemeralValue } from "@langchain/langgraph/channels";
import { Pregel, NodeBuilder } from "@langchain/langgraph/pregel";

const node1 = new NodeBuilder()
  .subscribeOnly("a")
  .do((x: string) => x + x)
  .writeTo("b");

const app = new Pregel({
  nodes: { node1 },
  channels: {
    a: new EphemeralValue<string>(),
    b: new EphemeralValue<string>(),
  },
  inputChannels: ["a"],
  outputChannels: ["b"],
});

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

{ b: 'foofoo' }

​
高级 API

LangGraph 提供了两个用于创建 Pregel 应用程序的高级 API：StateGraph（图 API）和函数式 API。
  • StateGraph（图 API）
  • 函数式 API
StateGraph（图 API）是一个更高级别的抽象，它简化了 Pregel 应用程序的创建。它允许您定义节点和边的图。当您编译图时，StateGraph API 会自动为您创建 Pregel 应用程序。
import { START, StateGraph } from "@langchain/langgraph";

interface Essay {
  topic: string;
  content?: string;
  score?: number;
}

const writeEssay = (essay: Essay) => {
  return {
    content: `Essay about ${essay.topic}`,
  };
};

const scoreEssay = (essay: Essay) => {
  return {
    score: 10
  };
};

const builder = new StateGraph<Essay>({
  channels: {
    topic: null,
    content: null,
    score: null,
  }
})
  .addNode("writeEssay", writeEssay)
  .addNode("scoreEssay", scoreEssay)
  .addEdge(START, "writeEssay")
  .addEdge("writeEssay", "scoreEssay");

// Compile the graph.
// This will return a Pregel instance.
const graph = builder.compile();
编译后的 Pregel 实例将与节点和通道列表相关联。您可以通过打印它们来检查节点和通道。
console.log(graph.nodes);
您将看到类似以下内容
{
  __start__: PregelNode { ... },
  writeEssay: PregelNode { ... },
  scoreEssay: PregelNode { ... }
}

console.log(graph.channels);
您应该会看到类似以下内容
{
  topic: LastValue { ... },
  content: LastValue { ... },
  score: LastValue { ... },
  __start__: EphemeralValue { ... },
  writeEssay: EphemeralValue { ... },
  scoreEssay: EphemeralValue { ... },
  'branch:__start__:__self__:writeEssay': EphemeralValue { ... },
  'branch:__start__:__self__:scoreEssay': EphemeralValue { ... },
  'branch:writeEssay:__self__:writeEssay': EphemeralValue { ... },
  'branch:writeEssay:__self__:scoreEssay': EphemeralValue { ... },
  'branch:scoreEssay:__self__:writeEssay': EphemeralValue { ... },
  'branch:scoreEssay:__self__:scoreEssay': EphemeralValue { ... },
  'start:writeEssay': EphemeralValue { ... }
}
在 GitHub 上编辑此页面源文件。
以编程方式连接这些文档到 Claude、VSCode 等，通过 MCP 获取实时答案。
© . This site is unofficial and not affiliated with LangChain, Inc.