使用函数式 API

函数式 API 允许您以最少的代码改动，将 LangGraph 的关键特性（持久化、记忆、人在环中和流式传输）添加到您的应用程序中。

有关函数式 API 的概念性信息，请参阅函数式 API。

创建一个简单的工作流

定义 entrypoint 时，输入仅限于函数的第一个参数。要传递多个输入，可以使用字典。

const checkpointer = new MemorySaver();

const myWorkflow = entrypoint(
  { checkpointer, name: "myWorkflow" },
  async (inputs: { value: number; anotherValue: number }) => {
    const value = inputs.value;
    const anotherValue = inputs.anotherValue;
    // ...
  }
);

await myWorkflow.invoke({ value: 1, anotherValue: 2 });

扩展示例：简单工作流

import { v7 as uuid7 } from "uuid";
import { entrypoint, task, MemorySaver } from "@langchain/langgraph";

// 检查一个数字是否为偶数的任务
const isEven = task("isEven", async (number: number) => {
  return number % 2 === 0;
});

// 格式化消息的任务
const formatMessage = task("formatMessage", async (isEven: boolean) => {
  return isEven ? "The number is even." : "The number is odd.";
});

// 创建一个用于持久化的检查点器
const checkpointer = new MemorySaver();

const workflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (inputs: { number: number }) => {
    // 简单的工作流，用于对数字进行分类
    const even = await isEven(inputs.number);
    return await formatMessage(even);
  }
);

// 使用唯一的线程 ID 运行工作流
const config = { configurable: { thread_id: uuid7() } };
const result = await workflow.invoke({ number: 7 }, config);
console.log(result);

扩展示例：使用 LLM 撰写文章

此示例展示了如何在语法上使用 @task 和 @entrypoint 装饰器。假设提供了检查点器，工作流结果将被持久化到检查点器中。

import { v7 as uuid7 } from "uuid";
import { ChatOpenAI } from "@langchain/openai";
import { entrypoint, task, MemorySaver } from "@langchain/langgraph";

const model = new ChatOpenAI({ model: "gpt-3.5-turbo" });

// 任务：使用 LLM 生成文章
const composeEssay = task("composeEssay", async (topic: string) => {
  // 生成一篇关于给定主题的文章
  const response = await model.invoke([
    { role: "system", content: "You are a helpful assistant that writes essays." },
    { role: "user", content: `Write an essay about ${topic}.` }
  ]);
  return response.content as string;
});

// 创建一个用于持久化的检查点器
const checkpointer = new MemorySaver();

const workflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (topic: string) => {
    // 使用 LLM 生成文章的简单工作流
    return await composeEssay(topic);
  }
);

// 执行工作流
const config = { configurable: { thread_id: uuid7() } };
const result = await workflow.invoke("the history of flight", config);
console.log(result);

并行执行

任务可以通过并发调用并等待结果来并行执行。这对于提高 I/O 密集型任务（例如调用 LLM API）的性能非常有用。

const addOne = task("addOne", async (number: number) => {
  return number + 1;
});

const graph = entrypoint(
  { checkpointer, name: "graph" },
  async (numbers: number[]) => {
    return await Promise.all(numbers.map(addOne));
  }
);

扩展示例：并行 LLM 调用

此示例展示了如何使用 @task 并行运行多个 LLM 调用。每个调用生成关于不同主题的段落，并将结果合并为单个文本输出。

import { v7 as uuid7 } from "uuid";
import { ChatOpenAI } from "@langchain/openai";
import { entrypoint, task, MemorySaver } from "@langchain/langgraph";

// 初始化 LLM 模型
const model = new ChatOpenAI({ model: "gpt-3.5-turbo" });

// 生成关于给定主题的段落的任务
const generateParagraph = task("generateParagraph", async (topic: string) => {
  const response = await model.invoke([
    { role: "system", content: "You are a helpful assistant that writes educational paragraphs." },
    { role: "user", content: `Write a paragraph about ${topic}.` }
  ]);
  return response.content as string;
});

