使用函数式 API - 以编程方式使用文档

该 **函数式 API** 允许您添加 LangGraph 的关键功能 (持久化, 内存, human-in-the-loop、和流式输出) 到您的应用程序中，只需对现有代码进行少量更改。

创建简单工作流

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

Extended example: simple workflow

  // Task that checks if a number is even
  const isEven = task("isEven", async (number: number) => {
    return number % 2 === 0;
  });

  // Task that formats a message
  const formatMessage = task("formatMessage", async (isEven: boolean) => {
    return isEven ? "The number is even." : "The number is odd.";
  });

  // Create a checkpointer for persistence
  const checkpointer = new MemorySaver();

  const workflow = entrypoint(
    { checkpointer, name: "workflow" },
    async (inputs: { number: number }) => {
      // Simple workflow to classify a number
      const even = await isEven(inputs.number);
      return await formatMessage(even);
    }
  );

  // Run the workflow with a unique thread ID
  const config = { configurable: { thread_id: uuid7() } };
  const result = await workflow.invoke({ number: 7 }, config);
  console.log(result);

Extended example: Compose an essay with an LLM

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

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

  // Task: generate essay using an LLM
  const composeEssay = task("composeEssay", async (topic: string) => {
    // Generate an essay about the given topic
    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;
  });

  // Create a checkpointer for persistence
  const checkpointer = new MemorySaver();

  const workflow = entrypoint(
    { checkpointer, name: "workflow" },
    async (topic: string) => {
      // Simple workflow that generates an essay with an LLM
      return await composeEssay(topic);
    }
  );

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

并行执行

任务可以通过并发调用并等待结果来实现并行执行。这对于提高 IO 密集型任务的性能很有帮助（例如调用 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));
  }
);

Extended example: parallel LLM calls

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

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

  // Task that generates a paragraph about a given topic
  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;
  });

  // Create a checkpointer for persistence
  const checkpointer = new MemorySaver();

  const workflow = entrypoint(
    { checkpointer, name: "workflow" },
    async (topics: string[]) => {
      // Generates multiple paragraphs in parallel and combines them
      const paragraphs = await Promise.all(topics.map(generateParagraph));
      return paragraphs.join("\n\n");
    }
  );

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

This example uses LangGraph's concurrency model to improve execution time, especially when tasks involve I/O like LLM completions.

调用图

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

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

const someWorkflow = entrypoint(
  { name: "someWorkflow" },
  async (someInput: Record<string, any>) => {
    // Call a graph defined using the graph API
    const result1 = await someGraph.invoke(/* ... */);
    // Call another graph defined using the graph API
    const result2 = await anotherGraph.invoke(/* ... */);
    return {
      result1,
      result2,
    };
  }
);

Extended example: calling a simple graph from the functional API

  // Define the shared state type
  const State = new StateSchema({
    foo: z.number(),
  });

  // Build the graph using the Graph API
  const builder = new StateGraph(State)
    .addNode("double", (state) => {
      return { foo: state.foo * 2 };
    })
    .addEdge("__start__", "double");
  const graph = builder.compile();

  // Define the functional API workflow
  const checkpointer = new MemorySaver();

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

  // Execute the workflow
  const config = { configurable: { thread_id: uuid7() } };
  console.log(await workflow.invoke(5, config)); // Output: { bar: 10 }

调用其他入口点

您可以调用其他 **入口点** 从 **入口点** or a **任务中调用。**.

// Will automatically use the checkpointer from the parent 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;
  }
);

Extended example: calling another entrypoint

  // Initialize a checkpointer
  const checkpointer = new MemorySaver();

  // A reusable sub-workflow that multiplies a number
  const multiply = entrypoint(
    { name: "multiply" },
    async (inputs: { a: number; b: number }) => {
      return inputs.a * inputs.b;
    }
  );

  // Main workflow that invokes the sub-workflow
  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 };
    }
  );

  // Execute the main workflow
  const config = { configurable: { thread_id: uuid7() } };
  console.log(await main.invoke({ x: 6, y: 7 }, config)); // Output: { product: 42 }

流式输出

该 **函数式 API** 使用与 **图 API**相同的流式处理机制。请阅读 **流式输出指南** 部分了解更多详情。

使用流式 API 从工作流运行中流式传输值块的示例。

在计算完结果后发送另一条自定义消息。
使用
处理流式输出。 streamEvents() 迭代
以接收值快照。 stream.values 重试策略

设置任务和入口超时时间

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

// This variable is just used for demonstration purposes to simulate a network failure.
// It's not something you will have in your actual code.
let attempts = 0;

// Let's configure the RetryPolicy to retry on ValueError.
// The default RetryPolicy is optimized for retrying specific network errors.
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'

缓存任务

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

