Functional API 概述

函数式 API 允许您通过对现有代码进行最少的修改，将 LangGraph 的关键特性（持久化、记忆、人机协作 和 流式输出）添加到您的应用程序中。

函数式 API 允许您通过对现有代码进行最少的修改，将 LangGraph 的关键特性（持久化、记忆、人机协作 和 流式输出）添加到您的应用程序中。

它旨在将这些特性集成到可能使用标准语言原语进行分支和控制流（例如 if 语句、for 循环和函数调用）的现有代码中。与许多需要将代码重构为显式管道或 DAG 的数据编排框架不同，函数式 API 允许您在不强制执行严格执行模型的情况下，将这些功能集成到应用程序中。

函数式 API 使用两个关键构建块：

:::python

• @entrypoint：将函数标记为工作流的起点，封装逻辑并管理执行流，包括处理长时间运行的任务和中断。
• @[@task]：表示一个离散的工作单元，例如 API 调用或数据处理步骤，可以在入口点内异步执行。任务返回一个类似未来的对象，可以异步等待或同步解析。 :::

:::js

• entrypoint：一个入口点封装工作流逻辑并管理执行流，包括处理长时间运行的任务和中断。
• task：表示一个离散的工作单元，例如 API 调用或数据处理步骤，可以在入口点内异步执行。任务返回一个类似未来的对象，可以异步等待或同步解析。 :::

这提供了一个最小的抽象，用于构建具有状态管理和流式输出的工作流。

有关如何使用函数式 API 的信息，请参阅 使用函数式 API。

函数式 API 与图 API

对于更喜欢更声明式方法的用户，LangGraph 的 图 API 允许您使用图范式定义工作流。这两个 API 共享相同的底层运行时，因此您可以在同一应用程序中使用它们。

以下是一些关键差异：

• 控制流：函数式 API 不需要考虑图结构。您可以使用标准的 Python 构造来定义工作流。这将通常会减少您需要编写的代码量。
• 短期记忆：图 API 需要声明一个 状态，并且可能需要定义 归约器 来管理图状态的更新。@entrypoint 和 @tasks 不需要显式状态管理，因为它们的状态仅限于函数内部，并且不会在函数之间共享。
• 检查点：这两个 API 都会生成和使用检查点。在 图 API 中，每个 超级步骤 后都会生成一个新的检查点。在 函数式 API 中，当任务执行时，其结果会保存到与给定入口点关联的现有检查点中，而不是创建新的检查点。
• 可视化：图 API 可以轻松地将工作流可视化为图，这对于调试、理解工作流以及与他人分享非常有用。函数式 API 不支持可视化，因为图是在运行时动态生成的。

示例

下面我们演示了一个简单的应用程序，该应用程序编写一篇论文并 中断 以请求人工审核。

:::python

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt

@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1) # A placeholder for a long-running task.
    return f"An essay about topic: {topic}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt({
        # Any json-serializable payload provided to interrupt as argument.
        # It will be surfaced on the client side as an Interrupt when streaming data
        # from the workflow.
        "essay": essay, # The essay we want reviewed.
        # We can add any additional information that we need.
        # For example, introduce a key called "action" with some instructions.
        "action": "Please approve/reject the essay",
    })

    return {
        "essay": essay, # The essay that was generated
        "is_approved": is_approved, # Response from HIL
    }

:::

:::js

import { MemorySaver, entrypoint, task, interrupt } from "@langchain/langgraph";

const writeEssay = task("writeEssay", async (topic: string) => {
  // A placeholder for a long-running task.
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return `An essay about topic: ${topic}`;
});

const workflow = entrypoint(
  { checkpointer: new MemorySaver(), name: "workflow" },
  async (topic: string) => {
    const essay = await writeEssay(topic);
    const isApproved = interrupt({
      // Any json-serializable payload provided to interrupt as argument.
      // It will be surfaced on the client side as an Interrupt when streaming data
      // from the workflow.
      essay, // The essay we want reviewed.
      // We can add any additional information that we need.
      // For example, introduce a key called "action" with some instructions.
      action: "Please approve/reject the essay",
    });

    return {
      essay, // The essay that was generated
      isApproved, // Response from HIL
    };
  }
);