// 创建一个用于持久化的检查点器
const checkpointer = new MemorySaver();

const workflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (topics: string[]) => {
    // 并行生成多个段落并进行合并
    const paragraphs = await Promise.all(topics.map(generateParagraph));
    return paragraphs.join("\n\n");
  }
);

// 运行工作流
const config = { configurable: { thread_id: uuid7() } };
const result = await workflow.invoke(["quantum computing", "climate change", "history of aviation"], config);
console.log(result);

此示例使用 LangGraph 的并发模型来提高执行时间，特别是当任务涉及像 LLM 补全这样的 I/O 时。

调用图

函数式 API 和 图 API 可以在同一个应用程序中一起使用，因为它们共享相同的底层运行时。

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

const builder = new StateGraph(/* ... */);
// ...
const someGraph = builder.compile();

const someWorkflow = entrypoint(
  { name: "someWorkflow" },
  async (someInput: Record<string, any>) => {
    // 调用一个使用图 API 定义的图
    const result1 = await someGraph.invoke(/* ... */);
    // 调用另一个使用图 API 定义的图
    const result2 = await anotherGraph.invoke(/* ... */);
    return {
      result1,
      result2,
    };
  }
);

扩展示例：从函数式 API 调用一个简单图

import { v7 as uuid7 } from "uuid";
import { entrypoint, MemorySaver, StateGraph, StateSchema } from "@langchain/langgraph";
import * as z from "zod";

// 定义共享状态类型
const State = new StateSchema({
  foo: z.number(),
});

// 使用图 API 构建图
const builder = new StateGraph(State)
  .addNode("double", (state) => {
    return { foo: state.foo * 2 };
  })
  .addEdge("__start__", "double");
const graph = builder.compile();

// 定义函数式 API 工作流
const checkpointer = new MemorySaver();

const workflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (x: number) => {
    const result = await graph.invoke({ foo: x });
    return { bar: result.foo };
  }
);

// 执行工作流
const config = { configurable: { thread_id: uuid7() } };
console.log(await workflow.invoke(5, config)); // 输出: { bar: 10 }

调用其他入口点

您可以在一个 entrypoint 或 task 内部调用其他 entrypoint。

// 将自动使用父入口点的检查点器
const someOtherWorkflow = entrypoint(
  { name: "someOtherWorkflow" },
  async (inputs: { value: number }) => {
    return inputs.value;
  }
);

const myWorkflow = entrypoint(
  { checkpointer, name: "myWorkflow" },
  async (inputs: { value: number }) => {
    const value = await someOtherWorkflow.invoke({ value: 1 });
    return value;
  }
);

扩展示例：调用另一个入口点

import { v7 as uuid7 } from "uuid";
import { entrypoint, MemorySaver } from "@langchain/langgraph";

// 初始化一个检查点器
const checkpointer = new MemorySaver();

// 一个可重用的子工作流，用于将数字相乘
const multiply = entrypoint(
  { name: "multiply" },
  async (inputs: { a: number; b: number }) => {
    return inputs.a * inputs.b;
  }
);

// 调用子工作流的主工作流
const main = entrypoint(
  { checkpointer, name: "main" },
  async (inputs: { x: number; y: number }) => {
    const result = await multiply.invoke({ a: inputs.x, b: inputs.y });
    return { product: result };
  }
);

// 执行主工作流
const config = { configurable: { thread_id: uuid7() } };
console.log(await main.invoke({ x: 6, y: 7 }, config)); // 输出: { product: 42 }

流式传输

函数式 API 使用与 图 API 相同的流式传输机制。更多详情请阅读 流式传输指南 部分。使用流式 API 从工作流运行中流式传输数据块的示例。

const config = {
  configurable: { thread_id: "functional-api-stream-custom-data" },
};

