> ## 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/python/langgraph/persistence)、[记忆](/oss/python/langgraph/add-memory)、[人机协作](/oss/python/langgraph/interrupts)和[流式输出](/oss/python/langgraph/streaming)）添加到你的应用程序中。

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

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

* **`@entrypoint`**：将函数标记为工作流的起点，封装逻辑并管理执行流程，包括处理长时间运行的任务和中断。
* **[`@task`](https://reference.langchain.com/python/langgraph/func/task)**：表示一个离散的工作单元，如 API 调用或数据处理步骤，可以在入口点内异步执行。任务返回一个类似 future 的对象，可以被等待或同步解析。

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

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

## 函数式 API 与图 API

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

以下是一些关键区别：

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

## 示例

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

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt

@task
def write_essay(topic: str) -> str:
    """撰写一篇关于给定主题的文章。"""
    time.sleep(1) # 模拟长时间运行任务的占位符。
    return f"An essay about topic: {topic}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """一个简单的工作流，撰写文章并请求审核。"""
    essay = write_essay("cat").result()
    is_approved = interrupt({
        # 作为参数提供给 interrupt 的任何可 JSON 序列化的载荷。
        # 它将在客户端作为 Interrupt 在流式传输工作流数据时浮现。
        "essay": essay, # 我们希望被审核的文章。
        # 我们可以添加任何需要的附加信息。
        # 例如，引入一个名为 "action" 的键并附带一些指令。
        "action": "Please approve/reject the essay",
    })

    return {
        "essay": essay, # 生成的文章
        "is_approved": is_approved, # 来自人机协作的响应
    }
```

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

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

  ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  import time
  from langchain_core.utils.uuid import uuid7
  from langgraph.func import entrypoint, task
  from langgraph.types import interrupt
  from langgraph.checkpoint.memory import InMemorySaver


  @task
  def write_essay(topic: str) -> str:
      """撰写一篇关于给定主题的文章。"""
      time.sleep(1)  # 这是模拟长时间运行任务的占位符。
      return f"An essay about topic: {topic}"

  @entrypoint(checkpointer=InMemorySaver())
  def workflow(topic: str) -> dict:
      """一个简单的工作流，撰写文章并请求审核。"""
      essay = write_essay("cat").result()
      is_approved = interrupt(
          {
              "essay": essay,
              "action": "Please approve/reject the essay",
          }
      )
      return {
          "essay": essay,
          "is_approved": is_approved,
      }


  thread_id = str(uuid7())
  config = {"configurable": {"thread_id": thread_id}}
  for item in workflow.stream("cat", config):
      print(item)
  # > {'write_essay': 'An essay about topic: cat'}
  # > {
  # >     '__interrupt__': (
  # >        Interrupt(
  # >            value={
  # >                'essay': 'An essay about topic: cat',
  # >                'action': 'Please approve/reject the essay'
  # >            },
  # >            id='b9b2b9d788f482663ced6dc755c9e981'
  # >        ),
  # >    )
  # > }
  ```

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

  ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  from langgraph.types import Command

  # 从用户获取审核（例如，通过 UI）
  # 在此例中，我们使用布尔值，但可以是任何可 JSON 序列化的值。
  human_review = True

  for item in workflow.stream(Command(resume=human_review), config):
      print(item)
  ```

  ```pycon theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  {'workflow': {'essay': 'An essay about topic: cat', 'is_approved': False}}
  ```

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

## 入口点

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

### 定义

**入口点**通过使用 `@entrypoint` 装饰器装饰函数来定义。

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

使用 `entrypoint` 装饰函数会生成一个 [`Pregel`](https://reference.langchain.com/python/langgraph/pregel/#langgraph.pregel.Pregel.stream) 实例，它帮助管理工作流的执行（例如，处理流式输出、恢复和检查点）。

你通常会希望向 `@entrypoint` 装饰器传递一个**检查点器**以启用持久化并使用**人机协作**等功能。

<Tabs>
  <Tab title="同步">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.func import entrypoint

    @entrypoint(checkpointer=checkpointer)
    def my_workflow(some_input: dict) -> int:
        # 可能涉及长时间运行任务（如 API 调用）的逻辑，
        # 并且可能因人机协作而中断。
        ...
        return result
    ```
  </Tab>

  <Tab title="异步">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.func import entrypoint

    @entrypoint(checkpointer=checkpointer)
    async def my_workflow(some_input: dict) -> int:
        # 可能涉及长时间运行任务（如 API 调用）的逻辑，
        # 并且可能因人机协作而中断
        ...
        return result
    ```
  </Tab>
</Tabs>

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

### 可注入参数

声明 `entrypoint` 时，你可以请求访问将在运行时自动注入的附加参数。这些参数包括：

| 参数           | 描述                                                                                                                       |
| ------------ | ------------------------------------------------------------------------------------------------------------------------ |
| **previous** | 访问给定线程的前一个 `checkpoint` 关联的状态。参见[短期记忆](#short-term-memory)。                                                              |
| **store**    | \[`BaseStore`]\[langgraph.store.base.BaseStore] 的实例。用于[长期记忆](/oss/python/langgraph/use-functional-api#long-term-memory)。 |
| **writer**   | 在 Async Python \< 3.11 时用于访问 StreamWriter。详情请参阅[函数式 API 的流式输出](/oss/python/langgraph/use-functional-api#streaming)。      |
| **config**   | 用于访问运行时配置。参见 [RunnableConfig](https://python.langchain.com/docs/concepts/runnables/#runnableconfig) 了解信息。                |

<Warning>
  使用适当的名称和类型注解声明参数。
</Warning>

<Accordion title="请求可注入参数">
  ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
  from langchain_core.runnables import RunnableConfig
  from langgraph.func import entrypoint
  from langgraph.store.base import BaseStore
  from langgraph.store.memory import InMemoryStore
  from langgraph.checkpoint.memory import InMemorySaver
  from langgraph.types import StreamWriter

  in_memory_checkpointer = InMemorySaver(...)
  in_memory_store = InMemoryStore(...)  # 用于长期记忆的 InMemoryStore 实例

  @entrypoint(
      checkpointer=in_memory_checkpointer,  # 指定检查点器
      store=in_memory_store  # 指定存储
  )
  def my_workflow(
      some_input: dict,  # 输入（例如通过 `invoke` 传递）
      *,
      previous: Any = None, # 用于短期记忆
      store: BaseStore,  # 用于长期记忆
      writer: StreamWriter,  # 用于流式输出自定义数据
      config: RunnableConfig  # 用于访问传递给入口点的配置
  ) -> ...:
  ```
</Accordion>

### 执行

使用 [`@entrypoint`](#entrypoint) 会产生一个 [`Pregel`](https://reference.langchain.com/python/langgraph/pregel/#langgraph.pregel.Pregel.stream) 对象，可以使用 `invoke`、`ainvoke`、`stream` 和 `astream` 方法执行。

<Tabs>
  <Tab title="Invoke">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    config = {
        "configurable": {
            "thread_id": "some_thread_id"
        }
    }
    my_workflow.invoke(some_input, config)  # 同步等待结果
    ```
  </Tab>

  <Tab title="异步 Invoke">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    config = {
        "configurable": {
            "thread_id": "some_thread_id"
        }
    }
    await my_workflow.ainvoke(some_input, config)  # 异步等待结果
    ```
  </Tab>

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

    for chunk in my_workflow.stream(some_input, config):
        print(chunk)
    ```
  </Tab>

  <Tab title="异步 Stream">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    config = {
        "configurable": {
            "thread_id": "some_thread_id"
        }
    }

    async for chunk in my_workflow.astream(some_input, config):
        print(chunk)
    ```
  </Tab>
</Tabs>

### 恢复

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

<Tabs>
  <Tab title="Invoke">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.types import Command

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

    my_workflow.invoke(Command(resume=some_resume_value), config)
    ```
  </Tab>

  <Tab title="异步 Invoke">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.types import Command

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

    await my_workflow.ainvoke(Command(resume=some_resume_value), config)
    ```
  </Tab>

  <Tab title="Stream">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.types import Command

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

    for chunk in my_workflow.stream(Command(resume=some_resume_value), config):
        print(chunk)
    ```
  </Tab>

  <Tab title="异步 Stream">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.types import Command

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

    async for chunk in my_workflow.astream(Command(resume=some_resume_value), config):
        print(chunk)
    ```
  </Tab>
</Tabs>

**错误后恢复**

要在错误后恢复，使用 `None` 和相同的 **thread id**（config）运行 `entrypoint`。

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

<Tabs>
  <Tab title="Invoke">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    my_workflow.invoke(None, config)
    ```
  </Tab>

  <Tab title="异步 Invoke">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    await my_workflow.ainvoke(None, config)
    ```
  </Tab>

  <Tab title="Stream">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    for chunk in my_workflow.stream(None, config):
        print(chunk)
    ```
  </Tab>

  <Tab title="异步 Stream">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}

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

    async for chunk in my_workflow.astream(None, config):
        print(chunk)
    ```
  </Tab>
</Tabs>

### 短期记忆

当 `entrypoint` 使用 `checkpointer` 定义时，它会在同一 **thread id** 上的连续调用之间将信息存储在[检查点](/oss/python/langgraph/persistence#checkpoints)中。

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

默认情况下，`previous` 参数是上一次调用的返回值。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> int:
    previous = previous or 0
    return number + previous

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

my_workflow.invoke(1, config)  # 1（previous 为 None）
my_workflow.invoke(2, config)  # 3（previous 为上次调用的 1）
```

#### `entrypoint.final`

[`entrypoint.final`](https://reference.langchain.com/python/langgraph/func/entrypoint/final) 是一个特殊原语，可以从入口点返回，允许将**保存在检查点中的值**与**入口点的返回值\*\*\*\*解耦**。

第一个值是入口点的返回值，第二个值是将保存在检查点中的值。类型注解为 `entrypoint.final[return_type, save_type]`。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
@entrypoint(checkpointer=checkpointer)
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
    previous = previous or 0
    # 这将向调用者返回 previous 值，将
    # 2 * number 保存到检查点，在下次调用时
    # 用于 `previous` 参数。
    return entrypoint.final(value=previous, save=2 * number)

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

my_workflow.invoke(3, config)  # 0（previous 为 None）
my_workflow.invoke(1, config)  # 6（previous 为上次调用的 3 * 2）
```

## 任务

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

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

### 定义

任务使用 `@task` 装饰器定义，它包装一个普通的 Python 函数。

```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
from langgraph.func import task

@task()
def slow_computation(input_value):
    # 模拟长时间运行的操作
    ...
    return result
```

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

### 执行

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

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

当你调用一个**任务**时，它会\_立即\_返回一个 future 对象。future 是稍后可用结果的占位符。

要获取**任务**的结果，你可以同步等待（使用 `result()`）或异步等待（使用 `await`）。

<Tabs>
  <Tab title="同步调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    @entrypoint(checkpointer=checkpointer)
    def my_workflow(some_input: int) -> int:
        future = slow_computation(some_input)
        return future.result()  # 同步等待结果
    ```
  </Tab>

  <Tab title="异步调用">
    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    @entrypoint(checkpointer=checkpointer)
    async def my_workflow(some_input: int) -> int:
        return await slow_computation(some_input)  # 异步等待结果
    ```
  </Tab>
</Tabs>

## 何时使用任务

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

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

## 序列化

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

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

这些要求对于启用检查点和工作流恢复是必要的。使用 Python 原语（如字典、列表、字符串、数字和布尔值）来确保你的输入和输出是可序列化的。

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

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

## 确定性

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

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

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

## 幂等性

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

## 常见陷阱

### 处理副作用

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

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

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    @entrypoint(checkpointer=checkpointer)
    def my_workflow(inputs: dict) -> int:
        # 此代码在恢复工作流时会第二次执行。
        # 这很可能不是你想要的。
        with open("output.txt", "w") as f:  # [!code highlight]
            f.write("Side effect executed")  # [!code highlight]
        value = interrupt("question")
        return value
    ```
  </Tab>

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

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.func import task

    @task  # [!code highlight]
    def write_to_file():  # [!code highlight]
        with open("output.txt", "w") as f:
            f.write("Side effect executed")

    @entrypoint(checkpointer=checkpointer)
    def my_workflow(inputs: dict) -> int:
        # 副作用现在封装在任务中。
        write_to_file().result()
        value = interrupt("question")
        return value
    ```
  </Tab>
</Tabs>

### 非确定性控制流

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

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

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

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

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

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

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    from langgraph.func import entrypoint

    @entrypoint(checkpointer=checkpointer)
    def my_workflow(inputs: dict) -> int:
        t0 = inputs["t0"]
        t1 = time.time()  # [!code highlight]

        delta_t = t1 - t0

        if delta_t > 1:
            result = slow_task(1).result()
            value = interrupt("question")
        else:
            result = slow_task(2).result()
            value = interrupt("question")

        return {
            "result": result,
            "value": value
        }
    ```
  </Tab>

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

    ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}}
    import time

    from langgraph.func import task

    @task  # [!code highlight]
    def get_time() -> float:  # [!code highlight]
        return time.time()

    @entrypoint(checkpointer=checkpointer)
    def my_workflow(inputs: dict) -> int:
        t0 = inputs["t0"]
        t1 = get_time().result()  # [!code highlight]

        delta_t = t1 - t0

        if delta_t > 1:
            result = slow_task(1).result()
            value = interrupt("question")
        else:
            result = slow_task(2).result()
            value = interrupt("question")

        return {
            "result": result,
            "value": value
        }
    ```
  </Tab>
</Tabs>

## 了解更多

* [如何使用函数式 API](/oss/python/langgraph/use-functional-api)
* [图 API 概念概述](/oss/python/langgraph/graph-api)
* [选择图 API 还是函数式 API](/oss/python/langgraph/choosing-apis)

***

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