:::

Detailed Explanation

此工作流将编写关于“猫”主题的论文，然后暂停以获取人工审核。该工作流可以无限期地中断，直到提供审核。

当工作流恢复执行时，它将从最开始执行，但由于 writeEssay 任务的结果已经保存，任务结果将从检查点加载，而不是重新计算。

:::python

import time

from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import Command, interrupt


@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1)  # This is a placeholder for a long-running task.
    return f"An essay about topic: {topic}"


@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt(
        {
            # Any json-serializable payload provided to interrupt as argument.
            # It will be surfaced on the client side as an Interrupt when streaming data
            # from the workflow.
            "essay": essay,  # The essay we want reviewed.
            # We can add any additional information that we need.
            # For example, introduce a key called "action" with some instructions.
            "action": "Please approve/reject the essay",
        }
    )
    return {
        "essay": essay,  # The essay that was generated
        "is_approved": is_approved,  # Response from HIL
    }


thread_id = str(uuid7())
config = {"configurable": {"thread_id": thread_id}}
stream = workflow.stream_events("cat", config, version="v3")
_ = stream.output
print({"write_essay": stream.interrupts[0].value["essay"]})
print({"__interrupt__": stream.interrupts})
# {'write_essay': 'An essay about topic: cat'}
# {
#   '__interrupt__': [
#     Interrupt(
#       value={
#           'essay': 'An essay about topic: cat',
#           'action': 'Please approve/reject the essay'
#       },
#       id='369d44b3d93d4a631ae583367ac6b5cc'
#     )
#   ]
# }

一篇论文已经编写完成，准备进行审核。一旦提供审核，我们可以恢复工作流：

# Get review from a user (e.g., via a UI)
# In this case, we're using a bool, but this can be any json-serializable value.
human_review = True

resumed_stream = workflow.stream_events(Command(resume=human_review), config, version="v3")
print(resumed_stream.output)
# {'essay': 'An essay about topic: cat', 'is_approved': True}

工作流已完成，审核已添加到论文中。 :::

:::js

import { MemorySaver, entrypoint, interrupt, task } from "@langchain/langgraph";

const writeEssay = task("writeEssay", async (topic: string) => {
  // This is a placeholder for a long-running task.
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return `An essay about topic: ${topic}`;
});

const workflow = entrypoint(
  { checkpointer: new MemorySaver(), name: "workflow" },
  async (_topic: string) => {
    const essay = await writeEssay("cat");
    const isApproved = interrupt({
      // Any json-serializable payload provided to interrupt as argument.
      // It will be surfaced on the client side as an Interrupt when streaming data
      // from the workflow.
      essay, // The essay we want reviewed.
      // We can add any additional information that we need.
      // For example, introduce a key called "action" with some instructions.
      action: "Please approve/reject the essay",
    });

    return {
      essay, // The essay that was generated
      isApproved, // Response from HIL
    };
  },
);

const threadId = "functional-api-thread";
const config = {
  configurable: {
    thread_id: threadId,
  },
};

const stream = await workflow.streamEvents("cat", { ...config, version: "v2" });
const initialChunks: Record<string, unknown>[] = [];
for await (const event of stream) {
  const chunk = event.data?.chunk;
  if (chunk && typeof chunk === "object") {
    console.log(chunk);
    initialChunks.push(chunk as Record<string, unknown>);
  }
}
// { writeEssay: "An essay about topic: cat" }
// { __interrupt__: [Interrupt(...)] }

一篇论文已经编写完成，准备进行审核。一旦提供审核，我们可以恢复工作流：

import { Command } from "@langchain/langgraph";

// Get review from a user (e.g., via a UI)
// In this case, we're using a bool, but this can be any json-serializable value.
const humanReview = true;

