Functional API 允许你以对现有代码最小化修改的方式，将 LangGraph 的关键特性（持久化、记忆、人在回路和流式输出）添加到你的应用程序中。

它旨在将这些特性集成到现有代码中，这些代码可能使用标准的语言原语（例如 if 语句、for 循环和函数调用）来进行分支和控制流。与许多要求将代码重构为显式流水线或 DAG 的数据编排框架不同，Functional API 允许你引入这些能力，而无需强制采用僵化的执行模型。

Functional API 使用两个关键的构建模块：

• @entrypoint：将函数标记为工作流的起点，封装逻辑并管理执行流，包括处理长时间运行的 task 和中断。

• @task：表示一个离散的工作单元，例如 API 调用或数据处理步骤，可以在 entrypoint 内异步执行。Task 返回一个类 future 对象，可以被 await 或同步解析。

• entrypoint：一个 entrypoint 封装工作流逻辑并管理执行流，包括处理长时间运行的 task 和中断。

• task：表示一个离散的工作单元，例如 API 调用或数据处理步骤，可以在 entrypoint 内异步执行。Task 返回一个类 future 对象，可以被 await 或同步解析。

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

TIP

有关如何使用 functional API 的信息，请参阅使用 Functional API。

Functional API 与 Graph API 的对比

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

以下是一些关键区别：

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

示例

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

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) # 用于模拟长时间运行 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({
        # 作为参数传递给 interrupt 的任何 JSON 可序列化负载。
        # 在流式传输数据时，它会在客户端以 Interrupt 的形式呈现
        # 来自工作流。
        "essay": essay, # 我们希望被审阅的文章。
        # 我们可以添加任何需要的额外信息。
        # 例如，引入一个名为 "action" 的键并附带一些说明。
        "action": "Please approve/reject the essay",
    })

    return {
        "essay": essay, # 生成的文章
        "is_approved": is_approved, # 来自 HIL 的响应
    }

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

const writeEssay = task("writeEssay", async (topic: string) => {
  // 用于模拟长时间运行 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({
      // 作为参数传递给 interrupt 的任何 JSON 可序列化负载。
      // 在流式传输数据时，它会在客户端以 Interrupt 的形式呈现
      // 来自工作流。
      essay, // 我们希望被审阅的文章。
      // 我们可以添加任何需要的额外信息。
      // 例如，引入一个名为 "action" 的键并附带一些说明。
      action: "Please approve/reject the essay",
    });

    return {
      essay, // 生成的文章
      isApproved, // 来自 HIL 的响应
    };
  }
);

详细说明

该工作流将编写一篇关于主题"cat"的文章，然后暂停以获取人工审核。工作流可以被无限期中断，直到提供审核意见。

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

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}

工作流已完成，审核意见已添加到文章中。

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 }

工作流已完成，审核意见已添加到文章中。

Entrypoint

可以使用 @entrypoint 装饰器从函数创建工作流。它封装工作流逻辑并管理执行流，包括处理_长时间运行的 task_ 和中断。

可以使用 entrypoint 函数从函数创建工作流。它封装工作流逻辑并管理执行流，包括处理_长时间运行的 task_ 和中断。

定义

entrypoint 通过使用 @entrypoint 装饰器装饰函数来定义。

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

使用 entrypoint 装饰函数会生成一个 Pregel 实例，帮助管理工作流的执行（例如处理流式输出、恢复和检查点持久化）。

你通常希望向 @entrypoint 装饰器传递一个检查点以启用持久化，并使用人在回路等功能。

同步

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def my_workflow(some_input: dict) -> int:
    # 一些可能涉及 API 调用等长时间运行 task 的逻辑，
    # 并且可能为了人在回路而被中断。
    ...
    return result

异步

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
async def my_workflow(some_input: dict) -> int:
    # 一些可能涉及 API 调用等长时间运行 task 的逻辑，
    # 并且可能为了人在回路而被中断
    ...
    return result

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

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

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

你通常希望向 entrypoint 函数传递一个检查点以启用持久化，并使用人在回路等功能。

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

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (someInput: Record<string, any>): Promise<number> => {
    // 一些可能涉及 API 调用等长时间运行 task 的逻辑，
    // 并且可能为了人在回路而被中断
    return result;
  }
);

WARNING

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

可注入参数

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

参数	说明
previous	访问给定线程上一个 checkpoint 关联的状态。请参阅短期记忆。
store	一个 [BaseStore][langgraph.store.base.BaseStore] 实例。对于长期记忆很有用。
writer	用于在使用 Async Python < 3.11 时访问 StreamWriter。详情请参阅使用 functional API 进行流式输出。
config	用于访问运行时配置。信息请参阅 RunnableConfig。

WARNING

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

请求可注入参数

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(...)  # 用于长期记忆的 InMemoryStore 实例

@entrypoint(
    checkpointer=in_memory_checkpointer,  # 指定检查点器
    store=in_memory_store  # 指定 store
)
def my_workflow(
    some_input: dict,  # 输入（例如，通过 `invoke` 传递）
    *,
    previous: Any = None, # 用于短期记忆
    store: BaseStore,  # 用于长期记忆
    writer: StreamWriter,  # 用于流式传输自定义数据
    config: RunnableConfig  # 用于访问传递给 entrypoint 的配置
) -> ...:

执行

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

Invoke

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
my_workflow.invoke(some_input, config)  # 同步等待结果

Async Invoke

