类 CompiledGraph<N, State, Update, ConfigurableFieldType, InputType, OutputType>

Pregel 类是 LangGraph 的核心运行时引擎，它实现了一种受 Google Pregel 系统 启发的 MPC（消息传递计算）图计算模型。它为构建可靠、可控的智能体工作流奠定了基础，这些工作流可以随时间演变状态。

主要特性

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

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]

类型参数

  • N extends string

    节点名称到其 PregelNode 实现的映射

  • State = any

    通道名称到其 BaseChannel 或 ManagedValueSpec 实现的映射

  • Update = any

    可传递给图的可配置字段类型

  • ConfigurableFieldType extends Record<string, any> = Record<string, any>

    图接受的输入值类型

  • InputType = any

    图生成的输出值类型

  • OutputType = any

继承层级 (查看完整)

构造函数

constructor

属性

NodeType RunInput RunOutput autoValidate builder cache? channels checkpointer? config? debug inputChannels interruptAfter? interruptBefore? lc_kwargs lc_runnable lc_serializable name? nodes outputChannels retryPolicy? stepTimeout? store? streamChannels? streamMode triggerToNodes

访问器

lc_aliases lc_attributes lc_id lc_secrets lc_serializable_keys streamChannelsAsIs streamChannelsList

方法

_batchWithConfig _callWithConfig _getOptionsList _separateRunnableConfigFromCallOptions _streamLog _transformStreamWithConfig asTool assign attachBranch attachEdge attachNode batch bind clearCache getGraph getGraphAsync getName getState getStateHistory getSubgraphs getSubgraphsAsync invoke map pick pipe stream streamEvents streamLog toJSON toJSONNotImplemented transform updateState validate withConfig withFallbacks withListeners withRetry isRunnable

构造函数

constructor

属性

NodeType

NodeType: N

RunInput

RunInput: State

RunOutput

RunOutput: Update

autoValidate

: boolean

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

builder

: Graph<N, State, Update, NodeSpec<State, Update>, StateDefinition>

可选 cache

?: BaseCache<unknown>

图的可选缓存，用于缓存任务。

channels

: Record<string | N, BaseChannel<unknown, unknown, unknown>>

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

可选 checkpointer

?: false | BaseCheckpointSaver<number>

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

可选 config

?: LangGraphRunnableConfig<Record<string, any>>

图执行的默认配置，可以按每次调用进行覆盖

debug

: boolean

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

inputChannels

: string | N | (string | N)[]

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

可选 interruptAfter

?: "*" | ("__start__" | N)[]

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

可选 interruptBefore

?: "*" | ("__start__" | N)[]

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

lc_kwargs

: SerializedFields

受保护的 lc_runnable

: boolean

lc_serializable

: boolean

可选 name

?: string

nodes

: Record<"__start__" | N, PregelNode<State, Update>>

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

outputChannels

: string | N | (string | N)[]

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

可选 retryPolicy

?: RetryPolicy

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

可选 stepTimeout

?: number

每个超步执行的可选超时时间（毫秒）

可选 store

?: BaseStore

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

可选 streamChannels

?: string | N | (string | N)[]

可选的要流式传输的通道。如果未指定，将流式传输所有通道。可以是一个单一通道键，也可以是通道键的数组。

streamMode

: StreamMode[]

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

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

triggerToNodes

: Record<string, string[]>

访问器

lc_aliases

  • get lc_aliases(): undefined | {
        [key: string]: string;
    }

  • 构造函数参数的别名映射。键是属性名称，例如“foo”。值是将在序列化中替换键的别名。这用于例如使参数名称与 Python 匹配。

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

lc_attributes

  • get lc_attributes(): undefined | SerializedFields

  • 要与构造函数参数合并的额外属性映射。键是属性名称，例如“foo”。值是属性值，将被序列化。这些属性需要被构造函数接受为参数。

    返回 undefined | SerializedFields

lc_id

  • get lc_id(): string[]

  • 模块的最终序列化标识符。

    返回 string[]

lc_secrets

  • get lc_secrets(): undefined | {
        [key: string]: string;
    }

  • 秘密的映射，这些秘密将从序列化中省略。键是构造函数参数中秘密的路径，例如“foo.bar.baz”。值是秘密 ID，将在反序列化时使用。

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

lc_serializable_keys

  • get lc_serializable_keys(): undefined | string[]

  • 应被序列化的键的手动列表。如果未被覆盖，则所有传递给构造函数的字段都将被序列化。

    返回 undefined | string[]

streamChannelsAsIs

  • get streamChannelsAsIs(): keyof Channels | (keyof Channels)[]

  • 获取以原始格式流式传输的通道。如果指定了 streamChannels，则按原样返回（单个键或数组）。否则，将图中所有通道作为数组返回。

    返回 keyof Channels | (keyof Channels)[]

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

