LangGraph 运行时

Pregel 实现了 LangGraph 的运行时，管理 LangGraph 应用程序的执行。编译 StateGraph 或创建 entrypoint 会生成一个 Pregel 实例，该实例可以通过输入进行调用。本指南从高层次解释了运行时，并提供了直接使用 Pregel 实现应用程序的说明。

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

概览

在 LangGraph 中，Pregel 将 参与者 (actors) 和 通道 (channels) 结合到一个单一的应用程序中。参与者 从通道读取数据并向通道写入数据。Pregel 按照 Pregel 算法/批量同步并行 (Bulk Synchronous Parallel) 模型将应用程序的执行组织成多个步骤。每个步骤由三个阶段组成：

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

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

参与者 (Actors)

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

通道 (Channels)

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

LastValue：默认通道，存储发送到通道的最后一个值，对于输入和输出值或将数据从一个步骤发送到下一个步骤非常有用。
Topic：可配置的 PubSub 主题，对于在 参与者 之间发送多个值或累积输出非常有用。可以配置为去重值或在多个步骤的过程中累积值。
BinaryOperatorAggregate：存储持久值，通过将二元运算符应用于当前值和发送到通道的每个更新来更新，对于计算多个步骤的聚合非常有用；例如，total = BinaryOperatorAggregate(int, operator.add)

示例

虽然大多数用户将通过 StateGraph API 或 entrypoint 装饰器与 Pregel 交互，但也可以直接与 Pregel 交互。以下是几个不同的示例，让您了解 Pregel API。

单个节点
多个节点
主题 (Topic)
BinaryOperatorAggregate
循环 (Cycle)

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 }

import { LastValue, 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 node2 = new NodeBuilder()
  .subscribeOnly("b")
  .do((x: string) => x + x)
  .writeTo("c");

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

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

{ b: foofoo, c: foofoofoofoo }

import { EphemeralValue, Topic } from "@langchain/langgraph/channels";
import { Pregel, NodeBuilder } from "@langchain/langgraph/pregel";

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

const node2 = new NodeBuilder()
  .subscribeTo("b")
  .do((x: { b: string }) => x.b + x.b)
  .writeTo("c");

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

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

{ c: [foofoo, foofoofoofoo] }

此示例演示如何使用 BinaryOperatorAggregate 通道来实现归约器 (reducer)。

import { EphemeralValue, BinaryOperatorAggregate } from "@langchain/langgraph/channels";
import { Pregel, NodeBuilder } from "@langchain/langgraph/pregel";

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

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

const reducer = (current: string, update: string) => {
  if (current) {
    return current + " | " + update;
  } else {
    return update;
  }
};

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

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

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

import { EphemeralValue } from "@langchain/langgraph/channels";
import { Pregel, NodeBuilder, ChannelWriteEntry } from "@langchain/langgraph/pregel";

const exampleNode = new NodeBuilder()
  .subscribeOnly("value")
  .do((x: string) => x.length < 10 ? x + x : null)
  .writeTo(new ChannelWriteEntry("value", { skipNone: true }));

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

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

{ value: aaaaaaaaaaaaaaaa }

高级 API

LangGraph 提供了两个用于创建 Pregel 应用程序的高级 API：StateGraph (Graph API) 和 Functional API。

StateGraph (Graph API)
Functional API

StateGraph (Graph 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");

// 编译图。
// 这将返回一个 Pregel 实例。
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 { ... }
}

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

import { MemorySaver } from "@langchain/langgraph";
import { entrypoint } from "@langchain/langgraph/func";

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

const checkpointer = new MemorySaver();

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

console.log("Nodes: ");
console.log(writeEssay.nodes);
console.log("Channels: ");
console.log(writeEssay.channels);

Nodes:
{ writeEssay: PregelNode { ... } }
Channels:
{
  __start__: EphemeralValue { ... },
  __end__: LastValue { ... },
  __previous__: LastValue { ... }
}

在 GitHub 上编辑本页或提交问题.

将这些文档连接到 Claude、VSCode 等，通过 MCP 获取实时解答。

Get started

Capabilities

Production

LangGraph APIs

概览

参与者 (Actors)

通道 (Channels)

示例

高级 API

Get started

Capabilities

Production

LangGraph APIs

​概览

​参与者 (Actors)

​通道 (Channels)

​示例

​高级 API

概览

参与者 (Actors)

通道 (Channels)

示例

高级 API