尚拙
Search K

Appearance

LangGraph 运行时（Pregel Runtime） ​

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

编译 StateGraph 或创建 entrypoint 会生成一个可以使用输入调用的 Pregel 实例。

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

INFO

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

概览 ​

在 LangGraph 中，Pregel 将参与者（actors）和通道（channels）组合到单个应用程序中。参与者从通道读取数据并向通道写入数据。Pregel 遵循 Pregel 算法 / 批量同步并行（Bulk Synchronous Parallel）模型，将应用程序的执行组织为多个步骤。

每个步骤由三个阶段组成：

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

重复执行，直到没有选择任何参与者执行，或者达到最大步数。

参与者（Actors） ​

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

通道（Channels） ​

通道用于在参与者（PregelNode）之间进行通信。每个通道都有一个值类型、一个更新类型和一个更新函数——该函数接收一系列更新并修改存储的值。通道可用于将数据从一个链发送到另一个链，或者在将来的步骤中将数据从链发送到其自身。

LastValue ​

LastValue 是默认的通道类型。它存储写入它的最后一个值，覆盖任何先前的值。用它来表示输入和输出值，或者用于将数据从一个步骤传递到下一个步骤。

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

const channel = new LastValue<number>();

Topic ​

Topic 是一个可配置的发布订阅（PubSub）通道，可用于在参与者之间发送多个值或累加跨步骤的输出。它可以配置为去重值，或者累加运行期间写入的所有值。

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

// 累加跨步骤写入的所有值
const channel = new Topic<string>({ accumulate: true });

BinaryOperatorAggregate ​

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

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

// 运行总计：每次写入都会加到当前值上
const total = new BinaryOperatorAggregate<number>({ operator: (a, b) => a + b });

示例 ​

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

下面提供了几个不同的示例，以便你了解 Pregel API。

单节点（Single node） ​

typescript
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" });
console
{ b: 'foofoo' }

多节点（Multiple nodes） ​

typescript
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" });
console
{ b: 'foofoo', c: 'foofoofoofoo' }

Topic 通道 ​

typescript
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" });
console
{ c: ['foofoo', 'foofoofoofoo'] }

BinaryOperatorAggregate 通道 ​

此示例演示如何使用 BinaryOperatorAggregate 通道实现 reducer。

typescript
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" });

循环（Cycle） ​

此示例演示如何通过让链写入其订阅的通道在图中引入循环。执行将继续，直到向通道写入 null 值为止。

typescript
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" });
console
{ value: 'aaaaaaaaaaaaaaaa' }

高级 API ​

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

StateGraph (Graph API) ​

StateGraph (Graph API) 是一个简化 Pregel 应用程序创建的高阶抽象。它允许你定义节点和边的图。当你编译图时，StateGraph API 会自动为你创建 Pregel 应用程序。

typescript
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 实例将与节点和通道列表相关联。你可以通过打印节点和通道来检查它们。

typescript
console.log(graph.nodes);

你将看到类似以下内容：

console
{
  __start__: PregelNode { ... },
  writeEssay: PregelNode { ... },
  scoreEssay: PregelNode { ... }
}
typescript
console.log(graph.channels);

你应该看到类似以下内容：

console
{
  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 ​

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

typescript
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);
console
Nodes:
{ writeEssay: PregelNode { ... } }
Channels:
{
  __start__: EphemeralValue { ... },
  __end__: LastValue { ... },
  __previous__: LastValue { ... }
}