类 Pregel<Nodes, Channels, ConfigurableFieldType, InputType, OutputType>

Pregel 类是 LangGraph 的核心运行时引擎,实现了受 Google 的 Pregel 系统启发的基于消息传递的图计算模型。它为构建可靠、可控且能随时间演进状态的代理工作流提供了基础。

主要特性

  • 节点之间在离散的“超步”(supersteps)中传递消息
  • 通过检查点器内置持久化层
  • 对值、更新和事件的一流流式支持
  • 通过中断实现人机协作(human-in-the-loop)能力
  • 支持在超步内并行执行节点

Pregel 类不应由消费者直接实例化。请改用以下更高级别的 API

  • StateGraph:用于构建代理工作流的主要图类
  • 函数式 API:一种使用任务和入口点的声明式方法

示例

// Using StateGraph API
const graph = new StateGraph(annotation)
.addNode("nodeA", myNodeFunction)
.addEdge("nodeA", "nodeB")
.compile();

// The compiled graph is a Pregel instance
const result = await graph.invoke(input);

示例

// Using Functional API
import { task, entrypoint } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph-checkpoint";

// Define tasks that can be composed
const addOne = task("add", async (x: number) => x + 1);

// Create a workflow using the entrypoint function
const workflow = entrypoint({
name: "workflow",
checkpointer: new MemorySaver()
}, async (numbers: number[]) => {
// Tasks can be run in parallel
const results = await Promise.all(numbers.map(n => addOne(n)));
return results;
});

// The workflow is a Pregel instance
const result = await workflow.invoke([1, 2, 3]); // Returns [2, 3, 4]

类型参数

继承关系 (查看完整)

实现

属性

autoValidate: boolean

编译时是否自动验证图结构。默认为 true。

channels: Channels

图中的通道,将通道名称映射到其 BaseChannel 或 ManagedValueSpec 实例。

checkpointer?: false | BaseCheckpointSaver<number>

用于持久化图状态的可选检查点器。提供时,会在每个超步保存图状态的检查点。如果为 false 或 undefined,则禁用检查点功能,图将无法保存或恢复状态。

config?: LangGraphRunnableConfig<Record<string, any>>

图执行的默认配置,可以在每次调用时覆盖。

debug: boolean

是否启用调试日志记录。默认为 false。

inputChannels: keyof Channels | (keyof Channels)[]

图的输入通道。当图被调用时,这些通道接收初始输入。可以是单个通道键或通道键数组。

interruptAfter?: "*" | (keyof Nodes)[]

在执行这些节点后中断的可选节点名称数组或“all”。用于实现人机协作(human-in-the-loop)工作流。

interruptBefore?: "*" | (keyof Nodes)[]

在执行这些节点前中断的可选节点名称数组或“all”。用于实现人机协作(human-in-the-loop)工作流。

lc_kwargs: SerializedFields
lc_runnable: boolean
lc_serializable: boolean
name?: string
nodes: Nodes

图中的节点,将节点名称映射到其 PregelNode 实例。

outputChannels: keyof Channels | (keyof Channels)[]

图的输出通道。当图完成时,这些通道包含最终输出。可以是单个通道键或通道键数组。

retryPolicy?: RetryPolicy

处理节点执行失败的可选重试策略

stepTimeout?: number

每个超步执行的可选超时时间(毫秒)

store?: BaseStore

图的可选长期内存存储,允许在线程之间持久化和检索数据。

streamChannels?: keyof Channels | (keyof Channels)[]

要流式传输的可选通道。如果未指定,则流式传输所有通道。可以是单个通道键或通道键数组。

streamMode: StreamMode[]

为此图启用的流式模式。默认为 ["values"]。支持的模式:

  • "values":在每个步骤后流式传输完整状态
  • "updates":在每个步骤后流式传输状态更新
  • "messages":从节点内部流式传输消息
  • "custom":从节点内部流式传输自定义事件
  • "debug":流式传输与图执行相关的事件 - 有助于跟踪和调试图执行