const stream = await main.streamEvents({ x: 5 }, { ...config, version: "v3" });
for await (const chunk of stream.values) {
  console.log(chunk);
}
// 输出为10

在计算开始之前发出自定义数据。
在计算结果之后发出另一个自定义消息。
使用 streamEvents() 处理流式输出。
迭代 stream.values 以接收值快照。

重试策略

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

// 此变量仅用于演示目的，以模拟网络故障。
// 在您的实际代码中不会有此变量。
let attempts = 0;

// 让我们配置 RetryPolicy 以在 ValueError 时重试。
// 默认的 RetryPolicy 针对重试特定网络错误进行了优化。
const retryPolicy: RetryPolicy = { retryOn: (error) => error instanceof Error };

const getInfo = task(
  {
    name: "getInfo",
    retry: retryPolicy,
  },
  () => {
    attempts += 1;

    if (attempts < 2) {
      throw new Error("Failure");
    }
    return "OK";
  }
);

const checkpointer = new MemorySaver();

const main = entrypoint(
  { checkpointer, name: "main" },
  async (inputs: Record<string, any>) => {
    return await getInfo();
  }
);

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

await main.invoke({ any_input: "foobar" }, config);

'OK'

缓存任务

import {
  InMemoryCache,
  entrypoint,
  task,
  CachePolicy,
} from "@langchain/langgraph";

const slowAdd = task(
  {
    name: "slowAdd",
    cache: { ttl: 120 },
  },
  async (x: number) => {
    await new Promise((resolve) => setTimeout(resolve, 1000));
    return x * 2;
  }
);

const main = entrypoint(
  { cache: new InMemoryCache(), name: "main" },
  async (inputs: { x: number }) => {
    const result1 = await slowAdd(inputs.x);
    const result2 = await slowAdd(inputs.x);
    return { result1, result2 };
  }
);

const stream = await main.streamEvents({ x: 5 }, { version: "v3" });
for await (const snapshot of stream.values) {
  console.log(snapshot);
}

ttl 以秒为单位指定。缓存将在此时间后失效。

出错后恢复

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

// 此变量仅用于演示目的，以模拟网络故障。
// 在您的实际代码中不会有此变量。
let attempts = 0;

const getInfo = task("getInfo", async () => {
  /**
   * 模拟一个任务，在成功之前会失败一次。
   * 第一次尝试时抛出异常，然后在后续尝试中返回 "OK"。
   */
  attempts += 1;

  if (attempts < 2) {
    throw new Error("Failure"); // 模拟第一次尝试失败
  }
  return "OK";
});

// 初始化一个用于持久化的内存中检查点器
const checkpointer = new MemorySaver();

const slowTask = task("slowTask", async () => {
  /**
   * 通过引入 1 秒延迟来模拟运行缓慢的任务。
   */
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return "Ran slow task.";
});

const main = entrypoint(
  { checkpointer, name: "main" },
  async (inputs: Record<string, any>) => {
    /**
     * 主工作流函数，按顺序运行 slowTask 和 getInfo 任务。
     *
     * 参数:
     * - inputs: 包含工作流输入值的 Record<string, any>。
     *
     * 工作流首先执行 `slowTask`，然后尝试执行 `getInfo`，
     * 后者在第一次调用时将失败。
     */
    const slowTaskResult = await slowTask(); // 对 slowTask 的阻塞调用
    await getInfo(); // 第一次尝试时这里将抛出异常
    return slowTaskResult;
  }
);

// 工作流执行配置，带有唯一的线程标识符
const config = {
  configurable: {
    thread_id: "1", // 用于跟踪工作流执行的唯一标识符
  },
};

// 此次调用将耗时约 1 秒，因为 slowTask 执行
try {
  // 第一次调用将引发异常，因为 `getInfo` 任务失败
  await main.invoke({ any_input: "foobar" }, config);
} catch (err) {
  // 优雅地处理失败
}

