LangGraph 运行时

Pregel 实现了 LangGraph 的运行时，负责管理 LangGraph 应用的执行。编译一个状态图或创建一个入口点会生成一个 Pregel 实例，该实例可以使用输入进行调用。本指南从高层次解释了运行时，并提供了直接使用Pregel实现应用的说明。

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

概述

在LangGraph中，Pregel将actors和channels合并成一个单一的应用程序。Actors从channels读取数据并将数据写入channels。Pregel将应用程序的执行组织成多个步骤，遵循Pregel Algorithm/Bulk Synchronous Parallel模型。每个步骤包含三个阶段：

计划：确定在此步骤中要执行哪些演员。例如，在第一步中，选择订阅特殊输入通道的演员；在后续步骤中，选择订阅上一步更新通道的演员。
执行：并行执行所有选定的演员，直到所有完成、其中一个失败或达到超时。在此阶段，通道更新对演员来说是不可见的，直到下一个步骤。
更新：使用在此步骤中由演员写入的值更新通道。

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

演员

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

通道

通道用于在演员（PregelNodes）之间进行通信。每个通道都有一个值类型、一个更新类型和一个更新函数——该函数接受一系列更新并修改存储的值。通道可以用来将数据从一个链发送到另一个链，或者在未来步骤中将数据从链发送到链本身。LangGraph提供了一些内置通道：

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

示例

虽然大多数用户将通过StateGraph API或entrypoint装饰器与Pregel交互，但也可以直接与Pregel进行交互。以下是一些不同示例，以帮助您了解Pregel API。

Single node
Multiple nodes
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 通道来实现一个归约器。

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（图API）和功能API。

StateGraph (Graph API)
Functional API

状态图（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");

// 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 { ... }
}

在功能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上编辑此页面的源代码。

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

LangGraph v1.0

开始使用

功能特性

生产环境

LangGraph APIs

概述

演员

通道

示例

高级API

LangGraph v1.0

开始使用

功能特性

生产环境

LangGraph APIs

​概述

​演员

​通道

​示例

​高级API

概述

演员

通道

示例

高级API