Functional API 概述

Functional API 允许您以最少的代码更改，将 LangGraph 的关键功能（持久性、记忆、人机交互和流式传输）添加到您的应用程序中。它旨在将这些功能集成到现有代码中，这些代码可能使用标准的语言原语进行分支和控制流，例如 if 语句、for 循环和函数调用。与许多需要将代码重构为明确的流水线或 DAG 的数据编排框架不同，Functional API 允许您在不强制执行僵化执行模型的情况下整合这些能力。 Functional API 使用两个关键构建块：

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

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

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

Functional API 与 Graph API 对比

对于偏好更具声明式方法的用户，LangGraph 的 Graph API 允许您使用图范式定义工作流。这两种 API 共享相同的底层运行时，因此您可以在同一个应用程序中一起使用它们。以下是一些关键区别：

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

示例

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

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

详细解释

这个工作流将撰写一篇关于“猫”的论文，然后暂停以获取人工审阅。该工作流可以被中断任意长时间，直到提供审阅。当工作流恢复时，它会从头开始执行，但由于 writeEssay 任务的结果已经保存，任务结果将从检查点加载，而不是重新计算。

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

const writeEssay = task("writeEssay", async (topic: string) => {
  // 这是一个长时间运行任务的占位符。
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return `关于${topic}的作文`;
});

const workflow = entrypoint(
  { checkpointer: new MemorySaver(), name: "workflow" },
  async (_topic: string) => {
    const essay = await writeEssay("cat");
    const isApproved = interrupt({
      // 提供给 interrupt 作为参数的任意 JSON 可序列化负载。
      // 当从工作流流式传输数据时，它将在客户端作为 Interrupt 呈现。
      essay, // 我们希望审查的作文。
      // 我们可以添加任何需要的额外信息。
      // 例如，引入一个名为 "action" 的键并附带一些指示。
      action: "请批准/拒绝这篇作文",
    });

    return {
      essay, // 已生成的作文
      isApproved, // 来自 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: "关于cat的作文" }
// { __interrupt__: [Interrupt(...)] }

一篇论文已撰写完成，准备审阅。提供审阅后，我们可以恢复工作流：

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

// 获取用户评审（例如，通过用户界面）
// 在本例中，我们使用的是布尔值，但这可以是任何 JSON 可序列化的值。
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: "关于主题“猫”的文章", isApproved: true }

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

Entrypoint

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

定义

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

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

执行

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

调用
流

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

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

恢复

在 interrupt 之后恢复执行，可以通过将恢复值传递给 Command 原语来完成。

调用
流

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

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

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

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

出错后恢复 要在出错后恢复，请使用 null 和相同的线程 ID（config）运行 entrypoint。这假设根本的错误已解决，并且执行可以成功继续。

调用
流

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

await myWorkflow.invoke(null, config);

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

短期记忆

当使用 checkpointer 定义 entrypoint 时，它会在同一线程 ID 的连续调用之间将信息存储在检查点中。这允许使用 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`

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)

Task

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

异步执行：任务设计为可异步执行，允许多个操作并发运行而不阻塞。
检查点：任务结果被保存到检查点，使得工作流可以从最后保存的状态恢复。（更多详情参见持久性）。

定义

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

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

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

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

执行

任务只能从 entrypoint、另一个 task 或状态图节点内部调用。任务不能直接从主应用程序代码调用。当您调用一个 task 时，它会返回一个可以等待的 Promise。

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

何时使用 task

Tasks 在以下场景中很有用：

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

序列化

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

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

这些要求对于启用检查点和工作流恢复是必要的。使用基本类型，如对象、数组、字符串、数字和布尔值，以确保您的输入和输出是可序列化的。序列化确保工作流状态（例如任务结果和中间值）可以可靠地保存和恢复。这对于实现人机交互、容错和并行执行至关重要。当工作流配置了 checkpointer 时，提供不可序列化的输入或输出将导致运行时错误。

确定性

当您恢复工作流运行时，代码不会从执行停止的同一行代码恢复。执行返回到一个检查点边界，并且工作流回放直到再次到达暂停点。对于 Functional API，回放从 entrypoint 的开头开始，而 LangGraph 从检查器恢复已完成的 task 和子图结果，而不是重新计算它们。这保留了跨暂停的记录步骤顺序，包括长时间运行或非确定性的 task 输出。要使用人机交互等功能，您必须将非确定性工作（例如，随机值）和副作用（例如，文件写入或 API 调用）放在 tasks 中。工作流的不同运行可能产生不同的结果，但恢复特定线程应该回放相同的持久化 task 和子图结果。为确保您的工作流是确定性的并且可以一致地回放，请遵循以下准则：

避免重复工作：在 entrypoint 中，如果您链接了多个副作用（例如，日志记录、文件写入或网络调用），请为每个副作用创建单独的 task，以便恢复时从检查点恢复它们的输出，而不是再次运行。
封装非确定性操作：将可能在两次尝试之间变化的值（例如，随机数或系统时间读取）放在 tasks 内部，以便回放与检查点的内容对齐。
使用幂等操作：对于部分任务失败和重试，请参阅幂等性。

幂等性

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

常见陷阱

处理副作用

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

错误
正确

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

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

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

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

非确定性控制流

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

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

这在使用具有多个 interrupt 调用的人机交互工作流时尤其重要。LangGraph 为每个 task/entrypoint 维护一个恢复值列表。当遇到一个 interrupt 时，它会与相应的恢复值进行匹配。这种匹配是严格基于索引的，因此恢复值的顺序应与 interrupt 的顺序匹配。如果在恢复时执行顺序未被维持，一个 interrupt 调用可能会与错误的 resume 值匹配，导致错误的结果。请阅读确定性部分以了解更多细节。

错误
正确

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

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

了解更多

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Capabilities

Production

Frontend

LangGraph APIs

Functional API 与 Graph API 对比

示例

Entrypoint

定义

执行

恢复

短期记忆

`entrypoint.final`

Task

定义

执行

何时使用 task

序列化

确定性

幂等性

常见陷阱

处理副作用

非确定性控制流

了解更多

​Functional API 与 Graph API 对比

​示例

​Entrypoint

​定义

​执行

​恢复

​短期记忆

​entrypoint.final

​Task

​定义

​执行

​何时使用 task

​序列化

​确定性

​幂等性

​常见陷阱

​处理副作用

​非确定性控制流

​了解更多

Functional API 与 Graph API 对比

示例

Entrypoint

定义

执行

恢复

短期记忆

`entrypoint.final`

Task

定义

执行

何时使用 task

序列化

确定性

幂等性

常见陷阱

处理副作用

非确定性控制流

了解更多