当我们恢复执行时，我们不需要重新运行 slowTask，因为其结果已保存在检查点中。

await main.invoke(null, config);

'Ran slow task.'

人在环中

函数式 API 使用 interrupt 函数和 Command 原语支持人在环中工作流。

基本人在环中工作流

我们将创建三个任务：

附加 "bar"。
暂停以等待人工输入。恢复时，附加人工输入。
附加 "qux"。

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

const step1 = task("step1", async (inputQuery: string) => {
  // 附加 bar
  return `${inputQuery} bar`;
});

const humanFeedback = task("humanFeedback", async (inputQuery: string) => {
  // 附加用户输入
  const feedback = interrupt(`Please provide feedback: ${inputQuery}`);
  return `${inputQuery} ${feedback}`;
});

const step3 = task("step3", async (inputQuery: string) => {
  // 附加 qux
  return `${inputQuery} qux`;
});

我们现在可以在一个入口点中组合这些任务：

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

const checkpointer = new MemorySaver();

const graph = entrypoint(
  { checkpointer, name: "graph" },
  async (inputQuery: string) => {
    const result1 = await step1(inputQuery);
    const result2 = await humanFeedback(result1);
    const result3 = await step3(result2);

    return result3;
  }
);

在任务内部调用 interrupt()，使人类能够审查和编辑前一个任务的输出。先前任务的结果——在本例中是 step_1——被持久化，因此在 interrupt 之后不会再次运行。让我们发送一个查询字符串：

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

const stream = await graph.streamEvents("foo", config, { version: "v3" });
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

请注意，我们在 step_1 之后使用 interrupt 暂停了。中断提供了恢复运行的说明。要恢复，我们发出一个 Command，其中包含 human_feedback 任务预期的数据。

// 继续执行
const stream = await graph.streamEvents(
  new Command({ resume: "baz" }),
  config,
  { version: "v3" }
);
for await (const message of stream.messages) {
  for await (const token of message.text) {
    process.stdout.write(token);
  }
}

恢复后，运行将完成剩余的步骤并按预期终止。

审查工具调用

为了在执行前审查工具调用，我们添加了一个 review_tool_call 函数，该函数调用 interrupt。当调用此函数时，执行将暂停，直到我们发出恢复它的命令。给定一个工具调用，我们的函数将 interrupt 以供人工审查。此时我们可以：

接受工具调用
修改工具调用并继续
生成自定义工具消息（例如，指示模型重新格式化其工具调用）

import { ToolCall } from "@langchain/core/messages/tool";
import { ToolMessage } from "@langchain/core/messages";

function reviewToolCall(toolCall: ToolCall): ToolCall | ToolMessage {
  // 审查工具调用，返回验证后的版本
  const humanReview = interrupt({
    question: "Is this correct?",
    tool_call: toolCall,
  });

  const reviewAction = humanReview.action;
  const reviewData = humanReview.data;

  if (reviewAction === "continue") {
    return toolCall;
  } else if (reviewAction === "update") {
    const updatedToolCall = { ...toolCall, args: reviewData };
    return updatedToolCall;
  } else if (reviewAction === "feedback") {
    return new ToolMessage({
      content: reviewData,
      name: toolCall.name,
      tool_call_id: toolCall.id,
    });
  }

  throw new Error(`Unknown review action: ${reviewAction}`);
}

我们现在可以更新我们的入口点来审查生成的工具调用。如果工具调用被接受或修改，我们像以前一样执行。否则，我们只需附加人类提供的 ToolMessage。先前任务的结果——在本例中是初始模型调用——被持久化，因此在 interrupt 之后不会再次运行。

import {
  MemorySaver,
  entrypoint,
  interrupt,
  Command,
  addMessages,
} from "@langchain/langgraph";
import { ToolMessage, AIMessage, BaseMessage } from "@langchain/core/messages";