const resumedStream = await workflow.streamEvents(
  new Command({ resume: humanReview }),
  { ...config, version: "v2" },
);
const resumedChunks: Record<string, unknown>[] = [];
for await (const event of resumedStream) {
  const chunk = event.data?.chunk;
  if (chunk && typeof chunk === "object") {
    console.log(chunk);
    resumedChunks.push(chunk as Record<string, unknown>);
  }
}
// { essay: "An essay about topic: cat", isApproved: true }

工作流已完成，审核已添加到论文中。 :::

入口点

:::python 可以使用 @[@entrypoint] 装饰器从一个函数创建工作流。它封装工作流逻辑并管理执行流，包括处理 长时间运行的任务 和 中断。 :::

:::js 可以使用 @[entrypoint][entrypoint] 函数从一个函数创建工作流。它封装工作流逻辑并管理执行流，包括处理 长时间运行的任务 和 中断。 :::

定义

:::python 一个 入口点 是通过使用 @entrypoint 装饰器来定义的。

函数 必须接受一个位置参数，该参数用作工作流输入。如果您需要传递多个数据，请使用字典作为第一个参数的输入类型。

使用 entrypoint 装饰函数将生成一个 @[Pregel][Pregel.stream] 实例，它有助于管理工作流的执行（例如，处理流式输出、恢复和检查点）。

通常，您需要将一个 检查点器 传递给 @entrypoint 装饰器，以启用持久化并使用 人机协作 等功能。

Sync

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def my_workflow(some_input: dict) -> int:
    # some logic that may involve long-running tasks like API calls,
    # and may be interrupted for human-in-the-loop.
    ...
    return result

Async

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
async def my_workflow(some_input: dict) -> int:
    # some logic that may involve long-running tasks like API calls,
    # and may be interrupted for human-in-the-loop
    ...
    return result

:::

:::js 一个 entrypoint 是通过调用 entrypoint 函数并传入配置和一个函数来定义的。

该函数 必须接受一个位置参数，该参数用作工作流输入。如果你需要传递多份数据，请将第一个参数的类型设置为对象。

使用函数创建 entrypoint 会产生一个工作流实例，该实例有助于管理工作流的执行（例如，处理流式输出、恢复和检查点）。

你通常需要将 checkpointer 传递给 entrypoint 函数以启用持久化并使用 人机交互 等功能。

import { entrypoint } from "@langchain/langgraph";

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (someInput: Record<string, any>): Promise<number> => {
    // some logic that may involve long-running tasks like API calls,
    // and may be interrupted for human-in-the-loop
    return result;
  }
);

:::

序列化 entrypoint 的 输入 和 输出 必须是 JSON 可序列化的，以支持检查点。请参阅 序列化 部分以获取更多详细信息。

:::python

可注入参数

在声明一个 entrypoint 时，你可以请求访问在运行时自动注入的附加参数。这些参数包括：

参数	描述
previous	访问给定线程中与上一个 checkpoint 关联的状态。参见 短期记忆。
store	一个 [BaseStore][langgraph.store.base.BaseStore] 的实例。适用于 长期记忆。
writer	用于在异步 Python < 3.11 时访问 StreamWriter。参见 使用函数式 API 进行流式输出的详细信息。
config	用于访问运行时配置。参见 RunnableConfig 获取信息。

使用适当的名字和类型注解声明这些参数。

Requesting Injectable Parameters

from langchain_core.runnables import RunnableConfig
from langgraph.func import entrypoint
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import StreamWriter

in_memory_checkpointer = InMemorySaver(...)
in_memory_store = InMemoryStore(...)  # An instance of InMemoryStore for long-term memory

@entrypoint(
    checkpointer=in_memory_checkpointer,  # Specify the checkpointer
    store=in_memory_store  # Specify the store
)
def my_workflow(
    some_input: dict,  # The input (e.g., passed via `invoke`)
    *,
    previous: Any = None, # For short-term memory
    store: BaseStore,  # For long-term memory
    writer: StreamWriter,  # For streaming custom data
    config: RunnableConfig  # For accessing the configuration passed to the entrypoint
) -> ...:

:::

执行

:::python 使用 @entrypoint 会生成一个 @[Pregel][Pregel.stream] 对象，可以使用 invoke、ainvoke、stream 和 astream 方法执行。

Invoke

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
my_workflow.invoke(some_input, config)  # Wait for the result synchronously

Async Invoke

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
await my_workflow.ainvoke(some_input, config)  # Await result asynchronously

Stream

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = my_workflow.stream_events(some_input, config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

Async Stream

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = await my_workflow.astream_events(some_input, config, version="v3")
async for message in stream.messages:
    async for token in message.text:
        print(token, end="", flush=True)

:::

:::js 使用 entrypoint 函数将返回一个可以使用 invoke 和 stream 方法执行的对象。

Invoke

const config = {
  configurable: {
    thread_id: "some_thread_id"
  }
};
await myWorkflow.invoke(someInput, config); // Wait for the result

Stream

const config = {
  configurable: {
    thread_id: "some_thread_id"
  }
};

const stream = await myWorkflow.streamEvents(someInput, config, { version: "v3" });
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

:::

恢复

:::python 在 @[interrupt][interrupt] 后恢复执行可以通过向 @[Command] 原语传递一个 resume 值来实现。

Invoke

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(Command(resume=some_resume_value), config)

Async Invoke

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

await my_workflow.ainvoke(Command(resume=some_resume_value), config)

Stream

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = my_workflow.stream_events(Command(resume=some_resume_value), config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

Async Stream

from langgraph.types import Command

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = await my_workflow.astream_events(Command(resume=some_resume_value), config, version="v3")
async for message in stream.messages:
    async for token in message.text:
        print(token, end="", flush=True)

:::

:::js 在 @[interrupt][interrupt]后通过传递resume值给@[Command]原语可以恢复执行。

Invoke

import { Command } from "@langchain/langgraph";

const config = {
  configurable: {
    thread_id: "some_thread_id"
  }
};

await myWorkflow.invoke(new Command({ resume: someResumeValue }), config);

Stream

import { Command } from "@langchain/langgraph";

const config = {
  configurable: {
    thread_id: "some_thread_id"
  }
};

const stream = await myWorkflow.streamEvents(
  new Command({ resume: someResumableValue }),
  config,
  { version: "v3" },
);

for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

:::

:::python 在错误后恢复

要在错误后恢复，运行entrypoint并使用None和相同的thread id（config）。

这假设底层的错误已解决，执行可以成功继续。

Invoke

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(None, config)

Async Invoke

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

await my_workflow.ainvoke(None, config)

Stream

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = my_workflow.stream_events(None, config, version="v3")
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

Async Stream

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

stream = await my_workflow.astream_events(None, config, version="v3")
async for message in stream.messages:
    async for token in message.text:
        print(token, end="", flush=True)

:::

:::js 在错误后恢复

要在错误后恢复，运行entrypoint并使用null和相同的thread id（config）。

这假设底层的错误已解决，执行可以成功继续。

Invoke

const config = {
  configurable: {
    thread_id: "some_thread_id"
  }
};

await myWorkflow.invoke(null, config);

Stream

const config = {
  configurable: {
    thread_id: "some_thread_id"
  }
};

const stream = await myWorkflow.streamEvents(null, config, { version: "v3" });
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

:::

短期记忆

当entrypoint使用checkpointer定义时，它会在同一thread id的连续调用之间存储信息在检查点。

:::python 这允许使用previous参数访问前一次调用的状态。

默认情况下，previous参数是前一次调用的返回值。

@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> int:
    previous = previous or 0
    return number + previous

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}

my_workflow.invoke(1, config)  # 1 (previous was None)
my_workflow.invoke(2, config)  # 3 (previous was 1 from the previous invocation)

:::

:::js 这允许使用getPreviousState函数访问前一次调用的状态。

默认情况下，getPreviousState函数返回前一次调用的返回值。

import { entrypoint, getPreviousState } from "@langchain/langgraph";

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (number: number) => {
    const previous = getPreviousState<number>() ?? 0;
    return number + previous;
  }
);