访问器

  • get lc_aliases(): undefined | {
        [key: string]: string;
    }
  • 构造函数参数别名的映射。键是属性名,例如 "foo"。值是将在序列化中替换键的别名。这用于例如使参数名与 Python 匹配。

    返回 undefined | {
        [key: string]: string;
    }

  • get lc_attributes(): undefined | SerializedFields
  • 要与构造函数参数合并的附加属性的映射。键是属性名,例如 "foo"。值是要序列化的属性值。这些属性需要被构造函数接受为参数。

    返回 undefined | SerializedFields

  • get lc_id(): string[]
  • 模块的最终序列化标识符。

    返回 string[]

  • get lc_secrets(): undefined | {
        [key: string]: string;
    }
  • 密钥的映射,这些密钥将从序列化中省略。键是构造函数参数中密钥的路径,例如 "foo.bar.baz"。值是密钥 ID,将在反序列化时使用。

    返回 undefined | {
        [key: string]: string;
    }

  • get lc_serializable_keys(): undefined | string[]
  • 应序列化的手动键列表。如果未覆盖,则构造函数中传入的所有字段都将被序列化。

    返回 undefined | string[]

  • get streamChannelsAsIs(): keyof Channels | (keyof Channels)[]
  • 获取要以原始格式流式传输的通道。如果指定了 streamChannels,则按原样返回(单键或数组)。否则,返回图中的所有通道作为数组。

    返回 keyof Channels | (keyof Channels)[]

    要流式传输的通道键,可以是单个键或数组

  • get streamChannelsList(): (keyof Channels)[]
  • 获取应流式传输的所有通道的列表。如果指定了 streamChannels,则返回这些通道。否则,返回图中的所有通道。

    返回 (keyof Channels)[]

    要流式传输的通道键数组