const checkpointer = new MemorySaver();

const agent = entrypoint(
  { checkpointer, name: "agent" },
  async (
    messages: BaseMessage[],
    previous?: BaseMessage[]
  ): Promise<BaseMessage> => {
    if (previous !== undefined) {
      messages = addMessages(previous, messages);
    }

    let modelResponse = await callModel(messages);
    while (true) {
      if (!modelResponse.tool_calls?.length) {
        break;
      }

      // 审查工具调用
      const toolResults: ToolMessage[] = [];
      const toolCalls: ToolCall[] = [];

      for (let i = 0; i < modelResponse.tool_calls.length; i++) {
        const review = reviewToolCall(modelResponse.tool_calls[i]);
        if (review instanceof ToolMessage) {
          toolResults.push(review);
        } else {
          // 是一个验证后的工具调用
          toolCalls.push(review);
          if (review !== modelResponse.tool_calls[i]) {
            modelResponse.tool_calls[i] = review; // 更新消息
          }
        }
      }

      // 执行剩余的工具调用
      const remainingToolResults = await Promise.all(
        toolCalls.map((toolCall) => callTool(toolCall))
      );

      // 附加到消息列表
      messages = addMessages(messages, [
        modelResponse,
        ...toolResults,
        ...remainingToolResults,
      ]);

      // 再次调用模型
      modelResponse = await callModel(messages);
    }

    // 生成最终响应
    messages = addMessages(messages, modelResponse);
    return entrypoint.final({ value: modelResponse, save: messages });
  }
);

短期记忆

短期记忆允许在同一个线程 ID 的不同调用之间存储信息。更多详情请参阅短期记忆。

管理检查点

您可以查看和删除检查点器存储的信息。

查看线程状态

const config = {
  configurable: {
    thread_id: "1",
    // 可选地提供特定检查点的 ID，
    // 否则显示最新的检查点
    // checkpoint_id: "1f029ca3-1f5b-6704-8004-820c16b69a5a"
  },
};
await graph.getState(config);

StateSnapshot {
  values: {
    messages: [
      HumanMessage { content: "hi! I'm bob" },
      AIMessage { content: "Hi Bob! How are you doing today?" },
      HumanMessage { content: "what's my name?" },
      AIMessage { content: "Your name is Bob." }
    ]
  },
  next: [],
  config: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1f5b-6704-8004-820c16b69a5a' } },
  metadata: {
    source: 'loop',
    writes: { call_model: { messages: AIMessage { content: "Your name is Bob." } } },
    step: 4,
    parents: {},
    thread_id: '1'
  },
  createdAt: '2025-05-05T16:01:24.680462+00:00',
  parentConfig: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1790-6b0a-8003-baf965b6a38f' } },
  tasks: [],
  interrupts: []
}

查看线程历史记录

const config = {
  configurable: {
    thread_id: "1",
  },
};
const history = [];
for await (const state of graph.getStateHistory(config)) {
  history.push(state);
}

[
  StateSnapshot {
    values: {
      messages: [
        HumanMessage { content: "hi! I'm bob" },
        AIMessage { content: "Hi Bob! How are you doing today? Is there anything I can help you with?" },
        HumanMessage { content: "what's my name?" },
        AIMessage { content: "Your name is Bob." }
      ]
    },
    next: [],
    config: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1f5b-6704-8004-820c16b69a5a' } },
    metadata: { source: 'loop', writes: { call_model: { messages: AIMessage { content: "Your name is Bob." } } }, step: 4, parents: {}, thread_id: '1' },
    createdAt: '2025-05-05T16:01:24.680462+00:00',
    parentConfig: { configurable: { thread_id: '1', checkpoint_ns: '', checkpoint_id: '1f029ca3-1790-6b0a-8003-baf965b6a38f' } },
    tasks: [],
    interrupts: []
  },
  // ... 更多状态快照
]

解耦返回值与保存值