config = {
    "configurable": {
        "thread_id": "some_thread_id"
    }
}
await my_workflow.ainvoke(some_input, config)  # 异步等待结果

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)

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

Invoke

const config = {
  configurable: {
    thread_id: "some_thread_id"
  }
};
await myWorkflow.invoke(someInput, config); // 等待结果

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

恢复

可以通过向 Command 原语传递 resume 值来在 interrupt 之后恢复执行。

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)

可以通过向 Command 原语传递 resume 值来在 interrupt 之后恢复执行。

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

出错后恢复

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

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

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)

出错后恢复

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

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

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 上连续调用之间的信息。

这允许使用 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 为 None）
my_workflow.invoke(2, config)  # 3（previous 是上一次调用返回的 1）

这允许使用 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 为 undefined）
await myWorkflow.invoke(2, config); // 3（previous 是上一次调用返回的 1）

entrypoint.final

entrypoint.final 是一个特殊的原语，可以从 entrypoint 返回，它允许将保存在检查点中的值与 entrypoint 的返回值进行解耦。

第一个值是 entrypoint 的返回值，第二个值是将保存在检查点中的值。类型注解为 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
    # 这将把 previous 值返回给调用方，同时将
    # 2 * number 保存到检查点，它将在下一次调用中
    # 用于 `previous` 参数。
    return entrypoint.final(value=previous, save=2 * number)

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

my_workflow.invoke(3, config)  # 0（previous 为 None）
my_workflow.invoke(1, config)  # 6（previous 是上一次调用保存的 3 * 2）

entrypoint.final 是一个特殊的原语，可以从 entrypoint 返回，它允许将保存在检查点中的值与 entrypoint 的返回值进行解耦。

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

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

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (number: number) => {
    const previous = getPreviousState<number>() ?? 0;
    // 这将把 previous 值返回给调用方，同时将
    // 2 * number 保存到检查点，它将在下一次调用中
    // 用于 `previous` 参数。
    return entrypoint.final({
      value: previous,
      save: 2 * number,
    });
  }
);

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

await myWorkflow.invoke(3, config); // 0（previous 为 undefined）
await myWorkflow.invoke(1, config); // 6（previous 是上一次调用保存的 3 * 2）

Task

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

• 异步执行：task 被设计为异步执行，允许多个操作并发运行而不会阻塞。
• 检查点持久化：task 结果被保存到检查点中，使工作流可以从最后保存的状态恢复。（更多细节请参阅持久化）。

定义

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

from langgraph.func import task

@task()
def slow_computation(input_value):
    # 模拟长时间运行的操作
    ...
    return result

task 使用 task 函数定义，该函数包装一个常规函数。

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

const slowComputation = task("slowComputation", async (inputValue: any) => {
  // 模拟长时间运行的操作
  return result;
});

WARNING

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

执行

task 只能在entrypoint、另一个task 或状态图节点内部调用。

task_不能_直接从主应用程序代码中调用。

当你调用task 时，它会_立即_返回一个 future 对象。future 是一个稍后才会有结果的占位符。

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

同步调用

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

异步调用

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

当你调用task 时，它会返回一个可以被 await 的 Promise。

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

何时使用 task

task 在以下场景中很有用：

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

序列化

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

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

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

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

序列化确保工作流状态（例如 task 结果和中间值）可以被可靠地保存和恢复。这对于启用人在回路交互、容错和并行执行至关重要。

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

确定性

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

对于 Functional API，重放从 entrypoint 的开头开始，而 LangGraph 从检查点恢复已完成的 task 和 subgraph 结果，而不是重新计算它们。这样可以跨暂停保持记录下来的步骤顺序，包括长时间运行或非确定性的 task 输出。

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

工作流的不同运行可能会产生不同的结果，但恢复特定线程应该重放相同的、已持久化的 task 和 subgraph 结果。

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

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

幂等性

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

常见陷阱

处理副作用

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

不正确

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

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    # 恢复工作流时，这段代码会被第二次执行。
    # 这很可能不是你想要的结果。
    with open("output.txt", "w") as f:  
        f.write("Side effect executed")  
    value = interrupt("question")
    return value

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

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow },
  async (inputs: Record<string, any>) => {
    // 恢复工作流时，这段代码会被第二次执行。
    // 这很可能不是你想要的结果。
    fs.writeFileSync("output.txt", "Side effect executed");
    const value = interrupt("question");
    return value;
  }
);

正确

在此示例中，副作用被封装在 task 中，确保恢复时执行的一致性。

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:
    # 副作用现在被封装在 task 中。
    write_to_file().result()
    value = interrupt("question")
    return value

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>) => {
    // 副作用现在被封装在 task 中。
    await writeToFile();
    const value = interrupt("question");
    return value;
  }
);

非确定性控制流

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

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

当使用带有多个中断调用的人在回路工作流时，这一点尤其重要。LangGraph 为每个 task/entrypoint 保留一个 resume 值列表。当遇到中断时，会将其与对应的 resume 值匹配。这种匹配严格基于索引，因此 resume 值的顺序应该与中断的顺序一致。

如果在恢复时没有保持执行顺序，某个 interrupt 调用可能会与错误的 resume 值匹配，从而导致不正确的结果。

更多细节请阅读确定性部分。

不正确

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

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
    }

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

正确

在此示例中，工作流使用输入 `t0` 来决定执行哪个 task。这是确定性的，因为工作流的结果只取决于输入。

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
    }

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

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
• Graph API 概念概述
• 在 Graph API 与 Functional API 之间选择