LangGraph 运行时 - Docs by LangChain中文

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

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

概述

在 LangGraph 中，Pregel 将 actor 和 channel 组合成一个应用程序。Actor 从 channel 中读取数据，并向 channel 写入数据。Pregel 按照 Pregel Algorithm/Bulk Synchronous Parallel 模型，将应用程序的执行组织为多个步骤。每个步骤包含三个阶段：

计划：确定此步骤中要执行哪些 actor。例如，在第一个步骤中，选择订阅特殊 input channel 的 actor；在后续步骤中，选择订阅上一步骤中已更新 channel 的 actor。
执行：并行执行所有被选中的 actor，直到全部完成、某个 actor 失败，或达到超时时间。在此阶段，channel 的更新对 actor 不可见，直到下一个步骤才可见。
更新：使用此步骤中 actor 写入的值更新 channel。

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

Actor

actor 是一个 PregelNode。它订阅 channel，从中读取数据，并向其中写入数据。可以将它理解为 Pregel 算法中的一个 actor。PregelNodes 实现了 LangChain 的 Runnable 接口。

Channel

Channel 用于 actor（PregelNodes）之间的通信。每个 channel 都有一个值类型、一个更新类型，以及一个更新函数——该函数接收一系列更新并修改已存储的值。Channel 可用于将数据从一条 chain 发送到另一条 chain，或在未来某个步骤中将数据从一条 chain 发送回自身。

LastValue

LastValue 是默认的 channel 类型。它会存储最后一次写入的值，并覆盖任何先前的值。可将它用于输入和输出值，或用于将数据从一个步骤传递到下一个步骤。

import { LastValue } from "@langchain/langgraph/channels";

const channel = new LastValue<number>();

Topic

Topic 是一个可配置的 PubSub channel，适合在 actor 之间发送多个值，或跨步骤累积输出。它可以配置为对值去重，或累积一次运行期间写入的所有值。

import { Topic } from "@langchain/langgraph/channels";

// Accumulate all values written across steps
const channel = new Topic<string>({ accumulate: true });

BinaryOperatorAggregate

BinaryOperatorAggregate 存储一个持久值，并通过将二元运算符应用于当前值和每个新更新来更新该值。可使用它跨步骤计算运行中的聚合结果。

import { BinaryOperatorAggregate } from "@langchain/langgraph/channels";

// Running total: each write adds to the current value
const total = new BinaryOperatorAggregate<number>({ operator: (a, b) => a + b });

示例

虽然大多数用户会通过 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 channel 来实现 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" });

此示例演示如何在图中引入一个循环，方式是让一条 chain 向它所订阅的 channel 写入数据。执行会持续进行，直到向该 channel 写入一个 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");

// Compile the graph.
// This will return a Pregel instance.
const graph = builder.compile();

编译后的 Pregel 实例会关联一组节点和 channel。你可以通过打印它们来查看这些节点和 channel。

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

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

​概述

​Actor

​Channel

​LastValue

​Topic

​BinaryOperatorAggregate

​示例

​高级 API

概述

Actor

Channel

LastValue

Topic

BinaryOperatorAggregate

示例

高级 API