使用 entrypoint.final 将返回给调用方的内容与持久化在检查点中的内容解耦。这在以下情况下很有用：

您想要返回一个计算结果（例如，摘要或状态），但保存一个不同的内部值以供下次调用时使用。
您需要控制下次运行时传递给 previous 参数的内容。

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

const checkpointer = new MemorySaver();

const accumulate = entrypoint(
  { checkpointer, name: "accumulate" },
  async (n: number, previous?: number) => {
    const prev = previous || 0;
    const total = prev + n;
    // 将 *上一个* 值返回给调用方，但将 *新的* 总计保存到检查点。
    return entrypoint.final({ value: prev, save: total });
  }
);

const config = { configurable: { thread_id: "my-thread" } };

console.log(await accumulate.invoke(1, config)); // 0
console.log(await accumulate.invoke(2, config)); // 1
console.log(await accumulate.invoke(3, config)); // 3

聊天机器人示例

一个使用函数式 API 和 InMemorySaver 检查点器的简单聊天机器人示例。机器人能够记住之前的对话并从中断的地方继续。

import { BaseMessage } from "@langchain/core/messages";
import {
  addMessages,
  entrypoint,
  task,
  MemorySaver,
} from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" });

const callModel = task(
  "callModel",
  async (messages: BaseMessage[]): Promise<BaseMessage> => {
    const response = await model.invoke(messages);
    return response;
  }
);

const checkpointer = new MemorySaver();

const workflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (
    inputs: BaseMessage[],
    previous?: BaseMessage[]
  ): Promise<BaseMessage> => {
    let messages = inputs;
    if (previous) {
      messages = addMessages(previous, inputs);
    }

    const response = await callModel(messages);
    return entrypoint.final({
      value: response,
      save: addMessages(messages, response),
    });
  }
);

const config = { configurable: { thread_id: "1" } };
const inputMessage = { role: "user", content: "hi! I'm bob" };

const stream1 = await workflow.streamEvents([inputMessage], { ...config, version: "v3" });
for await (const snapshot of stream1.values) {
  console.log(snapshot);
}

const inputMessage2 = { role: "user", content: "what's my name?" };
const stream2 = await workflow.streamEvents([inputMessage2], { ...config, version: "v3" });
for await (const snapshot of stream2.values) {
  console.log(snapshot);
}

长期记忆

长期记忆允许在不同的线程 ID 之间存储信息。这对于在一次对话中学习关于某个用户的信息，并在另一次对话中使用它非常有用。

工作流

工作流和代理指南提供了更多关于如何使用函数式 API 构建工作流的示例。

与其他库集成

使用函数式 API 将 LangGraph 的功能添加到其他框架：将持久化、记忆和流式传输等 LangGraph 功能添加到不提供开箱即用支持的其他代理框架中。

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

创建一个简单的工作流

并行执行

调用图

调用其他入口点

流式传输

重试策略

缓存任务

出错后恢复

人在环中

基本人在环中工作流

审查工具调用

短期记忆

管理检查点

查看线程状态

查看线程历史记录

解耦返回值与保存值

聊天机器人示例

长期记忆

工作流

与其他库集成

​创建一个简单的工作流

​并行执行

​调用图

​调用其他入口点

​流式传输

​重试策略

​缓存任务

​出错后恢复

​人在环中

​基本人在环中工作流

​审查工具调用

​短期记忆

​管理检查点

​查看线程状态

​查看线程历史记录

​解耦返回值与保存值

​聊天机器人示例

​长期记忆

​工作流

​与其他库集成

创建一个简单的工作流

并行执行

调用图

调用其他入口点

流式传输

重试策略

缓存任务

出错后恢复

人在环中

基本人在环中工作流

审查工具调用

短期记忆

管理检查点

查看线程状态

查看线程历史记录

解耦返回值与保存值

聊天机器人示例

长期记忆

工作流

与其他库集成