> ## Documentation Index
> Fetch the complete documentation index at: https://nvd-54.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 函数式 API 概述

**函数式 API** 允许你以最小的代码更改将 LangGraph 的关键功能（[持久化](/oss/javascript/langgraph/persistence)、[记忆](/oss/javascript/langgraph/add-memory)、[人机协作](/oss/javascript/langgraph/interrupts)和[流式输出](/oss/javascript/langgraph/streaming)）添加到你的应用中。

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

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

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

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

<Tip>
  有关如何使用函数式 API 的信息，请参阅[使用函数式 API](/oss/javascript/langgraph/use-functional-api)。
</Tip>

## 函数式 API vs. 图 API

对于偏好更声明式方法的用户，LangGraph 的[图 API](/oss/javascript/langgraph/graph-api) 允许你使用图范式定义工作流。两种 API 共享相同的底层运行时，因此你可以在同一个应用中同时使用它们。

以下是一些关键区别：

* **控制流**：函数式 API 不需要考虑图结构。你可以使用标准的 JavaScript 结构来定义工作流。这通常会减少你需要编写的代码量。
* **短期记忆**：**图 API** 需要声明一个[**状态**](/oss/javascript/langgraph/graph-api#state)，并且可能需要定义[**归约器**](/oss/javascript/langgraph/graph-api#reducers)来管理图状态的更新。`@entrypoint` 和 `@tasks` 不需要显式的状态管理，因为它们的状态局限于函数且不跨函数共享。
* **检查点**：两种 API 都生成和使用检查点。在**图 API** 中，每个[超级步骤](/oss/javascript/langgraph/graph-api)后会生成新的检查点。在**函数式 API** 中，当任务执行时，其结果会保存到与给定入口点关联的现有检查点中，而不是创建新的检查点。
* **可视化**：图 API 使得将工作流可视化为图变得容易，这对调试、理解工作流和与他人共享很有用。函数式 API 不支持可视化，因为图是在运行时动态生成的。

## 示例

下面我们演示一个简单的应用，它写一篇文章并[中断](/oss/javascript/langgraph/interrupts)以请求人工审核。

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import { MemorySaver, entrypoint, task, interrupt } from "@langchain/langgraph";

const writeEssay = task("writeEssay", async (topic: string) => {
  // 长时间运行任务的占位符。
  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: "请批准/拒绝这篇文章",
    });

    return {
      essay, // 生成的文章
      isApproved, // 来自人机协作的响应
    };
  }
);
```

<Accordion title="详细说明">
  此工作流将写一篇关于主题"cat"的文章，然后暂停以从人工获取审核。工作流可以无限期中断，直到提供审核。

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

  ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  import { v7 as uuid7 } from "uuid";
  import { MemorySaver, entrypoint, task, interrupt } from "@langchain/langgraph";

  const writeEssay = task("writeEssay", async (topic: string) => {
    // 这是长时间运行任务的占位符。
    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({
        essay, // 我们想要审核的文章。
        action: "请批准/拒绝这篇文章",
      });

      return {
        essay,
        isApproved,
      };
    }
  );

  const threadId = uuid7();

  const config = {
    configurable: {
      thread_id: threadId
    }
  };

  for await (const item of workflow.stream("cat", config)) {
    console.log(item);
  }
  ```

  ```console theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  { writeEssay: 'An essay about topic: cat' }
  {
    __interrupt__: [{
      value: { essay: 'An essay about topic: cat', action: '请批准/拒绝这篇文章' },
      resumable: true,
      ns: ['workflow:f7b8508b-21c0-8b4c-5958-4e8de74d2684'],
      when: 'during'
    }]
  }
  ```

  文章已写好并准备好审核。一旦提供审核，我们可以恢复工作流：

  ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  import { Command } from "@langchain/langgraph";

  // 从用户获取审核（例如，通过 UI）
  // 在本例中，我们使用布尔值，但这可以是任何 JSON 可序列化的值。
  const humanReview = true;

  const stream = await workflow.stream(
    new Command({ resume: humanReview }),
    config
  );
  for await (const item of stream) {
    console.log(item);
  }
  ```

  ```console theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  { workflow: { essay: 'An essay about topic: cat', isApproved: true } }
  ```

  工作流已完成，审核已添加到文章中。
</Accordion>

## 入口点

[`entrypoint`](https://reference.langchain.com/javascript/langchain-langgraph/index/entrypoint) 函数可用于从函数创建工作流。它封装工作流逻辑并管理执行流程，包括处理\_长时间运行的任务\_和[中断](/oss/javascript/langgraph/interrupts)。

### 定义

**入口点**通过调用 `entrypoint` 函数并传入配置和函数来定义。

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

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

你通常需要将**检查点**传递给 `entrypoint` 函数以启用持久化和使用**人机协作**等功能。

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import { entrypoint } from "@langchain/langgraph";

const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (someInput: Record<string, any>): Promise<number> => {
    // 可能涉及长时间运行任务（如 API 调用）的逻辑，
    // 并且可能因人机协作而中断
    return result;
  }
);
```