const slowAdd = task(
  {
    name: "slowAdd",
    cache: { ttl: 120 },   // [!code highlight]
  },
  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 以秒为单位指定。缓存将在此时间后失效。

错误后恢复

// This variable is just used for demonstration purposes to simulate a network failure.
// It's not something you will have in your actual code.
let attempts = 0;

const getInfo = task("getInfo", async () => {
  /**
   * Simulates a task that fails once before succeeding.
   * Throws an exception on the first attempt, then returns "OK" on subsequent tries.
   */
  attempts += 1;

  if (attempts < 2) {
    throw new Error("Failure"); // Simulate a failure on the first attempt
  }
  return "OK";
});

// Initialize an in-memory checkpointer for persistence
const checkpointer = new MemorySaver();

const slowTask = task("slowTask", async () => {
  /**
   * Simulates a slow-running task by introducing a 1-second delay.
   */
  await new Promise((resolve) => setTimeout(resolve, 1000));
  return "Ran slow task.";
});

const main = entrypoint(
  { checkpointer, name: "main" },
  async (inputs: Record<string, any>) => {
    /**
     * Main workflow function that runs the slowTask and getInfo tasks sequentially.
     *
     * Parameters:
     * - inputs: Record<string, any> containing workflow input values.
     *
     * The workflow first executes `slowTask` and then attempts to execute `getInfo`,
     * which will fail on the first invocation.
     */
    const slowTaskResult = await slowTask(); // Blocking call to slowTask
    await getInfo(); // Exception will be raised here on the first attempt
    return slowTaskResult;
  }
);

// Workflow execution configuration with a unique thread identifier
const config = {
  configurable: {
    thread_id: "1", // Unique identifier to track workflow execution
  },
};

// This invocation will take ~1 second due to the slowTask execution
try {
  // First invocation will raise an exception due to the `getInfo` task failing
  await main.invoke({ any_input: "foobar" }, config);
} catch (err) {
  // Handle the failure gracefully
}

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

await main.invoke(null, config);

'Ran slow task.'

Human-in-the-loop

函数式 API 支持 human-in-the-loop 使用 interrupt 函数和工作流 Command primitive.

基本的人在回路工作流

我们将创建三个任务:

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

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

const humanFeedback = task("humanFeedback", async (inputQuery: string) => {
  // Append user input
  const feedback = interrupt(`Please provide feedback: ${inputQuery}`);
  return `${inputQuery} ${feedback}`;
});

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

我们现在可以在入口点:

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

注意，我们已在 interrupt 之后暂停 step_1。中断提供了恢复运行的指令。要恢复，我们发出 Command 包含预期数据 human_feedback task.

// Continue execution
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 进行人工审查。此时我们可以：

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

function reviewToolCall(toolCall: ToolCall): ToolCall | ToolMessage {
  // Review a tool call, returning a validated version
  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.

  MemorySaver,
  entrypoint,
  interrupt,
  Command,
  addMessages,
} from "@langchain/langgraph";


const checkpointer = new MemorySaver();

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

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

      // Review tool calls
      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 {
          // is a validated tool call
          toolCalls.push(review);
          if (review !== modelResponse.tool_calls[i]) {
            modelResponse.tool_calls[i] = review; // update message
          }
        }
      }

      // Execute remaining tool calls
      const remainingToolResults = await Promise.all(
        toolCalls.map((toolCall) => callTool(toolCall))
      );

      // Append to message list
      messages = addMessages(messages, [
        modelResponse,
        ...toolResults,
        ...remainingToolResults,
      ]);

      // Call model again
      modelResponse = await callModel(messages);
    }

    // Generate final response
    messages = addMessages(messages, modelResponse);
    return entrypoint.final({ value: modelResponse, save: messages });
  }
);

短期记忆

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

管理检查点

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

<a id="checkpoint"></a> #### 查看线程状态

const config = {
  configurable: {
    thread_id: "1",  // [!code highlight]
    // optionally provide an ID for a specific checkpoint,
    // otherwise the latest checkpoint is shown
    // checkpoint_id: "1f029ca3-1f5b-6704-8004-820c16b69a5a" [!code highlight]
  },
};
await graph.getState(config);  // [!code highlight]

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: []
}

<a id="checkpoints"></a> #### 查看线程历史

const config = {
  configurable: {
    thread_id: "1",  // [!code highlight]
  },
};
const history = [];  // [!code highlight]
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: []
  },
  // ... more state snapshots
]

将返回值与保存的值解耦

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

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

const checkpointer = new MemorySaver();

const accumulate = entrypoint(
  { checkpointer, name: "accumulate" },
  async (n: number, previous?: number) => {
    const prev = previous || 0;
    const total = prev + n;
    // Return the *previous* value to the caller but save the *new* total to the checkpoint.
    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 检查点的简单聊天机器人示例。

该机器人能够记住之前的对话并从上次中断处继续。

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


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

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

const checkpointer = new MemorySaver();

const workflow = entrypoint(
  { checkpointer, name: "workflow" },
  async (
    inputs: BaseMessage[],
    previous?: BaseMessage[]
  ): Promise => {
    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 的功能（如持久化、记忆和流式处理）添加到其他开箱即用的代理框架中。