方法

  • 内部方法,用于处理可运行对象的批处理和配置。它接受一个函数、输入值和可选配置,并返回一个解析为输出值的 Promise。

    类型参数

    • T

    参数

    返回 Promise<(Error | OutputType)[]>

    一个解析为输出值的 Promise。

  • 类型参数

    参数

    • options: Partial<O> | Partial<O>[]
    • 可选的 length: number

    返回 Partial<O>[]

  • 参数

    返回 AsyncGenerator<RunLogPatch, any, unknown>

  • 辅助方法,将输入值迭代器转换为输出值迭代器,并带有回调。使用此方法在 Runnable 子类中实现 stream()transform()

    类型参数

    • I
    • O

    参数

    • inputGenerator: AsyncGenerator<I, any, unknown>
    • transformer: ((generator, runManager?, options?) => AsyncGenerator<O, any, unknown>)
        • (generator, runManager?, options?): AsyncGenerator<I, any, unknown>
        • 参数

          返回 AsyncGenerator<O, any, unknown>

    • 可选的 options: Partial<PregelOptions<Nodes, Channels, ConfigurableFieldType>> & {
          runType?: string;
      }

    返回 AsyncGenerator<O, any, unknown>

  • 将可运行对象转换为工具。返回一个 RunnableToolLike 的新实例,其中包含可运行对象、名称、描述和 schema。

    类型参数

    参数

    • fields: {
          可选的 description?: string;
          可选的 name?: string;
          schema: ZodType<T, ZodTypeDef, T>;
      }
      • 可选的 description?: string

        工具的描述。如果未提供,则回退到 Zod schema 上的描述;如果两者都未提供,则为 undefined。

      • 可选的 name?: string

        工具的名称。如果未提供,则默认为可运行对象的名称。

      • schema: ZodType<T, ZodTypeDef, T>

        工具输入的 Zod schema。从可运行对象的输入类型推断 Zod 类型。

    返回 RunnableToolLike<ZodType<ToolCall | T, ZodTypeDef, ToolCall | T>, OutputType>

    一个 RunnableToolLike 的实例,它是一个可以用作工具的可运行对象。

  • 将新字段分配给此可运行对象的字典输出。返回一个新的可运行对象。

    参数

    • mapping: RunnableMapLike<Record<string, unknown>, Record<string, unknown>>

    返回 Runnable<any, any, RunnableConfig<Record<string, any>>>

  • 参数

    • 可选 _: RunnableConfig<Record<string, any>>

    返回 Graph

  • 获取图结构的可绘制表示。这是 getGraph() 的异步版本,是推荐使用的方法。

    参数

    • config: RunnableConfig<Record<string, any>>

      生成图可视化配置

    返回 Promise<Graph>

    一个可以可视化的图表示

  • 参数

    • 可选 suffix: string

    返回 string

  • 获取图的当前状态。需要配置检查点。

    参数

    • config: RunnableConfig<Record<string, any>>

      获取状态的配置

    • 可选 options: GetStateOptions

      附加选项

    返回 Promise<StateSnapshot>

    当前图状态的快照

    抛出

    如果未配置检查点

  • 获取图状态的历史记录。需要配置检查点。适用于

    • 调试执行历史
    • 实现时间旅行
    • 分析图的行为

    参数

    • config: RunnableConfig<Record<string, any>>

      获取历史记录的配置

    • 可选 options: CheckpointListOptions

      过滤历史记录的选项

    返回 AsyncIterableIterator<StateSnapshot>

    状态快照的异步迭代器

    抛出

    如果未配置检查点

  • 获取此图中的所有子图。子图是嵌套在此图节点中的 Pregel 实例。

    参数

    • 可选 namespace: string

      用于过滤子图的可选命名空间

    • 可选 recurse: boolean

      是否递归获取子图的子图

    返回 Generator<[string, Pregel<any, any, StrRecord<string, any>, any, any>], any, unknown>

    生成器,产生 [名称, 子图] 的元组

    已废弃

    请改用 getSubgraphsAsync。异步方法将在下一个次要版本中成为默认方法。

  • 异步获取此图中的所有子图。子图是嵌套在此图节点中的 Pregel 实例。

    参数

    • 可选 namespace: string

      用于过滤子图的可选命名空间

    • 可选 recurse: boolean

      是否递归获取子图的子图

    返回 AsyncGenerator<[string, Pregel<any, any, StrRecord<string, any>, any, any>], any, unknown>

    AsyncGenerator,产生 [名称, 子图] 的元组

  • 从此 runnable 的字典输出中选取键。返回一个新的 runnable。

    参数

    • keys: string | string[]

    返回 Runnable<any, any, RunnableConfig<Record<string, any>>>

  • 创建一个新的 runnable 序列,按顺序运行每个 runnable,将一个 runnable 的输出导入另一个 runnable 或类似 runnable 的对象。

    类型参数

    • NewRunOutput

    参数

    • coerceable: RunnableLike<OutputType, NewRunOutput, RunnableConfig<Record<string, any>>>

      一个 runnable、函数或其值为函数或 runnable 的对象。

    返回 Runnable<null | InputType | Command<unknown>, Exclude<NewRunOutput, Error>, RunnableConfig<Record<string, any>>>

    一个新的 runnable 序列。

  • 流式传输图的执行过程,并在状态更新发生时发出。这是实时观察图执行的主要方法。

    流模式

    • "values":每一步后发出完整状态
    • "updates":每一步后仅发出状态变化
    • "debug":发出详细的调试信息
    • "messages":发出节点内的消息

    更多详情,请参见 流传输操作指南

    参数

    返回 Promise<IterableReadableStream<any>>

    图状态更新的异步可迭代流

  • 参数

    返回 IterableReadableStream<StreamEvent>

    继承文档

  • 参数

    返回 IterableReadableStream<Uint8Array>

  • 从一个 runnable 流式传输所有输出,这些输出会报告给回调系统。这包括 LLMs、Retrievers、Tools 等所有内部运行。输出以 Log 对象的形式流式传输,Log 对象包含一个 jsonpatch 操作列表,该列表描述了运行状态在每个步骤中如何变化,以及运行的最终状态。可以按顺序应用 jsonpatch 操作来构建状态。

    参数

    返回 AsyncGenerator<RunLogPatch, any, unknown>

  • 返回 Serialized

  • 返回 SerializedNotImplemented

  • transform 的默认实现,它会缓冲输入然后调用 stream。如果子类可以在输入仍在生成时开始产生输出,则应覆盖此方法。

    参数

    返回 AsyncGenerator<OutputType, any, unknown>

  • 使用新值更新图的状态。需要配置一个检查点。

    此方法可用于

    • 实现人工参与(human-in-the-loop)的工作流程
    • 在断点处修改图状态
    • 将外部输入集成到图中

    参数

    • inputConfig: LangGraphRunnableConfig<Record<string, any>>

      更新的配置

    • values: unknown

      用于更新状态的值

    • 可选 asNode: string | keyof Nodes

      可选的节点名称,用于指定更新归属于哪个节点

    返回 Promise<RunnableConfig<Record<string, any>>>

    更新后的配置

    抛出

    如果未配置检查点

    抛出

    如果更新无法归属于某个节点

  • 验证图结构以确保其格式正确。检查以下项:

    • 没有孤立节点
    • 有效的输入/输出通道配置
    • 有效的中断配置

    返回 this

    this - 用于方法链的 Pregel 实例

    抛出

    如果图结构无效

  • 使用更新的配置创建 Pregel 图的新实例。此方法遵循不可变模式 - 它不会修改当前实例,而是返回一个包含合并配置的新实例。

    参数

    • config: RunnableConfig<Record<string, any>>

      要与当前配置合并的配置

    返回 Pregel<Nodes, Channels, ConfigurableFieldType, InputType, OutputType>

    一个带有合并配置的新 Pregel 实例

    示例

    // Create a new instance with debug enabled
    const debugGraph = graph.withConfig({ debug: true });

    // Create a new instance with a specific thread ID
    const threadGraph = graph.withConfig({
    configurable: { thread_id: "123" }
    });
  • 从当前 runnable 创建一个新的 runnable,如果初始调用失败,它将尝试调用其他传入的备用 (fallback) runnables。

    参数

    返回 RunnableWithFallbacks<null | InputType | Command<unknown>, OutputType>

    一个新的 RunnableWithFallbacks。

  • 将生命周期监听器绑定到一个 Runnable,返回一个新的 Runnable。Run 对象包含关于运行的信息,包括其 id、类型、输入、输出、错误、startTime、endTime 以及添加到运行中的任何标签或元数据。

    参数

    • params: {
          onEnd?: ((run, config?) => void | Promise<void>);
          onError?: ((run, config?) => void | Promise<void>);
          onStart?: ((run, config?) => void | Promise<void>);
      }

      包含回调函数的对象。

      • 可选 onEnd?: ((run, config?) => void | Promise<void>)

        runnable 运行结束时调用,传入 Run 对象。

          • (run, config?): void | Promise<void>
          • 参数

            • run: Run
            • 可选 config: RunnableConfig<Record<string, any>>

            返回 void | Promise<void>

      • 可选 onError?: ((run, config?) => void | Promise<void>)

        如果 runnable 抛出错误时调用,传入 Run 对象。

          • (run, config?): void | Promise<void>
          • 参数

            • run: Run
            • 可选 config: RunnableConfig<Record<string, any>>

            返回 void | Promise<void>

      • 可选 onStart?: ((run, config?) => void | Promise<void>)

        runnable 开始运行前调用,传入 Run 对象。

          • (run, config?): void | Promise<void>
          • 参数

            • run: Run
            • 可选 config: RunnableConfig<Record<string, any>>

            返回 void | Promise<void>

    返回 Runnable<null | InputType | Command<unknown>, OutputType, PregelOptions<Nodes, Channels, ConfigurableFieldType>>

  • 参数

    • thing: any

    返回 thing is Runnable<any, any, RunnableConfig<Record<string, any>>>