const config = {
  configurable: {
    thread_id: "some_thread_id",
  },
};

await myWorkflow.invoke(1, config); // 1 (previous was undefined)
await myWorkflow.invoke(2, config); // 3 (previous was 1 from the previous invocation)

:::

entrypoint.final

:::python @[entrypoint.final][entrypoint.final]是一个特殊原语，可以从入口点返回并允许解耦存储在检查点中的值与入口点的返回值。

第一个值是入口点的返回值，第二个值是将在检查点中保存的值。类型注解是entrypoint.final[return_type, save_type]。

@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
    previous = previous or 0
    # This will return the previous value to the caller, saving
    # 2 * number to the checkpoint, which will be used in the next invocation
    # for the `previous` parameter.
    return entrypoint.final(value=previous, save=2 * number)

config = {
    "configurable": {
        "thread_id": "1"
    }
}

my_workflow.invoke(3, config)  # 0 (previous was None)
my_workflow.invoke(1, config)  # 6 (previous was 3 * 2 from the previous invocation)

:::

:::js @[entrypoint.final][entrypoint.final]是一个特殊原语，可以从入口点返回并允许解耦存储在检查点中的值与入口点的返回值。

第一个值是入口点的返回值，第二个值是将在检查点中保存的值。

import { entrypoint, getPreviousState } from "@langchain/langgraph";

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (number: number) => {
    const previous = getPreviousState<number>() ?? 0;
    // This will return the previous value to the caller, saving
    // 2 * number to the checkpoint, which will be used in the next invocation
    // for the `previous` parameter.
    return entrypoint.final({
      value: previous,
      save: 2 * number,
    });
  }
);

const config = {
  configurable: {
    thread_id: "1",
  },
};

await myWorkflow.invoke(3, config); // 0 (previous was undefined)
await myWorkflow.invoke(1, config); // 6 (previous was 3 * 2 from the previous invocation)

:::

任务

任务代表一个离散的工作单元，例如API调用或数据处理步骤。它有两个关键特征：

• 异步执行：任务设计为异步执行，允许多个操作并发运行而不阻塞。
• 检查点：任务结果保存到检查点，从而可以从上次保存的状态恢复工作流程。（参见persistence了解更多详情）。

定义

:::python 任务使用@task装饰器定义，该装饰器包装一个常规的Python函数。

from langgraph.func import task

@task()
def slow_computation(input_value):
    # Simulate a long-running operation
    ...
    return result

:::

任务通过 task 函数定义，该函数包装了一个普通函数。

import { task } from "@langchain/langgraph";

const slowComputation = task("slowComputation", async (inputValue: any) => {
  // Simulate a long-running operation
  return result;
});

:::

序列化 任务的 输出 必须是 JSON 可序列化的，以支持检查点。

执行

任务 只能从 入口点、另一个 任务 或 状态图节点 中调用。

任务 不能 直接从主应用程序代码中调用。

:::python 调用 任务 时，它会立即返回一个 future 对象。Future 是一个占位符，用于表示稍后可用的结果。

要获取 任务 的结果，你可以同步等待它（使用 result()）或异步等待它（使用 await）。

Synchronous Invocation

@entrypoint(checkpointer=checkpointer)
def my_workflow(some_input: int) -> int:
    future = slow_computation(some_input)
    return future.result()  # Wait for the result synchronously

Asynchronous Invocation

@entrypoint(checkpointer=checkpointer)
async def my_workflow(some_input: int) -> int:
    return await slow_computation(some_input)  # Await result asynchronously

:::

:::js 调用 任务 时，它返回一个 Promise，该 Promise 可以被异步等待。

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (someInput: number): Promise<number> => {
    return await slowComputation(someInput);
  }
);

:::

何时使用任务

任务 在以下场景中很有用：

