类 CompiledStateGraph<S, U, N, I, O, C>

构建和编译 StateGraph 的最终结果。不应直接实例化,仅应使用 StateGraph 的 .compile() 实例方法。

类型参数

层次结构 (查看完整)

构造函数

属性

NodeType: N
RunInput: S
RunOutput: U
autoValidate: boolean

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

builder: StateGraph<unknown, S, U, N, I, O, C>
channels: Record<string | N, BaseChannel<unknown, unknown, unknown>>

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

checkpointer?: false | BaseCheckpointSaver<number>

用于持久化图状态的可选检查点程序。如果提供,则在每个超级步骤保存图状态的检查点。当为 false 或未定义时,禁用检查点,并且图将无法保存或恢复状态。

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<S, U>>

图中的节点,将节点名称映射到其 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":流式传输与图的执行相关的事件 - 对于追踪和调试图执行非常有用

访问器

  • 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,反序列化时将使用该 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,该 Promise 解析为输出值。

    类型参数

    参数

    • func: ((inputs, options?, runManagers?, batchOptions?) => Promise<(Error | StateType<O>)[]>)

      要为每个输入值执行的函数。

        • (inputs, options?, runManagers?, batchOptions?): Promise<(Error | StateType<O>)[]>
        • 参数

          • inputs: T[]
          • 可选 options: Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>>[]
          • 可选 runManagers: (undefined | CallbackManagerForChainRun)[]
          • 可选 batchOptions: RunnableBatchOptions

          返回 Promise<(Error | StateType<O>)[]>

    • inputs: T[]
    • 可选 options: Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>> & {
          runType?: string;
      }> | Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>> & {
          runType?: string;
      }>[]
    • 可选 batchOptions: RunnableBatchOptions

    返回 Promise<(Error | StateType<O>)[]>

    解析为输出值的 Promise。

  • 类型参数

    参数

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

    返回 Partial<O>[]

  • 参数

    返回值 AsyncGenerator<RunLogPatch, any, unknown>

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

    类型参数

    参数

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

          • generator: AsyncGenerator<I, any, unknown>
          • Optional runManager: CallbackManagerForChainRun
          • 可选 options: Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>>

          返回值 AsyncGenerator<O, any, unknown>

    • 可选 options: Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>> & {
          runType?: string;
      }

    返回值 AsyncGenerator<O, any, unknown>

  • 将 runnable 转换为工具。返回 RunnableToolLike 的新实例,其中包含 runnable、名称、描述和模式。

    类型参数

    参数

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

        工具的描述。如果未提供,则回退到 Zod 模式上的描述;如果两者都未提供,则为未定义。

      • Optional name?: string

        工具的名称。如果未提供,则默认为 runnable 的名称。

      • schema: ZodType<T, ZodTypeDef, T>

        工具输入的 Zod 模式。从 runnable 的输入类型推断 Zod 类型。

    返回值 RunnableToolLike<ZodType<ToolCall | T, ZodTypeDef, ToolCall | T>, StateType<O>>

    RunnableToolLike 的实例,它是一个可以用作工具的 runnable。

  • 为此 runnable 的 dict 输出分配新字段。返回一个新的 runnable。

    参数

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

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

  • 参数

    • start: "__start__" | N
    • name: string
    • branch: Branch<S, N, LangGraphRunnableConfig<Record<string, any>>>
    • Optional options: {
          withReader?: boolean;
      }
      • Optional withReader?: boolean

    返回值 void

  • 参数

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

    返回值 void

  • 参数

    • key: "__start__"
    • Optional node: undefined

    返回值 void

  • 参数

    • key: N
    • node: StateGraphNodeSpec<S, U>

    返回值 void

  • 批处理的默认实现,它调用 invoke N 次。如果子类可以更有效地进行批处理,则应重写此方法。

    参数

    • inputs: (null | Command<unknown> | UpdateType<I>)[]

      每次批处理调用的输入数组。

    • Optional options: Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>> | Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>>

      要应用于每个批处理调用的单个调用选项对象,或每次调用的数组。

    • Optional batchOptions: RunnableBatchOptions & {
          returnExceptions?: false;
      }

    返回值 Promise<StateType<O>[]>

    RunOutput 的数组,如果设置了 batchOptions.returnExceptions,则为混合的 RunOutput 和错误

  • 参数

    返回 Promise<(Error | StateType<O>)[]>

  • 参数

    返回 Promise<(Error | StateType<O>)[]>

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

    参数

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

    Returns Graph

    已弃用

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

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

    参数

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

    Returns Promise<Graph>

  • 参数

    • Optional suffix: string

    Returns string

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

    参数

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

      用于检索状态的配置

    • Optional options: GetStateOptions

      附加选项

    Returns Promise<StateSnapshot>

    当前图状态的快照

    Throws

    如果未配置检查点

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

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

    参数

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

      用于检索历史记录的配置

    • Optional options: CheckpointListOptions

      用于过滤历史记录的选项

    Returns AsyncIterableIterator<StateSnapshot>

    状态快照的异步迭代器

    Throws

    如果未配置检查点

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

    参数

    • Optional namespace: string

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

    • Optional recurse: boolean

      是否递归获取子图的子图

    Returns Generator<[string, Pregel<any, any, StrRecord<string, any>, any, any>], any, unknown>

    生成器,产生 [name, subgraph] 元组

    已弃用

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

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

    参数

    • Optional namespace: string

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

    • Optional recurse: boolean

      是否递归获取子图的子图

    Returns AsyncGenerator<[string, Pregel<any, any, StrRecord<string, any>, any, any>], any, unknown>

    AsyncGenerator,产生 [name, subgraph] 元组

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

    参数

    • keys: string | string[]

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

  • 创建一个新的 runnable 序列,它串行运行每个单独的 runnable,并将一个 runnable 的输出通过管道传输到另一个 runnable 或类似 runnable 的对象。

    类型参数

    • NewRunOutput

    参数

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

      一个 runnable、函数或对象,其值是函数或 runnables。

    Returns Runnable<null | Command<unknown> | UpdateType<I>, Exclude<NewRunOutput, Error>, RunnableConfig<Record<string, any>>>

    一个新的 runnable 序列。

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

    流模式

    • "values": 发出每个步骤后的完整状态
    • "updates": 仅发出每个步骤后的状态更改
    • "debug": 发出详细的调试信息
    • "messages": 发出来自节点内部的消息

    更多详情,请参阅Streaming how-to guides

    参数

    返回 Promise<IterableReadableStream<any>>

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

  • 生成由可运行对象的内部步骤发出的事件流。

    用于创建 StreamEvents 的迭代器,该迭代器提供关于可运行对象进度的实时信息,包括来自中间结果的 StreamEvents。

    StreamEvent 是一个具有以下模式的字典

    • event: string - 事件名称的格式为:on_[runnable_type]_(start|stream|end)。
    • name: string - 生成事件的可运行对象的名称。
    • run_id: string - 与发出事件的可运行对象的给定执行相关联的随机生成的 ID。作为父可运行对象执行的一部分而被调用的子可运行对象会被分配其自己唯一的 ID。
    • tags: string[] - 生成事件的可运行对象的标签。
    • metadata: Record<string, any> - 生成事件的可运行对象的元数据。
    • data: Record<string, any>

    下面是一个表格,说明了各种链可能发出的一些事件。为了简洁起见,表格中省略了元数据字段。链定义已包含在表格之后。

    注意 此参考表适用于模式的 V2 版本。

    +----------------------+-----------------------------+------------------------------------------+
    | event | input | output/chunk |
    +======================+=============================+==========================================+
    | on_chat_model_start | {"messages": BaseMessage[]} | |
    +----------------------+-----------------------------+------------------------------------------+
    | on_chat_model_stream | | AIMessageChunk("hello") |
    +----------------------+-----------------------------+------------------------------------------+
    | on_chat_model_end | {"messages": BaseMessage[]} | AIMessageChunk("hello world") |
    +----------------------+-----------------------------+------------------------------------------+
    | on_llm_start | {'input': 'hello'} | |
    +----------------------+-----------------------------+------------------------------------------+
    | on_llm_stream | | 'Hello' |
    +----------------------+-----------------------------+------------------------------------------+
    | on_llm_end | 'Hello human!' | |
    +----------------------+-----------------------------+------------------------------------------+
    | on_chain_start | | |
    +----------------------+-----------------------------+------------------------------------------+
    | on_chain_stream | | "hello world!" |
    +----------------------+-----------------------------+------------------------------------------+
    | on_chain_end | [Document(...)] | "hello world!, goodbye world!" |
    +----------------------+-----------------------------+------------------------------------------+
    | on_tool_start | {"x": 1, "y": "2"} | |
    +----------------------+-----------------------------+------------------------------------------+
    | on_tool_end | | {"x": 1, "y": "2"} |
    +----------------------+-----------------------------+------------------------------------------+
    | on_retriever_start | {"query": "hello"} | |
    +----------------------+-----------------------------+------------------------------------------+
    | on_retriever_end | {"query": "hello"} | [Document(...), ..] |
    +----------------------+-----------------------------+------------------------------------------+
    | on_prompt_start | {"question": "hello"} | |
    +----------------------+-----------------------------+------------------------------------------+
    | on_prompt_end | {"question": "hello"} | ChatPromptValue(messages: BaseMessage[]) |
    +----------------------+-----------------------------+------------------------------------------+

    “on_chain_*”事件是对于不属于上述任何类别的 Runnables 的默认事件。

    除了上面的标准事件之外,用户还可以分派自定义事件。

    自定义事件将仅在 API 的 v2 版本中公开!

    自定义事件具有以下格式

    +-----------+------+------------------------------------------------------------+
    | Attribute | Type | Description |
    +===========+======+============================================================+
    | name | str | A user defined name for the event. |
    +-----------+------+------------------------------------------------------------+
    | data | Any | The data associated with the event. This can be anything. |
    +-----------+------+------------------------------------------------------------+

    这是一个例子

    import { RunnableLambda } from "@langchain/core/runnables";
    import { dispatchCustomEvent } from "@langchain/core/callbacks/dispatch";
    // Use this import for web environments that don't support "async_hooks"
    // and manually pass config to child runs.
    // import { dispatchCustomEvent } from "@langchain/core/callbacks/dispatch/web";

    const slowThing = RunnableLambda.from(async (someInput: string) => {
    // Placeholder for some slow operation
    await new Promise((resolve) => setTimeout(resolve, 100));
    await dispatchCustomEvent("progress_event", {
    message: "Finished step 1 of 2",
    });
    await new Promise((resolve) => setTimeout(resolve, 100));
    return "Done";
    });

    const eventStream = await slowThing.streamEvents("hello world", {
    version: "v2",
    });

    for await (const event of eventStream) {
    if (event.event === "on_custom_event") {
    console.log(event);
    }
    }

    参数

    返回 IterableReadableStream<StreamEvent>

  • 参数

    • input: null | Command<unknown> | UpdateType<I>
    • options: Partial<PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>> & {
          encoding: "text/event-stream";
          version: "v1" | "v2";
      }
    • Optional streamOptions: Omit<EventStreamCallbackHandlerInput, "autoClose">

    返回 IterableReadableStream<Uint8Array>

  • 流式传输来自可运行对象的所有输出,如同报告给回调系统一样。这包括 LLM、检索器、工具等的所有内部运行。输出作为 Log 对象流式传输,其中包括 jsonpatch ops 列表,这些 ops 描述了运行状态在每个步骤中是如何变化的,以及运行的最终状态。可以应用 jsonpatch ops 以构造状态。

    参数

    返回值 AsyncGenerator<RunLogPatch, any, unknown>

  • 返回 Serialized

  • 返回 SerializedNotImplemented

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

    参数

    返回 AsyncGenerator<StateType<O>, any, unknown>

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

    此方法可用于

    • 实施人机循环工作流程
    • 在断点期间修改图状态
    • 将外部输入集成到图中

    参数

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

      更新的配置

    • values: unknown

      用于更新状态的值

    • Optional asNode: string | N

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

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

    更新的配置

    Throws

    如果未配置检查点

    Throws

    如果更新不能归因于节点

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

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

    返回 this

    this - 用于方法链的 Pregel 实例

    Throws

    如果图结构无效

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

    参数

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

      要与当前配置合并的配置

    返回值 CompiledStateGraph<S, U, N, I, O, C>

    具有合并配置的新 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" }
    });
  • 从当前可运行对象创建一个新的可运行对象,如果初始调用失败,它将尝试调用其他传递的后备可运行对象。

    参数

    返回 RunnableWithFallbacks<null | Command<unknown> | UpdateType<I>, StateType<O>>

    一个新的 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>);
      }

      包含回调函数的对象。

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

        当 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>

    Returns Runnable<null | Command<unknown> | UpdateType<I>, StateType<O>, PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>>

  • 为现有的 runnable 添加重试逻辑。

    参数

    • 可选 fields: {
          onFailedAttempt?: RunnableRetryFailedAttemptHandler;
          stopAfterAttempt?: number;
      }
      • 可选 onFailedAttempt?: RunnableRetryFailedAttemptHandler
      • 可选 stopAfterAttempt?: number

    返回值 RunnableRetry<null | Command<unknown> | UpdateType<I>, StateType<O>, PregelOptions<Record<"__start__" | N, PregelNode<S, U>>, Record<string | N, BaseChannel<unknown, unknown, unknown>>, StateType<C> & Record<string, any>>>

    一个新的 RunnableRetry,当被调用时,将根据参数进行重试。

  • 参数

    • thing: any

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