函数式 API 概述

函数式 API 允许你以最小的代码更改将 LangGraph 的关键功能（持久化、记忆、人机协作和流式输出）添加到你的应用中。

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

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

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

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

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

函数式 API vs. 图 API

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

以下是一些关键区别：

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

示例

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

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, // 来自人机协作的响应
    };
  }
);

详细说明

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

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

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

{ 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'
  }]
}

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

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

{ workflow: { essay: 'An essay about topic: cat', isApproved: true } }

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

入口点

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

定义

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

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

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

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

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

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

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

执行

使用 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"
  }
};

for await (const chunk of myWorkflow.stream(someInput, config)) {
  console.log(chunk);
}

恢复

在中断后恢复执行可以通过将恢复值传递给 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.stream(
  new Command({ resume: someResumableValue }),
  config,
)

for await (const chunk of stream) {
  console.log(chunk);
}

错误后恢复

要在错误后恢复，使用 null 和相同的线程 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"
  }
};

for await (const chunk of myWorkflow.stream(null, config)) {
  console.log(chunk);
}

短期记忆

当 entrypoint 使用 checkpointer 定义时，它在检查点中存储相同线程 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 为 undefined）
await myWorkflow.invoke(2, config); // 3（previous 为上次调用的 1）

entrypoint.final

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

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

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 调用或数据处理步骤。它有两个关键特征：

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

定义

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

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

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

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

执行

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

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

当你调用一个任务时，它返回一个可以等待的 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 可序列化的。

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

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

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

确定性

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

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

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

幂等性

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

常见陷阱

处理副作用

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

错误示例

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

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

正确示例

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

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

非确定性控制流

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

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

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

如果恢复时执行顺序未被维护，一个 interrupt 调用可能与错误的 resume 值匹配，导致不正确的结果。

请阅读确定性部分了解更多详情。

错误示例

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

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 来确定执行哪个任务。这是确定性的，因为工作流的结果仅取决于输入。

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

了解更多

• 如何使用函数式 API
• 图 API 概念概述
• 在图 API 和函数式 API 之间选择

---

将这些文档连接到 Claude、VSCode 等工具，通过 MCP 获取实时答案。

在 GitHub 上编辑此页面 或 提交问题。