• 检查点：当你需要将长时间运行操作的结果保存到检查点，以便在恢复工作流时无需重新计算它。
• 人机交互：如果你正在构建一个需要人工干预的工作流，你必须使用 任务 封装任何随机性（例如，API 调用），以确保工作流可以正确恢复。有关更多详细信息，请参阅 确定性 部分。
• 并行执行：对于 I/O 密集型任务，任务 支持并行执行，允许多个操作并发运行而不会阻塞（例如，调用多个 API）。
• 可观察性：将操作包装在 任务 中提供了一种跟踪工作流进度和监控单个操作执行情况的方法，使用 LangSmith。
• 可重试工作：当工作需要重试以处理失败或不一致时，任务 提供了一种封装和管理重试逻辑的方法。

序列化

LangGraph 中的序列化有两个关键方面：

1. entrypoint 输入和输出必须是 JSON 可序列化的。
2. task 输出必须是 JSON 可序列化的。

:::python 这些要求对于启用检查点和工作流恢复是必要的。使用字典、列表、字符串、数字和布尔值等 Python 原语来确保你的输入和输出是可序列化的。 :::

:::js 这些要求对于启用检查点和工作流恢复是必要的。使用对象、数组、字符串、数字和布尔值等原语来确保你的输入和输出是可序列化的。 :::

序列化确保工作流状态（例如任务结果和中间值）可以可靠地保存和恢复。这对于支持人机交互、容错和并行执行至关重要。

提供不可序列化的输入或输出会导致在配置了检查点的工作流运行时出现运行时错误。

确定性

当你恢复工作流运行时，代码 不 会从执行停止的 同一行代码 处恢复。执行返回到检查点边界，并且工作流 重放 前进，直到它再次到达暂停点。

对于 Functional API，重放从 入口点 的开头开始，而 LangGraph 从检查点恢复已完成的 任务 和 子图 结果，而不是重新计算它们。这保留了跨暂停的步骤记录顺序，包括对于长时间运行或非确定性 任务 输出。

要使用 人机交互 等功能，你必须将非确定性工作（例如，随机值）和副作用（例如，文件写入或 API 调用）放在 任务 中。

工作流的多次运行可能会产生不同的结果，但恢复 特定 线程应该重放相同的持久化 任务 和 子图 结果。

为确保你的工作流是确定性的并且可以一致地重放，请遵循以下指南：

• 避免重复工作：在 入口点 中，如果你链式调用多个副作用（例如，日志记录、文件写入或网络调用），请为每个副作用分配一个 任务，以便恢复时从检查点恢复它们的输出，而不是再次运行它们。
• 封装非确定性操作：将可能在多次尝试中变化的值（例如，随机数或时钟读取）保留在 任务 中，以便重放与检查点记录对齐。
• 使用幂等操作：对于部分任务失败和重试，请参阅 幂等性。

幂等性

幂等性确保多次运行相同操作会产生相同的结果。这有助于防止重复 API 调用和冗余处理，如果步骤因失败而重跑。始终将 API 调用放在 任务 函数内部进行检查点，并设计它们在重新执行时保持幂等性。 这对于导致数据写入的操作尤其重要。 当工作流恢复时，LangGraph 从检查点重放已完成的 任务 结果。一个已开始但未完成的 任务 可能在恢复时再次运行，因此请设计副作用以保持幂等性。使用幂等性键或验证现有结果以避免意外重复。

常见陷阱

处理副作用

将副作用（例如，写入文件、发送电子邮件）封装在任务中，以确保在恢复工作流时不会多次执行它们。

Incorrect

在此示例中，副作用（写入文件）直接包含在工作流中，因此在工作流恢复时会再次执行它。

:::python

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    # This code will be executed a second time when resuming the workflow.
    # Which is likely not what you want.
    with open("output.txt", "w") as f:  
        f.write("Side effect executed")  
    value = interrupt("question")
    return value

:::

:::js

import { entrypoint, interrupt } from "@langchain/langgraph";
import fs from "fs";

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow },
  async (inputs: Record<string, any>) => {
    // This code will be executed a second time when resuming the workflow.
    // Which is likely not what you want.
    fs.writeFileSync("output.txt", "Side effect executed");
    const value = interrupt("question");
    return value;
  }
);