streamChannelsList

  • get streamChannelsList(): (keyof Channels)[]

  • 获取应流式传输的所有通道列表。如果指定了 streamChannels，则返回这些通道。否则，返回图中所有通道。

    返回 (keyof Channels)[]

    要流式传输的通道键数组

方法

_batchWithConfig

受保护的 _callWithConfig

受保护的 _getOptionsList

受保护的 _separateRunnableConfigFromCallOptions

受保护 _streamLog

受保护 _transformStreamWithConfig

  • _transformStreamWithConfig<I, O>(inputGenerator, transformer, options?): AsyncGenerator<O, any, any>

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

    类型参数

    • I
    • O

    参数

    返回 AsyncGenerator<O, any, any>

asTool

  • asTool<T>(fields): RunnableToolLike<ZodType<ToolCall | T, ZodTypeDef, ToolCall | T>, OutputType>

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

    类型参数

    参数

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

        工具的描述。如果未提供，则回退到 Zod 模式上的描述；如果两者都未提供，则为 undefined。

      • 可选 name?: string

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

      • schema: ZodType<T, ZodTypeDef, T>

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

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

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

assign

  • assign(mapping): Runnable<any, any, RunnableConfig<Record<string, any>>>

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

    参数

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

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

attachBranch

  • attachBranch(start, name, branch): void

  • 参数

    返回 void

attachEdge

  • attachEdge(start, end): void

  • 参数

    • start: "__start__" | N
    • end: "__end__" | N

    返回 void

attachNode

  • attachNode(key, node): void

  • 参数

    返回 void

batch

bind

clearCache

  • clearCache(): Promise<void>

  • 返回 Promise<void>

getGraph

  • getGraph(config?): Graph

  • 返回计算图的可绘制表示。

    参数

    • 可选 config: RunnableConfig<Record<string, any>> & {
          xray?: number | boolean;
      }

    返回 Graph

    已弃用

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

getGraphAsync

  • getGraphAsync(config?): Promise<Graph>

  • 返回计算图的可绘制表示。

    参数

    • 可选 config: RunnableConfig<Record<string, any>> & {
          xray?: number | boolean;
      }

    返回 Promise<Graph>

getName

  • getName(suffix?): string

  • 参数

    • 可选 suffix: string

    返回 string

getState

  • getState(config, options?): Promise<StateSnapshot>

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

    参数

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

      用于检索状态的配置。

    • 可选 options: GetStateOptions

      附加选项

    返回 Promise<StateSnapshot>

    当前图状态的快照。

    抛出

    如果未配置检查点

getStateHistory

  • getStateHistory(config, options?): AsyncIterableIterator<StateSnapshot, any, any>

  • 获取图状态的历史记录。需要配置检查点。对于以下情况很有用：

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

    参数

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

      用于检索历史记录的配置。

    • 可选 options: CheckpointListOptions

      过滤历史记录的选项。

    返回 AsyncIterableIterator<StateSnapshot, any, any>

    状态快照的异步迭代器。

    抛出

    如果未配置检查点

getSubgraphs

  • getSubgraphs(namespace?, recurse?): Generator<[string, Pregel<any, any, StrRecord<string, any>, any, any, any, any>], any, any>

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

    参数

    • 可选 namespace: string

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

    • 可选 recurse: boolean

      是否递归获取子图的子图

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

    生成器，产生 [name, subgraph] 的元组。

    已弃用

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

getSubgraphsAsync

  • getSubgraphsAsync(namespace?, recurse?): AsyncGenerator<[string, Pregel<any, any, StrRecord<string, any>, any, any, any, any>], any, any>

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

    参数

    • 可选 namespace: string

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

    • 可选 recurse: boolean

      是否递归获取子图的子图

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

    异步生成器，产生 [name, subgraph] 的元组。

invoke

map

pick

  • pick(keys): Runnable<any, any, RunnableConfig<Record<string, any>>>

  • 从此可运行对象的字典输出中选择键。返回一个新的可运行对象。

    参数

    • keys: string | string[]

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

pipe

  • pipe<NewRunOutput>(coerceable): Runnable<null | InputType | Command<unknown, Record<string, unknown>, string>, Exclude<NewRunOutput, Error>, RunnableConfig<Record<string, any>>>

  • 创建一个新的可运行序列，该序列按顺序运行每个独立的运行对象，并将一个运行对象的输出作为另一个运行对象或类运行对象的输入。

    类型参数

    • NewRunOutput

    参数

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

      一个可运行对象、函数或其值为函数或可运行对象的对象。

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

    一个新的可运行序列。

stream

streamEvents

  • streamEvents(input, options, streamOptions?): IterableReadableStream<StreamEvent>

  • 参数

    返回 IterableReadableStream<StreamEvent>

    继承文档

  • streamEvents(input, options, streamOptions?): IterableReadableStream<Uint8Array<ArrayBufferLike>>

  • 参数

    返回 IterableReadableStream<Uint8Array<ArrayBufferLike>>

