Functional API 概览 - Docs by LangChain中文

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

Detailed Explanation

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

import { v7 as uuid7 } from "uuid";
import { MemorySaver, entrypoint, task, interrupt } 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(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
    };
  }
);

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: 'Please approve/reject the essay' },
    resumable: true,
    ns: ['workflow:f7b8508b-21c0-8b4c-5958-4e8de74d2684'],
    when: 'during'
  }]
}

文章已经写好，可以审阅了。一旦提供审阅，我们就可以恢复工作流：

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 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 函数并传入配置和一个函数来定义的。该函数必须接受一个单一的位置参数，该参数作为工作流输入。如果需要传递多个数据，请使用对象作为第一个参数的输入类型。使用函数创建入口点会生成一个工作流实例，该实例有助于管理工作流的执行（例如，处理流式传输、恢复和检查点）。你通常需要将检查点保存器传递给 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 方法执行。

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

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

恢复

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

Invoke
Stream

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.stream(
  new Command({ resume: someResumableValue }),
  config,
)

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

错误后恢复 要在错误后恢复，请使用 null 和相同的线程 id（配置）运行 entrypoint。这假设底层的错误已解决，并且可以继续成功执行。

Invoke
Stream

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

await myWorkflow.invoke(null, config);

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

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

短期记忆

当使用 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)

任务

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

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

定义

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

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

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

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

执行

任务只能在入口点、另一个任务或状态图节点内部调用。任务_不能_直接从主应用程序代码调用。当你调用一个任务时，它会返回一个可以被 await 的 Promise。

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

何时使用任务

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

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

序列化

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

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

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

确定性

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

避免重复工作：在入口点中，如果你链式调用多个副作用（例如，日志记录、文件写入或网络调用），为每个副作用赋予自己的任务，这样恢复时就可以从检查点保存器恢复其输出，而不是再次运行它们。
封装非确定性操作：将可能在尝试之间发生变化的值（例如，随机数或墙上时钟读数）保留在任务内部，以便重放与检查点记录的内容一致。
使用幂等操作：对于部分任务失败和重试，请参见幂等性。

幂等性

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

常见陷阱

处理副作用

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

Incorrect
Correct

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

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

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

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 值匹配，导致不正确的结果。请阅读确定性部分了解更多详细信息。

Incorrect
Correct

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

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

了解更多

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

Edit this page on GitHub or file an issue.

​Functional API 与 Graph API

​示例

​Entrypoint

​定义

​执行

​恢复

​短期记忆

​entrypoint.final

​任务

​定义

​执行

​何时使用任务

​序列化

​确定性

​幂等性

​常见陷阱

​处理副作用

​非确定性控制流

​了解更多

Functional API 与 Graph API

示例

Entrypoint

定义

执行

恢复

短期记忆

`entrypoint.final`

任务

定义

执行

何时使用任务

序列化

确定性

幂等性

常见陷阱

处理副作用

非确定性控制流

了解更多