:::

Correct

在此示例中，副作用被封装在任务中，确保恢复时执行结果保持一致。

:::python

from langgraph.func import task

@task
def write_to_file():  
    with open("output.txt", "w") as f:
        f.write("Side effect executed")

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    # The side effect is now encapsulated in a task.
    write_to_file().result()
    value = interrupt("question")
    return value

:::

:::js

import { entrypoint, task, interrupt } from "@langchain/langgraph";
import * as fs from "fs";

const writeToFile = task("writeToFile", async () => {
  fs.writeFileSync("output.txt", "Side effect executed");
});

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (inputs: Record<string, any>) => {
    // The side effect is now encapsulated in a task.
    await writeToFile();
    const value = interrupt("question");
    return value;
  }
);

:::

非确定性控制流

每次可能产生不同结果的操作（如获取当前时间或随机数）应该封装在任务中，以确保在恢复时返回相同的结果。

• 在任务中：获取随机数（5）→ 中断 → 恢复 → （再次返回5）→ …
• 不在任务中：获取随机数（5）→ 中断 → 恢复 → 获取新的随机数（7）→ …

在使用人机交互工作流且有多处中断调用时，这一点尤其重要。LangGraph 为每个任务/入口点维护一个恢复值列表。当遇到中断时，它会与相应的恢复值匹配。这种匹配是严格基于索引的，因此恢复值的顺序应与中断的顺序相匹配。

如果恢复时执行顺序不保持一致，一个 @[interrupt]调用可能会与错误的resume值匹配，导致结果不正确。

请阅读关于determinism的章节以获取更多详细信息。

Incorrect

在此示例中，工作流使用当前时间来确定要执行哪个任务。这是非确定性的，因为工作流的结果取决于其执行的时间。

:::python

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    t0 = inputs["t0"]
    t1 = time.time()  

    delta_t = t1 - t0

    if delta_t > 1:
        result = slow_task(1).result()
        value = interrupt("question")
    else:
        result = slow_task(2).result()
        value = interrupt("question")

    return {
        "result": result,
        "value": value
    }

:::

:::js

import { entrypoint, interrupt } from "@langchain/langgraph";

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (inputs: { t0: number }) => {
    const t1 = Date.now();

    const deltaT = t1 - inputs.t0;

    if (deltaT > 1000) {
      const result = await slowTask(1);
      const value = interrupt("question");
      return { result, value };
    } else {
      const result = await slowTask(2);
      const value = interrupt("question");
      return { result, value };
    }
  }
);

:::

Correct

:::python 在此示例中，工作流使用输入 t0 决定执行哪个任务。由于工作流的结果仅取决于输入，因此这是确定性的。

import time

from langgraph.func import task

@task
def get_time() -> float:  
    return time.time()

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    t0 = inputs["t0"]
    t1 = get_time().result()  

    delta_t = t1 - t0

    if delta_t > 1:
        result = slow_task(1).result()
        value = interrupt("question")
    else:
        result = slow_task(2).result()
        value = interrupt("question")

    return {
        "result": result,
        "value": value
    }

:::

:::js 在此示例中，工作流使用输入 t0 来确定要执行哪个任务。这是确定性的，因为工作流的结果仅取决于输入。

import { entrypoint, task, interrupt } from "@langchain/langgraph";

const getTime = task("getTime", () => Date.now());

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (inputs: { t0: number }): Promise<any> => {
    const t1 = await getTime();

    const deltaT = t1 - inputs.t0;

    if (deltaT > 1000) {
      const result = await slowTask(1);
      const value = interrupt("question");
      return { result, value };
    } else {
      const result = await slowTask(2);
      const value = interrupt("question");
      return { result, value };
    }
  }
);

:::

了解更多

• 如何使用 Functional API
• 图 API 概念概述
• 在 Graph API 和 Functional API 之间进行选择

---

官方原文：src/oss/langgraph/functional-api.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。