<Warning>
  **序列化**
  入口点的**输入**和**输出**必须是 JSON 可序列化的，以支持检查点。请参阅[序列化](#serialization)部分了解更多详情。
</Warning>

### 执行

使用 [`entrypoint`](#entrypoint) 函数将返回一个可以使用 `invoke` 和 `stream` 方法执行的对象。

<Tabs>
  <Tab title="Invoke">
    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    const config = {
      configurable: {
        thread_id: "some_thread_id"
      }
    };
    await myWorkflow.invoke(someInput, config); // 等待结果
    ```
  </Tab>

  <Tab title="Stream">
    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    const config = {
      configurable: {
        thread_id: "some_thread_id"
      }
    };

    for await (const chunk of myWorkflow.stream(someInput, config)) {
      console.log(chunk);
    }
    ```
  </Tab>
</Tabs>

### 恢复

在[中断](https://reference.langchain.com/javascript/langchain-langgraph/index/interrupt)后恢复执行可以通过将**恢复**值传递给 [`Command`](https://reference.langchain.com/javascript/langchain-langgraph/index/Command) 原语来完成。

<Tabs>
  <Tab title="Invoke">
    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    import { Command } from "@langchain/langgraph";

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

    await myWorkflow.invoke(new Command({ resume: someResumeValue }), config);
    ```
  </Tab>

  <Tab title="Stream">
    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    import { Command } from "@langchain/langgraph";

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

    const stream = await myWorkflow.stream(
      new Command({ resume: someResumableValue }),
      config,
    )

    for await (const chunk of stream) {
      console.log(chunk);
    }
    ```
  </Tab>
</Tabs>

**错误后恢复**

要在错误后恢复，使用 `null` 和相同的**线程 ID**（config）运行 `entrypoint`。

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

<Tabs>
  <Tab title="Invoke">
    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    const config = {
      configurable: {
        thread_id: "some_thread_id"
      }
    };

    await myWorkflow.invoke(null, config);
    ```
  </Tab>

  <Tab title="Stream">
    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    const config = {
      configurable: {
        thread_id: "some_thread_id"
      }
    };

    for await (const chunk of myWorkflow.stream(null, config)) {
      console.log(chunk);
    }
    ```
  </Tab>
</Tabs>

### 短期记忆

当 `entrypoint` 使用 `checkpointer` 定义时，它在[检查点](/oss/javascript/langgraph/persistence#checkpoints)中存储相同**线程 ID** 的连续调用之间的信息。

这允许使用 `getPreviousState` 函数访问上一次调用的状态。

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

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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`](https://reference.langchain.com/javascript/functions/_langchain_langgraph.index.entrypoint.html#final) 是一个特殊原语，可以从入口点返回，允许将**保存在检查点中的值**与**入口点的返回值** **解耦**。

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

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
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）
```

## 任务

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

* **异步执行**：任务设计为异步执行，允许多个操作并发运行而不阻塞。
* **检查点**：任务结果保存到检查点，使工作流能从最后保存的状态恢复。（更多详情请参阅[持久化](/oss/javascript/langgraph/persistence)）。

### 定义

任务使用 `task` 函数定义，它包装一个普通函数。

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
import { task } from "@langchain/langgraph";

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

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

### 执行

**任务**只能从**入口点**、另一个**任务**或[状态图节点](/oss/javascript/langgraph/graph-api#nodes)内部调用。

任务\_不能\_直接从主应用代码调用。

当你调用一个**任务**时，它返回一个可以等待的 Promise。

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
const myWorkflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (someInput: number): Promise<number> => {
    return await slowComputation(someInput);
  }
);
```

## 何时使用任务

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

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

## 序列化

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

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

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

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

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

## 确定性

要利用**人机协作**等功能，任何随机性都应封装在**任务**内。这保证当执行被暂停（例如，用于人机协作）然后恢复时，它将遵循相同的\_步骤序列\_，即使**任务**结果是非确定性的。

LangGraph 通过在执行时持久化**任务**和[**子图**](/oss/javascript/langgraph/use-subgraphs)结果来实现此行为。设计良好的工作流确保恢复执行遵循相同的\_步骤序列\_，允许正确检索之前计算的结果而无需重新执行它们。这对于长时间运行的**任务**或具有非确定性结果的**任务**特别有用，因为它避免了重复之前完成的工作，并允许从基本相同的位置恢复。

虽然工作流的不同运行可能产生不同的结果，但恢复**特定**运行应始终遵循相同的记录步骤序列。这允许 LangGraph 高效地查找在图被中断之前执行的**任务**和**子图**结果，并避免重新计算它们。

## 幂等性

幂等性确保多次运行相同操作产生相同的结果。如果由于故障而重新运行某个步骤，这有助于防止重复的 API 调用和冗余处理。始终将 API 调用放在**任务**函数中以进行检查点，并将它们设计为在重新执行时幂等。如果**任务**开始但未成功完成，则可能发生重新执行。然后，如果工作流恢复，**任务**将再次运行。使用幂等性键或验证现有结果以避免重复。

## 常见陷阱

### 处理副作用

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

<Tabs>
  <Tab title="错误示例">
    在此示例中，副作用（写入文件）直接包含在工作流中，因此在恢复工作流时会被第二次执行。

    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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;
      }
    );
    ```
  </Tab>

  <Tab title="正确示例">
    在此示例中，副作用封装在任务中，确保恢复时一致执行。

    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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>) => {
        // 副作用现在封装在任务中。
        await writeToFile();
        const value = interrupt("question");
        return value;
      }
    );
    ```
  </Tab>
</Tabs>

### 非确定性控制流

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

* 在任务中：获取随机数 (5) → 中断 → 恢复 → (再次返回 5) → ...
* 不在任务中：获取随机数 (5) → 中断 → 恢复 → 获取新的随机数 (7) → ...

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

如果恢复时执行顺序未被维护，一个 [`interrupt`](https://reference.langchain.com/javascript/langchain-langgraph/index/interrupt) 调用可能与错误的 `resume` 值匹配，导致不正确的结果。

请阅读[确定性](#determinism)部分了解更多详情。

<Tabs>
  <Tab title="错误示例">
    在此示例中，工作流使用当前时间来确定执行哪个任务。这是非确定性的，因为工作流的结果取决于执行时的时间。

    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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 };
        }
      }
    );
    ```
  </Tab>

  <Tab title="正确示例">
    在此示例中，工作流使用输入 `t0` 来确定执行哪个任务。这是确定性的，因为工作流的结果仅取决于输入。

    ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    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 };
        }
      }
    );
    ```
  </Tab>
</Tabs>

## 了解更多

* [如何使用函数式 API](/oss/javascript/langgraph/use-functional-api)
* [图 API 概念概述](/oss/javascript/langgraph/graph-api)
* [在图 API 和函数式 API 之间选择](/oss/javascript/langgraph/choosing-apis)

***

<div className="source-links">
  <Callout icon="terminal-2">
    [将这些文档连接](/use-these-docs)到 Claude、VSCode 等工具，通过 MCP 获取实时答案。
  </Callout>

  <Callout icon="edit">
    [在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/functional-api.mdx) 或 [提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。
  </Callout>
</div>