streamLog

  • streamLog(input, options?, streamOptions?): AsyncGenerator<RunLogPatch, any, any>

  • 流式传输可运行对象的所有输出，这些输出将报告给回调系统。这包括LLMs、Retrievers、Tools等的所有内部运行。输出以日志对象（Log objects）的形式流式传输，其中包含一个jsonpatch操作列表，描述了运行状态在每一步中如何变化，以及运行的最终状态。可以应用jsonpatch操作来构建状态。

    参数

    返回 AsyncGenerator<RunLogPatch, any, any>

toJSON

  • toJSON(): Serialized

  • 返回 Serialized

toJSONNotImplemented

  • toJSONNotImplemented(): SerializedNotImplemented

  • 返回 SerializedNotImplemented

transform

  • transform(generator, options): AsyncGenerator<OutputType, any, any>

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

    参数

    返回 AsyncGenerator<OutputType, any, any>

updateState

  • updateState(inputConfig, values, asNode?): Promise<RunnableConfig<Record<string, any>>>

  • 使用新值更新图的状态。需要配置检查点（checkpointer）。

    此方法可用于

    • 实现人在回路（human-in-the-loop）工作流
    • 在断点期间修改图状态
    • 将外部输入集成到图中

    参数

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

      更新的配置

    • values: unknown

      用于更新状态的值

    • 可选 asNode: string | N

      可选的节点名称，用于归因更新

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

    更新后的配置

    抛出

    如果未配置检查点

    抛出

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

validate

  • validate(): this

  • 验证图结构以确保其格式正确。检查以下内容：

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

    返回 this

    this - 用于方法链的 Pregel 实例

    抛出

    如果图结构无效

withConfig

  • withConfig(config): CompiledGraph<N, State, Update, ConfigurableFieldType, InputType, OutputType>

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

    参数

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

      要与当前配置合并的配置

    返回 CompiledGraph<N, State, Update, 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" }
    });

withFallbacks

  • withFallbacks(fields): RunnableWithFallbacks<null | InputType | Command<unknown, Record<string, unknown>, string>, OutputType>

  • 从当前可运行对象创建一个新的可运行对象，如果初始调用失败，该对象将尝试调用其他传入的备用可运行对象。

    参数

    • fields: {
          fallbacks: Runnable<null | InputType | Command<unknown, Record<string, unknown>, string>, OutputType, RunnableConfig<Record<string, any>>>[];
      } | Runnable<null | InputType | Command<unknown, Record<string, unknown>, string>, OutputType, RunnableConfig<Record<string, any>>>[]

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

    一个新的 RunnableWithFallbacks。

withListeners

  • withListeners(params): Runnable<null | InputType | Command<unknown, Record<string, unknown>, string>, OutputType, PregelOptions<Record<"__start__" | N, PregelNode<State, Update>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, ConfigurableFieldType & Record<string, any>, undefined | StreamMode | StreamMode[], boolean>>

  • 将生命周期监听器绑定到可运行对象，并返回一个新的可运行对象。Run 对象包含有关运行的信息，包括其 ID、类型、输入、输出、错误、开始时间、结束时间，以及添加到运行中的任何标签或元数据。

    参数

    • 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>)

        在可运行对象运行结束后调用，传入 Run 对象。

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

          • 参数

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

            返回 void | Promise<void>

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

        如果可运行对象抛出错误，则调用此函数，传入 Run 对象。

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

          • 参数

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

            返回 void | Promise<void>

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

        在可运行对象开始运行之前调用，传入 Run 对象。

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

          • 参数

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

            返回 void | Promise<void>

    返回 Runnable<null | InputType | Command<unknown, Record<string, unknown>, string>, OutputType, PregelOptions<Record<"__start__" | N, PregelNode<State, Update>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, ConfigurableFieldType & Record<string, any>, undefined | StreamMode | StreamMode[], boolean>>

withRetry

静态 isRunnable

  • isRunnable(thing): thing is Runnable<any, any, RunnableConfig<Record<string, any>>>

  • 参数

    • thing: any

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

设置

成员可见性

主题

在本页

constructorNodeTypeRunInputRunOutputautoValidatebuildercachechannelscheckpointerconfigdebuginputChannelsinterruptAfterinterruptBeforelc_kwargslc_runnablelc_serializablenamenodesoutputChannelsretryPolicystepTimeoutstorestreamChannelsstreamModetriggerToNodeslc_aliaseslc_attributeslc_idlc_secretslc_serializable_keysstreamChannelsAsIsstreamChannelsList_batchWithConfig_callWithConfig_getOptionsList_separateRunnableConfigFromCallOptions_streamLog_transformStreamWithConfigasToolassignattachBranchattachEdgeattachNodebatchbindclearCachegetGraphgetGraphAsyncgetNamegetStategetStateHistorygetSubgraphsgetSubgraphsAsyncinvokemappickpipestreamstreamEventsstreamLogtoJSONtoJSONNotImplementedtransformupdateStatevalidatewithConfigwithFallbackswithListenerswithRetryisRunnable